lark2lark/config.py

"""配置加载：优先环境变量，其次 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