diff --git a/api_man/tests.py b/api_man/tests.py index 58e4d14..ae9b426 100644 --- a/api_man/tests.py +++ b/api_man/tests.py @@ -828,6 +828,31 @@ class CustomerAPITestCase(TestCase): self.assertEqual(len(results), 1) self.assertEqual(results[0]['id'], customer_a.id) + def test_filter_customers_by_visible_employee_name_excludes_unbound_customers(self): + """测试 visible_employee_name 过滤不会返回没有绑定员工的客户""" + visible_employee = Employee.objects.create( + merchant=self.merchant, + name='李四业务', + ) + unbound_customer = Customer.objects.create( + merchant=self.merchant, + name='未绑定客户', + created_by=self.employee, + ) + bound_customer = Customer.objects.create( + merchant=self.merchant, + name='已绑定客户', + created_by=self.employee, + ) + bound_customer.visible_employees.add(visible_employee) + + response = self.client.get('/api/backend/customers/?visible_employee_name=李四') + self.assertEqual(response.status_code, status.HTTP_200_OK) + results = response.data['results'] if isinstance(response.data, dict) else response.data + returned_ids = {item['id'] for item in results} + self.assertIn(bound_customer.id, returned_ids) + self.assertNotIn(unbound_customer.id, returned_ids) + def test_filter_customers_by_visible_employee_name_returns_distinct_results(self): """测试 visible_employee_name 过滤不会因多名匹配员工导致客户重复""" customer = Customer.objects.create( diff --git a/api_v1/management/commands/sync_external_printing_records.py b/api_v1/management/commands/sync_external_printing_records.py new file mode 100644 index 0000000..ea4ae1d --- /dev/null +++ b/api_v1/management/commands/sync_external_printing_records.py @@ -0,0 +1,14 @@ +from django.core.management.base import BaseCommand + +from api_v1.tasks import sync_external_printing_records + + +class Command(BaseCommand): + help = '从外部 records 接口同步印染订单/任务' + + def add_arguments(self, parser): + parser.add_argument('--limit', type=int, default=100) + + def handle(self, *args, **options): + payload = sync_external_printing_records.run(limit=options['limit']) + self.stdout.write(self.style.SUCCESS(str(payload))) diff --git a/api_v1/migrations/0013_printingexternalsyncfailure_and_more.py b/api_v1/migrations/0013_printingexternalsyncfailure_and_more.py new file mode 100644 index 0000000..c3aa220 --- /dev/null +++ b/api_v1/migrations/0013_printingexternalsyncfailure_and_more.py @@ -0,0 +1,44 @@ +# Generated by Django 5.2.8 on 2026-03-20 00:00 + +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ('api_v1', '0012_datasync_merchant'), + ] + + operations = [ + migrations.CreateModel( + name='PrintingExternalSyncFailure', + fields=[ + ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')), + ('created_at', models.DateTimeField(auto_now_add=True, db_index=True, verbose_name='创建时间')), + ('updated_at', models.DateTimeField(auto_now=True, verbose_name='更新时间')), + ('run_date', models.DateField(db_index=True, verbose_name='任务日期')), + ('external_record_id', models.PositiveBigIntegerField(db_index=True, verbose_name='外部记录ID')), + ('external_order_id', models.CharField(blank=True, max_length=100, verbose_name='外部订单编号')), + ('product_name', models.CharField(blank=True, max_length=200, verbose_name='外部产品名')), + ('error', models.TextField(blank=True, verbose_name='错误摘要')), + ('raw', models.JSONField(blank=True, default=dict, verbose_name='原始数据')), + ('attempts', models.PositiveIntegerField(default=0, verbose_name='尝试次数')), + ('last_attempt_at', models.DateTimeField(blank=True, null=True, verbose_name='最后尝试时间')), + ], + options={ + 'verbose_name': '外部印染同步失败记录', + 'verbose_name_plural': '外部印染同步失败记录', + 'db_table': 'api_printing_external_sync_failure', + 'ordering': ['-created_at'], + }, + ), + migrations.AddConstraint( + model_name='printingexternalsyncfailure', + constraint=models.UniqueConstraint(fields=('run_date', 'external_record_id'), name='uniq_printing_ext_sync_fail_run_date_record'), + ), + migrations.AlterField( + model_name='datasync', + name='table_name', + field=models.CharField(choices=[('product', '产品'), ('customer', '客户'), ('plate_order', '开版（明道云）'), ('printing_external_record', '印染订单（外部 records）'), ('mdy_plate_order_staging_tiia_upload', '开版暂存-腾讯图库上传')], max_length=50, verbose_name='同步目标'), + ), + ] diff --git a/api_v1/models.py b/api_v1/models.py index 8e0dd6b..c185531 100644 --- a/api_v1/models.py +++ b/api_v1/models.py @@ -93,6 +93,7 @@ class DataSync(ModelBase): PRODUCT = 'product', '产品' CUSTOMER = 'customer', '客户' PLATE_ORDER = 'plate_order', '开版（明道云）' + PRINTING_EXTERNAL_RECORD = 'printing_external_record', '印染订单（外部 records）' MDY_PLATE_ORDER_STAGING_TIIA_UPLOAD = 'mdy_plate_order_staging_tiia_upload', '开版暂存-腾讯图库上传' table_name = models.CharField(max_length=50, choices=TableName.choices, verbose_name='同步目标') @@ -221,6 +222,38 @@ class MDYPlateOrderStagingTiiaUploadFailure(ModelBase): return f'{self.run_date} {self.mdy_rowid}' +class PrintingExternalSyncFailure(ModelBase): + """ + 外部印染 records 同步失败记录。 + + 以外部 record.ID 为追踪主键，允许同一天重复失败时累加 attempts。 + """ + + run_date = models.DateField(db_index=True, verbose_name='任务日期') + external_record_id = models.PositiveBigIntegerField(db_index=True, verbose_name='外部记录ID') + external_order_id = models.CharField(max_length=100, blank=True, verbose_name='外部订单编号') + product_name = models.CharField(max_length=200, blank=True, verbose_name='外部产品名') + error = models.TextField(blank=True, verbose_name='错误摘要') + raw = models.JSONField(default=dict, blank=True, verbose_name='原始数据') + attempts = models.PositiveIntegerField(default=0, verbose_name='尝试次数') + last_attempt_at = models.DateTimeField(null=True, blank=True, verbose_name='最后尝试时间') + + class Meta: + db_table = 'api_printing_external_sync_failure' + verbose_name = '外部印染同步失败记录' + verbose_name_plural = '外部印染同步失败记录' + ordering = ['-created_at'] + constraints = [ + models.UniqueConstraint( + fields=['run_date', 'external_record_id'], + name='uniq_printing_ext_sync_fail_run_date_record', + ) + ] + + def __str__(self): + return f'{self.external_record_id} @ {self.run_date}' + + class ApiAuditLog(models.Model): """API审计日志 - 记录创建操作的历史现场 diff --git a/api_v1/tasks.py b/api_v1/tasks.py index 9074fd7..62f527b 100644 --- a/api_v1/tasks.py +++ b/api_v1/tasks.py @@ -1,19 +1,27 @@ import asyncio +import base64 import json import logging +import mimetypes import os +import re import shutil import subprocess +from collections import defaultdict from datetime import datetime from decimal import Decimal, InvalidOperation from pathlib import Path +import requests from celery import shared_task from django.conf import settings +from django.contrib.auth import get_user_model +from django.core.files.base import ContentFile from django.utils import timezone from basic_info import models as basic_models from api_v1 import models as api_models +from api_v1.views.printing.services import PrintingJobService, PrintingOrderService from flower.utils import ( fetch_products_from_mingdaoyun, fetch_customers_from_mingdaoyun, @@ -23,9 +31,11 @@ from api_v1.mdy_plate_order_staging_tiia_upload import ( build_default_tiia_rate_limiter, upload_mdy_plate_order_staging_plate_images_to_tencent_tiia, ) +from printing import models as printing_models logger = logging.getLogger(__name__) +User = get_user_model() def _ensure_backup_dir(output_dir: str | None) -> Path: @@ -171,6 +181,417 @@ def _ensure_int(value): return None +def _parse_external_datetime(value: str | None): + if not value: + return None + text = str(value).strip() + if not text: + return None + try: + dt = datetime.fromisoformat(text.replace('Z', '+00:00')) + except ValueError: + return None + if timezone.is_naive(dt): + dt = timezone.make_aware(dt, timezone.get_current_timezone()) + return dt + + +def _extract_positive_int(value, *, field_name: str, allow_blank: bool = True): + if value is None or value == '': + return None if allow_blank else 0 + try: + number = Decimal(str(value)) + except (InvalidOperation, TypeError, ValueError) as exc: + raise RuntimeError(f'外部字段 {field_name} 无法解析为整数: {value!r}') from exc + if number != number.to_integral_value(): + raise RuntimeError(f'外部字段 {field_name} 不是整数: {value!r}') + parsed = int(number) + if parsed < 0: + raise RuntimeError(f'外部字段 {field_name} 不能为负数: {value!r}') + return parsed + + +def _extract_pieces(value): + if value is None or value == '': + return None + match = re.search(r'(\d+)', str(value)) + if not match: + return None + return int(match.group(1)) + + +def _normalize_unit(value): + text = str(value or '').strip() + return text.lstrip('/').strip() or text + + +def _split_fabric_source_and_craft(value: str | None): + text = str(value or '').strip() + if not text: + return '', '' + parts = re.split(r'\s+', text) + fabric_source = parts[0] if parts else '' + craft = ' '.join(parts[1:]).strip() if len(parts) > 1 else '' + return fabric_source, craft + + +def _build_external_url(path: str) -> str: + base_url = str(getattr(settings, 'PRINTING_EXTERNAL_RECORDS_BASE_URL', '') or '').rstrip('/') + if not base_url: + raise RuntimeError('未配置 PRINTING_EXTERNAL_RECORDS_BASE_URL') + return f'{base_url}{path}' + + +def _get_external_headers() -> dict[str, str]: + secret = str(getattr(settings, 'PRINTING_EXTERNAL_RECORDS_AUTHORIZATION', '') or '').strip() + if not secret: + raise RuntimeError('未配置 PRINTING_EXTERNAL_RECORDS_AUTHORIZATION') + return {'Authorization': secret} + + +def _get_printing_sync_user(): + user_id = int(getattr(settings, 'PRINTING_EXTERNAL_SYNC_USER_ID', 1) or 1) + user = User.objects.filter(id=user_id).select_related('employee__merchant').first() + if not user: + raise RuntimeError(f'未找到固定同步用户: {user_id}') + employee = getattr(user, 'employee', None) + merchant = getattr(employee, 'merchant', None) + if not merchant: + raise RuntimeError(f'固定同步用户 {user_id} 未绑定 employee.merchant，无法确定同步商户') + return user + + +def _get_printing_sync_product_category(merchant): + category_id = getattr(settings, 'PRINTING_EXTERNAL_SYNC_PRODUCT_CATEGORY_ID', None) + qs = basic_models.ProductCategory.objects.filter(merchant=merchant).order_by('id') + if category_id: + qs = qs.filter(id=category_id) + category = qs.first() + if not category: + raise RuntimeError( + f'未找到用于外部印染同步的产品类别，请先创建类别或配置 ' + f'PRINTING_EXTERNAL_SYNC_PRODUCT_CATEGORY_ID（merchant={merchant.id}）' + ) + return category + + +def _fetch_external_printing_records(*, limit: int = 100) -> dict: + url = _build_external_url('/api/v1/records') + response = requests.get( + url, + headers=_get_external_headers(), + params={'limit': max(1, int(limit)), 'update_cursor': 'false'}, + timeout=30, + ) + response.raise_for_status() + payload = response.json() + if not isinstance(payload, dict): + raise RuntimeError('外部 records 接口返回格式非法：非 JSON object') + if not isinstance(payload.get('records', []), list): + raise RuntimeError('外部 records 接口返回格式非法：records 不是数组') + return payload + + +def _advance_external_printing_cursor(*, cursor_value: int) -> dict: + url = _build_external_url('/api/v1/cursor/set') + response = requests.post( + url, + headers=_get_external_headers(), + json={'value': int(cursor_value)}, + timeout=30, + ) + response.raise_for_status() + payload = response.json() + if not isinstance(payload, dict): + raise RuntimeError('外部 cursor/set 接口返回格式非法：非 JSON object') + return payload + + +def _group_external_records(records: list[dict]) -> dict[str, list[dict]]: + grouped: dict[str, list[dict]] = defaultdict(list) + for record in records: + external_order_id = str(record.get('BianHaoID') or '').strip() + if not external_order_id: + raise RuntimeError(f'外部 record 缺少 BianHaoID: {record!r}') + grouped[external_order_id].append(record) + return grouped + + +def _resolve_external_customer(*, merchant, record: dict): + customer_payload = record.get('customer') or {} + customer_name = str(customer_payload.get('KhName') or '').strip() + external_customer_id = str(record.get('KhID') or customer_payload.get('KhID') or '').strip() + if not customer_name: + raise RuntimeError(f'外部 record 缺少 customer.KhName: {record!r}') + + customer = ( + basic_models.Customer.objects.filter(merchant=merchant, name=customer_name) + .order_by('id') + .first() + ) + if not customer: + raise RuntimeError(f'未找到客户: {customer_name}') + return customer, customer_name, external_customer_id + + +def _resolve_external_employee_user(*, merchant, external_employee_name: str, sync_user): + normalized = str(external_employee_name or '').strip() + if not normalized: + return sync_user, '' + + employee = ( + basic_models.Employee.objects.filter(merchant=merchant, name=normalized) + .select_related('sys_user') + .order_by('id') + .first() + ) + if employee and employee.sys_user_id: + return employee.sys_user, '' + return sync_user, normalized + + +def _encode_external_image_name_b64(name: str) -> str: + raw = str(name or '').encode('utf-8') + return base64.urlsafe_b64encode(raw).decode('ascii').rstrip('=') + + +def _decode_external_image_data(data: str) -> bytes: + text = str(data or '').strip() + if not text: + raise RuntimeError('外部图片接口返回空 data') + padding = '=' * (-len(text) % 4) + try: + return base64.b64decode(text + padding) + except Exception as exc: + raise RuntimeError('外部图片接口返回的 data 不是合法 base64') from exc + + +def _fetch_external_product_image(external_product_name: str) -> dict: + url = _build_external_url('/api/v1/image') + response = requests.get( + url, + headers=_get_external_headers(), + params={'name_b64': _encode_external_image_name_b64(external_product_name)}, + timeout=30, + ) + if response.status_code == 404: + raise RuntimeError(f'外部图片不存在: {external_product_name}') + response.raise_for_status() + payload = response.json() + if not isinstance(payload, dict): + raise RuntimeError('外部图片接口返回格式非法：非 JSON object') + data = _decode_external_image_data(payload.get('data', '')) + content_type = str(payload.get('content_type') or '').strip() or 'application/octet-stream' + image_name = str(payload.get('name') or '').strip() + return { + 'bytes': data, + 'content_type': content_type, + 'name': image_name or external_product_name, + } + + +def _build_product_image_filename(product_name: str, image_name: str, content_type: str) -> str: + suffix = Path(str(image_name or '').strip()).suffix + if not suffix: + suffix = mimetypes.guess_extension(content_type or '') or '.jpg' + safe_name = re.sub(r'[^0-9A-Za-z._-]+', '_', str(product_name or '').strip()) or 'external_product' + return f'{safe_name}{suffix}' + + +def _upload_product_image(product, image_payload: dict): + filename = _build_product_image_filename( + product_name=product.name, + image_name=image_payload.get('name', ''), + content_type=image_payload.get('content_type', ''), + ) + content = ContentFile(image_payload['bytes']) + product.image.save(filename, content, save=False) + product.save(update_fields=['image', 'updated_at']) + + +def _record_printing_external_sync_failure(*, run_date, record: dict, error: str): + external_record_id = _extract_positive_int(record.get('ID'), field_name='ID', allow_blank=False) + obj, _created = api_models.PrintingExternalSyncFailure.objects.get_or_create( + run_date=run_date, + external_record_id=external_record_id, + defaults={ + 'external_order_id': str(record.get('BianHaoID') or '').strip(), + 'product_name': str(record.get('YanSe') or '').strip(), + 'error': error or '', + 'raw': record, + 'attempts': 0, + 'last_attempt_at': timezone.now(), + }, + ) + obj.external_order_id = str(record.get('BianHaoID') or '').strip() + obj.product_name = str(record.get('YanSe') or '').strip() + obj.error = error or '' + obj.raw = record + obj.attempts = int(obj.attempts or 0) + 1 + obj.last_attempt_at = timezone.now() + obj.save( + update_fields=[ + 'external_order_id', + 'product_name', + 'error', + 'raw', + 'attempts', + 'last_attempt_at', + ] + ) + + +def _ensure_external_product(*, merchant, external_product_name: str, category): + normalized = str(external_product_name or '').strip() + if not normalized: + raise RuntimeError('外部产品名为空，无法创建/绑定产品') + + existing_product = ( + basic_models.Product.objects.filter(merchant=merchant, name=normalized) + .order_by('id') + .first() + ) + if existing_product: + return existing_product, False + + image_payload = _fetch_external_product_image(normalized) + product = basic_models.Product.objects.create( + merchant=merchant, + category=category, + name=normalized, + ) + try: + _upload_product_image(product, image_payload) + except Exception: + if product.image: + try: + product.image.delete(save=False) + except Exception: + logger.warning('回滚外部产品图片失败: product_id=%s', product.id, exc_info=True) + product.delete() + raise + return product, True + + +def _build_order_external_raw(group_records: list[dict]) -> dict: + first = group_records[0] if group_records else {} + return { + 'group_size': len(group_records), + 'record_ids': [record.get('ID') for record in group_records], + 'first_record': first, + } + + +def _build_external_order_data( + *, + merchant, + external_order_id: str, + group_records: list[dict], + sync_user, +): + first = group_records[0] + customer, external_customer_name, external_customer_id = _resolve_external_customer(merchant=merchant, record=first) + created_by_user, external_employee_name = _resolve_external_employee_user( + merchant=merchant, + external_employee_name=str(first.get('CaoZY') or '').strip(), + sync_user=sync_user, + ) + fabric_source, craft = _split_fabric_source_and_craft(first.get('SHDZ')) + return { + 'merchant': merchant, + 'customer': customer, + 'fabric': '', + 'width': str(first.get('SeHao') or '').strip(), + 'fabric_source': fabric_source or None, + 'craft': craft or None, + 'rolling_warn': str(first.get('BeiZhu') or '').strip() or None, + 'curve': str(first.get('MeoA') or '').strip() or None, + 'outgoing_date': _parse_external_datetime(first.get('KdRiQi')), + 'created_by': created_by_user, + 'external_order_id': external_order_id, + 'external_customer_id': external_customer_id or None, + 'external_customer_name': external_customer_name or None, + 'external_employee_name': external_employee_name or None, + 'external_raw': _build_order_external_raw(group_records), + } + + +def _upsert_external_printing_order(*, merchant, external_order_id: str, group_records: list[dict], sync_user): + order_data = _build_external_order_data( + merchant=merchant, + external_order_id=external_order_id, + group_records=group_records, + sync_user=sync_user, + ) + created_by_user = order_data['created_by'] + existing_order = ( + printing_models.PrintingOrder.objects.filter( + merchant=merchant, + external_order_id=external_order_id, + ) + .order_by('id') + .first() + ) + if existing_order: + success, message, order = PrintingOrderService.update_printing_order( + existing_order, + order_data, + created_by_user, + ) + if not success: + raise RuntimeError(f'更新印染订单失败: {message}') + return order, False + order = PrintingOrderService.create_printing_order(order_data, created_by_user) + return order, True + + +def _upsert_external_printing_job(*, merchant, printing_order, record: dict, sync_user, category): + record_id = _extract_positive_int(record.get('ID'), field_name='ID', allow_blank=False) + created_by_user, _external_employee_name = _resolve_external_employee_user( + merchant=merchant, + external_employee_name=str(record.get('CaoZY') or '').strip(), + sync_user=sync_user, + ) + product, _ = _ensure_external_product( + merchant=merchant, + external_product_name=str(record.get('YanSe') or '').strip(), + category=category, + ) + job_data = { + 'merchant': merchant, + 'printing_order': printing_order, + 'product': product, + 'quantity': _extract_positive_int(record.get('ShuLiang'), field_name='ShuLiang', allow_blank=False), + 'unit': _normalize_unit(record.get('JiJiaDW')), + 'size': str(record.get('ShuLiangZ') or '').strip() or None, + 'pieces': _extract_pieces(record.get('BeiZhuC')), + 'description': None, + 'original_id': record_id, + 'created_by': created_by_user, + 'external_product_name': None, + 'external_raw': record, + } + existing_job = ( + printing_models.PrintingJob.objects.filter( + printing_order=printing_order, + original_id=record_id, + ) + .order_by('id') + .first() + ) + if existing_job: + success, message, job = PrintingJobService.update_printing_job( + existing_job, + job_data, + created_by_user, + ) + if not success: + raise RuntimeError(f'更新印染任务失败: {message}') + return job, False + job = PrintingJobService.create_printing_job(job_data, created_by_user) + return job, True + + def _run_fetch(page: int, page_size: int): return asyncio.run(fetch_products_from_mingdaoyun(page=page, page_size=page_size)) @@ -716,6 +1137,165 @@ def sync_mdy_plate_orders_for_merchant( return payload +@shared_task(bind=True) +def sync_external_printing_records(self, limit: int = 100): + """ + 从外部 records 接口增量同步印染订单/任务。 + + 策略： + - 拉取时强制 update_cursor=false + - 按外部 record.ID 处理失败并记录 + - 本批处理结束后无论是否有部分失败，都推进远端 cursor + """ + limit = max(1, int(limit or 100)) + sync_user = _get_printing_sync_user() + merchant = sync_user.employee.merchant + category = _get_printing_sync_product_category(merchant) + payload = _fetch_external_printing_records(limit=limit) + records = payload.get('records', []) + 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) + ) + + 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 = {} + cursor_payload = None + + api_models.DataSync.objects.create( + table_name=api_models.DataSync.TableName.PRINTING_EXTERNAL_RECORD, + merchant=merchant, + page_index=1, + page_size=limit, + synced_rows=len(records), + total_count=int(payload.get('count') or 0), + note=f'cursor_after={payload.get("cursor_after")} last_record_id={last_record_id}', + ) + + result = { + 'task_id': self.request.id, + '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, + '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), + } + logger.info('外部印染 records 同步完成: %s', result) + return result + + def _record_mdy_plate_order_staging_tiia_failure( *, run_date, diff --git a/api_v1/test_external_printing_records_sync.py b/api_v1/test_external_printing_records_sync.py new file mode 100644 index 0000000..3461811 --- /dev/null +++ b/api_v1/test_external_printing_records_sync.py @@ -0,0 +1,194 @@ +from unittest.mock import patch + +from django.conf import settings +from django.contrib.auth.models import User +from django.test import TestCase + +from api_v1 import models as api_models +from api_v1.tasks import sync_external_printing_records +from basic_info import models as basic_models +from printing import models as printing_models +from stateflow.models import Process, State + + +class ExternalPrintingRecordsSyncTaskTest(TestCase): + def setUp(self): + self.merchant = basic_models.Merchant.objects.create( + name='测试印染厂', + type=basic_models.MerchantTypeEnum.FACTORY, + ) + self.sync_user = User.objects.create_user(username='sync-user', password='testpass') + self.sync_employee = basic_models.Employee.objects.create( + merchant=self.merchant, + sys_user=self.sync_user, + name='同步用户', + ) + self.operator_user = User.objects.create_user(username='operator-user', password='testpass') + self.operator_employee = basic_models.Employee.objects.create( + merchant=self.merchant, + sys_user=self.operator_user, + name='钰涵', + ) + self.customer = basic_models.Customer.objects.create( + merchant=self.merchant, + name='鸿烨服饰', + ) + self.category = basic_models.ProductCategory.objects.create( + merchant=self.merchant, + name='同步产品类', + product_prefix='TP', + ) + self.existing_product = basic_models.Product.objects.create( + merchant=self.merchant, + category=self.category, + name='Tj1712#12号色-24码', + ) + + state = State.objects.create(name='待生产') + self.process = Process.objects.create(name='默认印染流程') + self.process.replace_nodes([state]) + + settings.PRINTING_DEFAULT_PROCESS_ID = self.process.id + settings.PRINTING_EXTERNAL_SYNC_USER_ID = self.sync_user.id + settings.PRINTING_EXTERNAL_SYNC_PRODUCT_CATEGORY_ID = self.category.id + + def _build_payload(self, records): + return { + 'mode': 'incremental', + 'cursor_before': 10, + 'cursor_after': 12, + 'last_record_id': records[-1]['ID'], + 'updated': False, + 'count': len(records), + 'records': records, + } + + def _build_record(self, *, record_id: int, product_name: str): + return { + 'BeiZhu': '手感一定要柔软', + 'BeiZhuC': '10件', + 'BianHaoID': 'KD20432358', + 'BianHaoKD': '1.27', + 'CaoZY': '钰涵', + 'FidJ': r'\\fw\\2026年-LWQ15\\2026\\H鸿烨\\Tj1712#', + 'ID': record_id, + 'JiJiaDW': '/段', + 'KdRiQi': '2026-01-26T20:00:52Z', + 'KhID': 'KH00999', + 'MeoA': 'LWQ15', + 'RiQi': '2026-01-26T00:00:00Z', + 'SHDZ': '客户布 烧花', + 'SeHao': '1.51', + 'ShuLiang': '10.00', + 'ShuLiangZ': '2.68', + 'YanSe': product_name, + 'customer': { + 'KhID': 'KH00999', + 'KhName': '鸿烨服饰', + }, + } + + @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_existing_product_skips_image_fetch( + self, + mock_fetch_records, + mock_fetch_image, + mock_advance_cursor, + ): + mock_fetch_records.return_value = self._build_payload( + [self._build_record(record_id=1000000, product_name=self.existing_product.name)] + ) + mock_advance_cursor.return_value = {'updated': True} + + result = sync_external_printing_records.run(limit=100) + + self.assertEqual(result['jobs_created'], 1) + self.assertEqual(result['failed_records'], 0) + self.assertEqual(mock_fetch_image.call_count, 0) + mock_advance_cursor.assert_called_once_with(cursor_value=1000000) + + job = printing_models.PrintingJob.objects.get(original_id=1000000) + self.assertEqual(job.product, self.existing_product) + self.assertIsNone(job.external_product_name) + + @patch('api_v1.tasks._advance_external_printing_cursor') + @patch('api_v1.tasks._upload_product_image') + @patch('api_v1.tasks._fetch_external_product_image') + @patch('api_v1.tasks._fetch_external_printing_records') + def test_missing_product_creates_product_and_job( + self, + mock_fetch_records, + mock_fetch_image, + mock_upload_product_image, + mock_advance_cursor, + ): + new_product_name = 'Tj1712#12号色-25码' + mock_fetch_records.return_value = self._build_payload( + [self._build_record(record_id=1000001, product_name=new_product_name)] + ) + mock_fetch_image.return_value = { + 'bytes': b'fake-image', + 'content_type': 'image/jpeg', + 'name': 'abc#123.jpg', + } + mock_advance_cursor.return_value = {'updated': True} + + def _fake_upload(product, image_payload): + product.image = 'product_images/generated.jpg' + product.save(update_fields=['image']) + + mock_upload_product_image.side_effect = _fake_upload + + result = sync_external_printing_records.run(limit=100) + + self.assertEqual(result['jobs_created'], 1) + self.assertEqual(result['failed_records'], 0) + mock_fetch_image.assert_called_once_with(new_product_name) + + product = basic_models.Product.objects.get(name=new_product_name, merchant=self.merchant) + self.assertIsNotNone(product) + self.assertEqual(product.category, self.category) + + job = printing_models.PrintingJob.objects.get(original_id=1000001) + self.assertEqual(job.product, product) + + @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_partial_failure_records_external_record_id_and_advances_cursor( + self, + mock_fetch_records, + mock_fetch_image, + mock_advance_cursor, + ): + success_record = self._build_record(record_id=1000000, product_name=self.existing_product.name) + failed_record = self._build_record(record_id=1000001, product_name='不存在图片的产品#1') + mock_fetch_records.return_value = self._build_payload([success_record, failed_record]) + mock_advance_cursor.return_value = {'updated': True} + + def _fake_fetch_image(product_name): + raise RuntimeError(f'外部图片不存在: {product_name}') + + mock_fetch_image.side_effect = _fake_fetch_image + + result = sync_external_printing_records.run(limit=100) + + self.assertEqual(result['jobs_created'], 1) + self.assertEqual(result['failed_records'], 1) + self.assertEqual(result['failed_record_ids'], [1000001]) + mock_advance_cursor.assert_called_once_with(cursor_value=1000001) + + self.assertTrue( + printing_models.PrintingJob.objects.filter(original_id=1000000).exists() + ) + self.assertFalse( + printing_models.PrintingJob.objects.filter(original_id=1000001).exists() + ) + + failure = api_models.PrintingExternalSyncFailure.objects.get( + external_record_id=1000001 + ) + self.assertEqual(failure.external_order_id, 'KD20432358') + self.assertIn('外部图片不存在', failure.error) diff --git a/docs/printing_external_records_sync_2026-03-20.md b/docs/printing_external_records_sync_2026-03-20.md new file mode 100644 index 0000000..522134d --- /dev/null +++ b/docs/printing_external_records_sync_2026-03-20.md @@ -0,0 +1,170 @@ +# 外部 records 同步到 PrintingOrder / PrintingJob 设计 + +日期：2026-03-20 + +## 目标 + +将外部服务 `GET http://43.139.183.222:18080/api/v1/records` 返回的增量 records， +按每 5 分钟一次的频率同步到本系统： + +- 一组相同 `BianHaoID` 的 records -> 1 条 `printing.PrintingOrder` +- 每条 record -> 1 条 `printing.PrintingJob` + +本设计文档用于固化本次实现的字段映射和兜底规则，后续若业务修正，以此为追溯基线。 + +## 外部 records 样例判断 + +外部 `records` 不是“订单主表”，而是“印花明细表”。 + +判断依据： + +- 同一个 `BianHaoID` 下会出现多条 record +- 每条 record 含数量、尺寸、单位、产品名等明细信息 +- 因此单条 record 更接近本系统的 `PrintingJob` + +## 同步主规则 + +1. 按 `BianHaoID` 对本次拉到的 records 分组 +2. 每个分组 upsert 一条 `PrintingOrder` +3. 分组内每条 record 按 `ID` upsert 一条 `PrintingJob` +4. 仅当本地全部写入成功后，才推进外部 cursor + +## 字段映射 + +### 一、外部分组 -> PrintingOrder + +| 外部字段 | 含义 | 本系统字段 | 规则 | +| --- | --- | --- | --- | +| `BianHaoID` | 订单编号 | `PrintingOrder.external_order_id` | 分组键；原样保存 | +| `customer.KhName` | 客户名 | `PrintingOrder.customer` | 按本商户 `Customer.name` 精确匹配，取第一条 | +| `KhID` | 外部客户 ID | `PrintingOrder.external_customer_id` | 原样保存 | +| `customer.KhName` | 外部客户名 | `PrintingOrder.external_customer_name` | 原样保存 | +| `CaoZY` | 外部操作员/业务员 | `PrintingOrder.created_by` / `PrintingOrder.external_employee_name` | 若能匹配到内部员工且员工绑定了系统用户，则 `created_by` 取该用户，`external_employee_name` 置空；否则 `created_by` 取固定同步用户，`external_employee_name` 保留原文 | +| `SHDZ` | 布料来源 + 工艺 | `PrintingOrder.fabric_source` / `PrintingOrder.craft` | 以空白字符 split；第 1 段为 `fabric_source`，剩余重新拼接为 `craft` | +| `SeHao` | 幅宽 | `PrintingOrder.width` | 原样保存 | +| `MeoA` | 曲线 | `PrintingOrder.curve` | 原样保存 | +| `BeiZhu` | 滚筒注意事项 | `PrintingOrder.rolling_warn` | 原样保存 | +| `KdRiQi` | 下单日期 | `PrintingOrder.outgoing_date` | 按 ISO 时间解析并保存 | +| 分组首条原始数据 | 外部来源快照 | `PrintingOrder.external_raw` | 保存 order 级别的原始数据，便于追溯 | + +补充说明： + +- 外部没有可靠的 “面料” 字段，因此 `PrintingOrder.fabric` 本次同步固定写空字符串 `""` +- `BianHaoKD`、`RiQi`、`FidJ` 等当前未单独属性化的字段，保留在 `external_raw` 中 +- `BianHaoKD` 当前样例值类似 `"1.27"`，不按日期强解析 + +### 二、单条 record -> PrintingJob + +| 外部字段 | 含义 | 本系统字段 | 规则 | +| --- | --- | --- | --- | +| `ID` | 明细顺序 ID | `PrintingJob.original_id` | 幂等键；原样保存 | +| `YanSe` | 外部产品名 | `PrintingJob.product` / `PrintingJob.external_product_name` | 若能按本商户 `Product.name` 精确匹配，直接绑定该产品且不再拉图；若本地不存在，则先创建产品、拉图并上传到 storage，成功后绑定该产品 | +| `ShuLiang` | 下单数量 | `PrintingJob.quantity` | 按整数解析；当前预期外部值为整数字符串或 `10.00` 这类可安全转整数的值 | +| `JiJiaDW` | 单位 | `PrintingJob.unit` | 去掉前导 `/` 后保存，例如 `/段` -> `段` | +| `ShuLiangZ` | 一段尺寸 | `PrintingJob.size` | 原样保存 | +| `BeiZhuC` | 件数 | `PrintingJob.pieces` | 提取其中的整数，例如 `10件` -> `10` | +| `YanSe` | 外部产品名 | `PrintingJob.description` | 本次不单独映射到 description；原始值由 `product.name` 或 `external_product_name` / `external_raw` 表达 | +| 单条原始数据 | 外部来源快照 | `PrintingJob.external_raw` | 保存 job 级别完整原始 record | + +## 员工绑定规则 + +1. 读取外部 `CaoZY` +2. 在同步用户所属商户下按 `Employee.name` 精确匹配 +3. 若匹配到的 `Employee.sys_user` 存在： + - `PrintingOrder.created_by = employee.sys_user` + - `PrintingOrder.external_employee_name = ""` +4. 若未匹配到，或员工未绑定系统用户： + - `PrintingOrder.created_by = settings.PRINTING_EXTERNAL_SYNC_USER_ID` + - `PrintingOrder.external_employee_name = 外部原文` + +## 客户绑定规则 + +1. 读取 `customer.KhName` +2. 在同步用户所属商户下按 `Customer.name` 精确匹配 +3. 取第一条结果作为 `PrintingOrder.customer` +4. 同时保留： + - `external_customer_id` + - `external_customer_name` + +保护规则： + +- 若本地找不到客户，则本次同步失败，不推进外部 cursor +- 原因：`PrintingOrder.customer` 在业务上不可空，且当前未授权把客户兜底到固定客户 + +## 产品绑定与图片规则 + +1. 读取 `YanSe` +2. 在同步用户所属商户下按 `Product.name` 精确匹配 +3. 匹配成功： + - `PrintingJob.product = 匹配到的产品` + - 不请求外部图片 API + - `PrintingJob.external_product_name = ""` +4. 匹配失败： + - 先创建本地 `Product` + - 再调用外部图片 API： + - `GET /api/v1/image?name_b64=` + - 将返回的 base64 图片内容写入 `Product.image` + - 由 Django storage 自动上传到七牛 + - 成功后 `PrintingJob.product = 新建产品` + - `PrintingJob.external_product_name = ""` +5. 若产品创建或图片拉取/上传失败： + - 该条外部 `record.ID` 记为失败 + - 该 record 不创建 `PrintingJob` + - 本批次其它成功 records 继续处理 + - cursor 仍然推进 + +## 外部接口与游标策略 + +外部配置写入 `settings.py`，不经 `.env`： + +- `PRINTING_EXTERNAL_RECORDS_BASE_URL` +- `PRINTING_EXTERNAL_RECORDS_AUTHORIZATION` +- `PRINTING_EXTERNAL_SYNC_USER_ID` +- `PRINTING_EXTERNAL_SYNC_PRODUCT_CATEGORY_ID` + +游标策略： + +1. 拉取 records 时显式使用 `update_cursor=false` +2. 本地成功 records / 失败 records 都处理完成后，调用 `/api/v1/cursor/set` 推进到 `last_record_id` +3. 失败 records 需单独落库，便于后续按外部 `record.ID` 补偿 + +## 调度策略 + +Celery Beat 每 5 分钟触发一次。 + +单次请求固定： + +- `limit=100` +- 不传范围参数 + +即依赖外部 cursor 做增量拉取。 + +## 本次新增字段计划 + +### PrintingOrder + +- `external_order_id` +- `external_customer_id` +- `external_customer_name` +- `external_employee_name` +- `external_raw` + +### PrintingJob + +- `external_product_name` +- `external_raw` + +### API v1 失败记录 + +- `PrintingExternalSyncFailure` + - `external_record_id` + - `external_order_id` + - `product_name` + - `error` + - `raw` + +## 已接受的业务风险 + +1. `BianHaoID` 是否跨外部系统唯一，不由本系统负责纠偏 +2. 客户按名称取第一条，存在误绑风险，但这是本期接受的业务策略 +3. 产品若本地不存在，则依赖“创建产品 + 拉图上传”流程；该流程失败时只影响对应 `record.ID` diff --git a/flower/settings.py b/flower/settings.py index b96bc24..76ac131 100644 --- a/flower/settings.py +++ b/flower/settings.py @@ -476,6 +476,10 @@ STATEFLOW_CURRENT_STATE_MODE = 'NEXT' # Printing module settings PRINTING_DEFAULT_PROCESS_ID = 1 # 默认印染流程ID +PRINTING_EXTERNAL_RECORDS_BASE_URL = 'http://43.139.183.222:18080' +PRINTING_EXTERNAL_RECORDS_AUTHORIZATION = 'your-fixed-authorization-secret' +PRINTING_EXTERNAL_SYNC_USER_ID = 1 +PRINTING_EXTERNAL_SYNC_PRODUCT_CATEGORY_ID = None PLATE_ORDER_DEFAULT_PROCESS_ID = 2 # 默认开版流程ID # 销售品自动创建配置（PrintingJob 流程完成时） @@ -581,6 +585,13 @@ CELERY_BEAT_SCHEDULE = { 'max_records': MDY_SYNC_MAX_RECORDS, }, }, + 'printing_external_records_sync': { + 'task': 'api_v1.tasks.sync_external_printing_records', + 'schedule': crontab(minute='*/5'), + 'kwargs': { + 'limit': 100, + }, + }, # 明道云开版：同步到“暂存表” # - 仅在 02:00-08:00 时间窗内持续运行（每 N 分钟触发一次） # - 通过 request_interval_seconds 控制单次任务对明道云 API 的请求节奏，避免超 QPS diff --git a/printing/migrations/0036_printing_external_fields.py b/printing/migrations/0036_printing_external_fields.py new file mode 100644 index 0000000..f1d0d4e --- /dev/null +++ b/printing/migrations/0036_printing_external_fields.py @@ -0,0 +1,48 @@ +# Generated by Django 5.2.8 on 2026-03-20 00:00 + +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ('printing', '0035_merge_20260309_2247'), + ] + + operations = [ + migrations.AddField( + model_name='printingjob', + name='external_product_name', + field=models.CharField(blank=True, max_length=200, null=True, verbose_name='外部产品名'), + ), + migrations.AddField( + model_name='printingjob', + name='external_raw', + field=models.JSONField(blank=True, default=dict, help_text='保存外部 record 的完整原始数据', verbose_name='外部原始数据'), + ), + migrations.AddField( + model_name='printingorder', + name='external_customer_id', + field=models.CharField(blank=True, max_length=100, null=True, verbose_name='外部客户ID'), + ), + migrations.AddField( + model_name='printingorder', + name='external_customer_name', + field=models.CharField(blank=True, max_length=100, null=True, verbose_name='外部客户名'), + ), + migrations.AddField( + model_name='printingorder', + name='external_employee_name', + field=models.CharField(blank=True, max_length=100, null=True, verbose_name='外部员工名'), + ), + migrations.AddField( + model_name='printingorder', + name='external_order_id', + field=models.CharField(blank=True, db_index=True, max_length=100, null=True, verbose_name='外部订单编号'), + ), + migrations.AddField( + model_name='printingorder', + name='external_raw', + field=models.JSONField(blank=True, default=dict, help_text='保存外部 records 聚合后的 order 级原始数据快照', verbose_name='外部原始数据'), + ), + ] diff --git a/printing/models.py b/printing/models.py index 6624f2d..d9d4f15 100644 --- a/printing/models.py +++ b/printing/models.py @@ -363,6 +363,37 @@ class PrintingOrder(ModelBase): related_name='printing_orders', verbose_name='创建人', ) + external_order_id = models.CharField( + max_length=100, + blank=True, + null=True, + db_index=True, + verbose_name='外部订单编号', + ) + external_customer_id = models.CharField( + max_length=100, + blank=True, + null=True, + verbose_name='外部客户ID', + ) + external_customer_name = models.CharField( + max_length=100, + blank=True, + null=True, + verbose_name='外部客户名', + ) + external_employee_name = models.CharField( + max_length=100, + blank=True, + null=True, + verbose_name='外部员工名', + ) + external_raw = models.JSONField( + default=dict, + blank=True, + verbose_name='外部原始数据', + help_text='保存外部 records 聚合后的 order 级原始数据快照', + ) def __str__(self): return self.human_id @@ -462,6 +493,18 @@ class PrintingJob(ModelBase): related_name='printing_jobs', verbose_name='创建人', ) + external_product_name = models.CharField( + max_length=200, + blank=True, + null=True, + verbose_name='外部产品名', + ) + external_raw = models.JSONField( + default=dict, + blank=True, + verbose_name='外部原始数据', + help_text='保存外部 record 的完整原始数据', + ) def __str__(self): return self.printing_order.human_id