"""Validated application configuration loaded from environment / .env.""" from __future__ import annotations import ipaddress import re from pathlib import Path from typing import Annotated, Any from pydantic import Field, SecretStr, field_validator, model_validator from pydantic_settings import BaseSettings, NoDecode, SettingsConfigDict _DOMAIN_RE = re.compile(r"^(?!-)[a-z0-9-]{1,63}(? 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("]"): import json 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()] def normalize_domain(raw: str) -> str: """Normalize an allow-list domain and reject unsafe/ambiguous forms.""" value = raw.strip().lower().rstrip(".") if not value: raise ValueError("empty domain") if "://" in value or "/" in value or "?" in value or "#" in value or "@" in value: raise ValueError(f"domain must not contain scheme, path, credentials: {raw!r}") if "*" in value or "[" in value or "]" in value: raise ValueError(f"wildcards and bracketed literals are not supported: {raw!r}") try: ip = ipaddress.ip_address(value) except ValueError: pass else: if not ip.is_global: raise ValueError(f"non-public address is not allowed in ALLOWED_DOMAINS: {raw!r}") return ip.compressed if re.fullmatch( r"(?:\d+|\d+(?:\.\d+){1,4}|0[xX][0-9a-fA-F]+(?:\.[0-9a-fA-F]+)*)", value, ): raise ValueError(f"ambiguous numeric hostname is not allowed: {raw!r}") if value in {"localhost", "local", "internal"} or value.endswith( (".localhost", ".local", ".internal", ".home.arpa") ): raise ValueError(f"internal hostname is not allowed: {raw!r}") try: ascii_value = value.encode("idna").decode("ascii") except UnicodeError as exc: raise ValueError(f"invalid internationalized domain: {raw!r}") from exc if ":" in ascii_value or not _DOMAIN_RE.fullmatch(ascii_value): raise ValueError(f"invalid domain name: {raw!r}") return ascii_value 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: Annotated[list[int], NoDecode] default_chat_id: int | None = None telegram_api_base_url: str | None = None proxy_url: str | None = None database_path: Path = Path("./data/bot.db") log_file: Path = Path("./logs/bot.log") log_level: str = "INFO" allowed_domains: Annotated[list[str], NoDecode] max_pages_per_job: int = Field(default=1000000, ge=1) max_videos_per_job: int = Field(default=1000000000, ge=1) max_concurrent_downloads: int = Field(default=4, ge=1, le=50) max_file_size_mb: int = Field(default=4000, ge=1) request_timeout: int = Field(default=120, ge=5) retry_count: int = Field(default=5, ge=0) max_job_seconds: int = Field(default=2592000, ge=30) page_delay_seconds: float = Field(default=0.2, ge=0) max_html_mb: int = Field(default=50, ge=1) respect_robots_txt: bool = True http_user_agent: str = "TelegramVideoBot/1.0 (authorized-content-publisher)" playwright_headless: bool = True browser_fallback_enabled: bool = False ffmpeg_enabled: bool = True send_thumbnails: bool = True telegram_max_upload_mb: int = Field(default=2000, ge=1) telegram_caption_limit: int = Field(default=1024, ge=50, le=4096) telegram_upload_timeout: int = Field(default=7200, ge=30) progress_update_interval: int = Field(default=15, ge=3, le=600) user_rate_limit_seconds: float = Field(default=0.5, ge=0, le=30) @field_validator("bot_token", mode="before") @classmethod def _validate_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; expected the token exactly as issued by BotFather" ) return token @field_validator("proxy_url", mode="before") @classmethod def _validate_proxy_url(cls, value: Any) -> str | None: 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 proxy URL such as http://127.0.0.1:10809" ) return text @field_validator("admin_ids", mode="before") @classmethod def _parse_admin_ids(cls, value: Any) -> list[int]: raw_items = _split_csv(value) if not raw_items: raise ValueError("ADMIN_IDS is required (comma-separated numeric Telegram user IDs)") result: list[int] = [] for item in raw_items: 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("allowed_domains", mode="before") @classmethod def _parse_domains(cls, value: Any) -> list[str]: return [normalize_domain(item) for item in _split_csv(value)] @field_validator("log_level") @classmethod def _normalize_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 @field_validator("telegram_api_base_url") @classmethod def _validate_api_base(cls, value: str | None) -> str | None: if value is None: return None text = value.strip().rstrip("/") if not text: return None if not (text.startswith("http://") or text.startswith("https://")): raise ValueError("TELEGRAM_API_BASE_URL must start with http:// or https://") return text @field_validator("database_path", "log_file") @classmethod def _validate_path(cls, value: Path) -> Path: if not str(value).strip(): raise ValueError("path must not be empty") return value @model_validator(mode="after") def _cross_validate(self) -> Settings: if not self.allowed_domains: raise ValueError( "ALLOWED_DOMAINS must contain at least one domain; refusing to crawl without " "an explicit allow-list" ) return self @property def max_file_size_bytes(self) -> int: return self.max_file_size_mb * 1024 * 1024 @property def telegram_upload_limit_bytes(self) -> int: return self.telegram_max_upload_mb * 1024 * 1024 @property def max_html_bytes(self) -> int: return self.max_html_mb * 1024 * 1024 def ensure_directories(self) -> None: for directory in (self.database_path.parent, self.log_file.parent): directory.mkdir(parents=True, exist_ok=True) def safe_summary(self) -> dict[str, Any]: """Non-secret values suitable for logs and /admin.""" return { "admin_ids": self.admin_ids, "default_chat_id": self.default_chat_id, "database_path": str(self.database_path), "allowed_domains": self.allowed_domains, "media_storage": "disabled (direct stream)", "max_pages_per_job": self.max_pages_per_job, "max_videos_per_job": self.max_videos_per_job, "max_concurrent_downloads": self.max_concurrent_downloads, "max_file_size_mb": self.max_file_size_mb, "telegram_max_upload_mb": self.telegram_max_upload_mb, "browser_fallback_enabled": self.browser_fallback_enabled, "proxy_configured": bool(self.proxy_url), "ffmpeg_enabled": self.ffmpeg_enabled, "respect_robots_txt": self.respect_robots_txt, }