62 lines
2.2 KiB
Python
62 lines
2.2 KiB
Python
"""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
|