feat: hpname and external_order_id + product_name unique together

api_v1/management/commands/backfill_external_product_images.py

@@ -0,0 +1,90 @@
+from django.core.management.base import BaseCommand, CommandError
+from django.db.models import Q
+
+from api_v1.tasks import _fetch_external_product_image, _upload_product_image
+from basic_info.models import Product
+
+
+class Command(BaseCommand):
+    help = '为没有图片的 Product 通过外部 image API 回补图片（使用 name_b64=base64url(product.name)）'
+
+    def add_arguments(self, parser):
+        parser.add_argument('--merchant-id', type=int, help='仅处理指定商户的产品')
+        parser.add_argument('--product-id', type=int, help='仅处理指定产品')
+        parser.add_argument('--limit', type=int, help='最多处理多少条产品')
+        parser.add_argument(
+            '--dry-run',
+            action='store_true',
+            help='仅输出将处理的产品，不实际请求外部 API 或写库',
+        )
+
+    def handle(self, *args, **options):
+        queryset = Product.objects.filter(
+            Q(image__isnull=True) | Q(image=''),
+        ).exclude(
+            name__isnull=True,
+        ).exclude(
+            name='',
+        ).select_related('merchant')
+
+        merchant_id = options.get('merchant_id')
+        if merchant_id:
+            queryset = queryset.filter(merchant_id=merchant_id)
+
+        product_id = options.get('product_id')
+        if product_id:
+            queryset = queryset.filter(id=product_id)
+
+        queryset = queryset.order_by('id')
+
+        limit = options.get('limit')
+        if limit is not None:
+            if int(limit) <= 0:
+                raise CommandError('--limit 必须大于 0')
+            queryset = queryset[: int(limit)]
+
+        products = list(queryset)
+        self.stdout.write(f'待处理产品数: {len(products)}')
+
+        if options.get('dry_run'):
+            for product in products:
+                self.stdout.write(
+                    f'[DRY-RUN] product_id={product.id} merchant_id={product.merchant_id} name={product.name}'
+                )
+            self.stdout.write(self.style.SUCCESS('dry-run 完成'))
+            return
+
+        success_count = 0
+        failed_count = 0
+        skipped_count = 0
+
+        for product in products:
+            if product.image:
+                skipped_count += 1
+                continue
+
+            try:
+                image_payload = _fetch_external_product_image(product.name)
+                _upload_product_image(product, image_payload)
+            except Exception as exc:
+                failed_count += 1
+                self.stderr.write(
+                    self.style.ERROR(
+                        f'补图失败: product_id={product.id} merchant_id={product.merchant_id} '
+                        f'name={product.name} error={exc}'
+                    )
+                )
+                continue
+
+            success_count += 1
+            self.stdout.write(
+                self.style.SUCCESS(
+                    f'补图成功: product_id={product.id} merchant_id={product.merchant_id} name={product.name}'
+                )
+            )
+
+        self.stdout.write(
+            self.style.SUCCESS(
+                f'完成: total={len(products)} success={success_count} failed={failed_count} skipped={skipped_count}'
+            )
+        )

api_v1/tasks.py

