vent_agent_2.0/config.py

"""全局配置：.env 读取与类型化访问的唯一入口。

设计取舍：不用 pydantic-settings，只用 ``python-dotenv`` + 三个 ``get_*``
帮助函数。理由是骨架阶段配置项还在快速增删，引入 settings 模型会让每次
加一个开关都要改三处；这里改一处即可，且每个配置项都带默认值——默认值
必须等于「什么都不配也能跑起来」的那套（沿用 1.5 的工程纪律：默认 = 可用的
最小配置，任何新开关的默认值都不应改变既有行为）。

## 关于环境变量与 .env 的优先级（踩过的坑，别改回去）

机器上若存在全局的 ``OPENAI_API_KEY``（实测遇到过一把无效的），进程环境会
**静默盖掉项目 .env**。症状是：配好的密钥明明是对的，服务却报 401，且报错
里显示的密钥后几位和 .env 里的对不上——足以浪费半天。

所以优先级做成**显式可配**，而不是隐含假定某一方赢：

===================  ====================================================
``DEEP_AGENT_ENV_PRIORITY``  行为
===================  ====================================================
``dotenv``（默认）   项目 .env 优先。防「被污染的 shell 静默改掉项目配置」
``process``          进程环境优先。用于**跑变体实例**：不同端口、不同库、
                     不同方言（``DB_DIALECT=mysql PORT=8081 python main.py``）
``none``             完全忽略 .env，只认进程环境与代码默认值（测试/CI 用）
===================  ====================================================

无论哪一档，**两侧值不同都会打 WARNING 并把键名暴露给 /health**，
让「我改了配置为什么不生效」有迹可循。

为什么默认不是 ``process``：两种失败模式的代价不对等。``dotenv`` 的代价是
「有意的环境变量覆盖不生效」（一条命令行加个前缀就绕过了，且启动日志会告诉
你），``process`` 的代价是「本机某个无关的全局变量静默改掉项目配置」（表现为
身份诡异的 401 或连错数据库）。
"""

from __future__ import annotations

import os
from pathlib import Path

from dotenv import dotenv_values, load_dotenv

# 项目根目录（本文件所在目录）。所有相对路径以此为基准解析，
# 避免「从哪个目录启动服务」影响资源定位。
ROOT = Path(__file__).resolve().parent
APP_VERSION = (ROOT / "VERSION").read_text(encoding="utf-8").strip()

_ENV_PATH = ROOT / ".env"
# dotenv（默认）| process | none
_ENV_PRIORITY = (os.getenv("DEEP_AGENT_ENV_PRIORITY", "dotenv") or "dotenv").strip().lower()
if _ENV_PRIORITY not in {"dotenv", "process", "none"}:
    _ENV_PRIORITY = "dotenv"

# 直接读文件（不经 os.environ），用于检测冲突
_FILE_VALUES: dict[str, str] = (
    {} if _ENV_PRIORITY == "none" else {k: v for k, v in dotenv_values(_ENV_PATH).items() if v}
)
# 冲突记录：键名 → 说明。供 log_config_sources() 与 /health 使用
CONFLICTS: dict[str, str] = {}

for _key, _file_value in _FILE_VALUES.items():
    _process_value = os.environ.get(_key)
    if _process_value and _process_value != _file_value:
        CONFLICTS[_key] = (
            "项目 .env 覆盖了同名进程环境变量"
            if _ENV_PRIORITY == "dotenv"
            else "进程环境变量覆盖了项目 .env"
        )

# 见模块 docstring 的「环境变量与 .env 的优先级」一节
if _ENV_PRIORITY != "none":
    load_dotenv(_ENV_PATH, override=(_ENV_PRIORITY == "dotenv"))


def _mask(value: str) -> str:
    """给日志用的脱敏显示：只留前 6 后 4 位。"""
    if not value:
        return "(空)"
    if len(value) <= 12:
        return "*" * len(value)
    return f"{value[:6]}...{value[-4:]}"


def log_config_sources() -> None:
    """把配置来源与冲突写进日志。须在 setup_logging() 之后调用。"""
    from obs import get_logger

    logger = get_logger("config")
    if _ENV_PRIORITY == "none":
        logger.warning(
            "DEEP_AGENT_ENV_PRIORITY=none：已忽略 %s，全部配置取自进程环境变量与代码默认值",
            _ENV_PATH,
        )
    elif _ENV_PATH.exists():
        logger.info(
            "配置来源：%s（%d 项）| 优先级=%s | 冲突 %d 项",
            _ENV_PATH, len(_FILE_VALUES), _ENV_PRIORITY, len(CONFLICTS),
        )
    else:
        logger.warning("未找到 %s，全部配置取自进程环境变量与代码默认值", _ENV_PATH)

    for key, reason in CONFLICTS.items():
        # 只暴露键名与脱敏后的值，不打印完整密钥
        effective = os.getenv(key, "")
        logger.warning(
            "配置冲突 %s：%s（生效值 %s）%s",
            key,
            reason,
            _mask(effective),
            "；如需让进程环境优先，加 DEEP_AGENT_ENV_PRIORITY=process"
            if _ENV_PRIORITY == "dotenv"
            else "",
        )


def get_str(key: str, default: str = "") -> str:
    value = os.getenv(key)
    return default if value is None or value == "" else value


def get_int(key: str, default: int = 0) -> int:
    raw = os.getenv(key)
    if raw is None or raw.strip() == "":
        return default
    try:
        return int(raw.strip())
    except ValueError:
        return default


