import os import runpy import shutil import sys import traceback from pathlib import Path from typing import Optional CODE_FILES = ["app.py", "engine.py", "engine_hedge.py", "mtm.py"] CODE_DIR = Path("_code") CODE_PREFIX = "code" def _secret_or_env(key: str, default=None): try: import streamlit as st try: val = st.secrets.get(key) except Exception: val = None except Exception: val = None if val is None: val = os.environ.get(key) return val if val is not None else default def _place(src: Path, dst: Path) -> None: tmp = dst.with_suffix(dst.suffix + ".tmp") shutil.copyfile(src, tmp) os.replace(tmp, dst) def _ensure_code() -> Optional[str]: """把程式檔備妥於 _code/。成功回傳 None,失敗回傳錯誤訊息。""" CODE_DIR.mkdir(exist_ok=True) if all((CODE_DIR / f).exists() for f in CODE_FILES): return None local_src = os.environ.get("CODE_LOCAL_DIR") if local_src: for f in CODE_FILES: _place(Path(local_src) / f, CODE_DIR / f) return None repo = _secret_or_env("HF_DATASET_REPO") token = _secret_or_env("HF_TOKEN") branch = _secret_or_env("HF_BRANCH", "main") if not (repo and token): return "尚未設定 HF_DATASET_REPO / HF_TOKEN(Streamlit Secrets 或環境變數)。" try: from huggingface_hub import hf_hub_download for f in CODE_FILES: cached = hf_hub_download( repo_id=repo, filename=f"{CODE_PREFIX}/{f}", repo_type="dataset", revision=branch, token=token, ) _place(Path(cached), CODE_DIR / f) return None except Exception: traceback.print_exc() return "程式碼載入失敗,請確認私有 dataset 的 code/ 資料夾與 Secrets 設定後重試。" _err = _ensure_code() if _err is not None: import streamlit as st st.error(_err) st.stop() _code_path = str(CODE_DIR.resolve()) if _code_path not in sys.path: sys.path.insert(0, _code_path) runpy.run_path(str(CODE_DIR / "app.py"), run_name="__main__")