148 lines
5.2 KiB
Python
148 lines
5.2 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from app.models.base import async_session
|
|
from app.services.credit.expiration_service import (
|
|
archive_expired_user_balances,
|
|
list_expired_balance_user_limits,
|
|
)
|
|
from app.services.credit.subscription_service import (
|
|
expire_subscription_by_id,
|
|
grant_due_subscription_period_by_id,
|
|
list_due_subscription_ids,
|
|
list_due_subscription_period_candidates,
|
|
)
|
|
from app.services.credit.utils import utc_now
|
|
from app.services.operation_log_service import log_operation_event
|
|
from app.tasks.async_runner import run_async
|
|
from app.tasks.celery_app import celery_app
|
|
|
|
|
|
async def _run_credit_maintenance_once(batch_size: int = 500) -> dict[str, Any]:
|
|
checked_at = utc_now()
|
|
limit = max(1, min(int(batch_size or 500), 2000))
|
|
errors: list[dict[str, str]] = []
|
|
log_operation_event(
|
|
domain="billing",
|
|
module="credit_maintenance",
|
|
event_type="CREDIT_MAINTENANCE_STARTED",
|
|
event_status="started",
|
|
source="app.tasks.credit_tasks._run_credit_maintenance_once",
|
|
message="积分维护批次开始",
|
|
detail={"checked_at": checked_at.isoformat(), "batch_size": limit},
|
|
)
|
|
|
|
async with async_session() as scan_db:
|
|
grant_candidates = await list_due_subscription_period_candidates(
|
|
scan_db, request_time=checked_at, batch_size=limit
|
|
)
|
|
subscription_ids = await list_due_subscription_ids(
|
|
scan_db, request_time=checked_at, batch_size=limit
|
|
)
|
|
expired_user_limits = await list_expired_balance_user_limits(
|
|
scan_db, request_time=checked_at, batch_size=limit
|
|
)
|
|
await scan_db.rollback()
|
|
|
|
def record_failure(stage: str, item_id: str, exc: Exception) -> None:
|
|
errors.append({"stage": stage, "id": item_id, "error": str(exc)})
|
|
log_operation_event(
|
|
domain="billing",
|
|
module="credit_maintenance",
|
|
event_type="CREDIT_MAINTENANCE_ITEM_FAILED",
|
|
event_status="failed",
|
|
source="app.tasks.credit_tasks._run_credit_maintenance_once",
|
|
task_id=item_id,
|
|
message="积分维护单条处理失败",
|
|
error=str(exc),
|
|
detail={"stage": stage, "item_id": item_id},
|
|
)
|
|
|
|
granted = 0
|
|
for period_id, subscription_id, user_id in grant_candidates:
|
|
async with async_session() as db:
|
|
try:
|
|
changed = await grant_due_subscription_period_by_id(
|
|
db,
|
|
period_id=period_id,
|
|
subscription_id=subscription_id,
|
|
user_id=user_id,
|
|
request_time=checked_at,
|
|
)
|
|
await db.commit()
|
|
granted += int(changed)
|
|
except Exception as exc:
|
|
await db.rollback()
|
|
record_failure("subscription_grant", period_id, exc)
|
|
|
|
expired_subscriptions = 0
|
|
for subscription_id in subscription_ids:
|
|
async with async_session() as db:
|
|
try:
|
|
changed = await expire_subscription_by_id(
|
|
db, subscription_id=subscription_id, request_time=checked_at
|
|
)
|
|
await db.commit()
|
|
expired_subscriptions += int(changed)
|
|
except Exception as exc:
|
|
await db.rollback()
|
|
record_failure("subscription_expire", subscription_id, exc)
|
|
|
|
expired = 0
|
|
for user_id, user_limit in expired_user_limits:
|
|
async with async_session() as db:
|
|
try:
|
|
changed = await archive_expired_user_balances(
|
|
db, user_id=user_id, request_time=checked_at, limit=user_limit
|
|
)
|
|
await db.commit()
|
|
expired += int(changed)
|
|
except Exception as exc:
|
|
await db.rollback()
|
|
record_failure("credit_expire", user_id, exc)
|
|
|
|
result = {
|
|
"checked_at": checked_at.isoformat(),
|
|
"subscription_periods_granted": granted,
|
|
"subscriptions_expired": expired_subscriptions,
|
|
"credit_balances_archived": expired,
|
|
"failed_count": len(errors),
|
|
"errors": errors[:50],
|
|
}
|
|
log_operation_event(
|
|
domain="billing",
|
|
module="credit_maintenance",
|
|
event_type="CREDIT_MAINTENANCE_COMPLETED",
|
|
event_status="failed" if errors else "success",
|
|
source="app.tasks.credit_tasks._run_credit_maintenance_once",
|
|
message="积分维护批次完成",
|
|
detail=result,
|
|
error=(f"{len(errors)} 条处理失败" if errors else None),
|
|
)
|
|
return result
|
|
|
|
|
|
if celery_app:
|
|
|
|
@celery_app.task(
|
|
name="credit.maintenance_once",
|
|
bind=True,
|
|
soft_time_limit=540,
|
|
time_limit=600,
|
|
ignore_result=True,
|
|
)
|
|
def credit_maintenance_once(self, batch_size: int = 500) -> dict[str, Any]:
|
|
return run_async(_run_credit_maintenance_once(batch_size))
|
|
|
|
else:
|
|
|
|
class _DisabledTask:
|
|
def delay(self, *args: Any, **kwargs: Any) -> None:
|
|
raise RuntimeError("Celery is disabled")
|
|
|
|
def apply_async(self, *args: Any, **kwargs: Any) -> None:
|
|
raise RuntimeError("Celery is disabled")
|
|
|
|
credit_maintenance_once = _DisabledTask()
|