fix(pl): 安排运输购买证失败提示补全客户名与产品名。

无匹配购买证时点名异地客户与产品，便于对照建档；来源 snapshot 补充展示用产品名且不参与 fingerprint。

Co-authored-by: Cursor <cursoragent@cursor.com>

backend/app/services/pl_quota_customer_select.py

@@ -88,6 +88,48 @@ def _customer_id_from_source(source) -> str:
     raise HTTPException(409, '用证段次缺少客户，无法按客户选证')
 
 
+def _snap_text(source, *keys) -> str:
+    snap = source.snapshot or {}
+    for key in keys:
+        value = snap.get(key)
+        if value is not None and str(value).strip():
+            return str(value).strip()
+    return ''
+
+
+def _source_customer_label(source) -> str:
+    return _snap_text(
+        source, 'customerName', 'customer_name', 'customerNo', 'customer_no'
+    ) or _customer_id_from_source(source)
+
+
+def _source_product_label(db, user, source) -> str:
+    name = _snap_text(source, 'productName', 'product_name')
+    if name:
+        return name
+    product_id = getattr(source, 'product_id', None)
+    if not product_id:
+        return ''
+    try:
+        from app.models.crm_models import CrmProduct
+
+        row = (
+            db.query(CrmProduct)
+            .filter(
+                CrmProduct.id == product_id,
+                CrmProduct.tenant_no == user.tenant_no,
+                CrmProduct.del_flag == 0,
+            )
+            .first()
+        )
+    except Exception:
+        return ''
+    product_name = getattr(row, 'product_name', None) if row is not None else None
+    if isinstance(product_name, str) and product_name.strip():
+        return product_name.strip()
+    return str(product_id)
+
+
 def _purchase_candidates(db, user, customer_id, categories):
     # MySQL ASC 会把 NULL 排最前；无到期证应排在末尾，优先消耗更早到期的证（D5）
     return (
@@ -144,10 +186,13 @@ def _diagnose_select_failure(db, user, source, *, vehicle_no: str, on_date: date
     place = '同市' if not source.requires_transport else '异地'
     plate = (vehicle_no or '').strip()
     need_qty = Decimal(source.quantity)
+    customer_label = _source_customer_label(source)
+    product_label = _source_product_label(db, user, source)
 
     if not candidates:
+        product_part = f'产品「{product_label}」' if product_label else '该产品'
         return (
-            f'{place}段客户下没有匹配品类的有效购买证，'
+            f'{place}段客户「{customer_label}」下没有{product_part}匹配品类的有效购买证，'
             '请先建档购买证或核对客户/证用品类后再安排'
         )
 
@@ -345,9 +390,10 @@ def select_whole_cover_certs(db, user, sources, *, vehicle_no: str, on_date: dat
         customer_id = _customer_id_from_source(source)
         categories = category_ids_for_product(db, user.tenant_no, source.product_id)
         if not categories:
+            product_label = _source_product_label(db, user, source) or '该产品'
             raise HTTPException(
                 409,
-                '产品未配置证用品类，请先在证用品类中关联系统产品后再安排运输',
+                f'产品「{product_label}」未配置证用品类，请先在证用品类中关联系统产品后再安排运输',
             )
 
         candidates = _purchase_candidates(db, user, customer_id, categories)
@@ -471,9 +517,10 @@ def resolve_customer_category_cover_certs(db, user, sources):
         )
         categories = category_ids_for_product(db, user.tenant_no, source.product_id)
         if not categories:
+            product_label = _source_product_label(db, user, source) or '该产品'
             raise HTTPException(
                 409,
-                '产品未配置证用品类，请先在证用品类中关联系统产品后再归档',
+                f'产品「{product_label}」未配置证用品类，请先在证用品类中关联系统产品后再归档',
             )
         purchases = _purchase_candidates(db, user, customer_id, categories)
         needs_transport = getattr(source, 'requires_transport', True)

backend/app/services/pl_quota_sources.py

@@ -168,6 +168,10 @@ def extract_sources(db, user, plan_id, mapping=None, *, actual=False, lock=False
             snapshot['vehicle'] = preview_vehicle.get('licensePlate', plan.license_plate)
             snapshot['trailer'] = preview_vehicle.get('trailerNo', plan.trailer_no)
         fingerprint = hashlib.sha256(json.dumps(snapshot, sort_keys=True, ensure_ascii=False).encode()).hexdigest()
+        # 展示字段不参与 fingerprint，避免仅补产品名就触发来源指纹变化
+        product_name = getattr(row, 'product_name', None)
+        if product_name is not None and str(product_name).strip():
+            snapshot = {**snapshot, 'productName': str(product_name).strip()}
         source = UsageSource(source_stop_type=side, source_stop_id=stop.id, source_stop_product_id=row.id,
                              plan_product_id=row.plan_product_id, contract_id=row.contract_id,
                              contract_product_id=row.contract_product_id, product_id=row.product_id,

backend/tests/test_pl_customer_cert_quota.py

@@ -945,6 +945,37 @@ def test_whole_cover_purchase_insufficient_message_is_precise(monkeypatch):
     assert '余量足够、有效期内且匹配品类/车号' not in detail
 
 
+def test_whole_cover_no_purchase_message_includes_customer_and_product(monkeypatch):
+    """无匹配购买证时，提示须点名异地客户与产品，便于对照建档。"""
+    from app.services import pl_quota_customer_select as sel
+
+    db = _mock_select_db([], [], [])
+    monkeypatch.setattr(sel, 'category_ids_for_product', lambda *a, **k: {'cat1'})
+
+    with pytest.raises(HTTPException) as exc:
+        sel.select_whole_cover_certs(
+            db,
+            SimpleNamespace(tenant_no='t'),
+            [
+                _remote_source(
+                    snapshot={
+                        'customerId': 'cust1',
+                        'customerName': '滁州康华电子材料有限公司',
+                        'productName': '98% 工业硫酸',
+                    }
+                )
+            ],
+            vehicle_no='皖A9BE91',
+            on_date=date(2026, 11, 14),
+        )
+    detail = str(exc.value.detail)
+    assert exc.value.status_code == 409
+    assert '异地段客户' in detail
+    assert '滁州康华电子材料有限公司' in detail
+    assert '98% 工业硫酸' in detail
+    assert '匹配品类的有效购买证' in detail
+
+
 def test_resolve_customer_category_cover_certs_lists_pool(monkeypatch):
     from app.services import pl_quota_customer_select as sel