import base64
import hashlib

from django.contrib.auth.hashers import BasePasswordHasher, mask_hash from django.utils.crypto import constant_time_compare from django.utils.translation import gettext_noop as _

def _b64url_encode(raw: bytes) -> str: return base64.urlsafe_b64encode(raw).decode("ascii").rstrip("=")

def _b64url_decode(value: str) -> bytes: return base64.urlsafe_b64decode(value + ("=" * (-len(value) % 4)))

class JellyfinPBKDF2SHA512PasswordHasher(BasePasswordHasher): """Compatibility hasher for Jellyfin PBKDF2-SHA512 local passwords.

Imported authentik format:
    jf512$ITERATIONS$SALT_BASE64URL$DIGEST_BASE64URL

must_update() deliberately returns True so a successful login causes
Django/authentik to replace jf512 with the preferred local hasher.
"""

algorithm = "jf512"

def salt(self):
    raise NotImplementedError(
        "jf512 is an import/verification hasher and must not create new salts"
    )

def encode(self, password, salt, iterations=None):
    if password is None:
        raise TypeError("password must not be None")

    iterations = int(iterations or 210000)
    salt_bytes = _b64url_decode(salt)
    digest = hashlib.pbkdf2_hmac(
        "sha512",
        password.encode("utf-8"),
        salt_bytes,
        iterations,
        dklen=64,
    )
    return f"{self.algorithm}${iterations}${salt}${_b64url_encode(digest)}"

def verify(self, password, encoded):
    try:
        algorithm, iterations, salt, _digest = encoded.split("$", 3)
        if algorithm != self.algorithm:
            return False
        encoded_2 = self.encode(password, salt, int(iterations))
    except (TypeError, ValueError):
        return False
    return constant_time_compare(encoded, encoded_2)

def safe_summary(self, encoded):
    algorithm, iterations, salt, digest = encoded.split("$", 3)
    return {
        _("algorithm"): algorithm,
        _("iterations"): iterations,
        _("salt"): mask_hash(salt),
        _("hash"): mask_hash(digest),
    }

def must_update(self, encoded):
    return True</code></pre>

"""Custom authentik settings for the Jellyfin password transition.

Copy to /data/user_settings.py (host path in this guide: /mnt/user/appdata/compose/authentik/data/user_settings.py).

Do not remove the jf512 hasher while any migrated authentik user still has a password whose algorithm is jf512. """

PASSWORD_HASHERS = [ "django.contrib.auth.hashers.PBKDF2PasswordHasher", "data.jellyfin_hashers.JellyfinPBKDF2SHA512PasswordHasher", "django.contrib.auth.hashers.PBKDF2SHA1PasswordHasher", "django.contrib.auth.hashers.Argon2PasswordHasher", "django.contrib.auth.hashers.BCryptSHA256PasswordHasher", "django.contrib.auth.hashers.ScryptPasswordHasher", ]

import json
import sys

from authentik.core.models import Group, User, UserTypes from django.contrib.auth.hashers import identify_hasher

GROUP_NAME = "jellyfin_users" MARKER_KEY = "jellyfin_migration" MARKER_VERSION = 1

def emit(payload): print(json.dumps(payload, ensure_ascii=False, sort_keys=True))

def get_algorithm(user): if not user.password: return "EMPTY" if user.password.startswith("!"): return "UNUSABLE" try: return identify_hasher(user.password).algorithm except Exception: return user.password.split("$", 1)[0]

def exact_and_case_matches(username): exact = User.objects.filter(username=username).first() case_matches = list( User.objects.filter(username__iexact=username).values_list("username", flat=True) ) return exact, case_matches

def inspect_user(username): exact, case_matches = exact_and_case_matches(username) if exact is None: return { "ok": True, "exists": False, "username": username, "case_matches": case_matches, }

return {
    "ok": True,
    "exists": True,
    "username": exact.username,
    "uuid": str(exact.uuid),
    "type": exact.type,
    "is_active": exact.is_active,
    "algorithm": get_algorithm(exact),
    "groups": sorted(g.name for g in exact.groups.all()),
    "attributes": exact.attributes,
    "case_matches": case_matches,
}

def create_user(username, encoded_hash, jellyfin_id): exact, case_matches = exact_and_case_matches(username)

if exact is not None:
    raise RuntimeError(f"User already exists exactly: {username!r}")

if case_matches:
    raise RuntimeError(
        f"Case-insensitive username collision for {username!r}: {case_matches!r}"
    )

group = Group.objects.get(name=GROUP_NAME)
if group.is_superuser:
    raise RuntimeError(f"Refusing to use superuser group {GROUP_NAME!r}")

hasher = identify_hasher(encoded_hash)
if hasher.algorithm != "jf512":
    raise RuntimeError(
        f"Imported password is not recognised as jf512 (got {hasher.algorithm!r})"
    )

marker = {
    MARKER_KEY: {
        "version": MARKER_VERSION,
        "jellyfin_id": jellyfin_id,
    }
}

user = User.objects.create(
    username=username,
    name=username,
    type=UserTypes.INTERNAL.value,
    is_active=True,
    path="users",
    attributes=marker,
)
user.set_password_from_hash(encoded_hash, signal=False)
user.save()
user.groups.add(group)

result = inspect_user(username)
result["created"] = True
return result

def rollback_user(username, jellyfin_id): exact, _case_matches = exact_and_case_matches(username)

if exact is None:
    return {
        "ok": True,
        "deleted": False,
        "reason": "already_absent",
        "username": username,
    }

marker = (exact.attributes or {}).get(MARKER_KEY) or {}
if (
    marker.get("version") != MARKER_VERSION
    or marker.get("jellyfin_id") != jellyfin_id
):
    raise RuntimeError(
        "Refusing rollback: authentik account does not carry the expected "
        "migration marker/Jellyfin ID."
    )

exact.delete()
return {
    "ok": True,
    "deleted": True,
    "username": username,
    "jellyfin_id": jellyfin_id,
}

def main(): request = json.loads(sys.stdin.read()) action = request.get("action") username = request.get("username")

if not isinstance(username, str):
    raise RuntimeError("username must be a string")

if action == "inspect":
    emit(inspect_user(username))
    return

if action == "create":
    encoded_hash = request.get("encoded_hash")
    jellyfin_id = request.get("jellyfin_id")
    if not isinstance(encoded_hash, str) or not encoded_hash:
        raise RuntimeError("encoded_hash missing")
    if not isinstance(jellyfin_id, str) or not jellyfin_id:
        raise RuntimeError("jellyfin_id missing")
    emit(create_user(username, encoded_hash, jellyfin_id))
    return

if action == "rollback":
    jellyfin_id = request.get("jellyfin_id")
    if not isinstance(jellyfin_id, str) or not jellyfin_id:
        raise RuntimeError("jellyfin_id missing")
    emit(rollback_user(username, jellyfin_id))
    return

