mobile.de/mobilede_scraper/api/routes/tasks.py

# Роуты запуска задач и истории sync-run.

import json

from fastapi import APIRouter, Depends, Query
from pydantic import BaseModel
from redis import Redis
from sqlalchemy import select, func

from ...core.config import Settings
from ..deps import get_persistence
from ...storage.db import PersistenceService
from ...storage.models import SyncRun
from ...worker.celery_app import celery_app
from ...worker.constants import MOBILEDE_SYNC_QUEUE
from ...worker.tasks import (
    mobilede_enrich_images_batch_task,
    mobilede_sync_detail_task,
    mobilede_sync_runtime_segments_task,
    mobilede_sync_search_task,
)

router = APIRouter()


class MobileDeSyncSearchRequest(BaseModel):
    start_page: int = 1
    max_pages: int = 5
    lane: str = "mobile_de_cars"
    search_url: str | None = None
    make_id: str | None = None
    model_id: str | None = None
    price_min: str | None = None
    price_max: str | None = None
    year_min: str | None = None
    year_max: str | None = None
    delay_seconds: float = 0.7
    use_cursor: bool = False
    continuous: bool | None = False


class MobileDeSyncDetailRequest(BaseModel):
    listing_id: str
    lane: str = "mobile_de_cars"


class MobileDeEnrichImagesRequest(BaseModel):
    limit: int = 50
    lane: str = "mobile_de_cars"
    max_existing_images: int = 1
    delay_seconds: float = 1.0


class MobileDeRuntimeSegmentsRequest(BaseModel):
    lane: str = "mobile_de_cars"
    delay_seconds: float = 0.7
    use_cursor: bool = True
    continuous: bool | None = False


@router.post("/mobilede/tasks/sync-search")
def start_mobilede_sync_search(body: MobileDeSyncSearchRequest):
    result = mobilede_sync_search_task.apply_async(
        kwargs=body.model_dump(),
        queue=MOBILEDE_SYNC_QUEUE,
    )
    return {
        "task_id": result.id,
        "status": "queued",
        "queue": MOBILEDE_SYNC_QUEUE,
    }


@router.post("/mobilede/tasks/sync-detail")
def start_mobilede_sync_detail(body: MobileDeSyncDetailRequest):
    result = mobilede_sync_detail_task.apply_async(
        kwargs=body.model_dump(),
        queue=MOBILEDE_SYNC_QUEUE,
    )
    return {
        "task_id": result.id,
        "status": "queued",
        "queue": MOBILEDE_SYNC_QUEUE,
        "listing_id": body.listing_id,
    }


@router.post("/mobilede/tasks/enrich-images")
def start_mobilede_enrich_images(body: MobileDeEnrichImagesRequest):
    result = mobilede_enrich_images_batch_task.apply_async(
        kwargs=body.model_dump(),
        queue=MOBILEDE_SYNC_QUEUE,
    )
    return {
        "task_id": result.id,
        "status": "queued",
        "queue": MOBILEDE_SYNC_QUEUE,
    }


@router.post("/mobilede/tasks/sync-runtime-segments")
def start_mobilede_runtime_segments(body: MobileDeRuntimeSegmentsRequest):
    result = mobilede_sync_runtime_segments_task.apply_async(
        kwargs=body.model_dump(),
        queue=MOBILEDE_SYNC_QUEUE,
    )
    return {
        "task_id": result.id,
        "status": "queued",
        "queue": MOBILEDE_SYNC_QUEUE,
    }


@router.get("/tasks/{task_id}")
def get_task_status(task_id: str):
    result = celery_app.AsyncResult(task_id)

    payload: dict = {
        "task_id": task_id,
        "state": result.state,
    }

    if result.successful():
        payload["result"] = result.result
    elif result.failed():
        payload["error"] = str(result.result)
    elif result.info is not None:
        payload["meta"] = result.info

    progress = _read_task_progress(task_id)
    if progress is not None:
        payload["progress"] = progress

    return payload


def _read_task_progress(task_id: str) -> dict | None:
    redis_client = None
    try:
        settings = Settings()
        redis_client = Redis.from_url(
            settings.redis.url,
            decode_responses=True,
            socket_connect_timeout=settings.redis.socket_connect_timeout_seconds,
            socket_timeout=settings.redis.socket_timeout_seconds,
            health_check_interval=settings.redis.health_check_interval_seconds,
            retry_on_timeout=True,
        )
        raw = redis_client.get(f"mobilede:state:task_progress:{task_id}")
        if not raw:
            return None
        data = json.loads(raw)
        return data if isinstance(data, dict) else None
    except Exception:
        return None
    finally:
        if redis_client is not None:
            try:
                redis_client.close()
            except Exception:
                pass


@router.get("/sync-runs")
def list_sync_runs(
    page: int = Query(1, ge=1),
    per_page: int = Query(20, ge=1, le=100),
    persistence: PersistenceService = Depends(get_persistence),
):
    # История запусков.
    with persistence.session_scope() as session:
        total = session.execute(select(func.count(SyncRun.id))).scalar() or 0
        offset = (page - 1) * per_page
        runs = session.execute(
            select(SyncRun).order_by(SyncRun.started_at.desc()).offset(offset).limit(per_page)
        ).scalars().all()

        return {
            "total": total,
            "page": page,
            "per_page": per_page,
            "items": [
                {
                    "id": r.id,
                    "started_at": r.started_at.isoformat() if r.started_at else None,
                    "finished_at": r.finished_at.isoformat() if r.finished_at else None,
                    "status": r.status,
                    "lane": r.lane,
                    "ids_fetched": r.ids_fetched,
                    "cars_upserted": r.cars_upserted,
                    "cars_failed": r.cars_failed,
                    "images_upserted": r.images_upserted,
                    "error_summary": r.error_summary,
                }
                for r in runs
            ],
        }