feat: retry haobuye sync order

api_v1/tasks.py

@@ -8,7 +8,7 @@ import re
 import shutil
 import subprocess
 from collections import defaultdict
-from datetime import datetime
+from datetime import date, datetime
 from decimal import Decimal, InvalidOperation
 from pathlib import Path
 
@@ -767,6 +767,122 @@ def _sync_external_printing_records_batch(
     }
 
 
+def retry_external_printing_sync_failures_impl(
+    *,
+    limit: int | None = None,
+    run_date_text: str | None = None,
+    record_ids: list[int] | None = None,
+) -> dict:
+    if limit is not None:
+        limit = max(1, int(limit))
+    record_ids = record_ids or []
+
+    sync_user = _get_printing_sync_user()
+    merchant = sync_user.employee.merchant
+    category = _get_printing_sync_product_category(merchant)
+
+    failures_qs = api_models.PrintingExternalSyncFailure.objects.all().order_by(
+        'external_record_id',
+        '-created_at',
+        '-id',
+    )
+    if run_date_text:
+        failures_qs = failures_qs.filter(run_date=date.fromisoformat(run_date_text))
+    if record_ids:
+        failures_qs = failures_qs.filter(external_record_id__in=record_ids)
+
+    selected_failures = []
+    selected_record_ids = set()
+    for failure in failures_qs:
+        if failure.external_record_id in selected_record_ids:
+            continue
+        selected_failures.append(failure)
+        selected_record_ids.add(failure.external_record_id)
+        if limit is not None and len(selected_failures) >= limit:
+            break
+
+    if not selected_failures:
+        return {'limit': limit, 'retried_records': 0, 'message': '没有可重试的失败记录'}
+
+    retry_records: list[dict] = []
+    failure_by_record_id: dict[int, api_models.PrintingExternalSyncFailure] = {}
+    skipped_records = 0
+
+    for failure in selected_failures:
+        raw = failure.raw or {}
+        if not isinstance(raw, dict) or not raw:
+            failure.error = '失败记录缺少原始 raw 数据，无法重试'
+            failure.attempts = int(failure.attempts or 0) + 1
+            failure.last_attempt_at = timezone.now()
+            failure.save(update_fields=['error', 'attempts', 'last_attempt_at', 'updated_at'])
+            skipped_records += 1
+            continue
+
+        retry_records.append(raw)
+        failure_by_record_id[failure.external_record_id] = failure
+
+    def _update_failure(*, record: dict, error: str):
+        external_record_id = int(record.get('ID') or 0)
+        failure = failure_by_record_id.get(external_record_id)
+        if not failure:
+            return
+        failure.error = error or ''
+        failure.raw = record
+        failure.attempts = int(failure.attempts or 0) + 1
+        failure.last_attempt_at = timezone.now()
+        failure.save(update_fields=['error', 'raw', 'attempts', 'last_attempt_at', 'updated_at'])
+
+    batch_result = _sync_external_printing_records_batch(
+        records=retry_records,
+        merchant=merchant,
+        sync_user=sync_user,
+        category=category,
+        record_failure=_update_failure,
+    )
+
+    failed_ids = set(batch_result['failed_record_ids'])
+    succeeded_ids = [record_id for record_id in failure_by_record_id.keys() if record_id not in failed_ids]
+
+    deleted_failures = 0
+    if succeeded_ids:
+        deleted_failures, _deleted_detail = api_models.PrintingExternalSyncFailure.objects.filter(
+            external_record_id__in=succeeded_ids
+        ).delete()
+
+    remaining_failures = api_models.PrintingExternalSyncFailure.objects.count()
+
+    return {
+        'limit': limit,
+        'retried_records': len(retry_records),
+        'skipped_records': skipped_records,
+        'orders_created': batch_result['orders_created'],
+        'orders_updated': batch_result['orders_updated'],
+        'jobs_created': batch_result['jobs_created'],
+        'jobs_updated': batch_result['jobs_updated'],
+        'failed_records': batch_result['failed_records'],
+        'failed_record_ids': batch_result['failed_record_ids'],
+        'deleted_failures': deleted_failures,
+        'remaining_failures': remaining_failures,
+    }
+
+
+@shared_task(bind=True)
+def retry_external_printing_sync_failures(
+    self,
+    limit: int = 100,
+    run_date_text: str | None = None,
+    record_ids: list[int] | None = None,
+):
+    payload = retry_external_printing_sync_failures_impl(
+        limit=max(1, int(limit or 100)),
+        run_date_text=run_date_text,
+        record_ids=record_ids or [],
+    )
+    payload['task_id'] = self.request.id
+    logger.info('重试外部印染同步失败记录完成: %s', payload)
+    return payload
+
+
 def _run_fetch(page: int, page_size: int):
     return asyncio.run(fetch_products_from_mingdaoyun(page=page, page_size=page_size))