InvestingTest / scripts /set_hf_space_env.py
Mbonea's picture
Add economy intelligence pipeline
1e12801
from __future__ import annotations
import argparse
import os
import re
import subprocess
from pathlib import Path
from huggingface_hub import HfApi
DEFAULT_LOCAL_ENV = Path(__file__).resolve().parents[1] / ".env.local"
def _read_local_env(path: Path) -> dict[str, str]:
values: dict[str, str] = {}
if not path.exists():
return values
for raw_line in path.read_text(encoding="utf-8").splitlines():
line = raw_line.strip()
if not line or line.startswith("#") or "=" not in line:
continue
key, value = line.split("=", 1)
values[key.strip().lstrip("\ufeff")] = value.strip().strip('"').strip("'")
return values
def _remote_url() -> str:
try:
return subprocess.check_output(
["git", "remote", "get-url", "hf-space"],
cwd=Path(__file__).resolve().parents[1],
text=True,
stderr=subprocess.DEVNULL,
).strip()
except Exception:
return ""
def _token_from_remote(remote: str) -> str | None:
match = re.search(r"https://[^:]+:(hf_[^@]+)@huggingface\.co/spaces/", remote)
return match.group(1) if match else None
def _repo_id_from_remote(remote: str) -> str | None:
match = re.search(r"huggingface\.co/spaces/([^/\s]+)/([^/\s]+)", remote)
if not match:
return None
return f"{match.group(1)}/{match.group(2)}"
def main() -> None:
parser = argparse.ArgumentParser(description="Set Uwekezaji economy variables on Hugging Face Space.")
parser.add_argument("--repo-id", default=None, help="Space repo id, e.g. Mbonea/InvestingTest")
parser.add_argument("--env-file", default=str(DEFAULT_LOCAL_ENV), help="Local env file to read values from")
args = parser.parse_args()
env_file = Path(args.env_file)
values = _read_local_env(env_file)
remote = _remote_url()
repo_id = args.repo_id or _repo_id_from_remote(remote)
token = values.get("HF_SPACE_API_TOKEN") or os.getenv("HF_SPACE_API_TOKEN") or _token_from_remote(remote)
if not repo_id:
raise SystemExit("Could not resolve Space repo id. Pass --repo-id.")
if not token:
raise SystemExit("Could not resolve HF token from local env, environment, or hf-space remote.")
required = {
"HF_SPACE_UPLOAD_URL": values.get("HF_SPACE_UPLOAD_URL") or os.getenv("HF_SPACE_UPLOAD_URL"),
"HF_SPACE_REPO_ID": values.get("HF_SPACE_REPO_ID") or os.getenv("HF_SPACE_REPO_ID") or repo_id,
"ECONOMY_PIPELINE_ENABLED": values.get("ECONOMY_PIPELINE_ENABLED") or "true",
"ECONOMY_DAILY_RUN_TIME": values.get("ECONOMY_DAILY_RUN_TIME") or "18:30",
"ECONOMY_DOCUMENT_ROOT": values.get("ECONOMY_DOCUMENT_ROOT") or "/data/InvestingTest/PDFs",
}
missing = [key for key, value in required.items() if not value]
if missing:
raise SystemExit(f"Missing required values: {', '.join(missing)}")
api = HfApi(token=token)
for key, value in required.items():
api.add_space_variable(repo_id=repo_id, key=key, value=value, token=token)
print(f"set variable {key}")
api.add_space_secret(repo_id=repo_id, key="HF_SPACE_API_TOKEN", value=token, token=token)
print("set secret HF_SPACE_API_TOKEN")
print(f"updated Space environment for {repo_id}")
if __name__ == "__main__":
main()