config.py

import json
import os
import re
import logging
import queue
from logging.handlers import RotatingFileHandler, QueueHandler, QueueListener
from typing import Any, Dict, List


DEFAULT_CONFIG: Dict[str, Any] = {
    "poll_interval_sec": 15,
    "max_emails_per_poll_per_account": 20,
    "initial_fetch_mode": "all",  # all | unseen
    "initial_fetch_limit": 200,
    "initial_fetch_on_empty_only": True,
    "code_patterns": [
        r"\\b\\d{6}\\b",
        r"(?:code|CODE|Code)[:：]\\s*([0-9]{4,8})",
        r"(?:验证码)[:：]\\s*([0-9]{4,8})",
        r"One-Time Password[:：]?\\s*([0-9]{4,8})",
    ],
    "subject_keywords": [],
    "sender_keywords": [],
    # logging
    "log_level": "DEBUG",
    "log_to_console": True,
    "log_file": "email_tools.log",
    "log_max_bytes": 2 * 1024 * 1024,
    "log_backup_count": 3,
    # UI
    "refresh_on_account_switch": False,
    "auto_start_poll": False,
    # proxy
    "proxy": {
        "enabled": False,
        "type": "socks5",  # socks5 | socks4 | http
        "host": "127.0.0.1",
        "port": 7890,
        "username": "",
        "password": "",
    },
}


def ensure_config_file(config_path: str) -> Dict[str, Any]:
    if not os.path.exists(config_path):
        with open(config_path, "w", encoding="utf-8") as f:
            json.dump(DEFAULT_CONFIG, f, ensure_ascii=False, indent=2)
        return DEFAULT_CONFIG
    with open(config_path, "r", encoding="utf-8") as f:
        data = json.load(f)
    # merge defaults
    merged = {**DEFAULT_CONFIG, **data}
    return merged


def save_config(config_path: str, data: Dict[str, Any]) -> None:
    with open(config_path, "w", encoding="utf-8") as f:
        json.dump(data, f, ensure_ascii=False, indent=2)


def compile_code_patterns(patterns: List[str]) -> List[re.Pattern]:
    compiled = []
    for p in patterns:
        try:
            compiled.append(re.compile(p))
        except re.error:
            continue
    return compiled


def setup_logging(cfg: Dict[str, Any]) -> None:
    """Configure rotative file logging and optional console logging.

    Idempotent: subsequent calls won't duplicate handlers for the root logger.
    """
    level_name = str(cfg.get("log_level", "INFO")).upper()
    level = getattr(logging, level_name, logging.INFO)

    root = logging.getLogger()
    root.setLevel(level)

    # Avoid duplicate handlers when reloading
    if getattr(root, "_email_tools_logging_configured", False):
        return

    log_dir = os.path.dirname(os.path.abspath(__file__))
    log_file = os.path.join(log_dir, str(cfg.get("log_file", "email_tools.log")))
    max_bytes = int(cfg.get("log_max_bytes", 2 * 1024 * 1024))
    backup_count = int(cfg.get("log_backup_count", 3))

    formatter = logging.Formatter(
        fmt="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
        datefmt="%Y-%m-%d %H:%M:%S",
    )

    # Real handlers managed by a QueueListener
    file_handler = RotatingFileHandler(log_file, maxBytes=max_bytes, backupCount=backup_count, encoding="utf-8")
    file_handler.setLevel(level)
    file_handler.setFormatter(formatter)

    handlers = [file_handler]
    if bool(cfg.get("log_to_console", True)):
        console = logging.StreamHandler()
        console.setLevel(level)
        console.setFormatter(formatter)
        handlers.append(console)

    q: queue.Queue = queue.Queue(maxsize=10000)
    queue_listener = QueueListener(q, *handlers, respect_handler_level=True)
    queue_listener.daemon = True
    queue_listener.start()

    queue_handler = QueueHandler(q)
    queue_handler.setLevel(level)
    # Remove any pre-existing handlers to avoid duplicate output
    for h in list(root.handlers):
        root.removeHandler(h)
    root.addHandler(queue_handler)

    root._email_tools_logging_configured = True  # type: ignore[attr-defined]
    root._email_tools_queue_listener = queue_listener  # type: ignore[attr-defined]


__all__ = [
    "ensure_config_file",
    "save_config",
    "compile_code_patterns",
    "DEFAULT_CONFIG",
    "setup_logging",
]