wren
49 lines
1.5 KiB
Python
49 lines
1.5 KiB
Python
"""页级图片质量任务入口。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
from typing import Any
|
|
|
|
from fastapi_common.fastapi_common_logger import logger
|
|
|
|
from fastapi_admin.celery_app import celery_app
|
|
from fastapi_admin.config import (
|
|
LEAUDIT_PAGE_QUALITY_QUEUE_NORMAL,
|
|
LEAUDIT_PAGE_QUALITY_QUEUE_URGENT,
|
|
)
|
|
from fastapi_modules.fastapi_leaudit.page_quality.runner import PageQualityRunner
|
|
|
|
log = logger
|
|
|
|
|
|
def resolve_page_quality_queue(speed: str = "normal") -> str:
|
|
"""根据优先级返回页级模糊检测队列。"""
|
|
if (speed or "").strip().lower() in {"urgent", "high", "fast", "紧急"}:
|
|
return LEAUDIT_PAGE_QUALITY_QUEUE_URGENT
|
|
return LEAUDIT_PAGE_QUALITY_QUEUE_NORMAL
|
|
|
|
|
|
def dispatch_page_quality_task(run_id: int, *, speed: str = "normal") -> str:
|
|
"""投递页级模糊检测任务。"""
|
|
queue = resolve_page_quality_queue(speed)
|
|
task = page_quality_process_document_task.apply_async(kwargs={"run_id": run_id}, queue=queue)
|
|
log.info("page_quality run_id=%s dispatched: queue=%s, task_id=%s", run_id, queue, task.id)
|
|
return str(task.id or "")
|
|
|
|
|
|
@celery_app.task(
|
|
bind=True,
|
|
name="leaudit.page_quality.process_document",
|
|
acks_late=True,
|
|
)
|
|
def page_quality_process_document_task(self, run_id: int) -> dict[str, Any]:
|
|
"""页级模糊检测 Celery 入口。"""
|
|
loop = asyncio.new_event_loop()
|
|
asyncio.set_event_loop(loop)
|
|
try:
|
|
runner = PageQualityRunner()
|
|
return loop.run_until_complete(runner.Execute(RunId=run_id))
|
|
finally:
|
|
loop.close()
|