raise RuntimeError(f"Unknown action: {action!r}")

try: main() except Exception as exc: emit({"ok": False, "error": str(exc)}) raise

#!/usr/bin/env python3
import csv
import datetime as dt
import hashlib
import json
import os
import urllib.request
from pathlib import Path

JF_URL = os.environ.get("JF_URL", "http://127.0.0.1:8096").rstrip("/") JF_API = os.environ.get("JF_API", "") ROOT = Path(file).resolve().parent SNAPSHOTS = ROOT / "snapshots"

Volatile/authentication fields intentionally excluded from the rights-only

fingerprint. Everything else in Policy plus all Configuration is protected.

RIGHTS_POLICY_EXCLUDE = { "AuthenticationProviderId", "PasswordResetProviderId", "InvalidLoginAttemptCount", }

def stamp(): return dt.datetime.now().strftime("%Y%m%d-%H%M%S")

def api_get(path): if not JF_API: raise SystemExit("❌ JF_API n'est pas exportée.")

req = urllib.request.Request(
    f"{JF_URL}{path}",
    headers={
        "Accept": "application/json",
        "Authorization": (
            'MediaBrowser '
            f'Token="{JF_API}", '
            'Client="jellyfin-authentik-audit", '
            'Device="Unraid", '
            'DeviceId="jellyfin-authentik-audit", '
            'Version="1.0"'
        ),
    },
)
with urllib.request.urlopen(req, timeout=30) as response:
    return json.load(response)

def fingerprint(obj): raw = json.dumps( obj, ensure_ascii=False, sort_keys=True, separators=(",", ":"), ).encode("utf-8") return hashlib.sha256(raw).hexdigest()

def rights_fingerprint(user): policy = dict(user.get("Policy") or {}) for key in RIGHTS_POLICY_EXCLUDE: policy.pop(key, None) return fingerprint( { "Policy": policy, "Configuration": user.get("Configuration") or {}, } )

def full_fingerprint(user): return fingerprint( { "Policy": user.get("Policy") or {}, "Configuration": user.get("Configuration") or {}, } )

def main(): out = SNAPSHOTS / stamp() out.mkdir(parents=True, exist_ok=False)

print("=" * 60)
print(" AUDIT JELLYFIN → AUTHENTIK")
print(" MODE : LECTURE SEULE")
print("=" * 60)
print()
print("Serveur :", JF_URL)
print("Dossier :", out.relative_to(ROOT))
print()

print("Lecture des utilisateurs...")
user_index = api_get("/Users")
users = []
for item in user_index:
    user = api_get(f"/Users/{item['Id']}")
    user["RightsFingerprint"] = rights_fingerprint(user)
    user["FullFingerprint"] = full_fingerprint(user)
    users.append(user)

print("Lecture des bibliothèques...")
folders = api_get("/Library/VirtualFolders")

print("Lecture des informations serveur...")
system = api_get("/System/Info")

policy_fields = set()
config_fields = set()
admin_count = 0
disabled_count = 0

# Preserve unresolved IDs rather than cleaning them. VirtualFolder ItemId is
# the best available reference for the library access IDs in the tested setup.
known_folder_ids = {
    f.get("ItemId") for f in folders if isinstance(f, dict) and f.get("ItemId")
}
unresolved_ids = set()

for user in users:
    policy = user.get("Policy") or {}
    config = user.get("Configuration") or {}
    policy_fields.update(policy.keys())
    config_fields.update(config.keys())

    if policy.get("IsAdministrator"):
        admin_count += 1
    if policy.get("IsDisabled"):
        disabled_count += 1

    for field in ("EnabledFolders", "BlockedMediaFolders"):
        for folder_id in policy.get(field) or []:
            if folder_id and folder_id not in known_folder_ids:
                unresolved_ids.add(folder_id)

payload = {
    "created_at": out.name,
    "server": JF_URL,
    "system": system,
    "folders": folders,
    "users": users,
    "summary": {
        "users": len(users),
        "libraries": len(folders),
        "policy_field_count": len(policy_fields),
        "configuration_field_count": len(config_fields),
        "admins": admin_count,
        "disabled": disabled_count,
        "unresolved_library_ids": sorted(unresolved_ids),
    },
}

(out / "jellyfin-audit.json").write_text(
    json.dumps(payload, indent=2, ensure_ascii=False),
    encoding="utf-8",
)
(out / "jellyfin-users-raw.json").write_text(
    json.dumps(users, indent=2, ensure_ascii=False),
    encoding="utf-8",
)
(out / "jellyfin-virtual-folders-raw.json").write_text(
    json.dumps(folders, indent=2, ensure_ascii=False),
    encoding="utf-8",
)
(out / "jellyfin-system-info-raw.json").write_text(
    json.dumps(system, indent=2, ensure_ascii=False),
    encoding="utf-8",
)

csv_fields = [
    "Name", "Id", "IsAdministrator", "IsDisabled",
    "AuthenticationProviderId", "PasswordResetProviderId",
    "RightsFingerprint", "FullFingerprint",
]
with (out / "jellyfin-audit.csv").open("w", encoding="utf-8", newline="") as fh:
    writer = csv.DictWriter(fh, fieldnames=csv_fields)
    writer.writeheader()
    for u in users:
        p = u.get("Policy") or {}
        writer.writerow(
            {
                "Name": u.get("Name"),
                "Id": u.get("Id"),
                "IsAdministrator": p.get("IsAdministrator"),
                "IsDisabled": p.get("IsDisabled"),
                "AuthenticationProviderId": p.get("AuthenticationProviderId"),
                "PasswordResetProviderId": p.get("PasswordResetProviderId"),
                "RightsFingerprint": u.get("RightsFingerprint"),
                "FullFingerprint": u.get("FullFingerprint"),
            }
        )

summary_lines = [
    f"Utilisateurs : {len(users)}",
    f"Bibliothèques : {len(folders)}",
    f"Policy :        {len(policy_fields)} champs distincts",
    f"Configuration: {len(config_fields)} champs distincts",
    f"IDs bibliothèque non résolus : {len(unresolved_ids)}",
    f"Administrateurs : {admin_count}",
    f"Comptes désactivés : {disabled_count}",
]
(out / "summary.txt").write_text("\n".join(summary_lines) + "\n", encoding="utf-8")

print()
print("✅ Audit terminé.")
print()
for line in summary_lines:
    print(line)
print()
print("Fichiers produits :")
for name in (
    "jellyfin-audit.csv", "jellyfin-audit.json", "jellyfin-system-info-raw.json",
    "jellyfin-users-raw.json", "jellyfin-virtual-folders-raw.json", "summary.txt"
):
    print(" -", out.relative_to(ROOT) / name)
