forked from erp-dev/erp
131 lines
5.1 KiB
Python
131 lines
5.1 KiB
Python
from datetime import date
|
||
|
||
from django.core.management.base import BaseCommand, CommandError
|
||
from django.utils import timezone
|
||
|
||
from api_v1 import models as api_models
|
||
from api_v1.tasks import (
|
||
_get_printing_sync_product_category,
|
||
_get_printing_sync_user,
|
||
_sync_external_printing_records_batch,
|
||
)
|
||
|
||
|
||
class Command(BaseCommand):
|
||
help = '基于失败记录重试外部印染 records 同步'
|
||
|
||
def add_arguments(self, parser):
|
||
parser.add_argument('--limit', type=int, default=None, help='最多重试多少条失败记录；不传则重试全部')
|
||
parser.add_argument(
|
||
'--run-date',
|
||
type=str,
|
||
default=None,
|
||
help='仅重试指定任务日期的失败记录，格式 YYYY-MM-DD',
|
||
)
|
||
parser.add_argument(
|
||
'--record-id',
|
||
action='append',
|
||
type=int,
|
||
default=None,
|
||
help='仅重试指定 external_record_id，可多次传入',
|
||
)
|
||
|
||
def handle(self, *args, **options):
|
||
limit = options.get('limit')
|
||
if limit is not None:
|
||
limit = max(1, int(limit))
|
||
run_date_text = options.get('run_date')
|
||
record_ids = options.get('record_id') 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:
|
||
try:
|
||
failures_qs = failures_qs.filter(run_date=date.fromisoformat(run_date_text))
|
||
except ValueError as exc:
|
||
raise CommandError(f'--run-date 格式非法: {run_date_text}') from exc
|
||
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:
|
||
self.stdout.write(self.style.SUCCESS(str({'retried_records': 0, 'message': '没有可重试的失败记录'})))
|
||
return
|
||
|
||
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()
|
||
|
||
result = {
|
||
'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,
|
||
}
|
||
self.stdout.write(self.style.SUCCESS(str(result))) |