@@ -500,12 +500,13 @@ def _build_external_order_data(
     return {
         'merchant': merchant,
         'customer': customer,
-        'fabric': '',
+        'fabric': str(first.get('HpName') or '').strip(),
         '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,
+        'position': str(first.get('FidJ') or '').strip() or None,
         'outgoing_date': _parse_external_datetime(first.get('KdRiQi')),
         'created_by': created_by_user,
         'external_order_id': external_order_id,
@@ -547,6 +548,7 @@ def _upsert_external_printing_order(*, merchant, external_order_id: str, group_r
 
 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)
+    external_product_name = str(record.get('YanSe') or '').strip()
     created_by_user, _external_employee_name = _resolve_external_employee_user(
         merchant=merchant,
         external_employee_name=str(record.get('CaoZY') or '').strip(),
@@ -554,7 +556,7 @@ def _upsert_external_printing_job(*, merchant, printing_order, record: dict, syn
     )
     product, _ = _ensure_external_product(
         merchant=merchant,
-        external_product_name=str(record.get('YanSe') or '').strip(),
+        external_product_name=external_product_name,
         category=category,
     )
     job_data = {
@@ -574,7 +576,7 @@ def _upsert_external_printing_job(*, merchant, printing_order, record: dict, syn
     existing_job = (
         printing_models.PrintingJob.objects.filter(
             printing_order=printing_order,
-            original_id=record_id,
+            product=product,
         )
         .order_by('id')
         .first()

api_v1/test_external_printing_records_sync.py

@@ -71,6 +71,7 @@ class ExternalPrintingRecordsSyncTaskTest(TestCase):
             'BianHaoKD': '1.27',
             'CaoZY': '钰涵',
             'FidJ': r'\\fw\\2026年-LWQ15\\2026\\H鸿烨\\Tj1712#',
+            'HpName': '120克本白四面弹单定',
             'ID': record_id,
             'JiJiaDW': '/段',
             'KdRiQi': '2026-01-26T20:00:52Z',
@@ -153,6 +154,36 @@ class ExternalPrintingRecordsSyncTaskTest(TestCase):
 
         job = printing_models.PrintingJob.objects.get(original_id=1000001)
         self.assertEqual(job.product, product)
+        self.assertEqual(job.printing_order.position, r'\\fw\\2026年-LWQ15\\2026\\H鸿烨\\Tj1712#')
+        self.assertEqual(job.printing_order.fabric, '120克本白四面弹单定')
+
+    @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_order_position_uses_first_record_fidj(
+        self,
+        mock_fetch_records,
+        mock_fetch_image,
+        mock_advance_cursor,
+    ):
+        first_record = self._build_record(record_id=1000000, product_name=self.existing_product.name)
+        second_record = self._build_record(record_id=1000001, product_name=self.existing_product.name)
+        first_record['FidJ'] = r'\\fw\\path\\first'
+        second_record['FidJ'] = r'\\fw\\path\\second'
+        mock_fetch_records.return_value = self._build_payload([first_record, second_record])
+        mock_advance_cursor.return_value = {'updated': True}
+
+        result = sync_external_printing_records.run(limit=100)
+
+        self.assertEqual(result['orders_created'], 1)
+        self.assertEqual(result['jobs_created'], 1)
+        self.assertEqual(result['jobs_updated'], 1)
+        self.assertEqual(mock_fetch_image.call_count, 0)
+
+        order = printing_models.PrintingOrder.objects.get(external_order_id='KD20432358')
+        self.assertEqual(order.position, r'\\fw\\path\\first')
+        self.assertEqual(order.fabric, '120克本白四面弹单定')
+        self.assertEqual(printing_models.PrintingJob.objects.count(), 1)
 
     @patch('api_v1.tasks._advance_external_printing_cursor')
     @patch('api_v1.tasks._fetch_external_product_image')
@@ -192,3 +223,40 @@ 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_same_order_and_product_overwrites_existing_job_instead_of_creating_new_one(
+        self,
+        mock_fetch_records,
+        mock_fetch_image,
+        mock_advance_cursor,
+    ):
+        initial_record = self._build_record(record_id=1000000, product_name=self.existing_product.name)
+        updated_record = self._build_record(record_id=1000001, product_name=self.existing_product.name)
+        updated_record['ShuLiang'] = '20.00'
+        updated_record['ShuLiangZ'] = '3.15'
+        updated_record['BeiZhuC'] = '20件'
+        mock_fetch_image.return_value = {'bytes': b'', 'content_type': 'image/jpeg', 'name': 'unused.jpg'}
+        mock_advance_cursor.return_value = {'updated': True}
+
+        mock_fetch_records.return_value = self._build_payload([initial_record])
+        first_result = sync_external_printing_records.run(limit=100)
+
+        mock_fetch_records.return_value = self._build_payload([updated_record])
+        second_result = sync_external_printing_records.run(limit=100)
+
+        self.assertEqual(first_result['jobs_created'], 1)
+        self.assertEqual(second_result['jobs_created'], 0)
+        self.assertEqual(second_result['jobs_updated'], 1)
+        self.assertEqual(mock_fetch_image.call_count, 0)
+
+        jobs = printing_models.PrintingJob.objects.all()
+        self.assertEqual(jobs.count(), 1)
+        job = jobs.get()
+        self.assertEqual(job.original_id, 1000001)
+        self.assertEqual(job.quantity, 20)
+        self.assertEqual(job.size, '3.15')
+        self.assertEqual(job.pieces, 20)
+        self.assertEqual(job.printing_order.fabric, '120克本白四面弹单定')

docs/printing_external_records_sync_2026-03-20.md

@@ -26,7 +26,7 @@
 
 1. 按 `BianHaoID` 对本次拉到的 records 分组
 2. 每个分组 upsert 一条 `PrintingOrder`
-3. 分组内每条 record 按 `ID` upsert 一条 `PrintingJob`
+3. 分组内每条 record 按 `BianHaoID + YanSe` 覆盖 upsert 一条 `PrintingJob`
 4. 仅当本地全部写入成功后，才推进外部 cursor
 
 ## 字段映射
@@ -39,25 +39,27 @@
 | `customer.KhName` | 客户名 | `PrintingOrder.customer` | 按本商户 `Customer.name` 精确匹配，取第一条 |
 | `KhID` | 外部客户 ID | `PrintingOrder.external_customer_id` | 原样保存 |
 | `customer.KhName` | 外部客户名 | `PrintingOrder.external_customer_name` | 原样保存 |
+| `HpName` | 布料名 | `PrintingOrder.fabric` | 原样保存到订单面料字段 |
 | `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` | 原样保存 |
+| `FidJ` | 电脑位置 | `PrintingOrder.position` | 取分组首条 record 的原始值 |
 | `BeiZhu` | 滚筒注意事项 | `PrintingOrder.rolling_warn` | 原样保存 |
 | `KdRiQi` | 下单日期 | `PrintingOrder.outgoing_date` | 按 ISO 时间解析并保存 |
 | 分组首条原始数据 | 外部来源快照 | `PrintingOrder.external_raw` | 保存 order 级别的原始数据，便于追溯 |
 
 补充说明：
 
-- 外部没有可靠的 “面料” 字段，因此 `PrintingOrder.fabric` 本次同步固定写空字符串 `""`
+- 外部 `HpName` 已作为布料名使用，直接写入 `PrintingOrder.fabric`
-- `BianHaoKD`、`RiQi`、`FidJ` 等当前未单独属性化的字段，保留在 `external_raw` 中
+- `BianHaoKD`、`RiQi` 等当前未单独属性化的字段，保留在 `external_raw` 中
 - `BianHaoKD` 当前样例值类似 `"1.27"`，不按日期强解析
 
 ### 二、单条 record -> PrintingJob
 
 | 外部字段 | 含义 | 本系统字段 | 规则 |
 | --- | --- | --- | --- |
-| `ID` | 明细顺序 ID | `PrintingJob.original_id` | 幂等键；原样保存 |
+| `ID` | 明细顺序 ID | `PrintingJob.original_id` | 保存最近一次覆盖该任务的外部明细 ID，便于追溯 |
 | `YanSe` | 外部产品名 | `PrintingJob.product` / `PrintingJob.external_product_name` | 若能按本商户 `Product.name` 精确匹配，直接绑定该产品且不再拉图；若本地不存在，则先创建产品、拉图并上传到 storage，成功后绑定该产品 |
 | `ShuLiang` | 下单数量 | `PrintingJob.quantity` | 按整数解析；当前预期外部值为整数字符串或 `10.00` 这类可安全转整数的值 |
 | `JiJiaDW` | 单位 | `PrintingJob.unit` | 去掉前导 `/` 后保存，例如 `/段` -> `段` |
@@ -66,6 +68,12 @@
 | `YanSe` | 外部产品名 | `PrintingJob.description` | 本次不单独映射到 description；原始值由 `product.name` 或 `external_product_name` / `external_raw` 表达 |
 | 单条原始数据 | 外部来源快照 | `PrintingJob.external_raw` | 保存 job 级别完整原始 record |
 
+补充说明：
+
+- `PrintingJob` 的覆盖唯一性按 `PrintingOrder.external_order_id + YanSe` 处理
+- 若同一外部订单下再次出现相同 `YanSe`，则更新已有 `PrintingJob`，不再新增
+- 更新时 `original_id` 会覆盖为最新外部 record 的 `ID`
+
 ## 员工绑定规则
 
 1. 读取外部 `CaoZY`

printing/admin.py

@@ -360,7 +360,12 @@ class PrintingJobAdmin(admin.ModelAdmin):
         'execution_status',
         'created_at',
     )
-    search_fields = ('printing_order__human_id', 'product__name')
+    search_fields = (
+        '=printing_order__id',
+        'printing_order__external_order_id',
+        'product__name',
+        'external_product_name',
+    )
     list_filter = ('created_at', 'printing_order__customer__name')
     actions = ['advance_to_next_state_action', 'step_back_one_state_action', 'reset_progress_action']
 

printing/tests.py

@@ -1,3 +1,71 @@
-from django.test import TestCase
+from django.contrib.admin.sites import AdminSite
+from django.contrib.auth import get_user_model
+from django.test import RequestFactory, TestCase
 
-# Create your tests here.
+from basic_info import models as basic_models
+from printing import admin as printing_admin
+from printing import models as printing_models
+from stateflow import models as stateflow_models
+
+User = get_user_model()
+
+
+class PrintingJobAdminSearchTestCase(TestCase):
+    def setUp(self):
+        self.site = AdminSite()
+        self.factory = RequestFactory()
+        self.admin_user = User.objects.create_superuser(
+            username='admin',
+            email='admin@example.com',
+            password='testpass123',
+        )
+
+        self.merchant = basic_models.Merchant.objects.create(
+            name='测试印花厂',
+            type=basic_models.MerchantTypeEnum.FACTORY,
+        )
+        self.customer = basic_models.Customer.objects.create(
+            merchant=self.merchant,
+            name='测试客户',
+        )
+        self.category = basic_models.ProductCategory.objects.create(
+            merchant=self.merchant,
+            name='测试分类',
+        )
+        self.product = basic_models.Product.objects.create(
+            merchant=self.merchant,
+            category=self.category,
+            name='测试产品',
+            human_id='TEST001',
+            unit=basic_models.ProductUnitEnum.METER,
+        )
+        self.process = stateflow_models.Process.objects.create(name='测试流程')
+        self.order = printing_models.PrintingOrder.objects.create(
+            customer=self.customer,
+            fabric='测试布料',
+            width='150cm',
+            process=self.process,
+            external_order_id='KD20432371',
+        )
+        self.job = printing_models.PrintingJob.objects.create(
+            printing_order=self.order,
+            product=self.product,
+            quantity=10,
+            unit='米',
+            size='10x10',
+            pieces=1,
+        )
+
+    def test_search_by_external_order_id_does_not_raise_and_returns_job(self):
+        model_admin = printing_admin.PrintingJobAdmin(printing_models.PrintingJob, self.site)
+        request = self.factory.get('/admin/printing/printingjob/', {'q': 'KD20432371'})
+        request.user = self.admin_user
+
+        queryset, may_have_duplicates = model_admin.get_search_results(
+            request,
+            printing_models.PrintingJob.objects.all(),
+            'KD20432371',
+        )
+
+        self.assertFalse(may_have_duplicates)
+        self.assertEqual(list(queryset), [self.job])