print()
print("🔒 AUCUNE ÉCRITURE N'A ÉTÉ EFFECTUÉE SUR JELLYFIN.")

if name == "main": main()

#!/usr/bin/env python3
import json
import os
import re
import sqlite3
from pathlib import Path

ROOT = Path(file).resolve().parent DB_PATH = Path(os.environ.get("JF_DB", "/mnt/user/appdata/Jellyfin/data/jellyfin.db"))

DEFAULT_PROVIDER = "Jellyfin.Server.Implementations.Users.DefaultAuthenticationProvider" LDAP_PROVIDER = os.environ.get( "JF_LDAP_PROVIDER", "Jellyfin.Plugin.LDAP_Auth.LdapAuthenticationProviderPlugin", )

HASH_RE = re.compile( r"^$PBKDF2-SHA512$iterations=(\d+)$([0-9A-Fa-f]+)$([0-9A-Fa-f]+)$" )

def latest_snapshot(): root = ROOT / "snapshots" dirs = sorted(p for p in root.iterdir() if p.is_dir()) if root.exists() else [] if not dirs: raise SystemExit("❌ Aucun snapshot. Lancez d'abord ./jellyfin_audit.py") return dirs[-1]

def inspect_hash(value): if not isinstance(value, str) or not value: return {"format": "MISSING", "migratable": False, "reason": "missing"} m = HASH_RE.match(value) if not m: return {"format": "OTHER", "migratable": False, "reason": "unexpected_format"}

iterations = int(m.group(1))
salt_hex = m.group(2)
digest_hex = m.group(3)

if len(salt_hex) != 32:
    return {
        "format": "PBKDF2-SHA512", "iterations": iterations,
        "migratable": False, "reason": f"salt_bytes={len(salt_hex)//2}",
    }
if len(digest_hex) != 128:
    return {
        "format": "PBKDF2-SHA512", "iterations": iterations,
        "migratable": False, "reason": f"digest_bytes={len(digest_hex)//2}",
    }
if iterations != 210000:
    return {
        "format": "PBKDF2-SHA512", "iterations": iterations,
        "migratable": False,
        "reason": "iteration_count_not_validated_by_this_guide",
    }

return {
    "format": "PBKDF2-SHA512",
    "iterations": iterations,
    "migratable": True,
    "reason": "ok",
}

def main(): snap = latest_snapshot() audit = json.loads((snap / "jellyfin-audit.json").read_text(encoding="utf-8")) api_by_id = {u["Id"]: u for u in audit["users"]}

uri = f"file:{DB_PATH}?mode=ro"
conn = sqlite3.connect(uri, uri=True)
conn.row_factory = sqlite3.Row
try:
    rows = conn.execute(
        """
        SELECT Id, Username, Password, AuthenticationProviderId, PasswordResetProviderId
        FROM Users
        """
    ).fetchall()
finally:
    conn.close()

records = []
formats = {}
ldap_count = 0
default_count = 0
non_migratable = 0
discrepancies = []
db_ids = set()

for row in rows:
    user_id = row["Id"]
    db_ids.add(user_id)
    parsed = inspect_hash(row["Password"])
    formats[parsed["format"]] = formats.get(parsed["format"], 0) + 1
    non_migratable += int(not parsed["migratable"])

    provider = row["AuthenticationProviderId"]
    if provider == LDAP_PROVIDER:
        ldap_count += 1
    elif provider == DEFAULT_PROVIDER:
        default_count += 1

    api_user = api_by_id.get(user_id)
    if api_user is None:
        discrepancies.append({"id": user_id, "username": row["Username"], "reason": "db_only"})
    elif api_user.get("Name") != row["Username"]:
        discrepancies.append(
            {
                "id": user_id,
                "db_username": row["Username"],
                "api_username": api_user.get("Name"),
                "reason": "username_mismatch",
            }
        )

    # Intentionally never save the full password hash.
    records.append(
        {
            "id": user_id,
            "username": row["Username"],
            "authentication_provider": provider,
            "password_reset_provider": row["PasswordResetProviderId"],
            **parsed,
        }
    )

for user_id, api_user in api_by_id.items():
    if user_id not in db_ids:
        discrepancies.append({"id": user_id, "username": api_user.get("Name"), "reason": "api_only"})

admins_default = sum(
    1
    for u in audit["users"]
    if (u.get("Policy") or {}).get("IsAdministrator")
    and (u.get("Policy") or {}).get("AuthenticationProviderId") == DEFAULT_PROVIDER
)

payload = {
    "database": str(DB_PATH),
    "snapshot": snap.name,
    "users_api": len(api_by_id),
    "users_db": len(rows),
    "formats": formats,
    "ldap": ldap_count,
    "default": default_count,
    "admins_still_default": admins_default,
    "non_migratable": non_migratable,
    "api_db_discrepancies": discrepancies,
    "users": records,
}
(snap / "jellyfin-hash-audit.json").write_text(
    json.dumps(payload, indent=2, ensure_ascii=False), encoding="utf-8"
)

lines = [
    f"Utilisateurs API : {len(api_by_id)}",
    f"Utilisateurs DB  : {len(rows)}",
    "",
    "Formats :",
]
for fmt, count in sorted(formats.items()):
    lines.append(f"  {fmt:<28} {count}")
lines += [
    "",
    f"LDAP : {ldap_count}",
    f"Default : {default_count}",
    "",
    f"Admins encore en Default : {admins_default}",
    f"Hashes non migrables : {non_migratable}",
    f"Écarts API/DB : {len(discrepancies)}",
]
(snap / "hash-summary.txt").write_text("\n".join(lines) + "\n", encoding="utf-8")

print("=" * 60)
print(" AUDIT HASHES JELLYFIN")
print(" MODE : LECTURE SEULE")
print("=" * 60)
print()
print("Base :", DB_PATH)
print("Snapshot :", snap.relative_to(ROOT))
print()
for line in lines:
    print(line)
print()
print("✅ Rapport :", snap.relative_to(ROOT) / "hash-summary.txt")
print()
print("🔒 Aucun hash complet sauvegardé.")
print("🔒 Aucune écriture SQLite.")

if name == "main": main()

import datetime as dt
import json
from pathlib import Path

from authentik.core.models import Group, User from django.contrib.auth.hashers import identify_hasher

GROUP_NAME = "jellyfin_users" OUT_ROOT = Path("/data/migration-audit")

def password_algorithm(user): value = user.password or "" if not value: return "EMPTY" if value.startswith("!"): return "UNUSABLE" try: return identify_hasher(value).algorithm except Exception: return value.split("$", 1)[0]

def main(): stamp = dt.datetime.now().strftime("%Y%m%d-%H%M%S") out = OUT_ROOT / stamp out.mkdir(parents=True, exist_ok=False)

users = list(User.objects.all().order_by("username"))
groups = list(Group.objects.all().order_by("name"))
group = Group.objects.filter(name=GROUP_NAME).first()

