"""配置加载:优先环境变量,其次 config.yaml,最后 config.example.yaml。""" from __future__ import annotations import os from dataclasses import dataclass, field from pathlib import Path from typing import List import yaml try: from dotenv import load_dotenv load_dotenv() # 若存在 .env 则自动加载 except ImportError: pass # python-dotenv 非必须 @dataclass class PanelConfig: host: str = "0.0.0.0" port: int = 8080 password: str = "" @dataclass class Config: app_id: str app_secret: str source_chat_ids: List[str] target_chat_ids: List[str] dedup_cache_size: int = 2000 max_qps: int = 4 max_retry: int = 2 log_level: str = "INFO" panel: PanelConfig = field(default_factory=PanelConfig) def validate(self) -> None: if not self.app_id or not self.app_secret: raise ValueError("app_id / app_secret 不能为空(请在 config.yaml 或环境变量中配置)") if not self.source_chat_ids: raise ValueError("source_chat_ids 至少配置 1 个源群 chat_id") if not self.target_chat_ids: raise ValueError("target_chat_ids 至少配置 1 个目标群 chat_id") if self.max_qps <= 0 or self.max_qps > 5: raise ValueError(f"max_qps 必须在 (0, 5] 区间,当前 {self.max_qps}") @property def panel_password(self) -> str: """面板密码:环境变量优先,其次配置文件。""" return os.getenv("LARK_PANEL_PASSWORD") or self.panel.password def _load_yaml(path: Path) -> dict: if not path.exists(): return {} with path.open("r", encoding="utf-8") as f: return yaml.safe_load(f) or {} def load_config(config_path: str | Path | None = None) -> Config: """加载顺序:环境变量 > config.yaml > config.example.yaml。""" base = Path(__file__).resolve().parent candidates = [] if config_path: candidates.append(Path(config_path)) candidates.append(base / "config.yaml") candidates.append(base / "config.example.yaml") merged: dict = {} for p in candidates: merged.update(_load_yaml(p)) # 环境变量覆盖(优先级最高) env_app_id = os.getenv("APP_ID") env_app_secret = os.getenv("APP_SECRET") if env_app_id: merged["app_id"] = env_app_id if env_app_secret: merged["app_secret"] = env_app_secret # 标准化白名单:去空白、去空串 for key in ("source_chat_ids", "target_chat_ids"): val = merged.get(key) or [] merged[key] = [str(x).strip() for x in val if str(x).strip()] # panel 段 panel_raw = merged.get("panel") or {} panel = PanelConfig( host=str(panel_raw.get("host", "0.0.0.0")), port=int(panel_raw.get("port", 8080)), password=str(panel_raw.get("password", "")), ) try: cfg = Config( app_id=merged.get("app_id", ""), app_secret=merged.get("app_secret", ""), source_chat_ids=merged["source_chat_ids"], target_chat_ids=merged["target_chat_ids"], dedup_cache_size=int(merged.get("dedup_cache_size", 2000)), max_qps=int(merged.get("max_qps", 4)), max_retry=int(merged.get("max_retry", 2)), log_level=str(merged.get("log_level", "INFO")).upper(), panel=panel, ) except KeyError as e: raise ValueError(f"配置缺失字段: {e}") from e cfg.validate() return cfg