121 lines
3.5 KiB
Python
121 lines
3.5 KiB
Python
# Роуты запуска задач синхронизации и просмотра истории sync-runs.
|
|
|
|
from fastapi import APIRouter, Depends, Query
|
|
from pydantic import BaseModel
|
|
from sqlalchemy import select, func
|
|
|
|
from ..deps import get_persistence
|
|
from ...storage.db import PersistenceService
|
|
from ...storage.models import SyncRun
|
|
from ...worker.celery_app import IAAI_SYNC_QUEUE, celery_app
|
|
from ...worker.tasks import sync_vehicle_task, sync_listing_task
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
class SyncVehicleRequest(BaseModel):
|
|
vehicle_url: str
|
|
lane: str = "iaai"
|
|
|
|
|
|
class SyncListingRequest(BaseModel):
|
|
make: str | None = None
|
|
model: str | None = None
|
|
lane: str = "iaai_cars"
|
|
limit: int | None = None
|
|
only_new: bool | None = None
|
|
|
|
|
|
@router.post("/tasks/sync-vehicle")
|
|
def start_sync_vehicle(
|
|
body: SyncVehicleRequest,
|
|
):
|
|
# Запустить задачу скрапинга одного автомобиля через Celery
|
|
result = sync_vehicle_task.apply_async(
|
|
kwargs={"vehicle_url": body.vehicle_url, "lane": body.lane},
|
|
queue=IAAI_SYNC_QUEUE,
|
|
)
|
|
|
|
return {
|
|
"task_id": result.id,
|
|
"status": "queued",
|
|
"vehicle_url": body.vehicle_url,
|
|
}
|
|
|
|
|
|
@router.post("/tasks/sync-listing")
|
|
def start_sync_listing(
|
|
body: SyncListingRequest,
|
|
):
|
|
# Запустить задачу полного цикла листинга через Celery
|
|
result = sync_listing_task.apply_async(
|
|
kwargs={
|
|
"make": body.make,
|
|
"model": body.model,
|
|
"lane": body.lane,
|
|
"limit": body.limit,
|
|
"only_new": body.only_new,
|
|
},
|
|
queue=IAAI_SYNC_QUEUE,
|
|
)
|
|
|
|
return {
|
|
"task_id": result.id,
|
|
"status": "queued",
|
|
}
|
|
|
|
|
|
@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
|
|
|
|
return payload
|
|
|
|
|
|
@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
|
|
],
|
|
}
|