algorithms = {}
type_counts = {}
records = []

for user in users:
    algo = password_algorithm(user)
    algorithms[algo] = algorithms.get(algo, 0) + 1
    type_counts[user.type] = type_counts.get(user.type, 0) + 1
    direct_groups = sorted(g.name for g in user.groups.all())
    effective_groups = sorted(g.name for g in user.all_groups())
    records.append(
        {
            "username": user.username,
            "uuid": str(user.uuid),
            "type": user.type,
            "is_active": user.is_active,
            "is_superuser": user.is_superuser,
            "algorithm": algo,
            "direct_groups": direct_groups,
            "effective_groups": effective_groups,
            "attributes": user.attributes,
        }
    )

direct_members = []
effective_members = []
parents = []
roles = []

if group is not None:
    direct_members = sorted(
        User.objects.filter(groups=group).values_list("username", flat=True)
    )
    effective_members = sorted(
        user.username
        for user in users
        if group.name in {g.name for g in user.all_groups()}
    )

    parent = getattr(group, "parent", None)
    if parent is not None:
        try:
            parents = [parent.name] if parent else []
        except Exception:
            parents = []

    try:
        roles = sorted(role.name for role in group.roles.all())
    except Exception:
        roles = []

payload = {
    "created_at": stamp,
    "summary": {
        "users": len(users),
        "groups": len(groups),
        "types": type_counts,
        "algorithms": algorithms,
    },
    "group": None if group is None else {
        "name": group.name,
        "is_superuser": group.is_superuser,
        "direct_members": direct_members,
        "effective_members": effective_members,
        "parents": parents,
        "roles": roles,
    },
    "users": records,
}

(out / "authentik-directory-audit.json").write_text(
    json.dumps(payload, indent=2, ensure_ascii=False), encoding="utf-8"
)

print("AUDIT ANNUAIRE AUTHENTIK")
print("=" * 70)
print()
print("MODE : LECTURE SEULE")
print()
print("Utilisateurs :", len(users))
print("Groupes      :", len(groups))
print()
print("TYPES D'UTILISATEURS")
print("-" * 70)
for type_name, count in sorted(type_counts.items()):
    print(f"{count:4d}  {type_name}")

print()
print("GROUPE", GROUP_NAME)
print("-" * 70)
if group is None:
    print("Présent                  : non")
else:
    print("Présent                  : oui")
    print("Superuser                :", group.is_superuser)
    print("Membres directs          :", len(direct_members))
    print("Membres effectifs        :", len(effective_members))
    print("Parents                  :", ", ".join(parents) if parents else "aucun")
    print("Rôles                    :", ", ".join(roles) if roles else "aucun")

print()
print("MEMBRES EFFECTIFS", GROUP_NAME)
print("-" * 70)
for username in effective_members:
    print(" -", username)

print()
print("SUPERUSERS AUTHENTIK")
print("-" * 70)
supers = [u for u in users if u.is_superuser]
if not supers:
    print(" Aucun")
else:
    for user in supers:
        print(" -", user.username)

print()
print("ALGORITHMES DE MOT DE PASSE")
print("-" * 70)
for algo, count in sorted(algorithms.items()):
    print(f"{count:4d}  {algo}")

print()
print("AUCUNE MODIFICATION AUTHENTIK EFFECTUÉE.")
print("AUCUN HASH DE MOT DE PASSE SAUVEGARDÉ.")
print()
print("JSON :", out / "authentik-directory-audit.json")

main()

#!/usr/bin/env python3
import csv
import datetime as dt
import json
import os
from pathlib import Path

ROOT = Path(file).resolve().parent AUTH_AUDIT_ROOT = Path( os.environ.get( "AUTH_AUDIT_ROOT", str(ROOT.parent / "data" / "migration-audit"), ) )

DEFAULT_PROVIDER = "Jellyfin.Server.Implementations.Users.DefaultAuthenticationProvider" LDAP_PROVIDER = os.environ.get( "JF_LDAP_PROVIDER", "Jellyfin.Plugin.LDAP_Auth.LdapAuthenticationProviderPlugin", )

PROTECTED = { value for value in os.environ.get("JF_PROTECTED_LOCAL_ADMINS", "admin").split(",") if value } TEST_ACCOUNTS = { value for value in os.environ.get("JF_TEST_ACCOUNTS", "hashtest").split(",") if value }

PROBLEM_STATUSES = { "CONFLICT_CASE", "LDAP_PASSWORD_STATE_REVIEW", "LDAP_INCONSISTENT", "EXISTING_AUTHENTIK_REVIEW", "HASH_NOT_MIGRATABLE", "UNKNOWN_PROVIDER", }

def latest_dir(root): dirs = sorted(p for p in root.iterdir() if p.is_dir()) if root.exists() else [] if not dirs: raise SystemExit(f"❌ Aucun audit trouvé dans {root}") return dirs[-1]

def main(): jf_dir = latest_dir(ROOT / "snapshots") auth_dir = latest_dir(AUTH_AUDIT_ROOT)

jf = json.loads((jf_dir / "jellyfin-audit.json").read_text(encoding="utf-8"))
hashes = json.loads((jf_dir / "jellyfin-hash-audit.json").read_text(encoding="utf-8"))
auth = json.loads((auth_dir / "authentik-directory-audit.json").read_text(encoding="utf-8"))

hash_by_id = {u["id"]: u for u in hashes["users"]}
auth_by_exact = {u["username"]: u for u in auth["users"]}
auth_casefold = {}
for u in auth["users"]:
    auth_casefold.setdefault(u["username"].casefold(), []).append(u["username"])

rows = []

