fix: create customer when external_order sync to database

api_v1/management/commands/backfill_external_product_images.py

@@ -35,7 +35,7 @@ class Command(BaseCommand):
         if product_id:
             queryset = queryset.filter(id=product_id)
 
-        queryset = queryset.order_by('id')
+        queryset = queryset.order_by('-id')
 
         limit = options.get('limit')
         if limit is not None:

api_v1/management/commands/retry_external_printing_sync_failures.py

@@ -0,0 +1,125 @@
+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=100, 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 = max(1, int(options.get('limit') or 100))
+        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 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()
+
+        result = {
+            '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,
+        }
+        self.stdout.write(self.style.SUCCESS(str(result)))

api_v1/tasks.py

@@ -330,7 +330,22 @@ def _resolve_external_customer(*, merchant, record: dict):
         .first()
     )
     if not customer:
-        raise RuntimeError(f'未找到客户: {customer_name}')
+        defaults = {
+            'merchant': merchant,
+            'name': customer_name,
+        }
+        sync_user = record.get('_sync_user')
+        employee = getattr(sync_user, 'employee', None)
+        if employee and employee.merchant_id == merchant.id:
+            defaults['created_by'] = employee
+        customer = basic_models.Customer.objects.create(**defaults)
+        logger.info(
+            '外部印染同步自动创建客户: merchant_id=%s customer_name=%s external_customer_id=%s customer_id=%s',
+            merchant.id,
+            customer_name,
+            external_customer_id,
+            customer.id,
+        )
     return customer, customer_name, external_customer_id
 
 
@@ -490,7 +505,12 @@ def _build_external_order_data(
     sync_user,
 ):
     first = group_records[0]
