"""Read-only administrator views. Callers supply the server OAuth HFAccount. Pages contain 50 rows. User is a case-insensitive substring of the displayed HF username; status is an exact displayed status. Date bounds are inclusive UTC instants (date-only until includes that whole day). Queue and History use submitted time for filtering. Positions precede all filters and count pending work only; maintenance-isolated work has no position. """ from __future__ import annotations from datetime import datetime, timezone import json import math import os from pathlib import Path import re import pandas as pd from src.storage import local from src.submission.details import _public_error from src.submission.hf_identity import HFAccount, require_hf_account from src.submission.links import read_links from src.submission.management_queue import active_requests from src.submission.public_metrics import _is_public_number, public_metric_groups, public_metric_values from src.submission.registry import _read from src.submission.status import display_submission_status, is_terminal_submission from src.submission.summary import render_submission_summary from src.tasks import TASKS, get_task PAGE_SIZE = 50 QUEUE_COLUMNS = ['Global Position', 'HF User', 'Task', 'Method', 'Submission ID', 'Status', 'Submitted At', 'Progress'] HISTORY_COLUMNS = ['HF User', 'Task', 'Method', 'Submission ID', 'Status', 'Submitted At', 'Completed At', 'Scenes Done', 'Scenes Failed', 'Score', 'Metrics', 'Errors'] _STRINGS = ('submission_id', 'task_name', 'hf_username_snapshot', 'display_name', 'submitted_at', 'started_at', 'completed_at') def require_admin(account: HFAccount) -> HFAccount: """Reject non-server account objects and subjects outside the live allowlist.""" require_hf_account(account) subjects = {value.strip() for value in os.environ.get('LEADERBOARD_ADMIN_HF_SUBJECTS', '').split(',') if value.strip()} if account.subject not in subjects: raise ValueError('Administrator access required.') return account def is_admin(account) -> bool: try: require_admin(account) return True except ValueError: return False def _tasks(task_names): if isinstance(task_names, str): task_names = [task_names] names = list(TASKS) if task_names is None else list(task_names) for name in names: get_task(name) return set(names) def _timestamp(value): if not isinstance(value, str) or not value: return None try: timestamp = datetime.fromisoformat(value.replace('Z', '+00:00')) if timestamp.tzinfo is None: timestamp = timestamp.replace(tzinfo=timezone.utc) return timestamp.astimezone(timezone.utc) except ValueError: return None def _filter_text(value): if value is None: return '' if not isinstance(value, str): raise ValueError('Text filters must be strings.') return value.strip() def _page_number(page): if page is None: return 1 if isinstance(page, bool) or not isinstance(page, int) or page < 1: raise ValueError('Page must be a positive integer.') return page def _filters(page, user, status, since, until): _page_number(page) user, status, since, until = map(_filter_text, (user, status, since, until)) bounds = [] for value in (since, until): timestamp = _timestamp(value) if value else None if value and timestamp is None: raise ValueError('Invalid date filter; use an ISO date or timestamp.') bounds.append(timestamp) lower, upper = bounds if upper and re.fullmatch(r'\d{4}-\d\d-\d\d', until): upper = upper.replace(hour=23, minute=59, second=59, microsecond=999999) if lower and upper and lower > upper: raise ValueError('Since must not be later than until.') def accept(record, date_key): timestamp = _timestamp(record.get(date_key)) return (user.casefold() in str(record.get('hf_username_snapshot') or '').casefold() and (not status or (display_submission_status(record) or 'unknown') == status.strip().lower()) and (lower is None or timestamp is not None and timestamp >= lower) and (upper is None or timestamp is not None and timestamp <= upper)) return accept def _hidden(task_names): hidden = {name: local.deleted_submission_ids(name) for name in task_names} requests = active_requests(local.DELETE_PENDING_DIR, local.DELETE_PROCESSING_DIR, Path(local.DELETE_PENDING_DIR).parent / 'complete') for _, request in requests: if request.get('action') not in {'cancel_pending', 'cancel_submission'} and request.get('task_name') in hidden: hidden[request['task_name']].add(request['submission_id']) return hidden def _records(directory): records = {} for filename in local._list_json(directory, strict=True): if not re.fullmatch(r'[0-9a-f]{10}\.json', filename): continue record = local._read_json(os.path.join(directory, filename), strict=True) if record and record.get('submission_id') == filename[:-5]: records[record['submission_id']] = record return records def _safe_record(record): safe = {key: record[key] for key in _STRINGS if isinstance(record.get(key), str)} metadata = read_links(record['task_name'], record['submission_id']) if isinstance(metadata.get('display_name'), str) and metadata['display_name']: safe['display_name'] = metadata['display_name'] safe['status'] = display_submission_status(record) or 'unknown' for key in ('scenes_total', 'scenes_done', 'scenes_failed'): value = record.get(key) if type(value) is int and value >= 0: safe[key] = value if _is_public_number(record.get('score')): safe['score'] = record['score'] return safe def _safe_metrics(record): task = get_task(record['task_name']) metrics = record.get('metrics') if not isinstance(metrics, dict): return {} keys = {key for _, group in public_metric_groups(task) for key in group} sources = set() for key, spec in getattr(task, 'excel_summary_specs', {}).items(): if key in keys and isinstance(spec, dict) and isinstance(spec.get('source'), str): sources.update((spec['source'], spec['source'] + '_std')) safe = {} for key in keys | sources: value = metrics.get(key) if isinstance(value, bool) or not isinstance(value, (int, float, str)): continue if key in sources: try: number = float(value) except (ValueError, OverflowError): continue if math.isfinite(number): safe[key] = number elif _is_public_number(value): safe[key] = value return safe def _metric_payload(record): return {title: dict(values) for title, values in public_metric_values(get_task(record['task_name']), _safe_metrics(record))} def _history_row(record): row = _row(record) row['Metrics'] = json.dumps(_metric_payload(record), ensure_ascii=False) errors = [] if record.get('error_message'): errors.append(_public_error(record['error_message'])) failures = record.get('scene_failures') if isinstance(failures, dict): errors.extend(_public_error(reason) for reason in failures.values()) row['Errors'] = json.dumps(list(dict.fromkeys(errors)), ensure_ascii=False) return row def _row(record): safe = _safe_record(record) return dict(zip(HISTORY_COLUMNS, (safe.get('hf_username_snapshot', ''), get_task(record['task_name']).display_name, safe.get('display_name', ''), safe.get('submission_id'), safe['status'], safe.get('submitted_at'), safe.get('completed_at'), safe.get('scenes_done'), safe.get('scenes_failed'), safe.get('score')))) def _queue_row(record, position): row = _row(record) safe = _safe_record(record) done = safe.get('scenes_done', 0) failed = safe.get('scenes_failed', 0) if record.get('status') in {'running', 'cancelling', 'cancel_requested'}: progress = local.read_progress(record['submission_id'], strict=True) or {} results = progress.get('scene_results') failures = progress.get('scene_failures') if isinstance(results, dict): done = len(results) if isinstance(failures, dict): failed = len(failures) row['Global Position'] = position row['Progress'] = f"{done}/{safe.get('scenes_total') or '?'}" + (f' (+{failed} fail)' if failed else '') return row def _page(rows, page, columns): page = _page_number(page) frame = pd.DataFrame(rows[(page - 1) * PAGE_SIZE:page * PAGE_SIZE], columns=columns) if 'Global Position' in frame: frame['Global Position'] = pd.array(frame['Global Position'], dtype='Int64') frame.attrs.update(page=page, page_size=PAGE_SIZE, total_rows=len(rows)) return frame def _terminal_matches(source, terminal, pending=False): if not terminal or not is_terminal_submission(terminal.get('status')) or not source.get('task_name'): return False if any(source.get(key) != terminal.get(key) for key in ('task_name', 'researcher_identity_id', 'submitted_at')): return False attempt, finished_attempt = source.get('worker_attempt', 0), terminal.get('worker_attempt', 0) return (type(attempt) is int and type(finished_attempt) is int and finished_attempt >= attempt and (pending or finished_attempt > attempt or all(source.get(key) == terminal.get(key) for key in ('worker_id', 'started_at')))) def queue_page(account: HFAccount, task_names=None, page=1, user='', status='', since='', until='') -> pd.DataFrame: """Processing first, then Worker FIFO; pending Position is global, not filtered.""" require_admin(account) tasks = _tasks(task_names) accept = _filters(page, user, status, since, until) pending = _records(local.PENDING_DIR) processing = _records(local.PROCESSING_DIR) complete = _records(local.COMPLETE_DIR) terminal = {sid for sid, record in {**pending, **processing}.items() if is_terminal_submission(record.get('status')) or _terminal_matches(record, complete.get(sid), pending=sid not in processing)} processing = {sid: record for sid, record in processing.items() if sid not in terminal and not is_terminal_submission(record.get('status'))} maintenance = _read().get('orcid_maintenance') or {} def isolated(record): return bool(maintenance and (maintenance.get('phase') == 'applying' or record.get('researcher_identity_id') in (maintenance.get('old_orcid'), maintenance.get('new_orcid')) or record.get('hf_subject') == maintenance.get('hf_subject'))) entries = [] for sid, record in pending.items(): if sid in terminal or sid in processing or is_terminal_submission(record.get('status')): continue timestamp = None if 'evaluator_dispatch_ready' in record: try: timestamp = datetime.strptime(record.get('submitted_at'), '%Y-%m-%dT%H:%M:%SZ').replace(tzinfo=timezone.utc) except (TypeError, ValueError): pass try: order = timestamp.timestamp() if timestamp else os.path.getmtime(os.path.join(local.PENDING_DIR, sid + '.json')) except FileNotFoundError: continue if not math.isfinite(order): raise ValueError('Queue ordering unavailable.') entries.append((order, sid, record)) entries.sort(key=lambda entry: (entry[0], entry[1])) ordered = [(record, None) for sid, record in sorted(processing.items())] position = 0 for _, sid, record in entries: current = None if not isolated(record): position += 1 current = position ordered.append((record, current)) hidden = _hidden(tasks) rows = [_queue_row(record, position) for record, position in ordered if record.get('task_name') in tasks and record['submission_id'] not in hidden[record['task_name']] and accept(record, 'submitted_at')] return _page(rows, page, QUEUE_COLUMNS) def history_page(account: HFAccount, task_names, page=1, user='', status='', since='', until='') -> pd.DataFrame: """Merged terminal History, newest completed first; hidden submissions excluded.""" require_admin(account) tasks = _tasks(task_names) accept = _filters(page, user, status, since, until) hidden = _hidden(tasks) records = {} for task in sorted(tasks): for record in local._raw_task_history(task): sid = record.get('submission_id') if (isinstance(sid, str) and re.fullmatch(r'[0-9a-f]{10}', sid) and record.get('task_name') == task): records[(task, sid)] = record records = [record for (task, sid), record in records.items() if sid not in hidden[task] and is_terminal_submission(record.get('status')) and accept(record, 'submitted_at')] minimum = datetime.min.replace(tzinfo=timezone.utc) records.sort(key=lambda record: (_timestamp(record.get('completed_at')) or minimum, record['submission_id'], record['task_name']), reverse=True) return _page([_history_row(record) for record in records], page, HISTORY_COLUMNS) def admin_details(account: HFAccount, sid, task_names) -> tuple[str, str]: """Return summary HTML and sanitized JSON text only; never write report files.""" require_admin(account) tasks = _tasks(task_names) if not isinstance(sid, str) or not re.fullmatch(r'[0-9a-f]{10}', sid): raise ValueError('Enter a valid 10-character submission ID.') hidden = _hidden(tasks) record = local._read_json(os.path.join(local.COMPLETE_DIR, sid + '.json'), strict=True) for directory in (local.PROCESSING_DIR, local.PENDING_DIR): candidate = local._read_json(os.path.join(directory, sid + '.json'), strict=True) if candidate and not _terminal_matches(candidate, record, pending=directory == local.PENDING_DIR): record = candidate break if not record: for task in sorted(tasks): record = local.find_task_history_record(task, sid) if record: break if (not record or record.get('submission_id') != sid or record.get('task_name') not in tasks or sid in hidden[record['task_name']]): raise ValueError('Submission not found for the selected tasks.') safe = _safe_record(record) task = get_task(record['task_name']) safe['metrics'] = _safe_metrics(record) progress = local.read_progress(sid, strict=True) or {} scene_results = progress.get('scene_results') if safe['status'] in {'running', 'cancelling'} and isinstance(scene_results, dict): safe['scenes_done'] = len(scene_results) failures = progress.get('scene_failures') or record.get('scene_failures') or {} snapshot = {} current = progress.get('current_scene') if isinstance(current, str) and re.fullmatch(r'[A-Za-z0-9_-]{1,160}', current): snapshot['current_scene'] = current retries = progress.get('scene_retries') or {} retry = retries.get(current) if isinstance(retries, dict) else None if isinstance(retry, dict) and retry.get('error'): snapshot['last_sanitized_error'] = _public_error(retry['error']) if record.get('error_message'): snapshot['error_summary'] = _public_error(record['error_message']) if isinstance(failures, dict) and failures: snapshot['scene_errors'] = [{'scene': scene if isinstance(scene, str) and re.fullmatch(r'[A-Za-z0-9_-]{1,160}', scene) else 'scene', 'reason': _public_error(reason)} for scene, reason in failures.items()] safe['scenes_failed'] = len(failures) payload = {key: value for key, value in safe.items() if key != 'metrics'} payload['public_metrics'] = {title: dict(values) for title, values in public_metric_values(task, safe['metrics'])} payload.update(snapshot) return render_submission_summary(safe, {}, snapshot), json.dumps(payload, ensure_ascii=False, indent=2)