"""配置加载：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