Roman
9148b5fb6a
- Détection de version via l'API Gitea, application via Watchtower - Router système, UI admin (intégrations), bandeau de mise à jour - Dockerfile multi-étapes, docker-compose, scripts/release.sh - Version centralisée dans app/version.py (pyproject ou OHM_VERSION au build)
137 lines
4.6 KiB
Python
137 lines
4.6 KiB
Python
"""Paramètres persistés en DB (activation des sources, santé, réglages UI…)."""
|
|
import json
|
|
import logging
|
|
import secrets
|
|
from datetime import UTC, datetime
|
|
|
|
from app.db import db
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
async def get_setting(key: str, default: object = None) -> object:
|
|
row = await db.fetchone("SELECT value FROM settings WHERE key = ?", (key,))
|
|
if row is None:
|
|
return default
|
|
try:
|
|
return json.loads(row["value"])
|
|
except json.JSONDecodeError:
|
|
logger.warning("Paramètre %r corrompu, valeur par défaut utilisée", key)
|
|
return default
|
|
|
|
|
|
async def set_setting(key: str, value: object) -> None:
|
|
await db.execute(
|
|
"INSERT INTO settings (key, value) VALUES (?, ?) "
|
|
"ON CONFLICT(key) DO UPDATE SET value = excluded.value",
|
|
(key, json.dumps(value, ensure_ascii=False)),
|
|
)
|
|
|
|
|
|
async def is_source_enabled(name: str) -> bool:
|
|
return bool(await get_setting(f"source:{name}:enabled", True))
|
|
|
|
|
|
async def set_source_enabled(name: str, enabled: bool) -> None:
|
|
await set_setting(f"source:{name}:enabled", enabled)
|
|
logger.info("Source %s %s", name, "activée" if enabled else "désactivée")
|
|
|
|
|
|
async def get_source_health(name: str) -> dict | None:
|
|
"""Dernier état de santé connu d'une source (None si jamais testée)."""
|
|
value = await get_setting(f"source:{name}:health")
|
|
return value if isinstance(value, dict) else None
|
|
|
|
|
|
async def set_source_health(name: str, healthy: bool, detail: str) -> dict:
|
|
state = {
|
|
"healthy": healthy,
|
|
"detail": detail,
|
|
"checked_at": datetime.now(UTC).isoformat(),
|
|
}
|
|
await set_setting(f"source:{name}:health", state)
|
|
return state
|
|
|
|
|
|
async def get_source_base_url(name: str) -> str | None:
|
|
"""URL personnalisée d'une source (None = valeur par défaut du code)."""
|
|
value = await get_setting(f"source:{name}:base_url")
|
|
return value if isinstance(value, str) and value else None
|
|
|
|
|
|
async def set_source_base_url(name: str, url: str | None) -> None:
|
|
"""Persiste l'URL personnalisée (None/'' → retour à la valeur par défaut)."""
|
|
key = f"source:{name}:base_url"
|
|
if url:
|
|
await set_setting(key, url)
|
|
else:
|
|
await db.execute("DELETE FROM settings WHERE key = ?", (key,))
|
|
|
|
|
|
async def apply_source_base_urls() -> None:
|
|
"""Applique les URL personnalisées aux instances de sources (au démarrage)."""
|
|
from app.scrapers.base import all_sources
|
|
|
|
for source in all_sources():
|
|
override = await get_source_base_url(source.name)
|
|
if override and override != type(source).base_url:
|
|
source.base_url = override
|
|
logger.info("URL personnalisée pour %s : %s", source.name, override)
|
|
|
|
# ---------------------------------------------------------------- intégrations *arr
|
|
|
|
TORZNAB_APIKEY_KEY = "torznab:apikey"
|
|
SONARR_URL_KEY = "sonarr:url"
|
|
SONARR_APIKEY_KEY = "sonarr:apikey"
|
|
|
|
|
|
async def get_torznab_apikey() -> str:
|
|
"""Clé API Torznab (générée au premier appel, persistée en DB)."""
|
|
key = await get_setting(TORZNAB_APIKEY_KEY)
|
|
if not isinstance(key, str) or not key:
|
|
key = secrets.token_hex(16)
|
|
await set_setting(TORZNAB_APIKEY_KEY, key)
|
|
logger.info("Clé API Torznab générée")
|
|
return key
|
|
|
|
|
|
async def reset_torznab_apikey() -> str:
|
|
key = secrets.token_hex(16)
|
|
await set_setting(TORZNAB_APIKEY_KEY, key)
|
|
logger.info("Clé API Torznab régénérée")
|
|
return key
|
|
|
|
|
|
async def get_sonarr_config() -> dict[str, str]:
|
|
return {
|
|
"url": await get_setting(SONARR_URL_KEY, ""),
|
|
"apikey": await get_setting(SONARR_APIKEY_KEY, ""),
|
|
}
|
|
|
|
|
|
async def set_sonarr_config(url: str, apikey: str) -> None:
|
|
await set_setting(SONARR_URL_KEY, url.rstrip("/"))
|
|
await set_setting(SONARR_APIKEY_KEY, apikey.strip())
|
|
logger.info("Configuration Sonarr enregistrée (%s)", url)
|
|
|
|
# ---------------------------------------------------------------- mise à jour logicielle
|
|
|
|
UPDATE_GITEA_URL_KEY = "update:gitea_url"
|
|
UPDATE_REPO_KEY = "update:repo"
|
|
UPDATE_TOKEN_KEY = "update:token"
|
|
|
|
|
|
async def get_update_config() -> dict[str, str]:
|
|
return {
|
|
"gitea_url": await get_setting(UPDATE_GITEA_URL_KEY, "https://git.lanro.eu"),
|
|
"repo": await get_setting(UPDATE_REPO_KEY, "Roman/ohm_streaming"),
|
|
"token": await get_setting(UPDATE_TOKEN_KEY, ""),
|
|
}
|
|
|
|
|
|
async def set_update_config(gitea_url: str, repo: str, token: str) -> None:
|
|
await set_setting(UPDATE_GITEA_URL_KEY, gitea_url.rstrip("/"))
|
|
await set_setting(UPDATE_REPO_KEY, repo.strip().strip("/"))
|
|
await set_setting(UPDATE_TOKEN_KEY, token.strip())
|
|
logger.info("Configuration de mise à jour enregistrée (%s)", repo)
|