Добавлены все базовые файлы для запуска проекта
This commit is contained in:
@@ -0,0 +1,61 @@
|
||||
"""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
|
||||
Reference in New Issue
Block a user