fix(pl): 客户品类规则预占不再依赖合同绑证 scope。

新规则自动选证后账本仍按合同绑证判定适用，导致预占报人工分配不适用；改为客户品类路径跳过 scope，明细合同绑证字段可空，并统一关联购买证文案。

backend/app/models/pl_quota_models.py

@@ -199,9 +199,9 @@ class PlPlanCertUsageItem(Base):
     __tablename__ = "pl_plan_cert_usage_item"
     id = Column(String(64), primary_key=True, comment='主键UUID')
     usage_id = Column(String(64), nullable=False, comment='pl_plan_cert_usage.id')
-    contract_cert_link_id = Column(String(64), nullable=False, comment='合同与证件绑定ID，须与usage的证件相同')
-    contract_cert_product_id = Column(String(64), nullable=False, comment='本产品的适用范围明细ID：pl_contract_cert_product.id')
-    contract_scope_revision = Column(Integer, nullable=False, comment='本次确认的合同证件适用范围版本')
+    contract_cert_link_id = Column(String(64), nullable=True, comment='合同与证件绑定ID；客户品类规则可空')
+    contract_cert_product_id = Column(String(64), nullable=True, comment='合同证件产品范围明细ID；客户品类规则可空')
+    contract_scope_revision = Column(Integer, nullable=False, default=0, comment='合同证件适用范围版本；客户品类规则为0')
     contract_id = Column(String(64), nullable=False, comment='本次履行的山峡合同ID')
     contract_product_id = Column(String(64), nullable=False, comment='对应合同产品行ID')
     plan_product_id = Column(String(64), nullable=False, comment='计划产品行ID')

backend/app/services/pl_quota_cert_bindings.py

@@ -129,7 +129,7 @@ def save_cert(db, user, cert_type, data, cert_id=None):
         raise HTTPException(409, '证件合同范围已变更，请重试')
     if cert_type == 'transport' and any(('purchase', pid) not in keys for pid in
             _reference_ids(db, user, effective_id, refs, current=True)):
-        raise HTTPException(409, '证面引用已变更，请重试')
+        raise HTTPException(409, '关联购买证已变更，请重试')
     prepared_face = certs.prepare_cert(db, user, cert_type, face, cert_id,
                                       locked=locked, effective_id=effective_id)
     # 用最终证面状态校验作废证不可扩大，不修改持久对象。

backend/app/services/pl_quota_cert_service.py

