"""Reviewed consolidation of accounts sharing a verified Jellyfin ID, entirely within Magent.""" import asyncio import json from contextlib import closing from datetime import datetime, timezone from fastapi import HTTPException from .. import db from ..feature_access import FEATURES from . import identity_review as review from .jellyfin_identity import source_key NAME_REFERENCES = { 'signup_invites': ('created_by',), 'portal_items': ('created_by_username', 'assignee_username'), 'portal_comments': ('author_username',), 'portal_item_activity': ('actor_username',), 'platform_issues': ('reporter_username',), 'platform_issue_events': ('author_username',), 'requests_cache': ('requested_by', 'requested_by_norm'), } def account_state(conn, ids): conn.row_factory = db.sqlite3.Row placeholders = ','.join('?' for _ in ids) return {table: [dict(row) for row in conn.execute( f'SELECT * FROM {table} WHERE {column} IN ({placeholders}) ORDER BY {column}', ids)] for table, column in [('users', 'id'), ('user_feature_permissions', 'user_id'), ('email_recap_subscriptions', 'user_id'), ('newsletter_subscriptions', 'user_id')]} def identity_group(report, target): identity = target['candidate_jellyfin_id'] return [row for row in report['rows'] if identity and row['candidate_jellyfin_id'] == identity] def build_preview(report, local, runtime, state, user_id, keep_id=None): target = next((row for row in report['rows'] if row['user']['id'] == user_id), None) if not target: raise HTTPException(404, 'This Magent account no longer exists. Run the check again.') group = identity_group(report, target) ids = {row['user']['id'] for row in group} if len(ids) < 2: raise HTTPException(409, 'No duplicate identity group remains. Run the account check again.') jf_id = target['candidate_jellyfin_id'] source = source_key(runtime.jellyfin_base_url) owned = {link['local_user_id'] for link in local['links'] if link['source'] == source and review.normalized_id(link['jellyfin_user_id']) == jf_id} recommended = min(ids, key=lambda identity: (identity not in owned, identity)) keep_id = keep_id or recommended if keep_id not in ids: raise HTTPException(400, 'Choose an account from this duplicate group to keep.') problems = [] if any(report['services'].get(service) != 'available' for service in ('jellyfin', 'seerr', 'jellystat')): problems.append('Restore all three media-service connections before consolidating accounts.') if not target['jellyfin'] or target['jellystat']['state'] != 'matched' or len(target['seerr']) != 1: problems.append('One Jellyfin identity and one Seerr account must be verified against Jellystat.') seerr_id = target['seerr'][0]['id'] if len(target['seerr']) == 1 else None for row in group: if row['basis'] not in {'confirmed_id', 'stored_jellyfin_id', 'stored_seerr_id'}: problems.append('Every account needs a stored Jellyfin or Seerr ID; names alone cannot authorize consolidation.') if any('different accounts' in issue or 'multiple distinct Jellyfin' in issue for issue in row['issues']): problems.append('A name and stored identity disagree. Resolve that mapping before consolidation.') if row['user']['role'] != 'user' or row['user']['auth_provider'] not in {'jellyfin', 'jellyseerr'}: problems.append('Only non-admin Jellyfin or Seerr accounts can use duplicate consolidation.') if not jf_id or row['candidate_jellyfin_id'] != jf_id or row['user']['jellyseerr_user_id'] not in (None, seerr_id): problems.append('These rows do not all resolve to the same Jellyfin and Seerr identity.') for link in local['links']: if link['local_user_id'] in ids: if link['source'] != source or review.normalized_id(link['jellyfin_user_id']) != jf_id: problems.append('A duplicate has a different saved Jellyfin identity or server.') elif link['source'] == source and review.normalized_id(link['jellyfin_user_id']) == jf_id: problems.append('Another account or orphaned reservation owns this Jellyfin identity.') for item in local['confirmations']: if item['local_user_id'] in ids: if (item['jellyfin_server_id'] != report['server_id'] or item['jellyfin_user_id'] != jf_id or item['jellyfin_source'] != source or item['seerr_source'] != source_key(runtime.jellyseerr_base_url) or item['seerr_user_id'] != seerr_id): problems.append('A saved confirmation points to a different identity or server.') elif item['jellyfin_server_id'] == report['server_id'] and item['jellyfin_user_id'] == jf_id: problems.append('Another confirmation owns this identity.') if any(row['user']['id'] not in ids and (row['candidate_jellyfin_id'] == jf_id or (seerr_id is not None and row['user']['jellyseerr_user_id'] == seerr_id)) for row in report['rows']): problems.append('An account outside this identity group also claims the identity.') accounts = [account for account in state['users'] if account['id'] in ids] kept = next(account for account in accounts if account['id'] == keep_id) overrides = {(entry['user_id'], entry['feature']): bool(entry['enabled']) for entry in state['user_feature_permissions']} features = {key: all(bool(account['invite_management_enabled']) if key == 'invites' else overrides.get((account['id'], key), True) for account in accounts) for key in FEATURES} expiries = [account['expires_at'] for account in accounts if account['expires_at']] try: expiry = min(expiries, key=lambda value: db._parse_datetime_value(value).timestamp()) if expiries else None except (ValueError, TypeError, AttributeError): expiry = kept['expires_at'] problems.append('An expiry date is invalid. Correct it before repairing duplicates.') proposed = {'id': keep_id, 'username': target['jellyfin']['name'] if target['jellyfin'] else kept['username'], 'email': kept['email'], 'profile_id': kept['profile_id'], 'expires_at': expiry, 'is_blocked': any(account['is_blocked'] for account in accounts), 'auto_search_enabled': all(account['auto_search_enabled'] for account in accounts), 'features': features, 'jellyfin_user_id': jf_id, 'seerr_user_id': seerr_id} public = [{key: account.get(key) for key in ('id', 'username', 'email', 'profile_id', 'last_login_at', 'created_at')} for account in accounts] return {'accounts': public, 'keep_id': keep_id, 'recommended_id': recommended, 'proposed': proposed, 'issues': sorted(set(problems)), 'can_confirm': not problems, 'revision': review.digest([report['revision'], state, keep_id, proposed])} async def prepare(user_id, keep_id=None): report, local, runtime = await review.review_identities() target = next((row for row in local['users'] if row['id'] == user_id), None) if not target: raise HTTPException(404, 'Account not found.') report_target = next(row for row in report['rows'] if row['user']['id'] == user_id) ids = sorted(row['user']['id'] for row in identity_group(report, report_target)) with closing(db._connect()) as conn: conn.execute('BEGIN') if review.digest(review.snapshot(conn)) != review.digest(local): raise HTTPException(409, 'Accounts changed during the check. Preview again.') state = account_state(conn, ids) return build_preview(report, local, runtime, state, user_id, keep_id), report, local, runtime, state def consolidate(preview, report, local, runtime, state, admin): if not preview['can_confirm']: raise HTTPException(409, 'This duplicate group cannot be consolidated. Review the listed conflicts.') ids = sorted(account['id'] for account in state['users']) keep = preview['keep_id'] removed = [identity for identity in ids if identity != keep] values = preview['proposed'] now = datetime.now(timezone.utc).isoformat() with closing(db._connect()) as conn, conn: conn.execute('BEGIN IMMEDIATE') if (review.digest(review.snapshot(conn)) != review.digest(local) or review.digest(account_state(conn, ids)) != review.digest(state) or review.config_digest(review.get_runtime_settings()) != review.config_digest(runtime)): raise HTTPException(409, 'Accounts, permissions or subscriptions changed. Preview again before saving.') for table in ('email_recap_deliveries', 'newsletter_deliveries'): if conn.execute(f"SELECT 1 FROM {table} WHERE user_id IN ({','.join('?' for _ in ids)}) AND state='sending'", ids).fetchone(): raise HTTPException(409, 'An account email is currently being sent. Wait for delivery to finish, then preview again.') archive = {**state, 'links': [entry for entry in local['links'] if entry['local_user_id'] in ids], 'confirmations': [entry for entry in local['confirmations'] if entry['local_user_id'] in ids], 'proposed': values} conn.execute('INSERT INTO user_duplicate_repairs(kept_user_id,archive_json,repaired_by,repaired_at) VALUES(?,?,?,?)', (keep, json.dumps(archive, sort_keys=True), admin['username'], now)) names = {account['username'] for account in state['users']} tables = {row[0] for row in conn.execute("SELECT name FROM sqlite_master WHERE type='table'")} for table, columns in NAME_REFERENCES.items(): if table not in tables: continue for column in columns: old_values = {review.name_key(name) for name in names} if column == 'requested_by_norm' else names for name in old_values: new_value = review.name_key(values['username']) if column == 'requested_by_norm' else values['username'] conn.execute(f'UPDATE {table} SET {column}=? WHERE {column}=? COLLATE BINARY', (new_value, name)) activity = [dict(row) for row in conn.execute('SELECT * FROM user_activity') if row['username'] in names] for entry in activity: conn.execute('DELETE FROM user_activity WHERE id=?', (entry['id'],)) for entry in activity: conn.execute('''INSERT INTO user_activity(username,ip,user_agent,first_seen_at,last_seen_at,hit_count) VALUES(?,?,?,?,?,?) ON CONFLICT(username,ip,user_agent) DO UPDATE SET first_seen_at=MIN(first_seen_at,excluded.first_seen_at),last_seen_at=MAX(last_seen_at,excluded.last_seen_at), hit_count=hit_count+excluded.hit_count''', (values['username'], entry['ip'], entry['user_agent'], entry['first_seen_at'], entry['last_seen_at'], entry['hit_count'])) for name in names: conn.execute('DELETE FROM password_reset_tokens WHERE username=? COLLATE NOCASE', (name,)) for identity in removed: # Duplicate subscriptions are not inherited. Preserve delivery history and cancel outstanding work. for table in ('email_recap_deliveries', 'newsletter_deliveries'): conn.execute(f"UPDATE {table} SET state='cancelled',detail='Duplicate account consolidated.' WHERE user_id=? AND state IN ('queued','retry','preparing')", (identity,)) conn.execute(f'UPDATE {table} SET user_id=? WHERE user_id=?', (keep, identity)) conn.execute('DELETE FROM jellyfin_user_links WHERE local_user_id=?', (identity,)) conn.execute('DELETE FROM user_identity_confirmations WHERE local_user_id=?', (identity,)) conn.execute('DELETE FROM users WHERE id=?', (identity,)) last_login = max((account['last_login_at'] for account in state['users'] if account['last_login_at']), default=None) conn.execute('''UPDATE users SET auth_provider='jellyfin',username=?,jellyseerr_user_id=?,is_blocked=?,auto_search_enabled=?, invite_management_enabled=?,expires_at=?,last_login_at=? WHERE id=?''', (values['username'], values['seerr_user_id'], values['is_blocked'], values['auto_search_enabled'], values['features']['invites'], values['expires_at'], last_login, keep)) for feature, enabled in values['features'].items(): if feature != 'invites': conn.execute('''INSERT INTO user_feature_permissions VALUES(?,?,?) ON CONFLICT(user_id,feature) DO UPDATE SET enabled=excluded.enabled''', (keep, feature, int(enabled))) conn.execute('''INSERT INTO jellyfin_user_links VALUES(?,?,?) ON CONFLICT(source,local_user_id) DO UPDATE SET jellyfin_user_id=excluded.jellyfin_user_id''', (source_key(runtime.jellyfin_base_url), keep, values['jellyfin_user_id'])) conn.execute('DELETE FROM user_identity_confirmations WHERE local_user_id=?', (keep,)) conn.execute('''INSERT INTO user_identity_confirmations VALUES(?,?,?,?,?,?,?,?)''', (keep, report['server_id'], values['jellyfin_user_id'], source_key(runtime.jellyfin_base_url), source_key(runtime.jellyseerr_base_url), values['seerr_user_id'], now, admin['username'])) return {'kept_user_id': keep, 'consolidated': len(removed), 'repaired_at': now} async def repair_duplicates(user_id, keep_id=None, revision=None, admin=None): preview, report, local, runtime, state = await prepare(user_id, keep_id) if revision is None: return preview if revision != preview['revision']: raise HTTPException(409, 'The duplicate-account preview changed. Preview again before saving.') return await asyncio.to_thread(consolidate, preview, report, local, runtime, state, admin)