276 lines
8.4 KiB
Python
276 lines
8.4 KiB
Python
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)
|