model-watchdog/config.py

"""配置加载：YAML -> ${ENV} 注入 -> 补默认值 -> 校验。

设计目标：config.yaml 里只需要按 API 端点分类填模型名，
端点地址、探测参数、判定字段全部由本文件的 ENDPOINTS 提供默认值。
"""
import logging
import os
import re

import yaml

from mailer import normalize_domain

log = logging.getLogger("config")

_ENV_RE = re.compile(r"\$\{([A-Za-z0-9_]+)\}")

# ============================================================
# 按 API 端点分类的默认值
# key 即 config.yaml 中 models 下的分组名，用户只填 models 列表
# ============================================================
ENDPOINTS = {
    # 文本对话（OpenAI 兼容模式）
    "chat": {
        "label": "文本对话",
        "url": "https://dashscope.aliyuncs.com/compatible-mode/v1/chat/completions",
        "timeout_seconds": 60,
        # 最省 token 的探测：只让模型回一个 ok
        "prompt": "ok",
        "max_tokens": 16,
        # qwen3 系列混合推理模型非流式调用需要关闭思考模式；
        # 若某模型不接受该参数，把这里清空即可
        "extra_params": {"enable_thinking": False},
    },
    # 文本转语音（WebSocket 协议）
    "tts": {
        "label": "文本转语音",
        "url": "wss://dashscope.aliyuncs.com/api-ws/v1/inference/",
        "timeout_seconds": 30,
        # 只建立连接并等 task-started，不发送待合成文本，因此不产生合成用量
        "voice": "longxiaochun_v2",
        "audio_format": "mp3",
        "sample_rate": 22050,
    },
    # 语音转文本（录音文件识别，异步提交）
    "asr": {
        "label": "语音转文本",
        "url": "https://dashscope.aliyuncs.com/api/v1/services/audio/asr/transcription",
        "timeout_seconds": 60,
        # 官方示例音频，只提交任务不轮询结果
        "sample_audio_url": (
            "https://dashscope.oss-cn-beijing.aliyuncs.com/samples/audio/paraformer/hello_world_female2.wav"
        ),
    },
    # 文生图（异步提交）
    "text2image": {
        "label": "文生图",
        "url": "https://dashscope.aliyuncs.com/api/v1/services/aigc/text2image/image-synthesis",
        "timeout_seconds": 60,
        "prompt": "a red apple",
        "size": "1024*1024",
        # validate=不传 prompt，靠服务端"先解析模型再校验参数"的顺序免费判断模型是否在线
        # submit=真实提交生成任务（会产生费用）
        "probe_mode": "validate",
    },
    # 文生视频（异步提交）—— 单次调用有真实费用，见 config.yaml 注释
    "text2video": {
        "label": "文生视频",
        "url": "https://dashscope.aliyuncs.com/api/v1/services/aigc/video-generation/video-synthesis",
        "timeout_seconds": 60,
        "prompt": "a red apple on a table",
        "size": "832*480",
        "duration": 5,
        "probe_mode": "validate",
    },
}


def expand_env(value):
    """递归把字符串中的 ${ENV} 替换为环境变量值（未设置则替换为空串）。"""
    if isinstance(value, str):
        return _ENV_RE.sub(lambda m: os.environ.get(m.group(1), ""), value)
    if isinstance(value, dict):
        return {k: expand_env(v) for k, v in value.items()}
    if isinstance(value, list):
        return [expand_env(v) for v in value]
    return value


def load_config(path):
    with open(path, "r", encoding="utf-8") as f:
        cfg = yaml.safe_load(f) or {}
    if not isinstance(cfg, dict):
        raise ValueError("配置文件顶层必须是映射（mapping）")
    cfg = expand_env(cfg)
    _apply_defaults(cfg)
    _validate(cfg)
    return cfg


