78 lines
2.5 KiB
Python
78 lines
2.5 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
统一输出路径解析 — 所有 scripts/ops/ 脚本共享。
|
||
|
||
使用方式：
|
||
from _env_paths import get_output_path
|
||
out_dir = get_output_path("SYSTEM_ANALYZE_ROOT")
|
||
|
||
规则：
|
||
- 先 load_dotenv（根 .env），再从 os.environ 读取
|
||
- 环境变量未定义时抛出 KeyError，强制要求 .env 配置
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
from pathlib import Path
|
||
|
||
from dotenv import load_dotenv
|
||
|
||
# 加载根 .env（仅首次生效，override=False 不覆盖已有环境变量）
|
||
_ROOT = Path(__file__).resolve().parents[2]
|
||
load_dotenv(_ROOT / ".env", override=False)
|
||
|
||
# CHANGE 2026-02-21 | 补充加载连接器级 .env，获取 API_BASE/API_TOKEN/STORE_ID 等
|
||
# override=False 保证根 .env 和已有环境变量优先
|
||
_FEIQIU_ENV = _ROOT / "apps" / "etl" / "connectors" / "feiqiu" / ".env"
|
||
if _FEIQIU_ENV.exists():
|
||
load_dotenv(_FEIQIU_ENV, override=False)
|
||
|
||
|
||
def ensure_repo_root() -> Path:
|
||
"""校验 cwd 是否为仓库根目录，不是则自动切换。
|
||
|
||
委托给 neozqyy_shared.repo_root（共享包），
|
||
未安装时 fallback 到本地实现。
|
||
"""
|
||
try:
|
||
from neozqyy_shared.repo_root import ensure_repo_root as _shared
|
||
return _shared()
|
||
except ImportError:
|
||
pass
|
||
# fallback：共享包未安装时的本地实现
|
||
cwd = Path.cwd()
|
||
if (cwd / "pyproject.toml").is_file() and (cwd / ".kiro").is_dir():
|
||
return cwd
|
||
root = Path(__file__).resolve().parents[2]
|
||
if (root / "pyproject.toml").is_file() and (root / ".kiro").is_dir():
|
||
os.chdir(root)
|
||
import warnings
|
||
warnings.warn(
|
||
f"cwd 不是仓库根目录，已自动切换: {cwd} → {root}",
|
||
stacklevel=2,
|
||
)
|
||
return root
|
||
raise RuntimeError(
|
||
f"无法定位仓库根目录。当前 cwd={cwd}，"
|
||
f"推断 root={root}，均未找到 pyproject.toml + .kiro。"
|
||
f"请在仓库根目录下运行脚本。"
|
||
)
|
||
|
||
|
||
def get_output_path(env_var: str) -> Path:
|
||
"""
|
||
从环境变量读取输出路径。
|
||
|
||
如果 .env 中未定义该变量，抛出 KeyError 并给出明确提示，
|
||
避免静默回退到错误路径。
|
||
"""
|
||
val = os.environ.get(env_var)
|
||
if not val:
|
||
raise KeyError(
|
||
f"环境变量 {env_var} 未定义。"
|
||
f"请在根 .env 中配置，参考 .env.template 和 docs/deployment/EXPORT-PATHS.md"
|
||
)
|
||
p = Path(val)
|
||
p.mkdir(parents=True, exist_ok=True)
|
||
return p
|