147 lines
3.9 KiB
Python
147 lines
3.9 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import socket
|
|
from dataclasses import dataclass
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
DISCORD_API = "https://discord.com/api/v10"
|
|
DEFAULT_INTERVAL_SECONDS = 60
|
|
DEFAULT_TIMEOUT_SECONDS = 10
|
|
MAX_DISCORD_EMBEDS = 10
|
|
MAX_REQUEST_BYTES = 8_000_000
|
|
SESSION_COOKIE = "archive_bot_session"
|
|
PBKDF2_ITERATIONS = 390_000
|
|
MEDIA_ITEMS_PER_EMBED = 10
|
|
DEFAULT_JELLYFIN_SYNC_INTERVAL_SECONDS = 900
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Service:
|
|
name: str
|
|
group: str
|
|
url: str
|
|
display_url: str
|
|
method: str
|
|
timeout: float
|
|
expected_statuses: set[int]
|
|
expected_min: int
|
|
expected_max: int
|
|
keyword: str | None
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class CheckResult:
|
|
service: Service
|
|
ok: bool
|
|
status: int | None
|
|
latency_ms: int | None
|
|
error: str | None
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class MediaItem:
|
|
title: str
|
|
media_type: str
|
|
source_id: str | None = None
|
|
tmdb_id: str | None = None
|
|
imdb_id: str | None = None
|
|
year: str | None = None
|
|
genres: str | None = None
|
|
rating: str | None = None
|
|
runtime: str | None = None
|
|
summary: str | None = None
|
|
seasons: int | None = None
|
|
episodes: int | None = None
|
|
|
|
|
|
class BotRuntime:
|
|
def __init__(
|
|
self,
|
|
token: str,
|
|
channel_id: str,
|
|
config_path: Path,
|
|
state_path: Path,
|
|
media_state_path: Path,
|
|
media_library_path: Path,
|
|
settings_path: Path,
|
|
dry_run: bool = False,
|
|
) -> None:
|
|
import threading
|
|
|
|
self.token = token
|
|
self.default_channel_id = channel_id
|
|
self.config_path = config_path
|
|
self.state_path = state_path
|
|
self.media_state_path = media_state_path
|
|
self.media_library_path = media_library_path
|
|
self.settings_path = settings_path
|
|
self.dry_run = dry_run
|
|
self.lock = threading.Lock()
|
|
self.last_results: list[CheckResult] = []
|
|
self.last_error: str | None = None
|
|
self.last_message_id: str | None = None
|
|
self.last_checked_at: datetime | None = None
|
|
|
|
|
|
def env(name: str, default: str | None = None) -> str:
|
|
value = os.getenv(name, default)
|
|
if value is None or not value.strip():
|
|
raise SystemExit(f"Missing required environment variable: {name}")
|
|
return value.strip()
|
|
|
|
|
|
def bool_env(name: str, default: bool = False) -> bool:
|
|
raw = os.getenv(name)
|
|
if raw is None:
|
|
return default
|
|
return raw.strip().lower() in {"1", "true", "yes", "on"}
|
|
|
|
|
|
def normalize_discord_token(token: str) -> str:
|
|
cleaned = token.strip().strip("\"'")
|
|
if cleaned.lower().startswith("bot "):
|
|
cleaned = cleaned[4:].strip()
|
|
return cleaned
|
|
|
|
|
|
def load_dotenv(path: Path = Path(".env")) -> None:
|
|
if not path.exists():
|
|
return
|
|
|
|
for line in path.read_text(encoding="utf-8").splitlines():
|
|
stripped = line.strip()
|
|
if not stripped or stripped.startswith("#") or "=" not in stripped:
|
|
continue
|
|
|
|
key, value = stripped.split("=", 1)
|
|
key = key.strip()
|
|
value = value.strip().strip("\"'")
|
|
if key and key not in os.environ:
|
|
os.environ[key] = value
|
|
|
|
|
|
def load_json(path: Path) -> dict[str, Any]:
|
|
try:
|
|
with path.open("r", encoding="utf-8") as handle:
|
|
data = json.load(handle)
|
|
except FileNotFoundError as exc:
|
|
raise ValueError(f"Config file not found: {path}") from exc
|
|
except PermissionError as exc:
|
|
raise ValueError(f"Config file is not readable: {path}") from exc
|
|
except json.JSONDecodeError as exc:
|
|
raise ValueError(f"Invalid JSON in {path}: {exc}") from exc
|
|
|
|
if not isinstance(data, dict):
|
|
raise ValueError(f"Config must be a JSON object: {path}")
|
|
return data
|
|
|
|
|
|
def clean_error(exc: BaseException) -> str:
|
|
reason = getattr(exc, "reason", None)
|
|
if reason:
|
|
return str(reason)[:120]
|
|
return str(exc)[:120] or exc.__class__.__name__ or socket.__name__
|