for user in jf["users"]:
    username = user["Name"]
    user_id = user["Id"]
    policy = user.get("Policy") or {}
    is_admin = bool(policy.get("IsAdministrator"))
    is_disabled = bool(policy.get("IsDisabled"))
    provider = policy.get("AuthenticationProviderId")
    reset_provider = policy.get("PasswordResetProviderId")
    h = hash_by_id.get(user_id) or {}
    auth_user = auth_by_exact.get(username)
    case_matches = [
        name for name in auth_casefold.get(username.casefold(), []) if name != username
    ]
    reasons = []

    if username in PROTECTED:
        status = "PROTECTED_LOCAL_ADMIN"
        reasons.append("Compte explicitement protégé et conservé en authentification locale.")

    elif username in TEST_ACCOUNTS:
        status = "TEST_ACCOUNT"
        reasons.append("Compte de test explicitement exclu des batches de production.")

    elif case_matches and auth_user is None:
        status = "CONFLICT_CASE"
        reasons.append(
            "Un compte Authentik existe avec une casse différente : "
            + ", ".join(repr(x) for x in case_matches)
        )

    elif provider == LDAP_PROVIDER:
        if auth_user is None:
            status = "LDAP_INCONSISTENT"
            reasons.append("Jellyfin est LDAP mais aucun compte Authentik exact n'existe.")
        elif "jellyfin_users" not in auth_user.get("effective_groups", []):
            status = "LDAP_INCONSISTENT"
            reasons.append("Compte Authentik absent du groupe effectif jellyfin_users.")
        elif auth_user.get("algorithm") == "jf512":
            status = "MIGRATED_PENDING_LOGIN"
            reasons.append("Migration technique terminée ; première connexion encore attendue.")
        elif auth_user.get("algorithm") == "pbkdf2_sha256":
            status = "ALREADY_MIGRATED"
            reasons.append("Compte LDAP cohérent et mot de passe déjà re-hashé par Authentik.")
        else:
            status = "LDAP_PASSWORD_STATE_REVIEW"
            reasons.append(
                "Compte LDAP cohérent mais algorithme de mot de passe inattendu : "
                f"{auth_user.get('algorithm')!r}"
            )

    elif provider == DEFAULT_PROVIDER:
        if is_disabled:
            if auth_user is None and h.get("migratable"):
                status = "READY_NEW_DISABLED"
                reasons.append("Compte désactivé, techniquement migrable, laissé hors batch.")
            elif auth_user is not None:
                status = "EXISTING_AUTHENTIK_REVIEW"
                reasons.append("Compte désactivé local mais déjà présent dans Authentik.")
            else:
                status = "HASH_NOT_MIGRATABLE"
                reasons.append("Compte désactivé avec hash non migrable.")
        elif auth_user is not None:
            status = "EXISTING_AUTHENTIK_REVIEW"
            reasons.append(
                "Compte encore local dans Jellyfin mais username exact déjà présent dans Authentik."
            )
        elif not h.get("migratable"):
            status = "HASH_NOT_MIGRATABLE"
            reasons.append(h.get("reason", "hash non migrable"))
        else:
            status = "READY_NEW"
            reasons.append("Compte local Jellyfin, hash compatible et absent d'Authentik.")

    else:
        status = "UNKNOWN_PROVIDER"
        reasons.append(f"Provider Jellyfin inattendu : {provider!r}")

    rows.append(
        {
            "username": username,
            "jellyfin_id": user_id,
            "status": status,
            "reasons": reasons,
            "is_admin": is_admin,
            "is_disabled": is_disabled,
            "authentication_provider": provider,
            "password_reset_provider": reset_provider,
            "hash_migratable": bool(h.get("migratable")),
            "rights_fingerprint": user.get("RightsFingerprint"),
            "full_fingerprint": user.get("FullFingerprint"),
            "authentik_exists": auth_user is not None,
            "authentik_type": None if auth_user is None else auth_user.get("type"),
            "authentik_password_algorithm": None if auth_user is None else auth_user.get("algorithm"),
            "authentik_effective_groups": [] if auth_user is None else auth_user.get("effective_groups", []),
            "authentik_in_jellyfin_users": False if auth_user is None else "jellyfin_users" in auth_user.get("effective_groups", []),
            "case_conflict": case_matches or None,
        }
    )

out_root = ROOT / "cross-audits"
out_root.mkdir(exist_ok=True)
out = out_root / dt.datetime.now().strftime("%Y%m%d-%H%M%S")
suffix = 1
while out.exists():
    out = out_root / f"{out.name}-{suffix:02d}"
    suffix += 1
out.mkdir()

payload = {
    "created_at": out.name,
    "jellyfin_snapshot": jf_dir.name,
    "authentik_snapshot": auth_dir.name,
    "jellyfin_users": len(jf["users"]),
    "authentik_users": len(auth["users"]),
    "users": rows,
}
(out / "cross-audit.json").write_text(
    json.dumps(payload, indent=2, ensure_ascii=False), encoding="utf-8"
)

fieldnames = [
    "username", "jellyfin_id", "status", "is_admin", "is_disabled",
    "authentication_provider", "password_reset_provider", "hash_migratable",
    "rights_fingerprint", "authentik_exists", "authentik_password_algorithm",
    "authentik_in_jellyfin_users", "case_conflict",
]
with (out / "cross-audit.csv").open("w", newline="", encoding="utf-8") as fh:
    writer = csv.DictWriter(fh, fieldnames=fieldnames)
    writer.writeheader()
    for row in rows:
        flat = dict(row)
        flat["case_conflict"] = json.dumps(row["case_conflict"], ensure_ascii=False)
        writer.writerow({k: flat.get(k) for k in fieldnames})

counts = {}
for row in rows:
    counts[row["status"]] = counts.get(row["status"], 0) + 1
problems = [row for row in rows if row["status"] in PROBLEM_STATUSES]

def section(title, selected):
    print()
    print(title)
    print("-" * 70)
    if not selected:
        print(" Aucun")
    else:
        for row in selected:
            print(" -", row["username"])

print("CROISEMENT JELLYFIN ↔ AUTHENTIK")
print("=" * 70)
print()
print("MODE : LECTURE SEULE")
print()
print("Utilisateurs Jellyfin :", len(jf["users"]))
print("Utilisateurs Authentik :", len(auth["users"]))
print()
print("STATUTS")
print("-" * 70)
for status in sorted(counts):
    print(f"{counts[status]:4d}  {status}")

print()
print("COMPTES À EXAMINER")
print("-" * 70)
if not problems:
    print(" Aucun")
else:
    for row in problems:
        print(f" - {row['username']!r}: {row['status']}")
        for reason in row["reasons"]:
            print("   ", reason)

section("COMPTES PROTÉGÉS", [r for r in rows if r["status"] == "PROTECTED_LOCAL_ADMIN"])
section("COMPTES TEST", [r for r in rows if r["status"] == "TEST_ACCOUNT"])
section("DÉJÀ MIGRÉS", [r for r in rows if r["status"] == "ALREADY_MIGRATED"])
section("COMPTES DÉSACTIVÉS MAIS MIGRABLES", [r for r in rows if r["status"] == "READY_NEW_DISABLED"])

jellyfin_names = {r["username"] for r in rows}
extras = [u for u in auth["users"] if u["username"] not in jellyfin_names]
print()
print("COMPTES AUTHENTIK SANS ÉQUIVALENT JELLYFIN")
print("-" * 70)
if not extras:
    print(" Aucun")
else:
    for u in extras:
        print(f" - {u['username']} [{u['type']}]")

print()
print("AUCUNE MODIFICATION EFFECTUÉE.")
print()
print("JSON :", out.relative_to(ROOT) / "cross-audit.json")
print("CSV  :", out.relative_to(ROOT) / "cross-audit.csv")

if name == "main": main()

#!/usr/bin/env python3
import argparse
import base64
import copy
import hashlib
import json
import os
import re
import sqlite3
import subprocess
import urllib.request
from pathlib import Path