-    customer, external_customer_name, external_customer_id = _resolve_external_customer(merchant=merchant, record=first)
+    first_with_context = dict(first)
+    first_with_context['_sync_user'] = sync_user
+    customer, external_customer_name, external_customer_id = _resolve_external_customer(
+        merchant=merchant,
+        record=first_with_context,
+    )
     created_by_user, external_employee_name = _resolve_external_employee_user(
         merchant=merchant,
         external_employee_name=str(first.get('CaoZY') or '').strip(),
@@ -594,6 +614,126 @@ def _upsert_external_printing_job(*, merchant, printing_order, record: dict, syn
     return job, True
 
 
+def _sync_external_printing_records_batch(
+    *,
+    records: list[dict],
+    merchant,
+    sync_user,
+    category,
+    record_failure,
+) -> dict:
+    orders_created = 0
+    orders_updated = 0
+    jobs_created = 0
+    jobs_updated = 0
+    failed_record_ids: list[int] = []
+    failed_records = 0
+    grouped_records: dict[str, list[dict]] = {}
+
+    if not records:
+        return {
+            'orders_created': 0,
+            'orders_updated': 0,
+            'jobs_created': 0,
+            'jobs_updated': 0,
+            'failed_records': 0,
+            'failed_record_ids': [],
+            'group_count': 0,
+        }
+
+    grouped_records = _group_external_records(records)
+    for external_order_id, group_records in grouped_records.items():
+        successful_records: list[dict] = []
+        for record in group_records:
+            try:
+                _ensure_external_product(
+                    merchant=merchant,
+                    external_product_name=str(record.get('YanSe') or '').strip(),
+                    category=category,
+                )
+            except Exception as exc:
+                error_message = str(exc)
+                logger.warning(
+                    '外部印染记录产品处理失败: external_record_id=%s external_order_id=%s error=%s',
+                    record.get('ID'),
+                    external_order_id,
+                    error_message,
+                )
+                record_failure(record=record, error=error_message)
+                failed_records += 1
+                failed_record_ids.append(
+                    _extract_positive_int(record.get('ID'), field_name='ID', allow_blank=False)
+                )
+                continue
+            successful_records.append(record)
+
+        if not successful_records:
+            continue
+
+        try:
+            order, order_created = _upsert_external_printing_order(
+                merchant=merchant,
+                external_order_id=external_order_id,
+                group_records=successful_records,
+                sync_user=sync_user,
+            )
+            if order_created:
+                orders_created += 1
+            else:
+                orders_updated += 1
+        except Exception as exc:
+            error_message = str(exc)
+            logger.warning(
+                '外部印染记录订单处理失败: external_order_id=%s error=%s',
+                external_order_id,
+                error_message,
+            )
+            for record in successful_records:
+                record_failure(record=record, error=error_message)
+                failed_records += 1
+                failed_record_ids.append(
+                    _extract_positive_int(record.get('ID'), field_name='ID', allow_blank=False)
+                )
+            continue
+
+        for record in successful_records:
+            try:
+                _job, job_created = _upsert_external_printing_job(
+                    merchant=merchant,
+                    printing_order=order,
+                    record=record,
+                    sync_user=sync_user,
+                    category=category,
+                )
+                if job_created:
+                    jobs_created += 1
+                else:
+                    jobs_updated += 1
+            except Exception as exc:
+                error_message = str(exc)
+                logger.warning(
+                    '外部印染记录任务处理失败: external_record_id=%s external_order_id=%s error=%s',
+                    record.get('ID'),
+                    external_order_id,
+                    error_message,
+                )
+                record_failure(record=record, error=error_message)
+                failed_records += 1
+                failed_record_ids.append(
+                    _extract_positive_int(record.get('ID'), field_name='ID', allow_blank=False)
+                )
+
+    return {
+        'orders_created': orders_created,
+        'orders_updated': orders_updated,
+        'jobs_created': jobs_created,
+        'jobs_updated': jobs_updated,
+        'failed_records': failed_records,
+        'failed_record_ids': failed_record_ids,
+        'group_count': len(grouped_records),
+    }
+
+
 def _run_fetch(page: int, page_size: int):
     return asyncio.run(fetch_products_from_mingdaoyun(page=page, page_size=page_size))
 
@@ -1158,113 +1298,32 @@ def sync_external_printing_records(self, limit: int = 100):
     last_record_id = payload.get('last_record_id')
     run_date = timezone.localdate()
 
-    orders_created = 0
-    orders_updated = 0
-    jobs_created = 0
-    jobs_updated = 0
-    failed_record_ids: list[int] = []
-    failed_records = 0
-
     if records:
-        grouped_records = _group_external_records(records)
+        batch_result = _sync_external_printing_records_batch(
-        for external_order_id, group_records in grouped_records.items():
+            records=records,
-            successful_records: list[dict] = []
-            for record in group_records:
-                try:
-                    _ensure_external_product(
             merchant=merchant,
-                        external_product_name=str(record.get('YanSe') or '').strip(),
-                        category=category,
-                    )
-                except Exception as exc:
-                    error_message = str(exc)
-                    logger.warning(
-                        '外部印染记录产品处理失败: external_record_id=%s external_order_id=%s error=%s',
-                        record.get('ID'),
-                        external_order_id,
-                        error_message,
-                    )
-                    _record_printing_external_sync_failure(
-                        run_date=run_date,
-                        record=record,
-                        error=error_message,
-                    )
-                    failed_records += 1
-                    failed_record_ids.append(
-                        _extract_positive_int(record.get('ID'), field_name='ID', allow_blank=False)
-                    )
-                    continue
-                successful_records.append(record)
-
-            if not successful_records:
-                continue
-
-            try:
-                order, order_created = _upsert_external_printing_order(
-                    merchant=merchant,
-                    external_order_id=external_order_id,
-                    group_records=successful_records,
-                    sync_user=sync_user,
-                )
-                if order_created:
-                    orders_created += 1
-                else:
-                    orders_updated += 1
-            except Exception as exc:
-                error_message = str(exc)
-                logger.warning(
-                    '外部印染记录订单处理失败: external_order_id=%s error=%s',
-                    external_order_id,
-                    error_message,
-                )
-                for record in successful_records:
-                    _record_printing_external_sync_failure(
-                        run_date=run_date,
-                        record=record,
-                        error=error_message,
-                    )
-                    failed_records += 1
-                    failed_record_ids.append(
-                        _extract_positive_int(record.get('ID'), field_name='ID', allow_blank=False)
-                    )
-                continue
-
-            for record in successful_records:
-                try:
-                    _job, job_created = _upsert_external_printing_job(
-                        merchant=merchant,
-                        printing_order=order,
-                        record=record,
             sync_user=sync_user,
             category=category,
-                    )
+            record_failure=lambda *, record, error: _record_printing_external_sync_failure(
-                    if job_created:
-                        jobs_created += 1
-                    else:
-                        jobs_updated += 1
-                except Exception as exc:
-                    error_message = str(exc)
-                    logger.warning(
-                        '外部印染记录任务处理失败: external_record_id=%s external_order_id=%s error=%s',
-                        record.get('ID'),
-                        external_order_id,
-                        error_message,
-                    )
-                    _record_printing_external_sync_failure(
                 run_date=run_date,
                 record=record,
-                        error=error_message,
+                error=error,
-                    )
+            ),
-                    failed_records += 1
-                    failed_record_ids.append(
-                        _extract_positive_int(record.get('ID'), field_name='ID', allow_blank=False)
         )
 
         cursor_payload = None
         if last_record_id is not None:
             cursor_payload = _advance_external_printing_cursor(cursor_value=int(last_record_id))
     else:
-        grouped_records = {}
+        batch_result = {
+            'orders_created': 0,
+            'orders_updated': 0,
+            'jobs_created': 0,
+            'jobs_updated': 0,
+            'failed_records': 0,
+            'failed_record_ids': [],
+            'group_count': 0,
+        }
         cursor_payload = None
 
     api_models.DataSync.objects.create(
@@ -1282,17 +1341,17 @@ def sync_external_printing_records(self, limit: int = 100):
         'limit': limit,
         'mode': payload.get('mode'),
         'count': len(records),
-        'orders_created': orders_created,
+        'orders_created': batch_result['orders_created'],
-        'orders_updated': orders_updated,
+        'orders_updated': batch_result['orders_updated'],
-        'jobs_created': jobs_created,
+        'jobs_created': batch_result['jobs_created'],
-        'jobs_updated': jobs_updated,
+        'jobs_updated': batch_result['jobs_updated'],
-        'failed_records': failed_records,
+        'failed_records': batch_result['failed_records'],
-        'failed_record_ids': failed_record_ids,
+        'failed_record_ids': batch_result['failed_record_ids'],
         'last_record_id': last_record_id,
         'cursor_before': payload.get('cursor_before'),
         'cursor_after': payload.get('cursor_after'),
         'cursor_updated': bool(cursor_payload),
-        'group_count': len(grouped_records),
+        'group_count': batch_result['group_count'],
     }
     logger.info('外部印染 records 同步完成: %s', result)
     return result

api_v1/test_external_printing_records_sync.py

@@ -1,8 +1,11 @@
+from io import StringIO
 from unittest.mock import patch
 
+from django.core.management import call_command
 from django.conf import settings
 from django.contrib.auth.models import User
 from django.test import TestCase
+from django.utils import timezone
 
 from api_v1 import models as api_models
 from api_v1.tasks import sync_external_printing_records
@@ -224,6 +227,70 @@ class ExternalPrintingRecordsSyncTaskTest(TestCase):
         self.assertEqual(failure.external_order_id, 'KD20432358')
         self.assertIn('外部图片不存在', failure.error)
 
+    @patch('api_v1.tasks._advance_external_printing_cursor')
+    @patch('api_v1.tasks._fetch_external_product_image')
+    @patch('api_v1.tasks._fetch_external_printing_records')
+    def test_missing_customer_creates_customer_and_syncs_successfully(
+        self,
+        mock_fetch_records,
+        mock_fetch_image,
+        mock_advance_cursor,
+    ):
+        missing_customer_name = '新补建客户'
+        record = self._build_record(record_id=1000002, product_name=self.existing_product.name)
+        record['customer']['KhName'] = missing_customer_name
+        record['KhID'] = 'KH01000'
+        mock_fetch_records.return_value = self._build_payload([record])
+        mock_advance_cursor.return_value = {'updated': True}
+
+        result = sync_external_printing_records.run(limit=100)
+
+        self.assertEqual(result['failed_records'], 0)
+        self.assertEqual(result['jobs_created'], 1)
+        self.assertEqual(mock_fetch_image.call_count, 0)
+
+        customer = basic_models.Customer.objects.get(merchant=self.merchant, name=missing_customer_name)
+        self.assertEqual(customer.created_by, self.sync_employee)
+
+        order = printing_models.PrintingOrder.objects.get(external_order_id='KD20432358')
+        self.assertEqual(order.customer, customer)
+        self.assertEqual(order.external_customer_name, missing_customer_name)
+
+    def test_retry_command_replays_failure_raw_and_clears_failure_rows(self):
+        missing_customer_name = '重试补建客户'
+        raw_record = self._build_record(record_id=1000003, product_name=self.existing_product.name)
+        raw_record['customer']['KhName'] = missing_customer_name
+        raw_record['KhID'] = 'KH01001'
+        api_models.PrintingExternalSyncFailure.objects.create(
+            run_date=timezone.localdate(),
+            external_record_id=1000003,
+            external_order_id='KD20432358',
+            product_name=self.existing_product.name,
+            error=f'未找到客户: {missing_customer_name}',
+            raw=raw_record,
+            attempts=1,
+            last_attempt_at=timezone.now(),
+        )
+
+        output = StringIO()
+        call_command(
+            'retry_external_printing_sync_failures',
+            '--record-id',
+            '1000003',
+            stdout=output,
+        )
+
+        self.assertTrue(
+            basic_models.Customer.objects.filter(merchant=self.merchant, name=missing_customer_name).exists()
+        )
+        self.assertTrue(
+            printing_models.PrintingJob.objects.filter(original_id=1000003).exists()
+        )
+        self.assertFalse(
+            api_models.PrintingExternalSyncFailure.objects.filter(external_record_id=1000003).exists()
+        )
+        self.assertIn("'retried_records': 1", output.getvalue())
+
     @patch('api_v1.tasks._advance_external_printing_cursor')
     @patch('api_v1.tasks._fetch_external_product_image')
     @patch('api_v1.tasks._fetch_external_printing_records')