def _apply_defaults(cfg):
    cfg.setdefault("api_key", "")
    cfg.setdefault("interval_minutes", 360)

    mail = cfg.setdefault("mail", {})
    mail.setdefault("imap_port", 993)
    mail.setdefault("inbox_folder", "INBOX")
    mail.setdefault("sent_folder", "Sent Messages")
    mail.setdefault("smtp_port", 465)
    mail.setdefault("smtp_ssl", True)
    mail.setdefault("append_to_sent", True)
    mail.setdefault("notify_emails", [])
    if mail["notify_emails"] is None:
        mail["notify_emails"] = []

    notice = cfg.setdefault("offline_notice", {})
    notice.setdefault("keyword", "模型下线通知")
    notice.setdefault("sender_domains", [])
    notice.setdefault("lookback_days", 3)
    if notice["sender_domains"] is None:
        notice["sender_domains"] = []
    if isinstance(notice["sender_domains"], str):   # 只写一个域名时允许不写成列表
        notice["sender_domains"] = [notice["sender_domains"]]
    # 兼容旧配置里的 sender（完整邮箱地址）：自动取其域名
    legacy = (notice.pop("sender", "") or "").strip()
    if legacy and not notice["sender_domains"]:
        notice["sender_domains"] = [legacy]
    notice["sender_domains"] = [
        d for d in (normalize_domain(x) for x in notice["sender_domains"]) if d
    ]

    dt = cfg.setdefault("dingtalk", {})
    dt.setdefault("enabled", False)
    dt.setdefault("access_token", "")
    dt.setdefault("webhook", "")
    dt.setdefault("secret", "")
    dt.setdefault("at_mobiles", [])
    dt.setdefault("at_all", False)
    dt.setdefault("timeout_seconds", 15)
    # yaml 里写了键但没写条目会解析成 None，统一规范成空列表
    if dt["at_mobiles"] is None:
        dt["at_mobiles"] = []

    # 模型分组：用端点默认值兜底，用户只需提供 models 列表
    groups = cfg.setdefault("models", {})
    for name, defaults in ENDPOINTS.items():
        g = groups.setdefault(name, {})
        if not isinstance(g, dict):
            raise ValueError(f"models.{name} 必须是映射（mapping）")
        g.setdefault("enabled", True)
        g.setdefault("models", [])
        for k, v in defaults.items():
            g.setdefault(k, v)
        g["endpoint"] = name


def _validate(cfg):
    if not cfg["api_key"]:
        raise ValueError("api_key 必填（百炼 DashScope Key，也可用 ${DASHSCOPE_API_KEY} 注入）")

    try:
        interval = int(cfg["interval_minutes"])
    except (TypeError, ValueError):
        raise ValueError("interval_minutes 必须是整数（分钟）") from None
    if interval < 1:
        raise ValueError("interval_minutes 必须 >= 1")
    cfg["interval_minutes"] = interval

    mail = cfg["mail"]
    for key, desc in (
        ("imap_host", "IMAP 服务器"),
        ("smtp_host", "SMTP 服务器"),
        ("username", "邮箱账号"),
        ("password", "邮箱密码/授权码"),
    ):
        if not mail.get(key):
            raise ValueError(f"mail.{key} 必填（{desc}）")
    if not mail["notify_emails"]:
        raise ValueError("mail.notify_emails 不能为空（告警群发对象）")

    if not cfg["offline_notice"]["sender_domains"]:
        log.warning(
            "offline_notice.sender_domains 为空：模块 A 将不校验发件人，"
            "任何人发含关键词的邮件都会触发转发，建议至少填一个域名"
        )

    dt = cfg["dingtalk"]
    if dt.get("enabled"):
        has_token = bool((dt.get("access_token") or "").strip())
        has_in_url = "access_token=" in (dt.get("webhook") or "")
        if not (has_token or has_in_url):
            raise ValueError(
                "dingtalk.enabled 为 true 时必须配置 access_token"
                "（或在 webhook 里自带 access_token）"
            )
    if not isinstance(dt.get("at_mobiles") or [], list):
        raise ValueError("dingtalk.at_mobiles 必须是列表")

    unknown = set(cfg["models"]) - set(ENDPOINTS)
    if unknown:
        raise ValueError(
            "models 下存在未知端点分组: %s（可用: %s）"
            % (", ".join(sorted(unknown)), ", ".join(ENDPOINTS))
        )
    for name, g in cfg["models"].items():
        if not isinstance(g["models"], list):
            raise ValueError(f"models.{name}.models 必须是列表")


def enabled_probe_targets(cfg):
    """展开成待探测列表: [(模型名, 分组配置), ...]，保持配置顺序。"""
    targets = []
    for name in ENDPOINTS:
        g = cfg["models"].get(name) or {}
        if not g.get("enabled"):
            continue
        for model in g["models"]:
            if model:
                targets.append((model, g))
    return targets