from __future__ import annotations """Validated application configuration. Secrets are read from environment variables (or a local .env file) and are never included in ``safe_summary`` or log messages. """ import json import re from pathlib import Path from typing import Optional, Any from zoneinfo import ZoneInfo, ZoneInfoNotFoundError from pydantic import Field, SecretStr, field_validator, model_validator from pydantic_settings import BaseSettings, SettingsConfigDict _TOKEN_RE = re.compile(r"^\d{5,15}:[A-Za-z0-9_-]{20,}$") _USERNAME_RE = re.compile(r"^[A-Za-z][A-Za-z0-9_]{3,31}$") def _split_csv(value: Any) -> list[str]: if value is None: return [] if isinstance(value, (list, tuple, set)): return [str(item).strip() for item in value if str(item).strip()] text = str(value).strip() if not text: return [] if text.startswith("[") and text.endswith("]"): try: parsed = json.loads(text) except ValueError as exc: raise ValueError("expected a comma-separated list") from exc if not isinstance(parsed, list): raise ValueError("expected a comma-separated list") return [str(item).strip() for item in parsed if str(item).strip()] return [item.strip() for item in text.split(",") if item.strip()] class Settings(BaseSettings): model_config = SettingsConfigDict( env_file=".env", env_file_encoding="utf-8", case_sensitive=False, extra="ignore", validate_default=True, ) bot_token: SecretStr admin_ids: list[int] bot_username: Optional[str] = None database_url: str = "sqlite+aiosqlite:///./data/publisher.db" auto_create_schema: bool = True timezone: str = "Asia/Tehran" log_level: str = "INFO" log_file: Path = Path("./logs/bot.log") data_dir: Path = Path("./data") telegram_api_base_url: Optional[str] = None proxy_url: Optional[str] = None max_media_size_mb: int = Field(default=2000, ge=1, le=4096) max_caption_length: int = Field(default=1024, ge=50, le=4096) delivery_rate_limit_requests: int = Field(default=5, ge=1, le=1000) delivery_rate_limit_window_seconds: int = Field(default=60, ge=1, le=86400) delivery_min_interval_seconds: int = Field(default=2, ge=0, le=3600) scheduler_poll_seconds: int = Field(default=5, ge=1, le=300) job_max_attempts: int = Field(default=5, ge=1, le=20) telegram_request_timeout_seconds: int = Field(default=60, ge=10, le=600) retry_max_attempts: int = Field(default=3, ge=1, le=10) retry_base_delay_seconds: float = Field(default=1.0, ge=0, le=30) telegram_upload_limit_mb: int = Field(default=50, ge=1, le=4000) # Broadcast delivery is a durable queue. The scheduler drains a bounded # batch on every tick; these values keep the bot inside Telegram's limits. broadcast_batch_size: int = Field(default=20, ge=1, le=200) broadcast_min_interval_seconds: float = Field(default=0.05, ge=0, le=10) broadcast_max_attempts: int = Field(default=3, ge=1, le=10) broadcast_retry_after_cap_seconds: int = Field(default=300, ge=1, le=3600) @field_validator("bot_token", mode="before") @classmethod def _validate_bot_token(cls, value: Any) -> str: token = value.get_secret_value() if isinstance(value, SecretStr) else str(value or "") token = token.strip() if not token: raise ValueError("BOT_TOKEN is required") if not _TOKEN_RE.fullmatch(token): raise ValueError("BOT_TOKEN format is invalid; use the exact token from BotFather") return token @field_validator("admin_ids", mode="before") @classmethod def _parse_admin_ids(cls, value: Any) -> list[int]: raw = _split_csv(value) if not raw: raise ValueError("ADMIN_IDS is required (comma-separated numeric Telegram user IDs)") result: list[int] = [] for item in raw: try: number = int(item) except ValueError as exc: raise ValueError(f"ADMIN_IDS contains a non-numeric value: {item!r}") from exc if number <= 0: raise ValueError("Telegram user IDs must be positive integers") result.append(number) return list(dict.fromkeys(result)) @field_validator("bot_username", mode="before") @classmethod def _validate_username(cls, value: Any) -> Optional[str]: if value is None: return None text = str(value).strip().lstrip("@") if not text: return None if not _USERNAME_RE.fullmatch(text): raise ValueError("BOT_USERNAME is not a valid Telegram bot username") return text @field_validator("database_url") @classmethod def _validate_database_url(cls, value: str) -> str: text = value.strip() if not text: raise ValueError("DATABASE_URL is required") if not ( text.startswith("sqlite+aiosqlite://") or text.startswith("postgresql+asyncpg://") ): raise ValueError( "DATABASE_URL must start with sqlite+aiosqlite:// or postgresql+asyncpg://" ) return text @field_validator("telegram_api_base_url") @classmethod def _validate_api_base(cls, value: Optional[str]) -> Optional[str]: if value is None: return None text = value.strip().rstrip("/") if not text: return None if not text.startswith(("http://", "https://")): raise ValueError("TELEGRAM_API_BASE_URL must start with http:// or https://") return text @field_validator("proxy_url") @classmethod def _validate_proxy(cls, value: Optional[str]) -> Optional[str]: if value is None: return None text = str(value).strip() if not text: return None if not text.lower().startswith(("http://", "https://")): raise ValueError("PROXY_URL must be an HTTP(S) proxy URL") return text @field_validator("timezone") @classmethod def _validate_timezone(cls, value: str) -> str: name = value.strip() try: ZoneInfo(name) except ZoneInfoNotFoundError as exc: raise ValueError(f"unknown IANA timezone: {name!r}") from exc return name @field_validator("log_level") @classmethod def _validate_log_level(cls, value: str) -> str: level = value.strip().upper() if level not in {"DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"}: raise ValueError("LOG_LEVEL must be DEBUG, INFO, WARNING, ERROR or CRITICAL") return level @model_validator(mode="after") def _cross_validate(self) -> Settings: if self.delivery_rate_limit_requests <= 0: raise ValueError("DELIVERY_RATE_LIMIT_REQUESTS must be positive") if self.telegram_upload_limit_mb > self.max_media_size_mb: pass return self @property def tzinfo(self) -> ZoneInfo: return ZoneInfo(self.timezone) @property def max_media_size_bytes(self) -> int: return self.max_media_size_mb * 1024 * 1024 @property def telegram_upload_limit_bytes(self) -> int: return self.telegram_upload_limit_mb * 1024 * 1024 def ensure_directories(self) -> None: self.data_dir.mkdir(parents=True, exist_ok=True) self.log_file.parent.mkdir(parents=True, exist_ok=True) if self.database_url.startswith("sqlite"): db_path = self.database_url.split("///", 1)[-1] if db_path and db_path != ":memory:": Path(db_path).expanduser().parent.mkdir(parents=True, exist_ok=True) def safe_summary(self) -> dict[str, Any]: return { "admin_ids": self.admin_ids, "bot_username": self.bot_username, "database_url": self.database_url.split("@")[-1], "auto_create_schema": self.auto_create_schema, "timezone": self.timezone, "log_level": self.log_level, "data_dir": str(self.data_dir), "max_media_size_mb": self.max_media_size_mb, "telegram_upload_limit_mb": self.telegram_upload_limit_mb, "broadcast_batch_size": self.broadcast_batch_size, "broadcast_min_interval_seconds": self.broadcast_min_interval_seconds, "proxy_configured": bool(self.proxy_url), }