This commit is contained in:
125
src/jiangchang_skill_core/config.py
Normal file
125
src/jiangchang_skill_core/config.py
Normal file
@@ -0,0 +1,125 @@
|
||||
"""三级优先级配置读取 + 首次 .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
|
||||
Reference in New Issue
Block a user