Allow CPA_AUTO_IMPORT_REMOTE from .env

Enable remote CPA import via env flag with priority over config.json, and load project .env when resolving remote settings.
This commit is contained in:
chaos committed 2026-07-11 18:06:17 +08:00
1 parent 0d8265de03
commit 06044551c7
4 files changed
+66 -9

No files matched your search

+54 -2
View File
@@ -22,6 +22,31 @@ from urllib.parse import quote, urlparse, urlunparse
_REG_DIR = Path(__file__).resolve().parent
_DEFAULT_OUT = _REG_DIR / "cpa_auths"
_DEFAULT_CPA = Path("") # empty = do not assume a machine-local CPA path
_DOTENV_LOADED = False
def _load_dotenv_once() -> None:
"""Load project .env into os.environ once (does not override existing env)."""
global _DOTENV_LOADED
if _DOTENV_LOADED:
return
_DOTENV_LOADED = True
env_path = _REG_DIR / ".env"
if not env_path.is_file():
return
try:
for raw in env_path.read_text(encoding="utf-8").splitlines():
line = raw.strip()
if not line or line.startswith("#") or "=" not in line:
continue
key, _, val = line.partition("=")
key = key.strip()
val = val.strip().strip('"').strip("'")
if key and key not in os.environ:
os.environ[key] = val
except Exception:
pass
def _ensure_cpa_xai_on_path(tools_dir: str | Path | None = None) -> Path:
@@ -90,10 +115,37 @@ def _normalize_cpa_remote_base(base: str) -> str:
return urlunparse((parsed.scheme, parsed.netloc, "", "", "", "")).rstrip("/")
def _env_flag(name: str) -> bool | None:
"""Parse env truthy/falsey flag. None if unset/empty."""
raw = (os.environ.get(name) or "").strip().lower()
if not raw:
return None
if raw in ("1", "true", "yes", "on", "y"):
return True
if raw in ("0", "false", "no", "off", "n"):
return False
return None
def resolve_cpa_remote_settings(cfg: dict | None = None) -> dict[str, Any]:
"""Resolve remote CPA import settings from config + env."""
"""Resolve remote CPA import settings from config + env.
Enable flag priority:
1) env CPA_AUTO_IMPORT_REMOTE / CPA_AUTO_IMPORT (if set)
2) config cpa_auto_import_remote
3) default false
Base/password: config first, then env fallbacks.
"""
_load_dotenv_once()
cfg = cfg or {}
enabled = bool(cfg.get("cpa_auto_import_remote", False))
env_enabled = _env_flag("CPA_AUTO_IMPORT_REMOTE")
if env_enabled is None:
env_enabled = _env_flag("CPA_AUTO_IMPORT")
if env_enabled is not None:
enabled = env_enabled
else:
enabled = bool(cfg.get("cpa_auto_import_remote", False))
base = (
(cfg.get("cpa_remote_base") or "").strip()
or (os.environ.get("CPA_REMOTE_BASE") or "").strip()