Spaces:
Running
Running
| """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) | |