From 315b53f5f72831732a289087bd26791e57a8a95a Mon Sep 17 00:00:00 2001 From: colaftc Date: Tue, 24 Mar 2026 22:14:41 +0800 Subject: [PATCH] fix: create customer when external_order sync to database --- .../backfill_external_product_images.py | 2 +- .../retry_external_printing_sync_failures.py | 125 ++++++++ api_v1/tasks.py | 279 +++++++++++------- api_v1/test_external_printing_records_sync.py | 67 +++++ 4 files changed, 362 insertions(+), 111 deletions(-) create mode 100644 api_v1/management/commands/retry_external_printing_sync_failures.py diff --git a/api_v1/management/commands/backfill_external_product_images.py b/api_v1/management/commands/backfill_external_product_images.py index 752a69d..71ec342 100644 --- a/api_v1/management/commands/backfill_external_product_images.py +++ b/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: diff --git a/api_v1/management/commands/retry_external_printing_sync_failures.py b/api_v1/management/commands/retry_external_printing_sync_failures.py new file mode 100644 index 0000000..15dac6b --- /dev/null +++ b/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))) \ No newline at end of file diff --git a/api_v1/tasks.py b/api_v1/tasks.py index 9f9e84d..b530300 100644 --- a/api_v1/tasks.py +++ b/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) - 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_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, - ) - 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, - ) - failed_records += 1 - failed_record_ids.append( - _extract_positive_int(record.get('ID'), field_name='ID', allow_blank=False) - ) + batch_result = _sync_external_printing_records_batch( + records=records, + merchant=merchant, + sync_user=sync_user, + category=category, + record_failure=lambda *, record, error: _record_printing_external_sync_failure( + run_date=run_date, + record=record, + error=error, + ), + ) 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_updated': orders_updated, - 'jobs_created': jobs_created, - 'jobs_updated': jobs_updated, - 'failed_records': failed_records, - 'failed_record_ids': failed_record_ids, + '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'], '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 diff --git a/api_v1/test_external_printing_records_sync.py b/api_v1/test_external_printing_records_sync.py index 06e335a..fdb6a4e 100644 --- a/api_v1/test_external_printing_records_sync.py +++ b/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')