import logging from dataclasses import dataclass from datetime import datetime, timedelta from uuid import uuid4 import django_rq from django.core.cache import cache from django.tasks import task from django.utils import timezone from games.models import EventGame from submissions.models import ImageCrop, SubmissionFile, OCRStatus from submissions.services.aandt_sync import ( AANDT_STATUS_IN_PROGRESS, sync_event_game_scopes, ) from PIL import Image import pytesseract from rq.exceptions import InvalidJobOperation, NoSuchJobError from rq.job import Job, JobStatus logger = logging.getLogger(__name__) # How often to repoll an EventGame's AAndT while it's "in progress", and while # it isn't yet (registrations, preparation, ...) or the run failed. SCOPE_SYNC_REPOLL_INTERVAL = timedelta(seconds=15) SCOPE_SYNC_IDLE_INTERVAL = timedelta(minutes=1) # AAndT statuses after which there's nothing left to sync, the loop stops. AANDT_FINISHED_STATUSES = {"9done", "11reviewed"} def _scope_sync_guard_key(event_game_id: int) -> str: return f"submissions:scope_sync:{event_game_id}" def _scope_sync_job_key(event_game_id: int) -> str: return f"submissions:scope_sync:{event_game_id}:job" def _scope_sync_state_key(event_game_id: int) -> str: return f"submissions:scope_sync:{event_game_id}:state" def _scope_sync_guard_timeout() -> int: return int(SCOPE_SYNC_IDLE_INTERVAL.total_seconds() * 3) # Last run's outcome, shown to admins in the sync modal. Kept around a while # after the loop stops, so they still see why it stopped. SCOPE_SYNC_STATE_TIMEOUT = int(timedelta(days=1).total_seconds()) @dataclass class ScopeSyncState: running: bool aandt_status: str | None = None last_run_at: datetime | None = None next_run_at: datetime | None = None error: str | None = None def get_scope_sync_state(event_game_id: int, /) -> ScopeSyncState: state = cache.get(_scope_sync_state_key(event_game_id)) or {} running = cache.get(_scope_sync_guard_key(event_game_id)) is not None return ScopeSyncState( running=running, aandt_status=state.get("aandt_status"), last_run_at=state.get("last_run_at"), next_run_at=state.get("next_run_at") if running else None, error=state.get("error"), ) def _enqueue_scope_sync(event_game_id: int, loop_id: str, run_after: datetime | None): task = ( sync_event_game_scopes_task.using(run_after=run_after) if run_after else sync_event_game_scopes_task ) result = task.enqueue(event_game_id, loop_id=loop_id) # Same id as the rq job (see django_tasks_rq), so stop_scope_sync can cancel it cache.set( _scope_sync_job_key(event_game_id), result.id, timeout=_scope_sync_guard_timeout(), ) def schedule_scope_sync(event_game_id: int, /) -> bool: """Start the sync loop of an EventGame, unless one is already running. Safe to call any number of times (admin button spam, multiple workers...). The loop holds a per-EventGame guard with its id (same idea as games.tasks.schedule_steam_refresh), so if the guard expires while an old loop is just late, the old one stops at its next run.""" loop_id = uuid4().hex if not cache.add( _scope_sync_guard_key(event_game_id), loop_id, timeout=_scope_sync_guard_timeout(), ): return False _enqueue_scope_sync(event_game_id, loop_id, run_after=None) return True def stop_scope_sync(event_game_id: int, /) -> bool: """Stop the sync loop of an EventGame: cancels its scheduled rq job and releases the guard (a run already in progress won't reschedule itself). Returns: Whether a loop was running. """ running = cache.get(_scope_sync_guard_key(event_game_id)) is not None job_id = cache.get(_scope_sync_job_key(event_game_id)) cache.delete_many( [_scope_sync_guard_key(event_game_id), _scope_sync_job_key(event_game_id)] ) if job_id: try: job = Job.fetch(job_id, connection=django_rq.get_connection()) if job.get_status() in (JobStatus.SCHEDULED, JobStatus.QUEUED): job.cancel() except (NoSuchJobError, InvalidJobOperation): pass return running def _process_ocr_file(obj, file, crop: ImageCrop | None = None) -> bool: if obj.ocr_status != OCRStatus.TO_PROCESS: return False obj.ocr_status = OCRStatus.PROCESSING obj.save(update_fields=["ocr_status"]) try: image = Image.open(file) if crop: image = image.crop((crop.x1, crop.y1, crop.x2, crop.y2)) obj.ocr_raw_data = pytesseract.image_to_string(image) obj.ocr_status = OCRStatus.PROCESSED except Exception as e: obj.ocr_status = OCRStatus.ERROR raise e finally: obj.save(update_fields=["ocr_status", "ocr_raw_data"]) return True @task def process_submission_files(): """Process uploaded submission files (OCR, validation, etc.)""" for file in SubmissionFile.objects.filter(ocr_status=OCRStatus.TO_PROCESS): process_submission_file.enqueue(str(file.id)) @task def process_submission_file(file_id: str): file = SubmissionFile.objects.filter(id=file_id).first() if not file: return _process_ocr_file(file, file.file) process_crops.enqueue(file_id) @task def process_crops(file_id: str): file = SubmissionFile.objects.filter(id=file_id).first() if not file: return for crop in file.crops.all(): process_crop.enqueue(str(crop.id)) @task def process_crop(crop_id: str): crop = ImageCrop.objects.filter(id=crop_id).first() if not crop: return _process_ocr_file(crop, crop.submission_file.file, crop=crop) @task def sync_all_event_games_task(): """Queue a Scope sync for every EventGame with an AAndT token configured.""" event_games = EventGame.objects.exclude(apibot_token__isnull=True).exclude( apibot_token="" ) for event_game in event_games: schedule_scope_sync(event_game.id) @task def sync_event_game_scopes_task(event_game_id: int, loop_id: str | None = None): """Sync an EventGame's Scopes, then re-enqueue itself until the AAndT is done. Started once (see schedule_scope_sync), it keeps going on its own without an external scheduler: every SCOPE_SYNC_REPOLL_INTERVAL while the AAndT is "in progress" (`7inprog`), every SCOPE_SYNC_IDLE_INTERVAL before that or when the run failed. Stops once the AAndT is done/reviewed, when the EventGame (or its token) is gone, or when another loop took the guard. Without loop_id it's a one-off run that leaves the loop alone. """ guard_key = _scope_sync_guard_key(event_game_id) if loop_id is not None and cache.get(guard_key) != loop_id: # Another loop owns the guard (or it expired/was stopped), stop this one return event_game = EventGame.objects.filter(id=event_game_id).first() if not event_game or not event_game.apibot_token: if loop_id is not None: cache.delete(guard_key) return status = None error = None try: status = sync_event_game_scopes(event_game).status except Exception as exc: if loop_id is None: raise # Retry later rather than letting the loop die (AAndT site down, bug, ...), # it'd otherwise only restart once the guard expires and someone restarts it logger.exception("Scope sync failed for EventGame %s", event_game_id) error = str(exc) or exc.__class__.__name__ if loop_id is None: return now = timezone.now() next_run_at = None # Re-checked: the loop may have been stopped while this run was syncing if status not in AANDT_FINISHED_STATUSES and cache.get(guard_key) == loop_id: next_run_at = now + ( SCOPE_SYNC_REPOLL_INTERVAL if status == AANDT_STATUS_IN_PROGRESS else SCOPE_SYNC_IDLE_INTERVAL ) cache.set( _scope_sync_state_key(event_game_id), { "aandt_status": status, "last_run_at": now, "next_run_at": next_run_at, "error": error, }, timeout=SCOPE_SYNC_STATE_TIMEOUT, ) if next_run_at is None: if cache.get(guard_key) == loop_id: cache.delete(guard_key) return cache.set(guard_key, loop_id, timeout=_scope_sync_guard_timeout()) _enqueue_scope_sync(event_game_id, loop_id, run_after=next_run_at)