ROOT = Path(file).resolve().parent AUTHENTIK_DIR = Path(os.environ.get("AUTHENTIK_DIR", "/mnt/user/appdata/compose/authentik")) DB_PATH = Path(os.environ.get("JF_DB", "/mnt/user/appdata/Jellyfin/data/jellyfin.db")) JF_URL = os.environ.get("JF_URL", "http://127.0.0.1:8096").rstrip("/") JF_API = os.environ.get("JF_API", "")

DEFAULT_PROVIDER = "Jellyfin.Server.Implementations.Users.DefaultAuthenticationProvider" LDAP_PROVIDER = os.environ.get( "JF_LDAP_PROVIDER", "Jellyfin.Plugin.LDAP_Auth.LdapAuthenticationProviderPlugin", ) PROTECTED = { value for value in os.environ.get("JF_PROTECTED_LOCAL_ADMINS", "admin").split(",") if value } TEST_ACCOUNTS = { value for value in os.environ.get("JF_TEST_ACCOUNTS", "hashtest").split(",") if value }

HASH_RE = re.compile( r"^$PBKDF2-SHA512$iterations=(\d+)$([0-9A-Fa-f]+)$([0-9A-Fa-f]+)$" ) RIGHTS_POLICY_EXCLUDE = { "AuthenticationProviderId", "PasswordResetProviderId", "InvalidLoginAttemptCount", }

def latest_dir(root): dirs = sorted(p for p in root.iterdir() if p.is_dir()) if root.exists() else [] if not dirs: raise RuntimeError(f"Aucun audit dans {root}") return dirs[-1]

def api_request(method, path, payload=None): if not JF_API: raise RuntimeError("JF_API n'est pas exportée.")

data = None
headers = {
    "Accept": "application/json",
    "Authorization": (
        'MediaBrowser '
        f'Token="{JF_API}", '
        'Client="jellyfin-authentik-migration", '
        'Device="Unraid", '
        'DeviceId="jellyfin-authentik-migration", '
        'Version="1.0"'
    ),
}
if payload is not None:
    data = json.dumps(payload).encode("utf-8")
    headers["Content-Type"] = "application/json"

req = urllib.request.Request(
    f"{JF_URL}{path}", data=data, headers=headers, method=method
)
with urllib.request.urlopen(req, timeout=30) as response:
    body = response.read()
    return None if not body else json.loads(body)

def canonical_hash(obj): raw = json.dumps( obj, ensure_ascii=False, sort_keys=True, separators=(",", ":") ).encode("utf-8") return hashlib.sha256(raw).hexdigest()

def rights_fingerprint(user): policy = dict(user.get("Policy") or {}) for key in RIGHTS_POLICY_EXCLUDE: policy.pop(key, None) return canonical_hash( {"Policy": policy, "Configuration": user.get("Configuration") or {}} )

def jellyfin_hash_to_jf512(user_id): uri = f"file:{DB_PATH}?mode=ro" conn = sqlite3.connect(uri, uri=True) conn.row_factory = sqlite3.Row try: row = conn.execute("SELECT Password FROM Users WHERE Id = ?", (user_id,)).fetchone() finally: conn.close()

if row is None:
    raise RuntimeError("Utilisateur absent de la base Jellyfin.")

match = HASH_RE.match(row["Password"] or "")
if not match:
    raise RuntimeError("Hash Jellyfin non PBKDF2-SHA512.")

iterations = int(match.group(1))
salt_hex = match.group(2)
digest_hex = match.group(3)

if iterations != 210000:
    raise RuntimeError(
        f"Nombre d'itérations non validé : {iterations}; 210000 attendu par ce guide."
    )
if len(salt_hex) != 32:
    raise RuntimeError("Salt Jellyfin inattendu : 16 octets attendus.")
if len(digest_hex) != 128:
    raise RuntimeError("Digest Jellyfin inattendu : 64 octets attendus.")

salt = base64.urlsafe_b64encode(bytes.fromhex(salt_hex)).decode("ascii").rstrip("=")
digest = base64.urlsafe_b64encode(bytes.fromhex(digest_hex)).decode("ascii").rstrip("=")
return f"jf512${iterations}${salt}${digest}"

def authentik_helper(payload): proc = subprocess.run( [ "docker", "compose", "exec", "-T", "server", "ak", "shell", "-c", 'exec(open("/data/jellyfin_migration_helper.py").read())', ], cwd=AUTHENTIK_DIR, input=json.dumps(payload, ensure_ascii=False) + "\n", text=True, capture_output=True, )

json_lines = [
    line.strip() for line in proc.stdout.splitlines() if line.strip().startswith("{")
]
if not json_lines:
    raise RuntimeError(
        "Aucun JSON retourné par le helper Authentik.\n"
        + proc.stdout[-2000:] + "\n" + proc.stderr[-2000:]
    )

result = json.loads(json_lines[-1])
if proc.returncode != 0 or not result.get("ok"):
    raise RuntimeError(
        "Helper Authentik en échec : " + json.dumps(result, ensure_ascii=False)
    )
return result

def load_reference(username): snap = latest_dir(ROOT / "snapshots") cross = latest_dir(ROOT / "cross-audits")

jf = json.loads((snap / "jellyfin-audit.json").read_text(encoding="utf-8"))
cross_data = json.loads((cross / "cross-audit.json").read_text(encoding="utf-8"))

jf_matches = [u for u in jf["users"] if u.get("Name") == username]
if len(jf_matches) != 1:
    raise RuntimeError("Utilisateur absent du snapshot Jellyfin.")

cross_matches = [u for u in cross_data["users"] if u.get("username") == username]
if len(cross_matches) != 1:
    raise RuntimeError("Utilisateur absent du cross-audit.")

return jf_matches[0], cross_matches[0], snap, cross

def main(): parser = argparse.ArgumentParser() parser.add_argument("--user", required=True, help="Username Jellyfin exact") parser.add_argument("--apply", action="store_true", help="Effectuer réellement la migration") args = parser.parse_args()

username = args.user

try:
    snapshot_user, cross_user, snap, cross = load_reference(username)
except Exception as exc:
    raise SystemExit(f"❌ {exc}")

if username in PROTECTED:
    raise SystemExit("❌ Compte protégé : migration refusée.")
if username in TEST_ACCOUNTS:
    raise SystemExit("❌ Compte de test : migration production refusée.")
if cross_user.get("is_disabled"):
    raise SystemExit("❌ Compte désactivé : migration automatique refusée.")
if cross_user.get("status") != "READY_NEW":
    raise SystemExit(f"❌ Statut cross-audit incompatible : {cross_user.get('status')}")

