jiangchang-platform-kit/src/jiangchang_skill_core/config.py

"""三级优先级配置读取 + 首次 .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 get_env_file_path() -> str | None:
    return _env_file_path


def get_example_path() -> str | None:
    return _example_path


def _iter_env_assignments(path: str) -> list[tuple[str, str]]:
    """按文件顺序返回 (key, 完整赋值行)。"""
    result: list[tuple[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.rstrip("\n\r")
            stripped = line.strip()
            if not stripped or stripped.startswith("#"):
                continue
            if "=" not in stripped:
                continue
            key, _, _ = stripped.partition("=")
            key = key.strip()
            if key:
                result.append((key, stripped))
    return result


def merge_missing_env_keys(
    example_path: str,
    dest_path: str,
    *,
    comment_skill: str | None = None,
) -> list[str]:
    """将 example 中有而用户 .env 没有的 key 追加到 dest 末尾，不修改已有项。"""
    if not example_path or not os.path.isfile(example_path):
        return []
    if not dest_path or not os.path.isfile(dest_path):
        return []

    dest_keys = set(_parse_env_file(dest_path).keys())
    to_append: list[tuple[str, str]] = []
    for key, assignment in _iter_env_assignments(example_path):
        if key not in dest_keys:
            to_append.append((key, assignment))

    if not to_append:
        return []

    header = (
        f"# Added by {comment_skill} from .env.example"
        if comment_skill
        else "# Added from .env.example"
    )
    with open(dest_path, "a", encoding="utf-8") as f:
        f.write("\n\n" + header + "\n")
        for _, assignment in to_append:
            f.write(assignment + "\n")

    reset_cache()
    return [key for key, _ in to_append]


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