139 lines
5.3 KiB
Python
139 lines
5.3 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import urllib.parse
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from .core import BotRuntime
|
|
from .db import load_document, save_document
|
|
|
|
|
|
def load_state(path: Path) -> dict[str, Any]:
|
|
data = load_document(path)
|
|
if data is None:
|
|
return {}
|
|
return data if isinstance(data, dict) else {}
|
|
|
|
|
|
def save_state(path: Path, state: dict[str, Any]) -> None:
|
|
save_document(path, state)
|
|
|
|
|
|
def validate_channel_id(value: str, label: str) -> str:
|
|
channel_id = value.strip()
|
|
if not channel_id:
|
|
raise ValueError(f"{label} channel ID is required")
|
|
if not channel_id.isdigit():
|
|
raise ValueError(f"{label} channel ID must be a Discord numeric channel ID")
|
|
return channel_id
|
|
|
|
|
|
def channel_settings(runtime: BotRuntime) -> dict[str, str]:
|
|
data = load_state(runtime.settings_path)
|
|
status_channel = str(data.get("status_channel_id", "")).strip() or runtime.default_channel_id
|
|
media_channel = str(data.get("media_channel_id", "")).strip() or status_channel
|
|
catalog_url = str(data.get("catalog_url", "")).strip() or os.getenv("PUBLIC_CATALOG_URL", "").strip()
|
|
return {
|
|
"statusChannelId": status_channel,
|
|
"mediaChannelId": media_channel,
|
|
"catalogUrl": catalog_url,
|
|
}
|
|
|
|
|
|
def validate_catalog_url(value: str) -> str:
|
|
catalog_url = value.strip()
|
|
if not catalog_url:
|
|
return ""
|
|
parsed = urllib.parse.urlparse(catalog_url)
|
|
if parsed.scheme not in {"http", "https"} or not parsed.netloc:
|
|
raise ValueError("Catalog URL must be a valid http(s) URL")
|
|
return catalog_url
|
|
|
|
|
|
def save_channel_settings(
|
|
runtime: BotRuntime,
|
|
status_channel_id: str,
|
|
media_channel_id: str,
|
|
catalog_url: str = "",
|
|
) -> dict[str, str]:
|
|
status_channel = validate_channel_id(status_channel_id, "Status")
|
|
media_channel = validate_channel_id(media_channel_id or status_channel, "Media")
|
|
state = load_state(runtime.settings_path)
|
|
state.update(
|
|
{
|
|
"status_channel_id": status_channel,
|
|
"media_channel_id": media_channel,
|
|
"catalog_url": validate_catalog_url(catalog_url),
|
|
"updated_at": datetime.now(timezone.utc).isoformat(),
|
|
}
|
|
)
|
|
save_state(runtime.settings_path, state)
|
|
return channel_settings(runtime)
|
|
|
|
|
|
def save_media_channel_setting(runtime: BotRuntime, media_channel_id: str) -> dict[str, str]:
|
|
media_channel = validate_channel_id(media_channel_id, "Media")
|
|
state = load_state(runtime.settings_path)
|
|
state["media_channel_id"] = media_channel
|
|
state["updated_at"] = datetime.now(timezone.utc).isoformat()
|
|
save_state(runtime.settings_path, state)
|
|
return channel_settings(runtime)
|
|
|
|
|
|
def jellyfin_settings(runtime: BotRuntime) -> dict[str, Any]:
|
|
data = load_state(runtime.settings_path)
|
|
api_key = str(data.get("jellyfin_api_key", "")).strip()
|
|
library_names = data.get("jellyfin_library_names", [])
|
|
if not isinstance(library_names, list):
|
|
library_names = []
|
|
return {
|
|
"url": str(data.get("jellyfin_url", "")).strip(),
|
|
"configured": bool(api_key),
|
|
"libraryNames": [str(name) for name in library_names if str(name).strip()],
|
|
"autoSync": bool(data.get("jellyfin_auto_sync", False)),
|
|
"lastSyncAt": data.get("jellyfin_last_sync_at"),
|
|
"lastSyncError": data.get("jellyfin_last_sync_error"),
|
|
"lastPublishedAt": data.get("jellyfin_last_published_at"),
|
|
"lastFingerprint": data.get("jellyfin_last_fingerprint"),
|
|
"lastPublishedFingerprint": data.get("jellyfin_last_published_fingerprint"),
|
|
"lastChanges": data.get("jellyfin_last_changes"),
|
|
}
|
|
|
|
|
|
def save_jellyfin_settings(runtime: BotRuntime, data: dict[str, Any]) -> dict[str, Any]:
|
|
state = load_state(runtime.settings_path)
|
|
url = str(data.get("url", "")).strip().rstrip("/")
|
|
if url:
|
|
parsed = urllib.parse.urlparse(url)
|
|
if parsed.scheme not in {"http", "https"} or not parsed.netloc:
|
|
raise ValueError("Jellyfin URL must be a valid http(s) URL")
|
|
|
|
api_key = str(data.get("apiKey", "")).strip()
|
|
state["jellyfin_url"] = url
|
|
if api_key:
|
|
state["jellyfin_api_key"] = api_key
|
|
elif data.get("clearApiKey"):
|
|
state["jellyfin_api_key"] = ""
|
|
raw_library_names = data.get("libraryNames", [])
|
|
if isinstance(raw_library_names, str):
|
|
library_names = [name.strip() for name in raw_library_names.split(",") if name.strip()]
|
|
elif isinstance(raw_library_names, list):
|
|
library_names = [str(name).strip() for name in raw_library_names if str(name).strip()]
|
|
else:
|
|
raise ValueError("Jellyfin libraries must be a list or comma-separated string")
|
|
state["jellyfin_library_names"] = library_names
|
|
state["jellyfin_auto_sync"] = bool(data.get("autoSync", False))
|
|
state["updated_at"] = datetime.now(timezone.utc).isoformat()
|
|
save_state(runtime.settings_path, state)
|
|
return jellyfin_settings(runtime)
|
|
|
|
|
|
def save_catalog_url_setting(runtime: BotRuntime, catalog_url: str) -> dict[str, str]:
|
|
state = load_state(runtime.settings_path)
|
|
state["catalog_url"] = validate_catalog_url(catalog_url)
|
|
state["updated_at"] = datetime.now(timezone.utc).isoformat()
|
|
save_state(runtime.settings_path, state)
|
|
return channel_settings(runtime)
|