user_id = snapshot_user["Id"]
original_policy = copy.deepcopy(snapshot_user.get("Policy") or {})
original_configuration = copy.deepcopy(snapshot_user.get("Configuration") or {})
original_rights = snapshot_user.get("RightsFingerprint")

try:
    live = api_request("GET", f"/Users/{user_id}")
except Exception as exc:
    raise SystemExit(f"❌ Impossible de lire Jellyfin live : {exc}")

if live.get("Name") != username:
    raise SystemExit(
        "❌ Drift : username live différent du snapshot : "
        f"{live.get('Name')!r} != {username!r}"
    )
if live.get("Id") != user_id:
    raise SystemExit("❌ Drift : Jellyfin ID différent.")
if (live.get("Policy") or {}) != original_policy:
    raise SystemExit("❌ Drift : Policy Jellyfin modifiée depuis le snapshot.")
if (live.get("Configuration") or {}) != original_configuration:
    raise SystemExit("❌ Drift : Configuration Jellyfin modifiée depuis le snapshot.")

try:
    encoded_hash = jellyfin_hash_to_jf512(user_id)
    auth = authentik_helper({"action": "inspect", "username": username})
except Exception as exc:
    raise SystemExit(f"❌ Précontrôle impossible : {exc}")

if auth.get("exists"):
    raise SystemExit("❌ Utilisateur Authentik déjà présent.")
case_matches = auth.get("case_matches") or []
if case_matches:
    raise SystemExit(f"❌ Collision de casse Authentik : {case_matches!r}")

print()
print("=" * 70)
print(" MIGRATION JELLYFIN → AUTHENTIK")
print("=" * 70)
print()
print("Utilisateur              :", username)
print("Jellyfin ID              :", user_id)
print("Admin                     :", bool(original_policy.get("IsAdministrator")))
print("Désactivé                 :", bool(original_policy.get("IsDisabled")))
print("Provider actuel           :", original_policy.get("AuthenticationProviderId"))
print("Provider cible            :", LDAP_PROVIDER)
print("PasswordResetProvider     :", original_policy.get("PasswordResetProviderId"))
print()
print("Hash Jellyfin             : PBKDF2-SHA512 compatible")
print("Hash Authentik transitoire: jf512")
print("Utilisateur Authentik     : absent")
print()
print("RightsFingerprint         :", original_rights)
print()
print("PasswordResetProviderId NE SERA PAS MODIFIÉ.")

if not args.apply:
    print()
    print("🔒 DRY-RUN : AUCUNE MODIFICATION EFFECTUÉE.")
    print()
    print("Pour migrer réellement cet utilisateur :")
    print(f"./migrate_one.py --user {username!r} --apply")
    return

print()
print("⚠️ MODE APPLY")
auth_created = False
jellyfin_changed = False

try:
    print()
    print("[1/4] Création Authentik...")
    created = authentik_helper(
        {
            "action": "create",
            "username": username,
            "encoded_hash": encoded_hash,
            "jellyfin_id": user_id,
        }
    )
    auth_created = True
    if created.get("algorithm") != "jf512":
        raise RuntimeError("Authentik n'a pas stocké jf512.")
    if "jellyfin_users" not in created.get("groups", []):
        raise RuntimeError("Groupe jellyfin_users absent après création.")
    print("      ✅ utilisateur créé")
    print("      ✅ hash jf512 importé")
    print("      ✅ groupe jellyfin_users")

    print()
    print("[2/4] Vérification Authentik...")
    check = authentik_helper({"action": "inspect", "username": username})
    if not check.get("exists"):
        raise RuntimeError("Utilisateur Authentik absent après création.")
    if check.get("algorithm") != "jf512":
        raise RuntimeError("Algorithm Authentik inattendu.")
    if "jellyfin_users" not in check.get("groups", []):
        raise RuntimeError("jellyfin_users absent.")
    print("      ✅ utilisateur présent")
    print("      ✅ algorithm = jf512")
    print("      ✅ jellyfin_users présent")

    print()
    print("[3/4] Changement AuthenticationProviderId...")
    new_policy = copy.deepcopy(original_policy)
    new_policy["AuthenticationProviderId"] = LDAP_PROVIDER
    # Send the FULL policy. Provider/reset-provider fields are required in
    # current Jellyfin DB paths and must not be accidentally nulled.
    api_request("POST", f"/Users/{user_id}/Policy", new_policy)
    jellyfin_changed = True

    print()
    print("[4/4] Contrôle intégral après migration...")
    after = api_request("GET", f"/Users/{user_id}")
    after_policy = after.get("Policy") or {}
    after_configuration = after.get("Configuration") or {}

    expected_policy = copy.deepcopy(original_policy)
    expected_policy["AuthenticationProviderId"] = LDAP_PROVIDER

    if after_policy != expected_policy:
        changed = sorted(
            key
            for key in set(original_policy) | set(after_policy)
            if original_policy.get(key) != after_policy.get(key)
        )
        raise RuntimeError(
            "Policy finale différente de l'attendu. Champs différents : "
            + ", ".join(changed)
        )
    if after_configuration != original_configuration:
        raise RuntimeError("Configuration Jellyfin modifiée.")
    if rights_fingerprint(after) != original_rights:
        raise RuntimeError("RightsFingerprint différent après migration.")

    print("      ✅ seul AuthenticationProviderId a changé")
    print(f"      ✅ {len(original_policy)} champs Policy protégés")
    print(f"      ✅ {len(original_configuration)} champs Configuration protégés")
    print("      ✅ RightsFingerprint identique")

except Exception as exc:
    print()
    print("🚨 ERREUR DE MIGRATION :", exc)
    print("ROLLBACK AUTOMATIQUE...")
    rollback_errors = []

    if jellyfin_changed:
        try:
            api_request("POST", f"/Users/{user_id}/Policy", original_policy)
            print("      ✅ Policy Jellyfin restaurée")
        except Exception as rb_exc:
            rollback_errors.append(f"Jellyfin: {rb_exc}")

    if auth_created:
        try:
            authentik_helper(
                {"action": "rollback", "username": username, "jellyfin_id": user_id}
            )
            print("      ✅ utilisateur Authentik de migration supprimé")
        except Exception as rb_exc:
            rollback_errors.append(f"Authentik: {rb_exc}")

    if rollback_errors:
        print("🚨 ROLLBACK INCOMPLET :", "; ".join(rollback_errors))
    raise SystemExit(1)

print()
print("=" * 70)
print(" ✅ MIGRATION TECHNIQUE RÉUSSIE")
print("=" * 70)
print()
print("Utilisateur :", username)
print("Jellyfin    : LDAP")
print("Authentik   : jf512")
print()
print("Il reste maintenant à TESTER UNE CONNEXION avec le mot de passe Jellyfin historique.")
print()
print("Après cette première connexion, Authentik doit convertir jf512 → pbkdf2_sha256.")

if name == "main": main()