@@ -211,13 +211,13 @@ def _validate_refs(db, user, cert_id, submitted, historical, purchases, current=
         if purchase_cert_id:
             purchase = purchases.get(purchase_cert_id)
             if purchase is None:
-                raise HTTPException(409, '证面引用已变更，请刷新后重新保存')
+                raise HTTPException(409, '关联购买证已变更，请刷新后重新保存')
         if historical and _historical_ref_locked(previous, item):
-            raise HTTPException(400, '已有历史用量，不能改变证面引用身份，请联系管理员')
+            raise HTTPException(400, '已有历史用量，不能改变关联购买证身份，请联系管理员')
         pending.append((previous, item))
     removed = [r for rid,r in old.items() if rid not in ids]
     if historical and removed:
-        raise HTTPException(400, '已有历史用量，不能移除证面引用，请联系管理员')
+        raise HTTPException(400, '已有历史用量，不能移除关联购买证，请联系管理员')
     return pending, removed
 
 
@@ -241,7 +241,7 @@ def prepare_cert(db, user, cert_type, data, cert_id=None, *, locked=None, effect
         purchases = {pid: (locked.get(('purchase', pid)) if locked is not None else
                           _get_row(db, user, 'purchase', pid, lock=True)) for pid in sorted(purchase_ids)}
         if any(row is None for row in purchases.values()):
-            raise HTTPException(409, '证面引用已变更，请重试')
+            raise HTTPException(409, '关联购买证已变更，请重试')
     row = (locked[(cert_type, cert_id)] if locked is not None else
            _get_row(db, user, cert_type, cert_id, lock=True)) if cert_id else None
     historical = bool(row and _has_history(db,user,cert_type,cert_id))

backend/app/services/pl_quota_diagnostics.py

@@ -89,7 +89,7 @@ def missing_purchase_link_message(demands, purchases, transports, cert_nos):
         seen.add(key)
         notices.append(
             f'运输证 {move_text} 没有引用合同正在使用的购买证 {buy_text}。'
-            '请先打开该运输证，在「证面引用」的「对应购买证档案」中选择这张购买证，再安排运输'
+            '请先打开该运输证，在「关联购买证」中选择这张购买证档案，再安排运输'
         )
     return '；'.join(notices)
 

backend/app/services/pl_quota_usage_service.py

@@ -198,6 +198,8 @@ def _candidate_context(db,user,r,plan,certs,usages,items,selected,preview=False,
         if product is None or product.product_id!=s.product_id: raise ValueError('合同产品归属不符')
     scopes={}; candidates={'purchase':[],'transport':[]}
     skipped_for_vehicle=False
+    from app.services.pl_cert_quota_rule_config_service import plan_uses_customer_category
+    customer_rule = plan_uses_customer_category(plan)
     for k in sorted(selected):
         kind,pid=k; c=certs[k]
         if not query(m.PlCertQtyLog,cert_type=kind,cert_id=pid,event_type='opening',del_flag=0).first(): raise ValueError('证件尚未确认期初')
@@ -216,11 +218,17 @@ def _candidate_context(db,user,r,plan,certs,usages,items,selected,preview=False,
             if c.opening_single_use_consumed or any(u.plan_id!=plan.id and (u.usage_status in ('reserved','in_transit','settled','exception') or (u.confirmation_snapshot or {}).get('executed')) for u in previous):
                 raise ValueError('一次有效证件已使用或已占用')
         eligible=set()
-        for sid,s in sources.items():
-            link=query(m.PlContractCertLink,cert_type=kind,cert_id=pid,contract_id=s.contract_id,del_flag=0,link_status='active').first()
-            scope=None if link is None else query(m.PlContractCertProduct,contract_cert_link_id=link.id,contract_product_id=s.contract_product_id,del_flag=0,scope_status='active').first()
-            if scope:
-                eligible.add(sid); scopes[(k,sid)]=(link,scope)
+        if customer_rule:
+            # 新规则已由客户+品类整笔选证；账本不再依赖合同绑证 scope。
+            eligible=set(sources)
+            for sid in eligible:
+                scopes[(k,sid)]=(None,None)
+        else:
+            for sid,s in sources.items():
+                link=query(m.PlContractCertLink,cert_type=kind,cert_id=pid,contract_id=s.contract_id,del_flag=0,link_status='active').first()
+                scope=None if link is None else query(m.PlContractCertProduct,contract_cert_link_id=link.id,contract_product_id=s.contract_product_id,del_flag=0,scope_status='active').first()
+                if scope:
+                    eligible.add(sid); scopes[(k,sid)]=(link,scope)
         refs=frozenset(x.purchase_cert_id for x in query(m.PlTransportCertPurchaseRef,transport_cert_id=pid,del_flag=0).all() if x.purchase_cert_id) if kind=='transport' else frozenset()
         own=usages[k].reserved_qty if k in usages else ZERO
         candidates[kind].append(PermitCandidate(pid,c.total_qty-c.used_qty-c.reserved_qty+own,c.quantity_unit,c.valid_to,frozenset(eligible),refs))
@@ -279,8 +287,10 @@ def _reserve(db,user,r,plan,certs,usages,items,selected,preview=False,lock_previ
                 'sourceId':a.source_id,'certId':a.permit_id,'sourceQuantity':_canonical(a.source_quantity),
                 'permitQuantity':_canonical(a.permit_quantity),'sourceUnit':a.source_unit,
                 'permitUnit':a.permit_unit,'allocationSeq':a.allocation_seq,
-                'contractCertLinkId':link.id,'contractCertProductId':scope.id,
-                'scopeRevision':link.scope_revision,'certRevision':certs[k].revision,
+                'contractCertLinkId':None if link is None else link.id,
+                'contractCertProductId':None if scope is None else scope.id,
+                'scopeRevision':0 if link is None else link.scope_revision,
+                'certRevision':certs[k].revision,
                 'ownReservedQty':_canonical(usages[k].reserved_qty if k in usages else ZERO),
             })
         return result
@@ -309,7 +319,27 @@ def _reserve(db,user,r,plan,certs,usages,items,selected,preview=False,lock_previ
             item=previous.get(a.source_id)
             if item is None:
                 item=m.PlPlanCertUsageItem(**_audit(user),usage_id=u.id); db.add(item); items[k].append(item)
-            values=dict(contract_cert_link_id=link.id,contract_cert_product_id=scope.id,contract_scope_revision=link.scope_revision,contract_id=s.contract_id,contract_product_id=s.contract_product_id,plan_product_id=s.plan_product_id,source_stop_type=s.source_stop_type,source_stop_id=s.source_stop_id,source_stop_product_id=s.source_stop_product_id,related_stop_products=s.related_stop_products,allocation_seq=a.allocation_seq,source_unit=s.unit,planned_source_qty=s.quantity,allocated_source_qty=a.source_quantity,planned_cert_qty=a.permit_quantity,reserved_qty=a.permit_quantity,used_qty=ZERO,item_status='reserved',source_fingerprint=s.fingerprint,source_snapshot={**s.snapshot, 'measurementVersion': s.snapshot.get('measurementVersion', 'legacy_single_side'), 'productId': s.product_id},del_flag=0,reserve_input_mode=_reserve_input_mode(s.unit, c.quantity_unit),reserve_conversion_note=notes.get((k[0],k[1],a.source_id)),reserve_confirmed_by=user.id,reserve_confirmed_time=datetime.now(),update_by=user.id)
+            values=dict(
+                contract_cert_link_id=None if link is None else link.id,
+                contract_cert_product_id=None if scope is None else scope.id,
+                contract_scope_revision=0 if link is None else link.scope_revision,
+                contract_id=s.contract_id,contract_product_id=s.contract_product_id,plan_product_id=s.plan_product_id,
+                source_stop_type=s.source_stop_type,source_stop_id=s.source_stop_id,source_stop_product_id=s.source_stop_product_id,
+                related_stop_products=s.related_stop_products,allocation_seq=a.allocation_seq,source_unit=s.unit,
+                planned_source_qty=s.quantity,allocated_source_qty=a.source_quantity,planned_cert_qty=a.permit_quantity,
+                reserved_qty=a.permit_quantity,used_qty=ZERO,item_status='reserved',source_fingerprint=s.fingerprint,
+                source_snapshot={
+                    **s.snapshot,
+                    'measurementVersion': s.snapshot.get('measurementVersion', 'legacy_single_side'),
+                    'productId': s.product_id,
+                    'quotaRuleVersion': getattr(plan, 'cert_quota_rule_version', None),
+                    'customerId': (s.snapshot or {}).get('customerId'),
+                    'customerName': (s.snapshot or {}).get('customerName'),
+                },
+                del_flag=0,reserve_input_mode=_reserve_input_mode(s.unit, c.quantity_unit),
+                reserve_conversion_note=notes.get((k[0],k[1],a.source_id)),
+                reserve_confirmed_by=user.id,reserve_confirmed_time=datetime.now(),update_by=user.id,
+            )
             for name,value in values.items(): setattr(item,name,value)
 
 

backend/tests/test_pl_customer_cert_quota.py

@@ -381,3 +381,85 @@ def test_same_city_source_skips_transport_requirement():
     assert purchase_ids == ['buy1']
     assert transport_ids == []
     assert manuals[0].permit_id == 'buy1'
+
+
+def test_candidate_context_customer_rule_skips_contract_scope(monkeypatch):
+    """新规则账本候选不以合同绑证 scope 判定适用，否则预占会误报人工分配不适用。"""
+    from datetime import date
+    from decimal import Decimal
+    from types import SimpleNamespace
+    from app.services import pl_quota_usage_service as usage
+    from app.schemas.pl.quota_usage import UsageSource
+
+    plan = SimpleNamespace(id='plan', cert_quota_rule_version='customer_category', plan_no='P1')
+    source = UsageSource(
+        source_stop_type='delivery',
+        source_stop_id='s1',
+        source_stop_product_id='sp1',
+        plan_product_id='pp1',
+        contract_id='c1',
+        contract_product_id='cp1',
+        product_id='prod1',
+        quantity=Decimal('50'),
+        unit='dun',
+        fingerprint='f' * 64,
+        snapshot={'customerId': 'cust1'},
+        requires_transport=True,
+    )
+    request = SimpleNamespace(
+        sources=[source],
+        on_date=date(2026, 11, 15),
+        vehicle_no='皖A9TEST1',
+        manual_allocations=[],
+    )
+    cert = SimpleNamespace(
+        id='buy1', status='active', verified_time=date(2026, 11, 1), verified_by='u',
+        valid_from=date(2026, 11, 1), valid_to=date(2027, 1, 1), valid_use_text='多次有效',
+        total_qty=Decimal('100'), used_qty=Decimal('0'), reserved_qty=Decimal('0'),
+        quantity_unit='dun', vehicle_numbers_json=['皖A9TEST1'], opening_single_use_consumed=0,
+        revision=1, cert_no='G1',
+    )
+
+    class Q:
+        def __init__(self, row=None, rows=None):
+            self._row = row
+            self._rows = rows or ([] if row is None else [row])
+
+        def first(self):
+            return self._row
+
+        def all(self):
+            return self._rows
+
+    def fake_query(db, user, model, lock=False, **filters):
+        name = getattr(model, '__name__', '')
+        if name == 'PlCertQtyLog':
+            return Q(SimpleNamespace(id='opening'))
+        if name == 'PlTransportCertPurchaseRef':
+            return Q(rows=[])
+        if name in ('PlContractCertLink', 'PlContractCertProduct'):
+            raise AssertionError('customer_category 不应查询合同绑证 scope')
+        return Q()
+
+    monkeypatch.setattr(usage, '_query', fake_query)
+    monkeypatch.setattr(usage, '_contract', lambda *a, **k: SimpleNamespace(status='executing', contract_type='sale'))
+    monkeypatch.setattr(
+        'app.services.pl_cert_quota_rule_config_service.plan_uses_customer_category',
+        lambda p: True,
+    )
+    # product ownership check uses _query for CrmContractProduct
+    def fake_query2(db, user, model, lock=False, **filters):
+        name = getattr(model, '__name__', '')
+        if name == 'CrmContractProduct':
+            return Q(SimpleNamespace(product_id='prod1'))
+        return fake_query(db, user, model, lock=lock, **filters)
+
+    monkeypatch.setattr(usage, '_query', fake_query2)
+
+    sources, scopes, candidates = usage._candidate_context(
+        MagicMock(), SimpleNamespace(tenant_no='t', id='u'), request, plan,
+        {('purchase', 'buy1'): cert}, {}, {}, {('purchase', 'buy1')},
+    )
+    assert 'delivery:sp1' in sources
+    assert scopes[(('purchase', 'buy1'), 'delivery:sp1')] == (None, None)
+    assert candidates['purchase'][0].product_ids == frozenset({'delivery:sp1'})

backend/tests/test_pl_quota_api.py

@@ -97,7 +97,7 @@ def test_plan_overview_accepts_web_funnel_aliases_and_validates_dates(api, monke
     assert filters == {'plan_no': '001', 'license_plate': '皖A',
         'plan_date_from': date(2026, 11, 1), 'plan_date_to': date(2026, 11, 3),
         'quota_status_in': 'reserved,settled', 'supplier_name': None, 'customer_name': None,
-        'product_name': None, 'follow_status_in': None}
+        'product_name': None, 'follow_status_in': None, 'conversion_only': False}
     assert client.get('/api/transport/quota-certs/plans', params={'planDateFrom': 'wrong'}).status_code == 422
     assert received == []
 

backend/tests/test_pl_quota_business.py

@@ -251,6 +251,11 @@ def approval_case(business, monkeypatch):
     c.db.commit()
     c.user.to_tenant_user_dict=lambda:{}
     monkeypatch.setattr(api,'get_tenant_no',lambda *a:'t')
+    # 安排入口要求已配置客户品类额度生效时间；本套用例聚焦共享运输/钉版本，不测切点本身
+    monkeypatch.setattr(
+        'app.services.pl_cert_quota_rule_config_service.assert_cert_quota_rule_configured',
+        lambda *a, **k: None,
+    )
     c.api=api
     return c
 

backend/tests/test_pl_quota_cert_archive.py

@@ -314,11 +314,11 @@ def test_unlinked_reference_can_bind_purchase_archive_after_opening(ctx):
     assert saved['purchaseRefs'][0]['purchaseCertId']==purchase_row['id']
     assert saved['purchaseRefs'][0]['referenceNo']=='G35260264950502'
     linked=saved['purchaseRefs'][0]
-    with pytest.raises(HTTPException, match='不能改变证面引用身份'):
+    with pytest.raises(HTTPException, match='不能改变关联购买证身份'):
         s.save_cert(db,u,'transport',{'purchaseRefs':[{'id':linked['id'],'referenceType':'unknown','referenceNo':purchase_row['certNo'],'purchaseCertId':purchase_row['id']}]},row['id'])
     with pytest.raises(HTTPException, match='不能移除'):
         s.save_cert(db,u,'transport',{'purchaseRefs':[]},row['id'])
-    with pytest.raises(HTTPException, match='不能改变证面引用身份'):
+    with pytest.raises(HTTPException, match='不能改变关联购买证身份'):
         s.save_cert(db,u,'transport',{'purchaseRefs':[
             {k:linked[k] for k in ('id','referenceType','referenceNo','purchaseCertId')},
             {'referenceType':'unknown','referenceNo':'NEW'},

docs/sql/2026-11-01-pl-contract-cert-quota.sql

@@ -29,6 +29,8 @@ CREATE TABLE `pl_purchase_cert` (
   `seller_legal_representative` varchar(100) DEFAULT NULL COMMENT '销售方法定代表人',
   `seller_phone` varchar(100) DEFAULT NULL COMMENT '销售方电话',
   `purchase_purpose` varchar(1000) DEFAULT NULL COMMENT '用途',
+  `customer_id` varchar(64) DEFAULT NULL COMMENT '系统客户ID；选证必填',
+  `category_id` varchar(64) DEFAULT NULL COMMENT '证用品类ID；选证必填',
   `tenant_no` varchar(64) NOT NULL COMMENT '租户编号，由服务端当前用户上下文设置',
   `order_num` int DEFAULT NULL COMMENT '公共排序号',
   `remark` varchar(1000) DEFAULT NULL COMMENT '系统备注，与证面备注分开',
@@ -74,6 +76,7 @@ CREATE TABLE `pl_transport_cert` (
   `route_text` text COMMENT '运输路线完整原文',
   `consignee_name` varchar(255) DEFAULT NULL COMMENT '收货单位名称或姓名',
   `applicant_name` varchar(255) DEFAULT NULL COMMENT '申请单位名称或姓名',
+  `customer_id` varchar(64) DEFAULT NULL COMMENT '系统客户ID；选证必填',
   `tenant_no` varchar(64) NOT NULL COMMENT '租户编号，由服务端当前用户上下文设置',
   `order_num` int DEFAULT NULL COMMENT '公共排序号',
   `remark` varchar(1000) DEFAULT NULL COMMENT '系统备注，与证面备注分开',

docs/sql/2026-11-15-pl-plan-cert-usage-item-nullable-contract-scope.sql

@@ -0,0 +1,7 @@
+-- 客户+品类规则下用证明细不再依赖合同绑证 scope；将合同绑定字段改为可空。
+SET NAMES utf8mb4;
+
+ALTER TABLE `pl_plan_cert_usage_item`
+  MODIFY COLUMN `contract_cert_link_id` varchar(64) DEFAULT NULL COMMENT '合同与证件绑定ID；客户品类规则可空',
+  MODIFY COLUMN `contract_cert_product_id` varchar(64) DEFAULT NULL COMMENT '合同证件产品范围明细ID；客户品类规则可空',
+  MODIFY COLUMN `contract_scope_revision` int(11) NOT NULL DEFAULT 0 COMMENT '合同证件适用范围版本；客户品类规则为0';

docs/sql/migrations/2026-11-15-pl-plan-cert-usage-item-nullable-contract-scope.sql

@@ -0,0 +1,7 @@
+-- 客户+品类规则下用证明细不再依赖合同绑证 scope；将合同绑定字段改为可空。
+SET NAMES utf8mb4;
+
+ALTER TABLE `pl_plan_cert_usage_item`
+  MODIFY COLUMN `contract_cert_link_id` varchar(64) DEFAULT NULL COMMENT '合同与证件绑定ID；客户品类规则可空',
+  MODIFY COLUMN `contract_cert_product_id` varchar(64) DEFAULT NULL COMMENT '合同证件产品范围明细ID；客户品类规则可空',
+  MODIFY COLUMN `contract_scope_revision` int(11) NOT NULL DEFAULT 0 COMMENT '合同证件适用范围版本；客户品类规则为0';

web-admin/src/views/m/pl/plan/PlanStopReadCard.vue

@@ -19,6 +19,10 @@
         <div
           ><small>联系方式</small><span>{{ stop.contact || '—' }}</span></div
         >
+        <div
+          ><small>是否同市</small
+          ><span>{{ Number(stop.sameCity) === 1 ? '是（同市只需购买证）' : '否' }}</span></div
+        >
         <div class="wide"
           ><small>{{ title }}地址</small><span>{{ displayAddress || '—' }}</span></div
         >