| """In-memory async task manager for captcha solving tasks."""
|
|
|
| from __future__ import annotations
|
|
|
| import asyncio
|
| import logging
|
| import uuid
|
| from dataclasses import dataclass, field
|
| from datetime import datetime, timedelta
|
| from enum import Enum
|
| from typing import Any, Protocol
|
|
|
| log = logging.getLogger(__name__)
|
|
|
|
|
| class TaskStatus(str, Enum):
|
| PROCESSING = "processing"
|
| READY = "ready"
|
| FAILED = "failed"
|
|
|
|
|
| @dataclass
|
| class Task:
|
| id: str
|
| type: str
|
| params: dict[str, Any]
|
| status: TaskStatus = TaskStatus.PROCESSING
|
| solution: dict[str, Any] | None = None
|
| error_code: str | None = None
|
| error_description: str | None = None
|
| created_at: datetime = field(default_factory=datetime.utcnow)
|
|
|
|
|
| class Solver(Protocol):
|
| async def solve(self, params: dict[str, Any]) -> dict[str, Any]: ...
|
|
|
|
|
| class TaskManager:
|
| TASK_TTL = timedelta(minutes=10)
|
|
|
| def __init__(self) -> None:
|
| self._tasks: dict[str, Task] = {}
|
| self._solvers: dict[str, Solver] = {}
|
|
|
| def register_solver(self, task_type: str, solver: Solver) -> None:
|
| self._solvers[task_type] = solver
|
|
|
| def create_task(self, task_type: str, params: dict[str, Any]) -> str:
|
| self._cleanup_expired()
|
| task_id = str(uuid.uuid4())
|
| task = Task(id=task_id, type=task_type, params=params)
|
| self._tasks[task_id] = task
|
| asyncio.create_task(self._process_task(task))
|
| return task_id
|
|
|
| def get_task(self, task_id: str) -> Task | None:
|
| return self._tasks.get(task_id)
|
|
|
| def supported_types(self) -> list[str]:
|
| return list(self._solvers.keys())
|
|
|
| async def _process_task(self, task: Task) -> None:
|
| solver = self._solvers.get(task.type)
|
| if not solver:
|
| task.status = TaskStatus.FAILED
|
| task.error_code = "ERROR_TASK_NOT_SUPPORTED"
|
| task.error_description = f"Task type '{task.type}' is not supported"
|
| return
|
|
|
| try:
|
| solution = await solver.solve(task.params)
|
| task.solution = solution
|
| task.status = TaskStatus.READY
|
| log.info("Task %s completed successfully", task.id)
|
| except Exception as exc:
|
| task.status = TaskStatus.FAILED
|
| task.error_code = "ERROR_CAPTCHA_UNSOLVABLE"
|
| task.error_description = str(exc)
|
| log.error("Task %s failed: %s", task.id, exc)
|
|
|
| def _cleanup_expired(self) -> None:
|
| now = datetime.utcnow()
|
| expired = [
|
| tid
|
| for tid, t in self._tasks.items()
|
| if now - t.created_at > self.TASK_TTL
|
| ]
|
| for tid in expired:
|
| del self._tasks[tid]
|
|
|
|
|
| task_manager = TaskManager()
|
|
|
