"""Server-side configuration for the n8n integration.""" from __future__ import annotations import os from dataclasses import dataclass from pathlib import Path from urllib.parse import urlparse from dotenv import dotenv_values DEFAULT_MAX_FILE_BYTES = 10 * 1024 * 1024 ENV_FILE = Path(__file__).resolve().parent / ".env" class ConfigurationError(ValueError): """A required webhook setting is missing or invalid.""" @dataclass(frozen=True) class Settings: chat_webhook_url: str = "" upload_webhook_url: str = "" chat_timeout_seconds: float = 60.0 upload_timeout_seconds: float = 120.0 max_file_bytes: int = DEFAULT_MAX_FILE_BYTES @classmethod def from_env(cls) -> Settings: # Resolve relative to the application, independently of the working directory. values = {**dotenv_values(ENV_FILE, encoding="utf-8-sig"), **os.environ} def setting(name: str, default: str) -> str: value = values.get(name) return default if value is None else value try: chat_timeout = float(setting("N8N_CHAT_TIMEOUT_SECONDS", "60")) upload_timeout = float(setting("N8N_UPLOAD_TIMEOUT_SECONDS", "120")) max_file_bytes = int(setting("N8N_MAX_FILE_BYTES", str(DEFAULT_MAX_FILE_BYTES))) except ValueError as exc: raise ConfigurationError("Некорректные числовые настройки n8n.") from exc if chat_timeout <= 0 or upload_timeout <= 0 or max_file_bytes <= 0: raise ConfigurationError("Таймауты и лимит файла должны быть положительными.") return cls( chat_webhook_url=setting("N8N_CHAT_WEBHOOK_URL", "").strip(), upload_webhook_url=setting("N8N_UPLOAD_WEBHOOK_URL", "").strip(), chat_timeout_seconds=chat_timeout, upload_timeout_seconds=upload_timeout, max_file_bytes=max_file_bytes, ) def require_webhook_url(value: str, label: str) -> str: parsed = urlparse(value) if parsed.scheme not in {"http", "https"} or not parsed.netloc: raise ConfigurationError(f"Не задан корректный полный URL webhook для {label}.") return value