134 lines
4.3 KiB
Python
134 lines
4.3 KiB
Python
# app/services/celery_download_recovery_service.py
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime
|
|
from typing import Any, Dict, Iterable, List, Optional, Union
|
|
|
|
from app.config import settings
|
|
from app.services.redis_registry_service import (
|
|
close_registry_redis,
|
|
datetime_to_epoch,
|
|
ensure_aware_utc,
|
|
get_registry_redis,
|
|
redis_get_due_registry_ids,
|
|
redis_get_registry_payloads,
|
|
redis_postpone_registry_item,
|
|
redis_remove_registry_item,
|
|
redis_upsert_registry_item,
|
|
utc_now,
|
|
)
|
|
|
|
|
|
# 说明：
|
|
# - 本文件保留旧函数名，作为下载容灾兼容层。
|
|
# - 底层 Redis Hash/ZSet 操作已迁移到 redis_registry_service.py。
|
|
# - 下载 active key、Redis URL 选择逻辑不变，避免影响已稳定下载模块。
|
|
|
|
|
|
def build_download_active_payload(
|
|
*,
|
|
record_id: str,
|
|
celery_task_id: Optional[str],
|
|
stage: str,
|
|
attempt: Optional[int] = None,
|
|
queue: str = "gen_result_download",
|
|
priority: Optional[int] = None,
|
|
enqueue_at: Optional[datetime] = None,
|
|
started_at: Optional[datetime] = None,
|
|
updated_at: Optional[datetime] = None,
|
|
lease_until: Optional[datetime] = None,
|
|
next_retry_at: Optional[datetime] = None,
|
|
check_at: Optional[datetime] = None,
|
|
reason: Optional[str] = None,
|
|
) -> Dict[str, Any]:
|
|
now = utc_now()
|
|
checked_updated_at = ensure_aware_utc(updated_at) or now
|
|
checked_enqueue_at = ensure_aware_utc(enqueue_at)
|
|
checked_started_at = ensure_aware_utc(started_at)
|
|
checked_lease_until = ensure_aware_utc(lease_until)
|
|
checked_next_retry_at = ensure_aware_utc(next_retry_at)
|
|
checked_check_at = ensure_aware_utc(check_at)
|
|
|
|
return {
|
|
"record_id": record_id,
|
|
"celery_task_id": celery_task_id,
|
|
"stage": stage,
|
|
"attempt": int(attempt or 0),
|
|
"queue": queue,
|
|
"priority": priority,
|
|
"enqueue_at": datetime_to_epoch(checked_enqueue_at) if checked_enqueue_at else None,
|
|
"started_at": datetime_to_epoch(checked_started_at) if checked_started_at else None,
|
|
"updated_at": datetime_to_epoch(checked_updated_at),
|
|
"lease_until": datetime_to_epoch(checked_lease_until) if checked_lease_until else None,
|
|
"next_retry_at": datetime_to_epoch(checked_next_retry_at) if checked_next_retry_at else None,
|
|
"check_at": datetime_to_epoch(checked_check_at) if checked_check_at else None,
|
|
"reason": reason,
|
|
}
|
|
|
|
|
|
async def upsert_download_active(
|
|
*,
|
|
record_id: str,
|
|
payload: Dict[str, Any],
|
|
check_at: Optional[Union[datetime, int, float]],
|
|
) -> None:
|
|
await redis_upsert_registry_item(
|
|
hash_key=settings.DOWNLOAD_ACTIVE_REDIS_HASH_KEY,
|
|
zset_key=settings.DOWNLOAD_ACTIVE_REDIS_ZSET_KEY,
|
|
item_id=record_id,
|
|
payload=payload,
|
|
check_at=check_at,
|
|
log_context="download_active",
|
|
)
|
|
|
|
|
|
async def remove_download_active(record_id: str) -> None:
|
|
await redis_remove_registry_item(
|
|
hash_key=settings.DOWNLOAD_ACTIVE_REDIS_HASH_KEY,
|
|
zset_key=settings.DOWNLOAD_ACTIVE_REDIS_ZSET_KEY,
|
|
item_id=record_id,
|
|
log_context="download_active",
|
|
)
|
|
|
|
|
|
async def get_due_download_record_ids(
|
|
*,
|
|
limit: Optional[int] = None,
|
|
now: Optional[datetime] = None,
|
|
) -> List[str]:
|
|
return await redis_get_due_registry_ids(
|
|
zset_key=settings.DOWNLOAD_ACTIVE_REDIS_ZSET_KEY,
|
|
limit=limit or int(settings.DOWNLOAD_RECOVERY_BATCH_SIZE or 100),
|
|
now=now,
|
|
log_context="download_active",
|
|
)
|
|
|
|
|
|
async def get_download_active_payloads(
|
|
record_ids: Iterable[str],
|
|
) -> Dict[str, Dict[str, Any]]:
|
|
return await redis_get_registry_payloads(
|
|
hash_key=settings.DOWNLOAD_ACTIVE_REDIS_HASH_KEY,
|
|
item_ids=record_ids,
|
|
log_context="download_active",
|
|
)
|
|
|
|
|
|
async def postpone_download_active_check(
|
|
*,
|
|
record_id: str,
|
|
payload: Optional[Dict[str, Any]] = None,
|
|
check_at: Optional[Union[datetime, int, float]] = None,
|
|
) -> None:
|
|
if check_at is None:
|
|
check_at = utc_now().timestamp() + int(settings.DOWNLOAD_TASK_QUEUE_TIMEOUT_SECONDS or 300)
|
|
|
|
await redis_postpone_registry_item(
|
|
hash_key=settings.DOWNLOAD_ACTIVE_REDIS_HASH_KEY,
|
|
zset_key=settings.DOWNLOAD_ACTIVE_REDIS_ZSET_KEY,
|
|
item_id=record_id,
|
|
payload=payload,
|
|
check_at=check_at,
|
|
log_context="download_active",
|
|
)
|