127 lines
3.9 KiB
Python
127 lines
3.9 KiB
Python
# vendored from jiangchang-platform-kit; do not edit here, edit the kit.
|
||
"""三级优先级配置读取 + 首次 .env 落盘（进程 env > 数据目录 .env > .env.example）。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
import shutil
|
||
from typing import Any
|
||
|
||
from .runtime_env import get_data_root, get_user_id
|
||
|
||
_skill_slug: str | None = None
|
||
_example_path: str | None = None
|
||
_env_file_path: str | None = None
|
||
_user_env_cache: dict[str, str] | None = None
|
||
_example_defaults_cache: dict[str, str] | None = None
|
||
|
||
|
||
def _parse_env_file(path: str) -> dict[str, str]:
|
||
"""标准库手写 .env 解析（不引 python-dotenv）。"""
|
||
result: dict[str, str] = {}
|
||
if not path or not os.path.isfile(path):
|
||
return result
|
||
with open(path, encoding="utf-8") as f:
|
||
for raw_line in f:
|
||
line = raw_line.strip()
|
||
if not line or line.startswith("#"):
|
||
continue
|
||
if "=" not in line:
|
||
continue
|
||
key, _, value = line.partition("=")
|
||
key = key.strip()
|
||
value = value.strip()
|
||
if not key:
|
||
continue
|
||
if len(value) >= 2 and value[0] == value[-1] and value[0] in ('"', "'"):
|
||
value = value[1:-1]
|
||
elif " #" in value:
|
||
value = value.split(" #", 1)[0].strip()
|
||
result[key] = value
|
||
return result
|
||
|
||
|
||
def _get_user_env() -> dict[str, str]:
|
||
global _user_env_cache
|
||
if _user_env_cache is not None:
|
||
return _user_env_cache
|
||
if _env_file_path and os.path.isfile(_env_file_path):
|
||
_user_env_cache = _parse_env_file(_env_file_path)
|
||
else:
|
||
_user_env_cache = {}
|
||
return _user_env_cache
|
||
|
||
|
||
def _get_example_defaults() -> dict[str, str]:
|
||
global _example_defaults_cache
|
||
if _example_defaults_cache is not None:
|
||
return _example_defaults_cache
|
||
if _example_path and os.path.isfile(_example_path):
|
||
_example_defaults_cache = _parse_env_file(_example_path)
|
||
else:
|
||
_example_defaults_cache = {}
|
||
return _example_defaults_cache
|
||
|
||
|
||
def reset_cache() -> None:
|
||
"""测试用：清空解析缓存。"""
|
||
global _user_env_cache, _example_defaults_cache
|
||
_user_env_cache = None
|
||
_example_defaults_cache = None
|
||
|
||
|
||
def ensure_env_file(skill_slug: str, example_path: str) -> str:
|
||
"""首次把 .env.example copy 到 {data_root}/{user}/{slug}/.env（已存在不覆盖），返回路径。"""
|
||
global _skill_slug, _example_path, _env_file_path
|
||
_skill_slug = skill_slug
|
||
_example_path = os.path.abspath(example_path)
|
||
dest_dir = os.path.join(get_data_root(), get_user_id(), skill_slug)
|
||
os.makedirs(dest_dir, exist_ok=True)
|
||
dest = os.path.join(dest_dir, ".env")
|
||
_env_file_path = dest
|
||
if not os.path.isfile(dest) and os.path.isfile(_example_path):
|
||
shutil.copy2(_example_path, dest)
|
||
reset_cache()
|
||
return dest
|
||
|
||
|
||
def get(key: str, default: Any = None) -> str | None:
|
||
"""进程环境变量 > 数据目录 .env > .env.example 默认值。"""
|
||
env_val = os.environ.get(key)
|
||
if env_val is not None and env_val != "":
|
||
return env_val
|
||
user_val = _get_user_env().get(key)
|
||
if user_val is not None and user_val != "":
|
||
return user_val
|
||
example_val = _get_example_defaults().get(key)
|
||
if example_val is not None and example_val != "":
|
||
return example_val
|
||
return default
|
||
|
||
|
||
def get_bool(key: str, default: bool = False) -> bool:
|
||
val = get(key)
|
||
if val is None:
|
||
return default
|
||
return str(val).strip().lower() in ("1", "true", "yes", "on")
|
||
|
||
|
||
def get_float(key: str, default: float) -> float:
|
||
val = get(key)
|
||
if val is None:
|
||
return default
|
||
try:
|
||
return float(val)
|
||
except (TypeError, ValueError):
|
||
return default
|
||
|
||
|
||
def get_int(key: str, default: int) -> int:
|
||
val = get(key)
|
||
if val is None:
|
||
return default
|
||
try:
|
||
return int(val)
|
||
except (TypeError, ValueError):
|
||
return default
|