def get_float(key: str, default: float = 0.0) -> float:
    raw = os.getenv(key)
    if raw is None or raw.strip() == "":
        return default
    try:
        return float(raw.strip())
    except ValueError:
        return default


def get_bool(key: str, default: bool = False) -> bool:
    """把 ``1/true/yes/on`` 视为真，``0/false/no/off`` 视为假，其余回落默认值。"""
    raw = os.getenv(key)
    if raw is None or raw.strip() == "":
        return default
    return raw.strip().lower() in {"1", "true", "yes", "on"}


# ============================================================
# 服务
# ============================================================
HOST: str = get_str("HOST", "0.0.0.0")
PORT: int = get_int("PORT", 8080)
LOG_LEVEL: str = get_str("LOG_LEVEL", "INFO")
LOG_FORMAT: str = get_str("LOG_FORMAT", "text")

# ============================================================
# 模型
# ============================================================
MODEL: str = get_str("DEEPAGENT_MODEL", "deepseek-flash")
OPENAI_BASE_URL: str = get_str("OPENAI_BASE_URL", "")
OPENAI_API_KEY: str = get_str("OPENAI_API_KEY", "")
OPENAI_TIMEOUT: int = get_int("OPENAI_TIMEOUT", 180)

# 思考级别：off | high | highest，烧进模型实例（运行时切换需重建 agent）
THINKING_LEVEL: str = get_str("DEEPAGENT_THINKING", "off")

# ============================================================
# 存储
# ============================================================
# mysql（默认，生产/联调）| sqlite（本地开发与自动化测试）
DB_DIALECT: str = get_str("DB_DIALECT", "mysql").lower()
DB_HOST: str = get_str("DB_HOST", "127.0.0.1")
DB_PORT: int = get_int("DB_PORT", 3306)
DB_USER: str = get_str("DB_USER", "root")
DB_PASSWORD: str = get_str("DB_PASSWORD", "")
DB_NAME: str = get_str("DB_NAME", "vent_agent_v2")
DB_PING_IDLE_SECONDS: int = get_int("DB_PING_IDLE_SECONDS", 60)
# sqlite 档位的库文件位置
DB_SQLITE_PATH: str = get_str("DB_SQLITE_PATH", "data/dev.sqlite")

# ============================================================
# 图状态（checkpoint）
# ============================================================
CHECKPOINTER: str = get_str("CHECKPOINTER", "sqlite").lower()  # sqlite | memory
CHECKPOINT_DB_PATH: str = get_str("CHECKPOINT_DB_PATH", "data/checkpoints.sqlite")

# ============================================================
# 上下文与压缩
# ============================================================
# 声明的上下文窗口（0 = 用模型登记值）。
# 这是压缩的**唯一调节旋钮**：框架的摘要中间件在「声明窗口的 85%」处触发，
# 所以想让它更早压缩，就把 CONTEXT_LIMIT 调到 (真实窗口 × 期望比例 / 0.85)。
CONTEXT_LIMIT: int = get_int("CONTEXT_LIMIT", 0)
# 自动压缩开关。关闭时通过 HarnessProfile.excluded_middleware 摘掉框架的
# 摘要中间件——注意这会同时失去「淘汰历史自动落盘到 /conversation_history/」
# 这个能力，长会话会硬撑到超出模型窗口后报错。
CTX_COMPACT_ENABLED: bool = get_bool("CTX_COMPACT_ENABLED", True)
# 冷启动时从会话历史补种的最近消息条数（热线程只喂当轮新消息）
CTX_RECENT_LIMIT: int = get_int("CTX_RECENT_LIMIT", 30)

# 框架固定的压缩触发比例。仅用于展示与断言，改它无效——真正的旋钮是
# CONTEXT_LIMIT。放在这里是为了让 /health 能如实报出「几成触发」。
FRAMEWORK_COMPACT_TRIGGER: float = 0.85

# ============================================================
# 并发闸
# ============================================================
MAX_CONCURRENT_TASKS: int = get_int("TFAGENTS_MAX_CONCURRENT_TASKS", 10)
MAX_USER_TASKS: int = get_int("TFAGENTS_MAX_USER_TASKS", 2)
TASK_WALL_TIMEOUT: int = get_int("TFAGENTS_TASK_WALL_TIMEOUT", 1800)

# 独立算法：单图与批量共享计算并行上限；批量排队、结果保留。
ALGORITHM_MAX_PARALLEL: int = max(1, get_int("ALGORITHM_MAX_PARALLEL", 2))
ALGORITHM_MAX_PENDING_IMAGES: int = max(1, get_int("ALGORITHM_MAX_PENDING_IMAGES", 100))
ALGORITHM_RESULT_TTL_HOURS: int = max(1, get_int("ALGORITHM_RESULT_TTL_HOURS", 24))
ALGORITHM_JOB_DIR: str = get_str("ALGORITHM_JOB_DIR", "data/algorithm_jobs")

# ============================================================
# 计费
# ============================================================
DAILY_TOKEN_LIMIT: int = get_int("TFAGENTS_DAILY_TOKEN_LIMIT", 0)  # 0 = 不限

# ============================================================
# 观测
# ============================================================
TRACE_MIDDLEWARE: bool = get_bool("TFA_TRACE_MIDDLEWARE", True)


def resolve_path(relative: str) -> Path:
    """把配置里的相对路径解析为项目根下的绝对路径，并确保父目录存在。"""
    path = Path(relative)
    if not path.is_absolute():
        path = ROOT / path
    path.parent.mkdir(parents=True, exist_ok=True)
    return path