198 lines
14 KiB
Python
198 lines
14 KiB
Python
"""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), key != 'ignore_profile_limits') 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)
|