ZNJJ-api-server/src/core/config.py

import os
from dotenv import load_dotenv

load_dotenv()

class Config:
    PROJECT_NAME = "Flexible Employment Analysis API"
    API_V1_STR = "/api/v1"
    SECRET_KEY = os.getenv("SECRET_KEY", "your-secret-key")
    DATABASE_URL = os.getenv("DATABASE_URL", "sqlite:///./test.db")
    DEBUG = os.getenv("DEBUG", "false").lower() in ("true", "1", "t")
    
    # FastGPT Configuration
    FASTGPT_API_KEY = os.getenv("ANALYSIS_AUTH_TOKEN")
    FASTGPT_BASE_URL = os.getenv("ANALYSIS_SERVICE_URL")
    FASTGPT_APP_ID = os.getenv("APP_ID")
    
    # Chat backend switch（作业 2）
    # "fastgpt" 保持现有行为；"langgraph" 使用内存图 fake 回复。
    AGENT_BACKEND = os.getenv("AGENT_BACKEND", "fastgpt")

    # 命名 LLM 提供商列表（均为 OpenAI 兼容接口），两种写法二选一：
    # ① JSON 字典（推荐，单变量管理）：
    #   LLM_PROVIDERS={"deepseek": {"base_url": "https://...", "api_key": "...",
    #                    "model": "deepseek-v4.1-flash"},
    #                  "kimi": {"base_url": "http://127.0.0.1:3425/v1",
    #                    "api_key": "magpie", "model": "kimi-code-cn/kimi-for-coding",
    #                    "extra_body": {"reasoning_effort": "none"}}}
    #   每组 base_url/api_key/model 必填；timeout_seconds/max_retries/
    #   temperature 可选；extra_body 为对象时直接合并进请求体（思考开关等）。
    # ② 逗号分隔名单 + 分散变量：
    #   LLM_PROVIDERS=deepseek,kimi + LLM_DEEPSEEK_BASE_URL/_API_KEY/_MODEL 等。
    # 运行时选择：LLM_PROVIDER=<名>；缺省或 fake = 离线脚本模型。
    LLM_PROVIDERS = os.getenv("LLM_PROVIDERS", "")
    LLM_PROVIDER = os.getenv("LLM_PROVIDER", "")
    # LLM 客户端是否信任系统代理（Windows 注册表）。缺省 False：
    # 本地网关（127.0.0.1）必须直连；外部网关若需代理设为 true。
    LLM_HTTP_TRUST_ENV = os.getenv("LLM_HTTP_TRUST_ENV", "false").lower() in (
        "true", "1", "t",
    )

    @classmethod
    def _declared_provider_names(cls) -> list[str]:
        """LLM_PROVIDERS 中声明的提供商名（JSON 字典取键，名单取逗号项）。"""
        raw = cls.LLM_PROVIDERS.strip()
        if not raw:
            return []
        if raw.startswith("{"):
            import json

            data = json.loads(raw)
            return [str(k).strip().lower() for k in data]
        return [n.strip().lower() for n in raw.split(",") if n.strip()]

    @classmethod
    def llm_profiles(cls) -> dict[str, dict]:
        """解析命名提供商列表为统一结构；声明了但缺必填项的组不收录
        （由 validate() 统一报错）。extra_body 统一为 JSON 字符串。"""
        raw = cls.LLM_PROVIDERS.strip()
        if not raw:
            return {}
        if raw.startswith("{"):
            return cls._profiles_from_json(raw)
        return cls._profiles_from_prefixed(raw)

    @classmethod
    def _profiles_from_json(cls, raw: str) -> dict[str, dict]:
        import json

        try:
            data = json.loads(raw)
        except ValueError as exc:
            raise ValueError(f"LLM_PROVIDERS is not valid JSON: {exc}") from exc
        if not isinstance(data, dict):
            raise ValueError("LLM_PROVIDERS JSON must be an object keyed by provider name")
        profiles: dict[str, dict] = {}
        for key, spec in data.items():
            name = str(key).strip().lower()
            if not name or not isinstance(spec, dict):
                continue
            base_url = str(spec.get("base_url", ""))
            api_key = str(spec.get("api_key", ""))
            model = str(spec.get("model", ""))
            if not (base_url and api_key and model):
                continue
            extra_body = spec.get("extra_body", "")
            if isinstance(extra_body, dict):
                extra_body = json.dumps(extra_body, ensure_ascii=False)
            temperature = spec.get("temperature")
            profiles[name] = {
                "base_url": base_url,
                "api_key": api_key,
                "model": model,
                "timeout_seconds": float(spec.get("timeout_seconds", 30)),
                "max_retries": int(spec.get("max_retries", 2)),
                "temperature": float(temperature) if temperature is not None else None,
                "extra_body": str(extra_body),
            }
        return profiles

    @classmethod
    def _profiles_from_prefixed(cls, raw: str) -> dict[str, dict]:
        profiles: dict[str, dict] = {}
        for name in cls._declared_provider_names():
            prefix = f"LLM_{name.upper()}_"
            base_url = os.getenv(prefix + "BASE_URL", "")
            api_key = os.getenv(prefix + "API_KEY", "")
            model = os.getenv(prefix + "MODEL", "")
            if not (base_url and api_key and model):
                continue
            temperature = os.getenv(prefix + "TEMPERATURE")
            profiles[name] = {
                "base_url": base_url,
                "api_key": api_key,
                "model": model,
                "timeout_seconds": float(os.getenv(prefix + "TIMEOUT_SECONDS", "30")),
                "max_retries": int(os.getenv(prefix + "MAX_RETRIES", "2")),
                "temperature": float(temperature) if temperature else None,
                "extra_body": os.getenv(prefix + "EXTRA_BODY", ""),
            }
        return profiles

    @classmethod
    def llm_profile(cls, name: str) -> dict | None:
        return cls.llm_profiles().get(name.strip().lower())
    
    @classmethod
    def validate(cls):
        """Validate required configuration."""
        # TODO(作业 2): 语音管线仍依赖 FastGPT。当前仅按文本后端开关校验；
        # 接入语音开关后，需要单独校验语音路径所需配置。
        if cls.AGENT_BACKEND not in ("fastgpt", "langgraph"):
            raise ValueError(
                f"AGENT_BACKEND must be 'fastgpt' or 'langgraph', got {cls.AGENT_BACKEND!r}"
            )
        profiles = cls.llm_profiles()  # JSON 非法时在此抛出明确错误
        incomplete = [n for n in cls._declared_provider_names() if n not in profiles]
        if incomplete:
            raise ValueError(
                "LLM providers missing BASE_URL/API_KEY/MODEL: "
                + ", ".join(incomplete)
            )
        if (
            cls.LLM_PROVIDER
            and cls.LLM_PROVIDER != "fake"
            and cls.LLM_PROVIDER.strip().lower() not in profiles
        ):
            raise ValueError(
                f"Unknown LLM_PROVIDER {cls.LLM_PROVIDER!r}; "
                f"configured: {sorted(profiles)}"
            )
        if cls.AGENT_BACKEND != "fastgpt":
            return
        required = [
            ("FASTGPT_BASE_URL (ANALYSIS_SERVICE_URL)", cls.FASTGPT_BASE_URL),
            ("FASTGPT_API_KEY (ANALYSIS_AUTH_TOKEN)", cls.FASTGPT_API_KEY),
            ("FASTGPT_APP_ID (APP_ID)", cls.FASTGPT_APP_ID),
        ]
        missing = [name for name, value in required if not value]
        if missing:
            raise ValueError(f"Missing required environment variables: {', '.join(missing)}")