#!/usr/bin/env python3
import argparse
import json
from pathlib import Path

ROOT = Path(file).resolve().parent

def latest_cross(): root = ROOT / "cross-audits" dirs = sorted(p for p in root.iterdir() if p.is_dir()) if root.exists() else [] if not dirs: raise SystemExit("❌ Aucun cross-audit.") return dirs[-1]

def main(): parser = argparse.ArgumentParser() parser.add_argument("--count", type=int, required=True) parser.add_argument("--output", required=True) args = parser.parse_args()

cross_dir = latest_cross()
data = json.loads((cross_dir / "cross-audit.json").read_text(encoding="utf-8"))
ready = sorted(
    (u["username"] for u in data["users"] if u["status"] == "READY_NEW"),
    key=str.casefold,
)

if len(ready) < args.count:
    raise SystemExit(
        f"❌ Seulement {len(ready)} READY_NEW pour {args.count} demandés."
    )

selected = ready[: args.count]
output = ROOT / args.output
output.write_text("\n".join(selected) + "\n", encoding="utf-8")

print("Cross-audit :", cross_dir.relative_to(ROOT))
print("READY_NEW total :", len(ready))
print("Batch créé :", output.relative_to(ROOT))
print()
for i, username in enumerate(selected, 1):
    print(f"{i:03d}. {username!r}")

if name == "main": main()

#!/usr/bin/env python3
import argparse
import datetime as dt
import subprocess
import sys
from pathlib import Path

ROOT = Path(file).resolve().parent

def main(): parser = argparse.ArgumentParser() parser.add_argument("--file", required=True) parser.add_argument("--count", type=int, required=True) args = parser.parse_args()

batch = Path(args.file)
if not batch.is_absolute():
    batch = ROOT / batch
users = [line for line in batch.read_text(encoding="utf-8").splitlines() if line != ""]

if len(users) != args.count:
    raise SystemExit(f"❌ {args.count} attendus, {len(users)} trouvés")

log_dir = ROOT / "dry-run-logs"
log_dir.mkdir(exist_ok=True)
log = log_dir / f"{batch.stem}-{dt.datetime.now().strftime('%Y%m%d-%H%M%S')}.log"

success = 0
with log.open("w", encoding="utf-8", buffering=1) as fh:
    for i, username in enumerate(users, 1):
        banner = f"\n{'='*70}\nDRY-RUN [{i}/{len(users)}] {username!r}\n{'='*70}\n"
        print(banner, end="", flush=True)
        fh.write(banner)
        fh.flush()

        proc = subprocess.Popen(
            [sys.executable, str(ROOT / "migrate_one.py"), "--user", username],
            cwd=ROOT,
            stdout=subprocess.PIPE,
            stderr=subprocess.STDOUT,
            text=True,
            bufsize=1,
        )
        assert proc.stdout is not None
        for line in proc.stdout:
            print(line, end="", flush=True)
            fh.write(line)
            fh.flush()
        rc = proc.wait()
        if rc != 0:
            print(f"❌ Dry-run échoué pour {username!r}. Arrêt.")
            print("Log :", log.relative_to(ROOT))
            raise SystemExit(1)
        success += 1

print()
print(f"✅ Dry-runs réussis : {success}/{len(users)}")
print("Log :", log.relative_to(ROOT))

if name == "main": main()

#!/usr/bin/env python3
import argparse
import datetime as dt
import re
import subprocess
import sys
from pathlib import Path

ROOT = Path(file).resolve().parent

def safe_filename(value): value = re.sub(r"[^A-Za-z0-9.-]+", "", value) return value[:80] or "user"

def main(): parser = argparse.ArgumentParser() parser.add_argument("--file", required=True) parser.add_argument("--count", required=True, type=int) args = parser.parse_args()

batch_file = Path(args.file)
if not batch_file.is_absolute():
    batch_file = ROOT / batch_file
if not batch_file.exists():
    raise SystemExit(f"❌ Batch absent : {batch_file}")

raw_lines = batch_file.read_text(encoding="utf-8").splitlines()

# CRITICAL: do not strip usernames. A real migration encountered a Jellyfin
# username with a trailing space. Removing it would silently change identity.
users = [line for line in raw_lines if line != ""]

if len(users) != args.count:
    raise SystemExit(
        f"❌ Sécurité : {args.count} comptes attendus, {len(users)} trouvés."
    )
if len(set(users)) != len(users):
    raise SystemExit("❌ Sécurité : usernames dupliqués dans le batch.")

print("BATCH JELLYFIN → AUTHENTIK")
print("=" * 70)
print()
print("Fichier :", batch_file)
print("Comptes :", len(users))
print()
for i, username in enumerate(users, 1):
    print(f"{i:03d}. {username!r}")

print()
confirmation = input(f"Tapez exactement MIGRATE {len(users)} pour continuer : ")
if confirmation != f"MIGRATE {len(users)}":
    raise SystemExit("Annulé.")

stamp = dt.datetime.now().strftime("%Y%m%d-%H%M%S")
log_dir = ROOT / "batch-logs" / stamp
log_dir.mkdir(parents=True, exist_ok=False)

results = []

for index, username in enumerate(users, 1):
    print()
    print("=" * 70)
    print(f"[{index}/{len(users)}] {username!r}")
    print("=" * 70)

    log_path = log_dir / f"{index:03d}-{safe_filename(username)}.log"
    proc = subprocess.Popen(
        [
            sys.executable,
            str(ROOT / "migrate_one.py"),
            "--user",
            username,
            "--apply",
        ],
        cwd=ROOT,
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
        text=True,
        bufsize=1,
    )

    with log_path.open("w", encoding="utf-8", buffering=1) as log:
        assert proc.stdout is not None
        for line in proc.stdout:
            print(line, end="", flush=True)
            log.write(line)
            log.flush()

    rc = proc.wait()
    if rc == 0:
        results.append((username, "MIGRATED", rc))
        print(f"✅ {username!r}: MIGRATED")
    else:
        results.append((username, "FAILED", rc))
        print(f"❌ {username!r}: FAILED")
        break

summary_path = log_dir / "summary.txt"
with summary_path.open("w", encoding="utf-8") as fh:
    fh.write("BATCH JELLYFIN → AUTHENTIK\n")
    fh.write("=" * 70 + "\n\n")
    for username, state, rc in results:
        fh.write(f"{username!r}: {state} (rc={rc})\n")

failed = [r for r in results if r[1] == "FAILED"]
print()
if failed:
    print("⚠️ Batch incomplet.")
    print("Ne pas poursuivre avant audit.")
elif len(results) == len(users):
    print("✅ Batch terminé avec succès.")
else:
    print("⚠️ Batch interrompu avant la fin.")
print()
print("Résumé :", summary_path.relative_to(ROOT))

if failed:
    raise SystemExit(1)

if name == "main": main()