import os from dataclasses import dataclass from functools import lru_cache from pathlib import Path from typing import Optional from dotenv import load_dotenv BASE_DIR = Path(__file__).resolve().parents[1] DOTENV_PATH = BASE_DIR / ".env" if DOTENV_PATH.exists(): load_dotenv(DOTENV_PATH) @dataclass class Settings: telegram_api_id: int telegram_api_hash: str telegram_session_path: str telegram_bot_username: str telegram_bot_token: Optional[str] host: str port: int enable_pagination: bool pagination_max_pages: int pagination_delay: float pagination_timeout: float pagination_keywords: list[str] database_url: str remote_upload_enabled: bool remote_ssh_host: str remote_ssh_user: str remote_ssh_password: str remote_ssh_target: str remote_upload_interval: int remote_upload_batch_size: int def __post_init__(self): if not self.database_url: raise ValueError("DATABASE_URL 未配置,无法初始化数据库连接") def _env_bool(name: str, default: bool = False) -> bool: return os.getenv(name, str(default)).strip().lower() in ("1", "true", "yes", "on") def _env_int(name: str, default: int) -> int: try: return int(os.getenv(name, str(default))) except ValueError: return default def _env_float(name: str, default: float) -> float: try: return float(os.getenv(name, str(default))) except ValueError: return default @lru_cache() def get_settings() -> Settings: keywords = os.getenv( "FUNSTAT_PAGINATION_KEYWORDS", "➡️,下一页,Next,更多,下页,›,>>" ) return Settings( telegram_api_id=_env_int("TELEGRAM_API_ID", 24660516), telegram_api_hash=os.getenv("TELEGRAM_API_HASH", "eae564578880a59c9963916ff1bbbd3a"), telegram_session_path=os.getenv( "TELEGRAM_SESSION_PATH", str(Path.home() / "telegram_sessions" / "funstat_bot") ), telegram_bot_username=os.getenv("TELEGRAM_BOT_USERNAME", "@ktqiangda_bot"), telegram_bot_token=os.getenv("TELEGRAM_BOT_TOKEN"), host=os.getenv("FUNSTAT_HOST", "127.0.0.1"), port=_env_int("FUNSTAT_PORT", 8094), enable_pagination=_env_bool("FUNSTAT_ENABLE_PAGINATION", True), pagination_max_pages=_env_int("FUNSTAT_PAGINATION_MAX_PAGES", 10), pagination_delay=_env_float("FUNSTAT_PAGINATION_DELAY", 2.0), pagination_timeout=_env_float("FUNSTAT_PAGINATION_TIMEOUT", 8.0), pagination_keywords=[ kw.strip() for kw in keywords.split(",") if kw.strip() ], database_url=os.getenv( "DATABASE_URL", "postgresql://funstat:funstat_dev_password@127.0.0.1:5433/funstat" ), remote_upload_enabled=_env_bool("REMOTE_UPLOAD_ENABLED", True), remote_ssh_host=os.getenv("REMOTE_SSH_HOST", ""), remote_ssh_user=os.getenv("REMOTE_SSH_USER", ""), remote_ssh_password=os.getenv("REMOTE_SSH_PASSWORD", ""), remote_ssh_target=os.getenv("REMOTE_SSH_TARGET", "/home/atai/funstat_data/inbox"), remote_upload_interval=_env_int("REMOTE_UPLOAD_INTERVAL", 120), remote_upload_batch_size=_env_int("REMOTE_UPLOAD_BATCH_SIZE", 200), )