278 lines
15 KiB
Python
278 lines
15 KiB
Python
"""Admin-reviewed account links. Live IDs are authoritative; names only suggest candidates."""
|
|
|
|
import asyncio
|
|
import hashlib
|
|
import json
|
|
import re
|
|
import sqlite3
|
|
from collections import defaultdict
|
|
from contextlib import closing
|
|
from datetime import datetime, timezone
|
|
|
|
from fastapi import HTTPException
|
|
|
|
from .. import db
|
|
from ..clients.jellyfin import JellyfinClient
|
|
from ..clients.jellyseerr import JellyseerrClient
|
|
from ..clients.jellystat import JellystatClient
|
|
from ..runtime import get_runtime_settings
|
|
from .jellyfin_identity import source_key
|
|
|
|
MAX_USERS = 3000
|
|
CONFIG_KEYS = ("jellyfin_base_url", "jellyfin_api_key", "jellyseerr_base_url",
|
|
"jellyseerr_api_key", "jellystat_base_url", "jellystat_api_key")
|
|
|
|
|
|
def normalized_id(value):
|
|
value = str(value or "").lower().replace("-", "")
|
|
return value if re.fullmatch(r"[a-f0-9]{32}", value) else None
|
|
|
|
|
|
def name_key(value):
|
|
return str(value or "").strip().casefold()
|
|
|
|
|
|
def digest(value):
|
|
return hashlib.sha256(json.dumps(value, sort_keys=True, separators=(",", ":")).encode()).hexdigest()
|
|
|
|
|
|
def config_digest(runtime):
|
|
return digest([getattr(runtime, key, None) for key in CONFIG_KEYS])
|
|
|
|
|
|
def snapshot(conn):
|
|
conn.row_factory = sqlite3.Row
|
|
return {
|
|
"users": [dict(row) for row in conn.execute(
|
|
"SELECT id, username, role, auth_provider, jellyseerr_user_id FROM users ORDER BY id")],
|
|
"links": [dict(row) for row in conn.execute(
|
|
"SELECT source, local_user_id, jellyfin_user_id FROM jellyfin_user_links ORDER BY source, local_user_id")],
|
|
"confirmations": [dict(row) for row in conn.execute(
|
|
"SELECT * FROM user_identity_confirmations ORDER BY local_user_id")],
|
|
# Detect settings changes between checking services and committing the reviewed links.
|
|
"config_revision": digest([tuple(row) for row in conn.execute(
|
|
"SELECT key, value FROM settings WHERE key IN (" + ",".join("?" for _ in CONFIG_KEYS) + ") ORDER BY key", CONFIG_KEYS)]),
|
|
}
|
|
|
|
|
|
def read_snapshot():
|
|
with closing(db._connect()) as conn:
|
|
return snapshot(conn)
|
|
|
|
|
|
async def jellyfin_directory(runtime):
|
|
client = JellyfinClient(runtime.jellyfin_base_url, runtime.jellyfin_api_key)
|
|
if not client.configured():
|
|
return {"state": "not_configured", "users": []}
|
|
try:
|
|
users, server = await asyncio.gather(client.get_users(), client.get_system_info())
|
|
server_id = normalized_id(server.get("Id")) if isinstance(server, dict) else None
|
|
if not server_id or not isinstance(users, list) or len(users) > MAX_USERS:
|
|
raise ValueError()
|
|
clean = []
|
|
seen = set()
|
|
for user in users:
|
|
user_id = normalized_id(user.get("Id"))
|
|
if not user_id or user_id in seen or normalized_id(user.get("ServerId")) != server_id:
|
|
raise ValueError()
|
|
seen.add(user_id)
|
|
clean.append({"id": user_id, "name": str(user.get("Name") or "")[:200]})
|
|
return {"state": "available", "server_id": server_id, "users": sorted(clean, key=lambda row: row["id"])}
|
|
except Exception:
|
|
return {"state": "unavailable", "users": []}
|
|
|
|
|
|
async def seerr_directory(runtime):
|
|
client = JellyseerrClient(runtime.jellyseerr_base_url, runtime.jellyseerr_api_key)
|
|
if not client.base_url or not client.api_key:
|
|
return {"state": "not_configured", "users": []}
|
|
try:
|
|
users = []
|
|
seen = set()
|
|
expected_total = None
|
|
async with asyncio.timeout(20):
|
|
for skip in range(0, MAX_USERS, 100):
|
|
page = await client.get_users(take=100, skip=skip)
|
|
total = page["pageInfo"]["results"]
|
|
batch = page["results"]
|
|
if type(total) is not int or total < 0 or total > MAX_USERS or not isinstance(batch, list):
|
|
raise ValueError()
|
|
if expected_total is not None and total != expected_total:
|
|
raise ValueError()
|
|
expected_total = total
|
|
for user in batch:
|
|
user_id = user.get("id")
|
|
if type(user_id) is not int or user_id <= 0 or user_id in seen:
|
|
raise ValueError()
|
|
seen.add(user_id)
|
|
users.append({"id": user_id, "name": str(user.get("displayName") or user.get("jellyfinUsername") or "")[:200],
|
|
"jellyfin_id": normalized_id(user.get("jellyfinUserId"))})
|
|
if len(users) == total:
|
|
return {"state": "available", "users": sorted(users, key=lambda row: row["id"])}
|
|
if len(batch) != 100 or len(users) > total:
|
|
raise ValueError()
|
|
except Exception:
|
|
pass
|
|
return {"state": "unavailable", "users": []}
|
|
|
|
|
|
def build_report(local, jellyfin, seerr, jellystat, runtime):
|
|
jf_by_id = {row["id"]: row for row in jellyfin["users"]}
|
|
jf_by_name = defaultdict(list)
|
|
for row in jellyfin["users"]:
|
|
jf_by_name[name_key(row["name"])].append(row["id"])
|
|
seerr_by_id = {row["id"]: row for row in seerr["users"]}
|
|
seerr_by_jf = defaultdict(list)
|
|
for row in seerr["users"]:
|
|
if row["jellyfin_id"]:
|
|
seerr_by_jf[row["jellyfin_id"]].append(row)
|
|
current_source = source_key(runtime.jellyfin_base_url)
|
|
seerr_source = source_key(runtime.jellyseerr_base_url)
|
|
links = {row["local_user_id"]: normalized_id(row["jellyfin_user_id"]) for row in local["links"] if row["source"] == current_source}
|
|
confirmed = {row["local_user_id"]: row for row in local["confirmations"]}
|
|
local_by_name, local_by_seerr = defaultdict(list), defaultdict(list)
|
|
for user in local["users"]:
|
|
local_by_name[name_key(user["username"])].append(user["id"])
|
|
if user["jellyseerr_user_id"] is not None:
|
|
local_by_seerr[user["jellyseerr_user_id"]].append(user["id"])
|
|
rows = []
|
|
for user in local["users"]:
|
|
issues = []
|
|
saved = confirmed.get(user["id"])
|
|
linked = links.get(user["id"])
|
|
stored_seerr = seerr_by_id.get(user["jellyseerr_user_id"])
|
|
by_name = jf_by_name.get(name_key(user["username"]), [])
|
|
basis = "none"
|
|
candidate = None
|
|
if saved:
|
|
candidate = saved["jellyfin_user_id"]
|
|
basis = "confirmed_id"
|
|
if saved["jellyfin_server_id"] != jellyfin.get("server_id") or saved["seerr_source"] != seerr_source:
|
|
issues.append("The confirmed server or Seerr connection has changed.")
|
|
elif linked:
|
|
candidate, basis = linked, "stored_jellyfin_id"
|
|
elif stored_seerr and stored_seerr["jellyfin_id"]:
|
|
candidate, basis = stored_seerr["jellyfin_id"], "stored_seerr_id"
|
|
elif user["auth_provider"] == "jellyfin" and len(by_name) == 1:
|
|
candidate, basis = by_name[0], "suggested_username"
|
|
if len(local_by_name[name_key(user["username"])]) > 1:
|
|
issues.append("Multiple Magent rows share this username after case and whitespace normalization.")
|
|
if len(local_by_seerr.get(user["jellyseerr_user_id"], [])) > 1:
|
|
issues.append("Multiple Magent rows share the stored Seerr ID.")
|
|
if len(by_name) > 1:
|
|
issues.append("This name matches multiple distinct Jellyfin IDs.")
|
|
if candidate and by_name and candidate not in by_name:
|
|
issues.append("The stored ID and current Jellyfin username point to different accounts.")
|
|
if linked and candidate and linked != candidate:
|
|
issues.append("The stored Jellyfin link conflicts with the confirmed identity.")
|
|
jf = jf_by_id.get(candidate)
|
|
if candidate and not jf and jellyfin["state"] == "available":
|
|
issues.append("The linked Jellyfin ID is absent from the current server.")
|
|
expected_seerr = seerr_by_jf.get(candidate, [])
|
|
if len(expected_seerr) > 1:
|
|
issues.append("Multiple Seerr users reference the same Jellyfin ID.")
|
|
if user["jellyseerr_user_id"] is not None and seerr["state"] == "available" and (
|
|
len(expected_seerr) != 1 or expected_seerr[0]["id"] != user["jellyseerr_user_id"]
|
|
):
|
|
issues.append("The stored Seerr ID does not match Seerr's Jellyfin ID mapping.")
|
|
if saved and user["jellyseerr_user_id"] != saved["seerr_user_id"]:
|
|
issues.append("The stored Seerr ID has changed since confirmation.")
|
|
js = jellystat.get(candidate, {"state": "not_checked"})
|
|
rows.append({"user": user, "jellyfin": jf, "candidate_jellyfin_id": candidate,
|
|
"stored_jellyfin_id": linked, "seerr": expected_seerr, "jellystat": js,
|
|
"basis": basis, "issues": issues, "confirmed_at": saved["confirmed_at"] if saved else None,
|
|
"can_confirm": False, "state": "unlinked"})
|
|
candidates = defaultdict(list)
|
|
for row in rows:
|
|
if row["candidate_jellyfin_id"]:
|
|
candidates[row["candidate_jellyfin_id"]].append(row)
|
|
for row in rows:
|
|
candidate = row["candidate_jellyfin_id"]
|
|
if len(candidates.get(candidate, [])) > 1:
|
|
row["issues"].append("Multiple Magent accounts resolve to this Jellyfin ID.")
|
|
# Also protect IDs already reserved by a link/confirmation whose local user was deleted.
|
|
if any(link["local_user_id"] != row["user"]["id"] and link["source"] == current_source
|
|
and normalized_id(link["jellyfin_user_id"]) == candidate for link in local["links"]) or any(
|
|
item["local_user_id"] != row["user"]["id"] and item["jellyfin_server_id"] == jellyfin.get("server_id")
|
|
and item["jellyfin_user_id"] == candidate for item in local["confirmations"]):
|
|
row["issues"].append("This Jellyfin ID is already reserved by another Magent account.")
|
|
if row["issues"]:
|
|
row["state"] = "conflict"
|
|
elif jellyfin["state"] != "available" or seerr["state"] != "available" or (candidate and row["jellystat"]["state"] in {"unavailable", "not_configured"}):
|
|
row["state"] = "unavailable"
|
|
elif not row["jellyfin"] or not row["seerr"] or row["jellystat"]["state"] != "matched":
|
|
row["state"] = "unlinked"
|
|
elif row["confirmed_at"] and row["stored_jellyfin_id"] == candidate:
|
|
row["state"] = "confirmed"
|
|
else:
|
|
row["state"] = "ready"
|
|
row["can_confirm"] = True
|
|
upstream = [{"platform": "Seerr", "id": str(row["id"]), "name": row["name"], "jellyfin_id": row["jellyfin_id"],
|
|
"detail": "No current Jellyfin account has this ID."} for row in seerr["users"]
|
|
if row["jellyfin_id"] not in jf_by_id and jellyfin["state"] == "available"]
|
|
upstream += [{"platform": "Jellyfin", "id": row["id"], "name": row["name"], "jellyfin_id": row["id"],
|
|
"detail": "No Magent account resolves to this ID."} for row in jellyfin["users"] if row["id"] not in candidates]
|
|
services = {"jellyfin": jellyfin["state"], "seerr": seerr["state"],
|
|
"jellystat": "not_configured" if not runtime.jellystat_base_url or not runtime.jellystat_api_key else
|
|
"not_checked" if not jellystat else
|
|
"unavailable" if any(r["state"] == "unavailable" for r in jellystat.values()) else "available"}
|
|
report = {"server_id": jellyfin.get("server_id"), "services": services, "rows": rows, "upstream": upstream,
|
|
"counts": {"magent": len(rows), "jellyfin": len(jellyfin["users"]), "seerr": len(seerr["users"]),
|
|
"jellystat_checked": sum(r["state"] in {"matched", "missing"} for r in jellystat.values()),
|
|
**{state: sum(row["state"] == state for row in rows) for state in ("ready", "confirmed", "conflict", "unlinked", "unavailable")}}}
|
|
report["revision"] = digest([report, digest(local), config_digest(runtime)])
|
|
report["checked_at"] = datetime.now(timezone.utc).isoformat()
|
|
return report
|
|
|
|
|
|
async def review_identities():
|
|
runtime = await asyncio.to_thread(get_runtime_settings)
|
|
local, jf, seerr = await asyncio.gather(asyncio.to_thread(read_snapshot), jellyfin_directory(runtime), seerr_directory(runtime))
|
|
if len(local["users"]) > MAX_USERS:
|
|
raise HTTPException(422, "The identity check supports up to 3,000 Magent accounts.")
|
|
ids = {row["id"] for row in jf["users"]}
|
|
ids.update(row["jellyfin_id"] for row in seerr["users"] if row["jellyfin_id"])
|
|
ids.update(normalized_id(row["jellyfin_user_id"]) for row in local["links"])
|
|
ids.discard(None)
|
|
if len(ids) > MAX_USERS:
|
|
raise HTTPException(422, "There are too many upstream IDs for one identity check.")
|
|
stats_client = JellystatClient(runtime.jellystat_base_url, runtime.jellystat_api_key)
|
|
js = await stats_client.check_user_ids(sorted(ids)) if stats_client.configured() else {key: {"state": "not_configured"} for key in ids}
|
|
return build_report(local, jf, seerr, js, runtime), local, runtime
|
|
|
|
|
|
def save_confirmations(report, local, runtime, user_ids, admin):
|
|
rows = {row["user"]["id"]: row for row in report["rows"]}
|
|
if any(user_id not in rows or not rows[user_id]["can_confirm"] for user_id in user_ids):
|
|
raise HTTPException(409, "Some selected accounts cannot be confirmed. Run the check again and review the conflicts.")
|
|
now = datetime.now(timezone.utc).isoformat()
|
|
try:
|
|
with closing(db._connect()) as conn, conn:
|
|
conn.execute("BEGIN IMMEDIATE")
|
|
if digest(snapshot(conn)) != digest(local) or config_digest(get_runtime_settings()) != config_digest(runtime):
|
|
raise HTTPException(409, "Accounts or settings changed during confirmation. Run the check again.")
|
|
for user_id in user_ids:
|
|
row = rows[user_id]
|
|
jf_id = row["jellyfin"]["id"]
|
|
seerr_id = row["seerr"][0]["id"]
|
|
conn.execute("""INSERT INTO jellyfin_user_links (source, local_user_id, jellyfin_user_id) VALUES (?, ?, ?)
|
|
ON CONFLICT(source,local_user_id) DO UPDATE SET jellyfin_user_id=excluded.jellyfin_user_id""",
|
|
(source_key(runtime.jellyfin_base_url), user_id, jf_id))
|
|
conn.execute("UPDATE users SET jellyseerr_user_id=? WHERE id=?", (seerr_id, user_id))
|
|
conn.execute("""INSERT INTO user_identity_confirmations
|
|
(local_user_id,jellyfin_server_id,jellyfin_user_id,jellyfin_source,seerr_source,seerr_user_id,confirmed_at,confirmed_by)
|
|
VALUES (?,?,?,?,?,?,?,?) ON CONFLICT(local_user_id) DO UPDATE SET
|
|
jellyfin_source=excluded.jellyfin_source,confirmed_at=excluded.confirmed_at,confirmed_by=excluded.confirmed_by""",
|
|
(user_id, report["server_id"], jf_id, source_key(runtime.jellyfin_base_url), source_key(runtime.jellyseerr_base_url),
|
|
seerr_id, now, admin["username"]))
|
|
except sqlite3.IntegrityError as exc:
|
|
raise HTTPException(409, "An identity is already linked to another account. Run the check again.") from exc
|
|
return {"confirmed": len(user_ids), "confirmed_at": now}
|
|
|
|
|
|
async def confirm_identities(revision, user_ids, admin):
|
|
report, local, runtime = await review_identities()
|
|
if report["revision"] != revision:
|
|
raise HTTPException(409, "The identity check has changed. Run it again before confirming accounts.")
|
|
return await asyncio.to_thread(save_confirmations, report, local, runtime, user_ids, admin)
|