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

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

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

backend/app/services/pl_quota_cert_bindings.py

@@ -155,4 +155,6 @@ def save_cert(db, user, cert_type, data, cert_id=None):
     for scope in prepared:
         scopes.apply_contract_bindings(db, user, scope)
     db.flush()
+    # 与无合同集合的 save_cert 一致：建档/保存后自动期初并启用；current=True 的读回不会再跑启用。
+    row = certs.ensure_zero_opening(db, user, cert_type, row.id, row=row)
     return get_cert(db, user, cert_type, row.id, current=True)

backend/app/services/pl_quota_cert_service.py

@@ -138,6 +138,8 @@ def _customer_display_name(db, user, customer_id):
 
 def get_cert(db, user, cert_type, cert_id, current=False):
     row = _get_row(db, user, cert_type, cert_id, lock=current)
+    # 只读列表/详情不得在未提交事务里改状态，否则会出现「页面已启用、库仍 draft」。
+    # 自动期初与启用仅在 save_cert / confirm 等写入路径执行。
     result = _row_to_dict(row)
     result['usagePolicy'] = usage_policy(row.valid_use_text)
     opening_query = db.query(PlCertQtyLog.id).filter_by(
@@ -151,7 +153,9 @@ def get_cert(db, user, cert_type, cert_id, current=False):
     result['customerName'] = _customer_display_name(db, user, getattr(row, 'customer_id', None))
     if cert_type == 'transport':
         result['purchaseRefs'] = [_row_to_dict(r) for r in _refs(db, user, cert_id, lock=current)]
-        result['referencesEditable'] = not _has_history(db, user, cert_type, cert_id, lock=current)
+        result['referencesEditable'] = not _has_usage_beyond_zero_opening(
+            db, user, cert_type, cert_id, lock=current
+        )
         result['derivedCategoryIds'] = _transport_derived_categories(db, user, cert_id)
     return result
 
@@ -224,8 +228,11 @@ def _refs_keep_verified(pending, removed):
 
 
 def _has_history(db, user, cert_type, cert_id, lock=True):
+    """任意额度流水或计划占用均视为已有历史（含自动期初），用于锁证号/核定量等身份。"""
     for model in (PlCertQtyLog, PlPlanCertUsage):
-        query = db.query(model.id).filter_by(tenant_no=user.tenant_no, cert_type=cert_type, cert_id=cert_id)
+        query = db.query(model.id).filter_by(
+            tenant_no=user.tenant_no, cert_type=cert_type, cert_id=cert_id, del_flag=0,
+        )
         if lock:
             query = query.with_for_update()
         if query.first() is not None:
@@ -233,6 +240,35 @@ def _has_history(db, user, cert_type, cert_id, lock=True):
     return False
 
 
+def _has_usage_beyond_zero_opening(db, user, cert_type, cert_id, lock=True):
+    """真实用量/占用才锁运输证引用；期初已用=0 的 opening 仍可补全关联购买证。"""
+    usage_q = db.query(PlPlanCertUsage.id).filter_by(
+        tenant_no=user.tenant_no, cert_type=cert_type, cert_id=cert_id, del_flag=0,
+    )
+    if lock:
+        usage_q = usage_q.with_for_update()
+    if usage_q.first() is not None:
+        return True
+    log_q = db.query(PlCertQtyLog).filter_by(
+        tenant_no=user.tenant_no, cert_type=cert_type, cert_id=cert_id, del_flag=0,
+    )
+    if lock:
+        log_q = log_q.with_for_update()
+    for log in log_q.all():
+        snap = log.event_snapshot or {}
+        # 仅建档自动期初（已用0）不锁引用；手工期初或其它流水仍锁。
+        if (
+            log.event_type == 'opening'
+            and snap.get('auto')
+            and (log.used_delta or 0) == 0
+            and (log.reserved_delta or 0) == 0
+            and (log.total_delta or 0) == 0
+        ):
+            continue
+        return True
+    return False
+
+
 def _validate_refs(db, user, cert_id, submitted, historical, purchases, current=False):
     if submitted is None:
         raise HTTPException(400, '引用集合不可为null；清空请提交[]')
@@ -337,7 +373,8 @@ def prepare_cert(db, user, cert_type, data, cert_id=None, *, locked=None, effect
     if duplicate and duplicate.id != cert_id:
         raise HTTPException(400, '证号已存在，删除后也不得重建额度池')
     effective_id = cert_id or effective_id or str(uuid4())
-    pending, removed = _validate_refs(db,user,effective_id,refs,historical,purchases,current=bool(cert_id)) if refs_present else ([],[])
+    refs_locked = bool(row and _has_usage_beyond_zero_opening(db, user, cert_type, cert_id))
+    pending, removed = _validate_refs(db,user,effective_id,refs,refs_locked,purchases,current=bool(cert_id)) if refs_present else ([],[])
     changed = {key for key, value in values.items() if row is None or getattr(row, key) != value}
     refs_changed = bool(removed) or any(previous is None or previous.del_flag == 1 or any(getattr(previous, key) != value for key, value in item.items() if key != 'id') for previous, item in pending)
     return dict(model=model, row=row, values=values, cert_id=cert_id, effective_id=effective_id,
@@ -384,19 +421,98 @@ def apply_cert(db, user, prepared):
     return row
 
 
+def ensure_zero_opening(db, user, cert_type, cert_id, *, row=None):
+    """建档后无需手工期初确认：缺 opening 流水时写入期初已用=0；draft 直接启用。"""
+    if row is None:
+        row = _get_row(db, user, cert_type, cert_id)
+    if row.status in ('void', 'closed'):
+        return row
+    existing = db.query(PlCertQtyLog.id).filter_by(
+        tenant_no=user.tenant_no, cert_type=cert_type, cert_id=cert_id, event_type='opening', del_flag=0,
+    ).first()
+    if existing:
+        return row
+    row = _get_row(db, user, cert_type, cert_id, lock=True)
+    if row.status in ('void', 'closed'):
+        return row
+    if db.query(PlCertQtyLog.id).filter_by(
+        tenant_no=user.tenant_no, cert_type=cert_type, cert_id=cert_id, event_type='opening', del_flag=0,
+    ).with_for_update().first():
+        return row
+    # 有效次数原文未识别时保留 draft，不自动启用；明确一次/多次后保存或读取再启用。
+    if usage_policy(row.valid_use_text) == 'needs_confirmation':
+        return row
+    used = Decimal(row.used_qty or 0)
+    op_id = f'auto-opening:{cert_type}:{cert_id}'
+    digest = hashlib.sha256(op_id.encode()).hexdigest()
+    operation, replay = _claim_operation(db, user, 'opening', op_id, digest)
+    if replay:
+        return row
+    db.add(PlCertQtyLog(
+        **_audit(user),
+        operation_id=op_id,
+        request_hash=digest,
+        cert_type=cert_type,
+        cert_id=cert_id,
+        event_type='opening',
+        quantity_unit=row.quantity_unit,
+        total_delta=Decimal(0),
+        reserved_delta=Decimal(0),
+        used_delta=used,
+        before_total_qty=row.total_qty,
+        after_total_qty=row.total_qty,
+        before_reserved_qty=row.reserved_qty,
+        after_reserved_qty=row.reserved_qty,
+        before_used_qty=used,
+        after_used_qty=used,
+        reason='建档自动启用',
+        evidence_urls=None,
+        event_snapshot={'singleUseConsumed': False, 'sourceNote': 'auto', 'auto': True},
+    ))
+    if row.status == 'draft':
+        row.status = 'active'
+        row.verified_by = user.id
+        row.verified_time = datetime.now()
+    row.update_by = user.id
+    operation.result_json = {'certType': cert_type, 'certId': cert_id, 'auto': True}
+    db.flush()
+    return row
+
+
 def save_cert(db, user, cert_type, data, cert_id=None):
     row = apply_cert(db, user, prepare_cert(db, user, cert_type, data, cert_id))
+    ensure_zero_opening(db, user, cert_type, row.id, row=row)
     return get_cert(db, user, cert_type, row.id)
 
 
 def delete_cert(db, user, cert_type, cert_id):
     row = _get_row(db,user,cert_type,cert_id,lock=True)
-    if _has_history(db,user,cert_type,cert_id) or row.reserved_qty or row.used_qty:
+    if row.reserved_qty or row.used_qty:
+        raise HTTPException(400, '已有用量或流水的证件只能停用、核销或作废')
+    if db.query(PlPlanCertUsage.id).filter_by(
+        tenant_no=user.tenant_no, cert_type=cert_type, cert_id=cert_id, del_flag=0,
+    ).with_for_update().first():
+        raise HTTPException(400, '已有用量或流水的证件只能停用、核销或作废')
+    logs = (
+        db.query(PlCertQtyLog)
+        .filter_by(tenant_no=user.tenant_no, cert_type=cert_type, cert_id=cert_id, del_flag=0)
+        .with_for_update()
+        .all()
+    )
+    # 仅建档自动期初（或无流水）仍允许删除；手工期初或其它额度流水则禁止。
+    def _auto_opening(log):
+        snap = log.event_snapshot or {}
+        return log.event_type == 'opening' and bool(snap.get('auto')) and (log.used_delta or 0) == 0
+
+    if any(not _auto_opening(log) for log in logs):
         raise HTTPException(400, '已有用量或流水的证件只能停用、核销或作废')
     if db.query(PlContractCertLink.id).filter_by(tenant_no=user.tenant_no,cert_type=cert_type,cert_id=cert_id,del_flag=0).with_for_update().first():
         raise HTTPException(400, '证件已绑定合同')
     if cert_type == 'purchase' and db.query(PlTransportCertPurchaseRef.id).filter_by(tenant_no=user.tenant_no,purchase_cert_id=cert_id,del_flag=0).with_for_update().first():
         raise HTTPException(400, '证件被运输证引用')
+    for log in logs:
+        log.del_flag = 1
+        log.update_by = user.id
     row.del_flag = 1
     row.update_by = user.id
     if cert_type == 'transport':
@@ -443,7 +559,24 @@ def _confirm_cert(db, user, cert_type, cert_id, data):
     _assert_known_usage_policy(row)
     if row.status == 'void': raise HTTPException(400, '作废证件不能启用')
     if row.status == 'closed': raise HTTPException(400, '已核销证件不能启用')
-    if row.verified_time or _has_history(db,user,cert_type,cert_id):
+    # 建档自动期初（used=0）可被手工/接口期初覆盖；已有真实用量或非自动期初则拒绝。
+    openings = (
+        db.query(PlCertQtyLog)
+        .filter_by(
+            tenant_no=user.tenant_no, cert_type=cert_type, cert_id=cert_id,
+            event_type='opening', del_flag=0,
+        )
+        .with_for_update()
+        .all()
+    )
+    for log in openings:
+        snap = log.event_snapshot or {}
+        if snap.get('auto') and (log.used_delta or 0) == 0:
+            log.del_flag = 1
+            log.update_by = user.id
+        else:
+            raise HTTPException(400, '期初已确认，不能重复初始化额度')
+    if _has_history(db, user, cert_type, cert_id):
         raise HTTPException(400, '期初已确认，不能重复初始化额度')
     if body.used_qty > row.total_qty:
         raise HTTPException(400, '期初已用量不能超过核定量')

backend/app/services/pl_quota_customer_select.py

@@ -43,6 +43,16 @@ def _has_opening(db, user, cert_type, cert_id) -> bool:
     )
 
 
+def _ensure_opening_quiet(db, user, cert_type, cert_id, row):
+    """选证前尽量自动补期初；单元测试 mock 库失败时静默跳过。"""
+    try:
+        from app.services.pl_quota_cert_service import ensure_zero_opening
+
+        ensure_zero_opening(db, user, cert_type, cert_id, row=row)
+    except Exception:
+        return
+
+
 def _vehicle_ok(cert, vehicle_no: str) -> bool:
     plates = cert.vehicle_numbers_json or []
     if not isinstance(plates, list):
@@ -53,6 +63,14 @@ def _vehicle_ok(cert, vehicle_no: str) -> bool:
     return plate in {str(p).strip() for p in plates if p}
 
 
+def _cert_no(cert) -> str:
+    return str(getattr(cert, 'cert_no', None) or getattr(cert, 'id', '') or '').strip()
+
+
+def _join_cert_nos(certs) -> str:
+    return '、'.join(dict.fromkeys(_cert_no(c) for c in certs if _cert_no(c)))
+
+
 def _permit_qty_for(source_qty: Decimal, source_unit: str, cert_unit: str) -> Decimal:
     if same_quantity_unit(source_unit, cert_unit):
         return source_qty
@@ -70,6 +88,292 @@ 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 (
+        db.query(PlPurchaseCert)
+        .filter(
+            PlPurchaseCert.tenant_no == user.tenant_no,
+            PlPurchaseCert.del_flag == 0,
+            PlPurchaseCert.status == 'active',
+            PlPurchaseCert.customer_id == customer_id,
+            PlPurchaseCert.category_id.in_(categories),
+        )
+        .order_by(
+            PlPurchaseCert.valid_to.is_(None).asc(),
+            PlPurchaseCert.valid_to.asc(),
+            PlPurchaseCert.id.asc(),
+        )
+        .all()
+    )
+
+
+def _linked_transports(db, user, customer_id, purchase_id):
+    refs = (
+        db.query(PlTransportCertPurchaseRef)
+        .filter_by(
+            tenant_no=user.tenant_no,
+            purchase_cert_id=purchase_id,
+            del_flag=0,
+        )
+        .all()
+    )
+    transport_cert_ids = [r.transport_cert_id for r in refs if r.transport_cert_id]
+    if not transport_cert_ids:
+        return []
+    return (
+        db.query(PlTransportCert)
+        .filter(
+            PlTransportCert.tenant_no == user.tenant_no,
+            PlTransportCert.del_flag == 0,
+            PlTransportCert.status == 'active',
+            PlTransportCert.customer_id == customer_id,
+            PlTransportCert.id.in_(transport_cert_ids),
+        )
+        .order_by(
+            PlTransportCert.valid_to.is_(None).asc(),
+            PlTransportCert.valid_to.asc(),
+            PlTransportCert.id.asc(),
+        )
+        .all()
+    )
+
+
+def _diagnose_select_failure(db, user, source, *, vehicle_no: str, on_date: date, candidates):
+    """选证失败时按证据优先给出单一可执行原因，避免笼统兜底句。"""
+    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}段客户「{customer_label}」下没有{product_part}匹配品类的有效购买证，'
+            '请先建档购买证或核对客户/证用品类后再安排'
+        )
+
+    unverified_buy = []
+    no_opening_buy = []
+    expired_buy = []
+    consumed_buy = []
+    unit_buy = []
+    short_buy = []
+    ready_buy = []
+    no_transport_link = []
+    unverified_move = []
+    no_opening_move = []
+    expired_move = []
+    consumed_move = []
+    vehicle_miss = []
+    unit_move = []
+    short_move = []
+
+    for purchase in candidates:
+        if not purchase.verified_by or not purchase.verified_time:
+            unverified_buy.append(purchase)
+            continue
+        _ensure_opening_quiet(db, user, 'purchase', purchase.id, purchase)
+        if not _has_opening(db, user, 'purchase', purchase.id):
+            no_opening_buy.append(purchase)
+            continue
+        if not _in_force(purchase, on_date):
+            expired_buy.append(purchase)
+            continue
+        if is_single_use_consumed(db, user, 'purchase', purchase):
+            consumed_buy.append(purchase)
+            continue
+        try:
+            need = _permit_qty_for(need_qty, source.unit, purchase.quantity_unit)
+        except HTTPException:
+            unit_buy.append(purchase)
+            continue
+        if _available(purchase) < need:
+            short_buy.append((purchase, need))
+            continue
+
+        if not source.requires_transport:
+            ready_buy.append(purchase)
+            continue
+
+        transports = _linked_transports(db, user, purchase.customer_id, purchase.id)
+        if not transports:
+            no_transport_link.append(purchase)
+            continue
+
+        purchase_had_ready_transport = False
+        for transport in transports:
+            if not transport.verified_by or not transport.verified_time:
+                unverified_move.append(transport)
+                continue
+            _ensure_opening_quiet(db, user, 'transport', transport.id, transport)
+            if not _has_opening(db, user, 'transport', transport.id):
+                no_opening_move.append(transport)
+                continue
+            if not _in_force(transport, on_date):
+                expired_move.append(transport)
+                continue
+            if is_single_use_consumed(db, user, 'transport', transport):
+                consumed_move.append(transport)
+                continue
+            if not _vehicle_ok(transport, vehicle_no):
+                vehicle_miss.append(transport)
+                continue
+            try:
+                t_need = _permit_qty_for(need_qty, source.unit, transport.quantity_unit)
+            except HTTPException:
+                unit_move.append(transport)
+                continue
+            if _available(transport) < t_need:
+                short_move.append((transport, t_need))
+                continue
+            purchase_had_ready_transport = True
+            break
+        if purchase_had_ready_transport:
+            ready_buy.append(purchase)
+
+    if source.requires_transport and not plate:
+        return (
+            f'{place}段安排运输须填写车牌号；当前计划未填写车牌，'
+            '无法匹配运输证车号，请先填写车牌后再安排'
+        )
+
+    if vehicle_miss and not short_move and not unit_move:
+        nos = _join_cert_nos(vehicle_miss)
+        listed = '、'.join(
+            dict.fromkeys(
+                str(p).strip()
+                for t in vehicle_miss
+                for p in (t.vehicle_numbers_json or [])
+                if str(p).strip()
+            )
+        )
+        suffix = f'（证上现有车号：{listed}）' if listed else ''
+        return (
+            f'{place}段运输证 {nos} 未包含车牌号「{plate}」{suffix}。'
+            '请到运输证中补充该车号，或更换已登记该车号的车辆后再安排'
+        )
+
+    if short_move:
+        parts = []
+        for cert, need in short_move[:3]:
+            left = _available(cert)
+            parts.append(
+                f'{_cert_no(cert)}剩余{left}{cert.quantity_unit}，本次需要{need}{cert.quantity_unit}'
+            )
+        return (
+            f'{place}段运输证{"；".join(parts)}。'
+            '运输证余量不足，请先补充额度或调整计划量后再安排'
+        )
+
+    if short_buy and not ready_buy:
+        parts = []
+        for cert, need in short_buy[:3]:
+            left = _available(cert)
+            parts.append(
+                f'{_cert_no(cert)}剩余{left}{cert.quantity_unit}，本次需要{need}{cert.quantity_unit}'
+            )
+        return (
+            f'{place}段购买证{"；".join(parts)}。'
+            '购买证余量不足，请先补充额度或调整计划量后再安排'
+        )
+
+    if no_transport_link and not vehicle_miss and not short_move:
+        nos = _join_cert_nos(no_transport_link)
+        return (
+            f'{place}段购买证 {nos} 尚未关联运输证。'
+            '请打开运输证，在「关联购买证」中选择该购买证后再安排'
+        )
+
+    if expired_move and not vehicle_miss:
+        nos = _join_cert_nos(expired_move)
+        return (
+            f'{place}段运输证 {nos} 不在计划日期 {on_date.isoformat()} 的有效期内，'
+            '请更换有效运输证后再安排'
+        )
+
+    if expired_buy and not ready_buy:
+        nos = _join_cert_nos(expired_buy)
+        return (
+            f'{place}段购买证 {nos} 不在计划日期 {on_date.isoformat()} 的有效期内，'
+            '请更换有效购买证后再安排'
+        )
+
+    if unverified_move and not vehicle_miss:
+        nos = _join_cert_nos(unverified_move)
+        return f'{place}段运输证 {nos} 尚未复核启用，请先复核后再安排'
+
+    if unverified_buy and not ready_buy:
+        nos = _join_cert_nos(unverified_buy)
+        return f'{place}段购买证 {nos} 尚未复核启用，请先复核后再安排'
+
+    if consumed_move:
+        nos = _join_cert_nos(consumed_move)
+        return f'{place}段运输证 {nos} 为一次有效且已使用，请更换证件后再安排'
+
+    if consumed_buy and not ready_buy:
+        nos = _join_cert_nos(consumed_buy)
+        return f'{place}段购买证 {nos} 为一次有效且已使用，请更换证件后再安排'
+
+    if unit_buy or unit_move:
+        return (
+            f'{place}段证件单位与计划单位无法自动换算，'
+            '请调整单位或到「计划段次用证」填写换算后再安排'
+        )
+
+    if no_opening_buy or no_opening_move:
+        return f'{place}段证件尚未确认期初用量，请先确认期初后再安排'
+
+    return (
+        f'{place}段找不到可用的购买证'
+        f'{"和运输证" if source.requires_transport else ""}，'
+        '请补充证件或标记证书齐备后再安排'
+    )
+
+
 def select_whole_cover_certs(db, user, sources, *, vehicle_no: str, on_date: date):
     """返回 purchase_cert_ids, transport_cert_ids, manual_allocations。
 
@@ -86,28 +390,13 @@ 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}」未配置证用品类，请先在证用品类中关联系统产品后再安排运输',
             )
 
-        # MySQL ASC 会把 NULL 排最前；无到期证应排在末尾，优先消耗更早到期的证（D5）
-        candidates = (
-            db.query(PlPurchaseCert)
-            .filter(
-                PlPurchaseCert.tenant_no == user.tenant_no,
-                PlPurchaseCert.del_flag == 0,
-                PlPurchaseCert.status == 'active',
-                PlPurchaseCert.customer_id == customer_id,
-                PlPurchaseCert.category_id.in_(categories),
-            )
-            .order_by(
-                PlPurchaseCert.valid_to.is_(None).asc(),
-                PlPurchaseCert.valid_to.asc(),
-                PlPurchaseCert.id.asc(),
-            )
-            .all()
-        )
+        candidates = _purchase_candidates(db, user, customer_id, categories)
 
         chosen_purchase = None
         chosen_transport = None
@@ -117,6 +406,7 @@ def select_whole_cover_certs(db, user, sources, *, vehicle_no: str, on_date: dat
         for purchase in candidates:
             if not purchase.verified_by or not purchase.verified_time:
                 continue
+            _ensure_opening_quiet(db, user, 'purchase', purchase.id, purchase)
             if not _has_opening(db, user, 'purchase', purchase.id):
                 continue
             if not _in_force(purchase, on_date):
@@ -135,37 +425,13 @@ def select_whole_cover_certs(db, user, sources, *, vehicle_no: str, on_date: dat
                 purchase_permit_qty = need
                 break
 
-            refs = (
-                db.query(PlTransportCertPurchaseRef)
-                .filter_by(
-                    tenant_no=user.tenant_no,
-                    purchase_cert_id=purchase.id,
-                    del_flag=0,
-                )
-                .all()
-            )
-            transport_cert_ids = [r.transport_cert_id for r in refs if r.transport_cert_id]
-            if not transport_cert_ids:
+            transports = _linked_transports(db, user, customer_id, purchase.id)
+            if not transports:
                 continue
-            transports = (
-                db.query(PlTransportCert)
-                .filter(
-                    PlTransportCert.tenant_no == user.tenant_no,
-                    PlTransportCert.del_flag == 0,
-                    PlTransportCert.status == 'active',
-                    PlTransportCert.customer_id == customer_id,
-                    PlTransportCert.id.in_(transport_cert_ids),
-                )
-                .order_by(
-                    PlTransportCert.valid_to.is_(None).asc(),
-                    PlTransportCert.valid_to.asc(),
-                    PlTransportCert.id.asc(),
-                )
-                .all()
-            )
             for transport in transports:
                 if not transport.verified_by or not transport.verified_time:
                     continue
+                _ensure_opening_quiet(db, user, 'transport', transport.id, transport)
                 if not _has_opening(db, user, 'transport', transport.id):
                     continue
                 if not _in_force(transport, on_date):
@@ -189,11 +455,11 @@ def select_whole_cover_certs(db, user, sources, *, vehicle_no: str, on_date: dat
                 break
 
         if chosen_purchase is None:
-            place = '同市' if not source.requires_transport else '异地'
             raise HTTPException(
                 409,
-                f'{place}段找不到余量足够、有效期内且匹配品类/车号的购买证'
-                f'{"和运输证" if source.requires_transport else ""}，请补充证件或标记证书齐备后再安排',
+                _diagnose_select_failure(
+                    db, user, source, vehicle_no=vehicle_no, on_date=on_date, candidates=candidates,
+                ),
             )
 
         purchase_ids.append(chosen_purchase.id)
@@ -222,3 +488,76 @@ def select_whole_cover_certs(db, user, sources, *, vehicle_no: str, on_date: dat
         return list(dict.fromkeys(values))
 
     return unique(purchase_ids), unique(transport_ids), manuals
+
+
+def resolve_customer_category_cover_certs(db, user, sources):
+    """归档超额补扣：按客户+品类收集证件池，不查合同绑证。
+
+    与 select_whole_cover_certs 不同，这里不按数量筛证，把客户下匹配品类的
+    有效购买证及其关联运输证全部纳入候选，由结转阶段结合本计划返还额度分配。
+    """
+    if not sources:
+        return [], []
+
+    purchase_ids = []
+    transport_ids = []
+    missing_both = []
+    missing_purchase = []
+    missing_transport = []
+
+    for source in sources:
+        customer_id = _customer_id_from_source(source)
+        snap = source.snapshot or {}
+        customer_label = (
+            snap.get('customerName')
+            or snap.get('customer_name')
+            or snap.get('customerNo')
+            or snap.get('customer_no')
+            or customer_id
+        )
+        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)
+        if not purchases:
+            (missing_both if needs_transport else missing_purchase).append(str(customer_label))
+            continue
+        purchase_ids.extend(p.id for p in purchases)
+        if not needs_transport:
+            continue
+        linked = []
+        for purchase in purchases:
+            linked.extend(_linked_transports(db, user, customer_id, purchase.id))
+        if not linked:
+            missing_transport.append(str(customer_label))
+            continue
+        transport_ids.extend(t.id for t in linked)
+
+    messages = []
+    if missing_both:
+        names = '、'.join(dict.fromkeys(missing_both))
+        messages.append(
+            f'客户{names}下尚未建档匹配品类的购买证和运输证，请先建档证件后再归档'
+        )
+    if missing_purchase:
+        names = '、'.join(dict.fromkeys(missing_purchase))
+        messages.append(
+            f'客户{names}下尚未建档匹配品类的购买证，请先建档证件后再归档'
+        )
+    if missing_transport:
+        names = '、'.join(dict.fromkeys(missing_transport))
+        messages.append(
+            f'客户{names}下购买证尚未关联可用运输证，请先关联运输证后再归档'
+        )
+    if messages:
+        raise HTTPException(409, '；'.join(messages))
+
+    def unique(values):
+        return list(dict.fromkeys(values))
+
+    return unique(purchase_ids), unique(transport_ids)

backend/app/services/pl_quota_planning.py

@@ -137,7 +137,9 @@ def build_usage_request(db, user, plan_id, operation, action, *, lock=False, che
         if customer_rule:
             # 新规则：禁止手工选证；按客户+品类整笔自动选证
             from app.services.pl_quota_customer_select import select_whole_cover_certs
-            vehicle = (preview_vehicle or {}).get('licensePlate', plan.license_plate) or ''
+            vehicle = str(
+                (preview_vehicle or {}).get('licensePlate', plan.license_plate) or ''
+            ).strip()
             on_date = max(date.today(), plan.plan_date or date.today())
             purchase_ids, transport_ids, manuals = select_whole_cover_certs(
                 db, user, sources, vehicle_no=vehicle, on_date=on_date,
@@ -171,7 +173,7 @@ def build_usage_request(db, user, plan_id, operation, action, *, lock=False, che
         certs = {key: _get_row(db, user, *key, lock=False) for key in sorted(keys)}
         if sources and not keys:
             msg = (
-                '找不到适用的购买证和运输证，请补充证件或标记证书齐备后再安排运输'
+                '未选到可用的购买证和运输证。请核对证件余量、有效期、品类与车牌号后补充证件，或标记证书齐备后再安排运输'
                 if customer_rule else
                 '请先在合同中绑定购买证和运输证后再安排运输'
             )
@@ -183,7 +185,7 @@ def build_usage_request(db, user, plan_id, operation, action, *, lock=False, che
         operation_id=op.operation_id, action=action, plan_id=plan_id, sources=sources,
         allow_supplement=op.allow_supplement,
         purchase_cert_ids=config.purchase_cert_ids, transport_cert_ids=config.transport_cert_ids,
-        vehicle_no=(preview_vehicle or {}).get('licensePlate', plan.license_plate) or '', on_date=max(date.today(), plan.plan_date or date.today()),
+        vehicle_no=str((preview_vehicle or {}).get('licensePlate', plan.license_plate) or '').strip(), on_date=max(date.today(), plan.plan_date or date.today()),
         executed=executed, manual_allocations=[UsageManual(**r.model_dump()) for r in config.manual_allocations],
         actual_by_source=actuals, manual_actuals=[UsageManual(**r.model_dump()) for r in op.manual_actuals],
         reason=op.reason, evidence_urls=op.evidence_urls or '', source_note=op.source_note or '',

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/app/services/pl_quota_usage_service.py

@@ -26,6 +26,26 @@ from app.services.pl_quota_units import same_quantity_unit, standard_conversion_
 ZERO = Decimal('0')
 
 
+def _normalize_vehicle_no(vehicle_no) -> str:
+    return str(vehicle_no or '').strip()
+
+
+def _vehicle_on_cert(cert, vehicle_no) -> bool:
+    """计划车牌与证上车号比对时两侧都去空白，避免尾部空格误判。"""
+    plate = _normalize_vehicle_no(vehicle_no)
+    if not plate:
+        return False
+    plates = getattr(cert, 'vehicle_numbers_json', None) or []
+    if isinstance(plates, str):
+        try:
+            plates = json.loads(plates)
+        except Exception:
+            plates = [plates]
+    if not isinstance(plates, list):
+        return False
+    return plate in {str(p).strip() for p in plates if p is not None and str(p).strip()}
+
+
 def _reserve_input_mode(source_unit, cert_unit):
     if same_quantity_unit(source_unit, cert_unit):
         return 'same_unit'
@@ -131,9 +151,8 @@ def _execute(db,user,r,digest,plan):
     selected={('purchase',x) for x in r.purchase_cert_ids}|{('transport',x) for x in r.transport_cert_ids}
     keys=selected|{(u.cert_type,u.cert_id) for u in old_rows}
     if r.action in ('settle', 'actual_adjust') and not r.allow_supplement and r.sources:
-        from app.services.pl_quota_planning import resolve_contract_bound_certs
         try:
-            purchase_ids, transport_ids = resolve_contract_bound_certs(db, user, r.sources)
+            purchase_ids, transport_ids = _settle_cover_cert_ids(db, user, plan, r.sources)
         except HTTPException:
             purchase_ids, transport_ids = [], []
         keys |= {('purchase', cert_id) for cert_id in purchase_ids} | {('transport', cert_id) for cert_id in transport_ids}
@@ -200,8 +219,10 @@ def _candidate_context(db,user,r,plan,certs,usages,items,selected,preview=False,
     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)
+    from app.services.pl_quota_cert_service import ensure_zero_opening
     for k in sorted(selected):
         kind,pid=k; c=certs[k]
+        c = certs[k] = ensure_zero_opening(db, user, kind, pid, row=c)
         if not query(m.PlCertQtyLog,cert_type=kind,cert_id=pid,event_type='opening',del_flag=0).first(): raise ValueError('证件尚未确认期初')
         if c.status=='closed': raise ValueError('已核销证件不能再安排用证')
         if c.status!='active' or not c.verified_time or not c.verified_by: raise ValueError('证件尚未启用复核')
@@ -210,7 +231,7 @@ def _candidate_context(db,user,r,plan,certs,usages,items,selected,preview=False,
         policy=usage_policy(c.valid_use_text)
         if policy=='needs_confirmation': raise ValueError('有效次数待确认')
         # 合同上常有多张运输证。车号对不上的证不参与本次分配，只要还有一张覆盖该主车号即可安排。
-        if kind=='transport' and (not r.vehicle_no or r.vehicle_no not in (c.vehicle_numbers_json or [])):
+        if kind=='transport' and not _vehicle_on_cert(c, r.vehicle_no):
             skipped_for_vehicle=True
             continue
         if policy=='once' and is_single_use_consumed(db, user, kind, c, exclude_plan_id=plan.id):
@@ -231,7 +252,21 @@ def _candidate_context(db,user,r,plan,certs,usages,items,selected,preview=False,
         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))
     if skipped_for_vehicle and not candidates['transport']:
-        raise ValueError('车号不在运输证范围')
+        plate = _normalize_vehicle_no(r.vehicle_no)
+        if not plate:
+            raise ValueError('安排运输须填写车牌号；当前未填写车牌，无法匹配运输证车号')
+        nos = '、'.join(
+            dict.fromkeys(
+                str(certs[k].cert_no or k[1])
+                for k in sorted(selected)
+                if k[0] == 'transport' and k in certs
+            )
+        )
+        suffix = f'（已选运输证 {nos}）' if nos else ''
+        raise ValueError(
+            f'车牌号「{plate}」不在运输证车号范围内{suffix}，'
+            '请到运输证中补充该车号或更换车辆后再安排'
+        )
     return sources,scopes,candidates
 
 def _reserve(db,user,r,plan,certs,usages,items,selected,preview=False,lock_preview=False):
@@ -360,28 +395,88 @@ def _validate_current_sources(r,active):
                 raise ValueError('当前权威来源身份、计划量、单位或指纹已变化，请重新核对用证')
 
 class ArchiveQuotaBlocked(ValueError):
-    """合同上的购买证、运输证仍不够本次实际量，归档必须整单回滚。"""
+    """可用购买证、运输证仍不够本次实际量，归档必须整单回滚。"""
 
 
 ARCHIVE_QUOTA_BLOCK = '合同绑定的购买证和运输证剩余额度不够覆盖本次实际数量，不能归档。请先补充证件额度或调整实际数量后再归档。'
+CUSTOMER_ARCHIVE_QUOTA_BLOCK = (
+    '客户名下购买证和运输证剩余额度不够覆盖本次实际数量，不能归档。'
+    '请先补充证件额度或调整实际数量后再归档。'
+)
+ARCHIVE_CONVERSION_EXCESS_BLOCK = (
+    '实际数量超出安排时的预占，且原证件剩余额度不够按原换算系数继续扣减，不能归档。'
+    '请把实际数量改小，或先补充证件额度后再归档。'
+)
+
+
+def _settle_cover_cert_ids(db, user, plan, sources):
+    """归档超额补扣证件池：新规则按客户+品类，旧规则按合同绑证。"""
+    from app.services.pl_cert_quota_rule_config_service import plan_uses_customer_category
+    if plan_uses_customer_category(plan):
+        from app.services.pl_quota_customer_select import resolve_customer_category_cover_certs
+        return resolve_customer_category_cover_certs(db, user, sources)
+    from app.services.pl_quota_planning import resolve_contract_bound_certs
+    return resolve_contract_bound_certs(db, user, sources)
 
 
 def _settlement_cover_cert(db, user, r, plan, kind, cert):
     """合同上另有证件时，不合格的证跳过，不因此挡住其余还能扣的证。"""
+    return _settlement_cover_reject_reason(db, user, r, plan, kind, cert) is None
+
+
+def _settlement_cover_reject_reason(db, user, r, plan, kind, cert):
+    """返回不可用于归档补扣的具体原因；可用时返回 None。"""
+    from app.services.pl_quota_cert_service import ensure_zero_opening
+    cert = ensure_zero_opening(db, user, kind, cert.id, row=cert)
+    label = '购买证' if kind == 'purchase' else '运输证'
+    no = str(cert.cert_no or cert.id)
     if not _query(db, user, m.PlCertQtyLog, cert_type=kind, cert_id=cert.id, event_type='opening', del_flag=0).first():
-        return False
+        return f'{label} {no} 尚未确认期初用量'
     if cert.status != 'active' or not cert.verified_time or not cert.verified_by:
-        return False
+        return f'{label} {no} 未启用或未复核'
     if cert.valid_from is None or cert.valid_to is None or not cert.valid_from <= r.on_date <= cert.valid_to:
-        return False
+        return f'{label} {no} 不在归档日期 {r.on_date.isoformat()} 的有效期内'
     policy = usage_policy(cert.valid_use_text)
     if policy == 'needs_confirmation':
-        return False
-    if kind == 'transport' and (not r.vehicle_no or r.vehicle_no not in (cert.vehicle_numbers_json or [])):
-        return False
+        return f'{label} {no} 有效次数待确认'
+    if kind == 'transport':
+        plate = _normalize_vehicle_no(r.vehicle_no)
+        if not plate:
+            return f'运输证 {no} 需要车牌号，当前计划未填写车牌'
+        if not _vehicle_on_cert(cert, plate):
+            raw = cert.vehicle_numbers_json or []
+            if isinstance(raw, str):
+                listed = raw
+            else:
+                listed = '、'.join(str(p).strip() for p in raw if str(p).strip())
+            suffix = f'（证上现有车号：{listed}）' if listed else ''
+            return f'运输证 {no} 未包含车牌号「{plate}」{suffix}'
     if policy == 'once' and is_single_use_consumed(db, user, kind, cert, exclude_plan_id=plan.id):
-        return False
-    return True
+        return f'{label} {no} 为一次有效且已使用'
+    return None
+
+
+def _archive_cover_block_message(db, user, r, plan, certs, candidate_keys):
+    """补证候选全被跳过时，汇总具体原因，避免笼统额度不足。"""
+    from app.services.pl_cert_quota_rule_config_service import plan_uses_customer_category
+    customer_rule = plan_uses_customer_category(plan)
+    reasons = []
+    for kind, cid in candidate_keys:
+        key = (kind, cid)
+        cert = certs.get(key)
+        if cert is None:
+            label = '购买证' if kind == 'purchase' else '运输证'
+            reasons.append(f'{label} {cid} 不存在或不可用')
+            continue
+        reason = _settlement_cover_reject_reason(db, user, r, plan, kind, cert)
+        if reason:
+            reasons.append(reason)
+    if not reasons:
+        return CUSTOMER_ARCHIVE_QUOTA_BLOCK if customer_rule else ARCHIVE_QUOTA_BLOCK
+    unique = list(dict.fromkeys(reasons))
+    detail = '；'.join(unique[:4])
+    scope = '客户名下' if customer_rule else '合同上'
+    return f'{detail}。{scope}没有其他可用证件覆盖本次实际数量，不能归档。请先处理上述问题后再归档。'
 
 
 _QTY = Decimal('0.000001')
@@ -438,7 +533,7 @@ def _prepare_saved_conversion(groups, actual_by_source, certs):
                 proposals[kind] = [row for row in rows if row.source_id != sid] + updated
                 mine = updated
                 if need > 0:
-                    return groups, [], ARCHIVE_QUOTA_BLOCK
+                    return groups, [], ARCHIVE_CONVERSION_EXCESS_BLOCK
             remaining = actual
             for row in sorted(mine, key=lambda item: item.allocation_seq):
                 take = min(remaining, row.source_quantity)
@@ -479,8 +574,8 @@ def _settle(db,user,r,certs,usages,items,plan):
             raise ArchiveQuotaBlocked(conversion_error)
         if generated:
             r = r.model_copy(update={'manual_actuals': generated})
-    # 同单位超出预占时，自动改用合同上已绑定的购买证和运输证，含原证剩余额度。
-    contract_cover = False
+    # 同单位超出预占时，自动改用补扣证件池（新规则客户+品类，旧规则合同绑证），含原证剩余额度。
+    auto_cover = False
     if not r.allow_supplement:
         for sid, value in r.actual_by_source.items():
             mine = [a for rows in groups.values() for a in rows if a.source_id == sid]
@@ -488,7 +583,7 @@ def _settle(db,user,r,certs,usages,items,plan):
                 continue
             capacity = sum((a.source_quantity for a in groups['purchase'] if a.source_id == sid), ZERO)
             if value > capacity:
-                contract_cover = True
+                auto_cover = True
                 break
     # 原始容量不随补量扩大；新增证件原始容量为零，后续纠错始终先用原分配。
     base_actual=dict(r.actual_by_source)
@@ -497,7 +592,7 @@ def _settle(db,user,r,certs,usages,items,plan):
         capacity=sum((a.source_quantity for a in groups['purchase'] if a.source_id==sid),ZERO)
         if value>capacity:
             excess[sid]=value-capacity
-            if r.allow_supplement or contract_cover: base_actual[sid]=capacity
+            if r.allow_supplement or auto_cover: base_actual[sid]=capacity
     # 人工显式补量可从当前实运来源拆出一段（如原证24、新证5），两段合计仍是29。
     # 两证有一侧同单位时由联合分配器配对；异单位补量仍须完整人工输入。
     if r.allow_supplement and r.manual_allocations:
@@ -549,25 +644,32 @@ def _settle(db,user,r,certs,usages,items,plan):
     settled={kind:settle_allocations(rows,base_actual,manual_actuals=[a for a in _manual(r.manual_actuals) if a.permit_kind==kind]) for kind,rows in groups.items()}
     supplemental=[]
     if excess:
-        if contract_cover:
-            from app.services.pl_quota_planning import resolve_contract_bound_certs
+        if auto_cover:
+            from app.services.pl_cert_quota_rule_config_service import plan_uses_customer_category
+            customer_rule = plan_uses_customer_category(plan)
             try:
-                purchase_ids, transport_ids = resolve_contract_bound_certs(db, user, r.sources)
+                purchase_ids, transport_ids = _settle_cover_cert_ids(db, user, plan, r.sources)
             except HTTPException as exc:
                 raise ArchiveQuotaBlocked(str(getattr(exc, 'detail', exc))) from exc
+            candidate_keys=[('purchase', x) for x in purchase_ids] + [('transport', x) for x in transport_ids]
             selected=set()
-            for kind, cid in [('purchase', x) for x in purchase_ids] + [('transport', x) for x in transport_ids]:
+            for kind, cid in candidate_keys:
                 key=(kind, cid)
                 if key not in certs:
-                    raise ArchiveQuotaBlocked(ARCHIVE_QUOTA_BLOCK)
+                    raise ArchiveQuotaBlocked(
+                        _archive_cover_block_message(db, user, r, plan, certs, candidate_keys)
+                    )
                 if _settlement_cover_cert(db, user, r, plan, kind, certs[key]):
                     selected.add(key)
             if not selected:
-                raise ArchiveQuotaBlocked(ARCHIVE_QUOTA_BLOCK)
+                raise ArchiveQuotaBlocked(
+                    _archive_cover_block_message(db, user, r, plan, certs, candidate_keys)
+                )
         else:
             if not r.reason.strip() or not (r.evidence_urls.strip() or r.source_note.strip()):
                 raise ValueError('补分配必须有理由及依据')
             selected={('purchase',x) for x in r.purchase_cert_ids}|{('transport',x) for x in r.transport_cert_ids}
+            customer_rule = False
         # 校验器看到的是扣除原量、返还本计划旧账后的净可用量。
         base_totals={k:ZERO for k in usages}
         for kind,rows in settled.items():
@@ -577,7 +679,7 @@ def _settle(db,user,r,certs,usages,items,plan):
         try:
             result=_reserve(db,user,extra_request,plan,certs,credits,{},selected,preview=True,lock_preview=True)
         except InsufficientQuota as exc:
-            if not contract_cover:
+            if not auto_cover:
                 raise
             from app.services.pl_quota_diagnostics import archive_quantity_shortfall_message
             cert_rows=[]
@@ -591,7 +693,8 @@ def _settle(db,user,r,certs,usages,items,plan):
             message=archive_quantity_shortfall_message(
                 getattr(exc, 'diagnostics', None), cert_rows, r.actual_by_source, excess,
                 {source.source_id: source.unit for source in r.sources})
-            raise ArchiveQuotaBlocked(message or ARCHIVE_QUOTA_BLOCK) from exc
+            fallback = CUSTOMER_ARCHIVE_QUOTA_BLOCK if customer_rule else ARCHIVE_QUOTA_BLOCK
+            raise ArchiveQuotaBlocked(message or fallback) from exc
         for kind in ('purchase','transport'):
             for row in result[kind+'Allocations']:
                 supplemental.append((kind,row))

backend/tests/test_pl_customer_cert_quota.py

@@ -752,3 +752,311 @@ def test_get_cert_customer_name_from_crm_not_consignee(monkeypatch):
     assert result['customerName'] == '滁州康华电子材料有限公司'
     assert result['consigneeName'] == '国能宿州热电有限公司'
     assert result['customerId'] == 'cust-kanghua'
+
+
+def _remote_source(**kw):
+    from app.schemas.pl.quota_usage import UsageSource
+
+    base = dict(
+        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('30'),
+        unit='吨',
+        fingerprint='r' * 64,
+        snapshot={'customerId': 'cust1'},
+        requires_transport=True,
+    )
+    base.update(kw)
+    return UsageSource(**base)
+
+
+def _ok_purchase(**kw):
+    data = dict(
+        id='buy1',
+        cert_no='G-BUY-1',
+        customer_id='cust1',
+        category_id='cat1',
+        status='active',
+        verified_by='u',
+        verified_time=datetime.now(),
+        valid_from=None,
+        valid_to=date(2027, 1, 1),
+        total_qty=Decimal('100'),
+        used_qty=Decimal('0'),
+        reserved_qty=Decimal('0'),
+        quantity_unit='吨',
+        opening_single_use_consumed=0,
+    )
+    data.update(kw)
+    return SimpleNamespace(**data)
+
+
+def _ok_transport(**kw):
+    data = dict(
+        id='tr1',
+        cert_no='Y-TR-1',
+        customer_id='cust1',
+        status='active',
+        verified_by='u',
+        verified_time=datetime.now(),
+        valid_from=None,
+        valid_to=date(2027, 1, 1),
+        total_qty=Decimal('100'),
+        used_qty=Decimal('0'),
+        reserved_qty=Decimal('0'),
+        quantity_unit='吨',
+        opening_single_use_consumed=0,
+        vehicle_numbers_json=['皖A11111'],
+    )
+    data.update(kw)
+    return SimpleNamespace(**data)
+
+
+def _mock_select_db(purchases, transports, refs):
+    from app.models.pl_quota_models import (
+        PlPurchaseCert,
+        PlTransportCert,
+        PlTransportCertPurchaseRef,
+        PlCertQtyLog,
+    )
+
+    class PurchaseQuery:
+        def filter(self, *a, **k):
+            return self
+
+        def order_by(self, *a, **k):
+            return self
+
+        def all(self):
+            return list(purchases)
+
+    class TransportQuery:
+        def filter(self, *a, **k):
+            return self
+
+        def order_by(self, *a, **k):
+            return self
+
+        def all(self):
+            return list(transports)
+
+    class RefQuery:
+        def filter_by(self, **k):
+            return self
+
+        def all(self):
+            return list(refs)
+
+    class LogQuery:
+        def filter_by(self, **k):
+            return self
+
+        def first(self):
+            return SimpleNamespace(id='opening')
+
+    db = MagicMock()
+
+    def q_side(model):
+        if model is PlPurchaseCert:
+            return PurchaseQuery()
+        if model is PlTransportCert:
+            return TransportQuery()
+        if model is PlTransportCertPurchaseRef:
+            return RefQuery()
+        if model is PlCertQtyLog:
+            return LogQuery()
+        return MagicMock()
+
+    db.query.side_effect = q_side
+    return db
+
+
+def test_whole_cover_vehicle_mismatch_message_is_precise(monkeypatch):
+    """异地段购买证够用但运输证未含该车牌时，应点名车牌与证号。"""
+    from app.services import pl_quota_customer_select as sel
+
+    purchase = _ok_purchase()
+    transport = _ok_transport(vehicle_numbers_json=['皖A00000'])
+    ref = SimpleNamespace(transport_cert_id='tr1', purchase_cert_id='buy1')
+    db = _mock_select_db([purchase], [transport], [ref])
+    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()],
+            vehicle_no='皖A9BE91',
+            on_date=date(2026, 11, 14),
+        )
+    detail = str(exc.value.detail)
+    assert exc.value.status_code == 409
+    assert '皖A9BE91' in detail
+    assert 'Y-TR-1' in detail
+    assert '车号' in detail or '车牌' in detail
+    assert '余量足够、有效期内且匹配品类/车号' not in detail
+
+
+def test_whole_cover_empty_vehicle_message_is_precise(monkeypatch):
+    from app.services import pl_quota_customer_select as sel
+
+    purchase = _ok_purchase()
+    transport = _ok_transport()
+    ref = SimpleNamespace(transport_cert_id='tr1', purchase_cert_id='buy1')
+    db = _mock_select_db([purchase], [transport], [ref])
+    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()],
+            vehicle_no='  ',
+            on_date=date(2026, 11, 14),
+        )
+    detail = str(exc.value.detail)
+    assert '车牌' in detail
+    assert '余量足够、有效期内且匹配品类/车号' not in detail
+
+
+def test_whole_cover_purchase_insufficient_message_is_precise(monkeypatch):
+    from app.services import pl_quota_customer_select as sel
+
+    purchase = _ok_purchase(total_qty=Decimal('10'), cert_no='G-SMALL')
+    db = _mock_select_db([purchase], [], [])
+    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(requires_transport=False, quantity=Decimal('30'))],
+            vehicle_no='皖A1',
+            on_date=date(2026, 11, 14),
+        )
+    detail = str(exc.value.detail)
+    assert 'G-SMALL' in detail
+    assert '余量' in detail or '不够' in detail
+    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
+
+    purchase = _ok_purchase()
+    transport = _ok_transport()
+    ref = SimpleNamespace(transport_cert_id='tr1', purchase_cert_id='buy1')
+    db = _mock_select_db([purchase], [transport], [ref])
+    monkeypatch.setattr(sel, 'category_ids_for_product', lambda *a, **k: {'cat1'})
+
+    purchase_ids, transport_ids = sel.resolve_customer_category_cover_certs(
+        db,
+        SimpleNamespace(tenant_no='t'),
+        [_remote_source(snapshot={'customerId': 'cust1', 'customerName': '昆腾'})],
+    )
+    assert purchase_ids == ['buy1']
+    assert transport_ids == ['tr1']
+
+
+def test_resolve_customer_category_cover_certs_missing_uses_customer_wording(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.resolve_customer_category_cover_certs(
+            db,
+            SimpleNamespace(tenant_no='t'),
+            [_remote_source(snapshot={'customerId': 'cust1', 'customerName': '昆腾'})],
+        )
+    detail = str(exc.value.detail)
+    assert '客户昆腾' in detail
+    assert '合同' not in detail
+
+
+def test_settle_cover_cert_ids_customer_rule_skips_contract_bindings(monkeypatch):
+    from app.services import pl_quota_usage_service as usage
+
+    plan = SimpleNamespace(id='plan', cert_quota_rule_version='customer_category')
+    called = {'contract': 0, 'customer': 0}
+
+    def fake_customer(db, user, sources):
+        called['customer'] += 1
+        return ['buy1'], ['tr1']
+
+    def fake_contract(db, user, sources):
+        called['contract'] += 1
+        raise AssertionError('customer_category 归档超额不应查合同绑证')
+
+    monkeypatch.setattr(
+        'app.services.pl_quota_customer_select.resolve_customer_category_cover_certs',
+        fake_customer,
+    )
+    monkeypatch.setattr(
+        'app.services.pl_quota_planning.resolve_contract_bound_certs',
+        fake_contract,
+    )
+    monkeypatch.setattr(
+        'app.services.pl_cert_quota_rule_config_service.plan_uses_customer_category',
+        lambda p: True,
+    )
+
+    purchase_ids, transport_ids = usage._settle_cover_cert_ids(
+        MagicMock(), SimpleNamespace(tenant_no='t'), plan, [_remote_source()],
+    )
+    assert purchase_ids == ['buy1']
+    assert transport_ids == ['tr1']
+    assert called == {'contract': 0, 'customer': 1}
+
+
+def test_archive_cover_block_message_customer_rule_wording(monkeypatch):
+    from app.services import pl_quota_usage_service as svc
+
+    plan = SimpleNamespace(id='plan', cert_quota_rule_version='customer_category')
+    r = SimpleNamespace(vehicle_no='皖A1', on_date=date(2026, 11, 15))
+    monkeypatch.setattr(
+        'app.services.pl_cert_quota_rule_config_service.plan_uses_customer_category',
+        lambda p: True,
+    )
+    message = svc._archive_cover_block_message(
+        MagicMock(), SimpleNamespace(tenant_no='t'), r, plan, {}, [],
+    )
+    assert '客户名下' in message
+    assert '合同' not in message

backend/tests/test_pl_quota_api.py

@@ -57,7 +57,7 @@ def test_permission_guards_validation_and_atomic_creation(api):
     saved = client.post(path, json=payload())
     assert saved.status_code == 200
     row = saved.json()
-    assert row['tenantNo'] == 't' and row['status'] == 'draft' and row['openingConfirmed'] is False
+    assert row['tenantNo'] == 't' and row['status'] == 'active' and row['openingConfirmed'] is True
     assert client.put(path + '/' + row['id'], json={'totalQty': '-1'}).status_code == 422
     assert client.get(path + '/' + row['id']).json()['totalQty'] == '30.000000'
     app.dependency_overrides[auth] = lambda: {'role': 'admin', 'user_id': 'u', 'tenant_no': 'other'}
@@ -130,7 +130,8 @@ def test_total_adjust_requires_provenance_and_replays_without_duplicate_log(api)
     assert response.status_code == 200
     assert client.post(path, json=body).status_code == 200
     logs = client.get(base + '/' + row['id'] + '/logs').json()
-    assert len(logs) == 1 and logs[0]['totalDelta'] == '5.000000'
+    adjust_logs = [item for item in logs if item.get('eventType') == 'total_adjust']
+    assert len(adjust_logs) == 1 and adjust_logs[0]['totalDelta'] == '5.000000'
     assert client.post(path, json={**body,'totalQty':'40'}).status_code == 409
     assert client.post(path, json={**body,'operationId':'adjust-2','sourceNote':''}).status_code == 409
     assert client.get(base + '/' + row['id']).json()['totalQty'] == '35.000000'

backend/tests/test_pl_quota_cert_archive.py

@@ -41,14 +41,31 @@ def transport(**kw):
     return data
 
 
+def test_save_auto_enables_with_zero_opening(ctx):
+    """建档即启用：期初已用默认 0，无需手工期初确认。"""
+    db, u, s, m = ctx
+    row = s.save_cert(db, u, 'purchase', purchase())
+    assert row['status'] == 'active'
+    assert row['openingConfirmed'] is True
+    assert row['usedQty'] == Decimal('0')
+    assert row['verifiedBy'] == 'u'
+    assert row['verifiedTime'] is not None
+    log = db.query(m.PlCertQtyLog).filter_by(del_flag=0, event_type='opening').one()
+    assert (log.event_snapshot or {}).get('auto') is True
+    transport_row = s.save_cert(db, u, 'transport', transport())
+    assert transport_row['status'] == 'active' and transport_row['openingConfirmed'] is True
+
+
 def test_manual_optional_image_complete_fields(ctx):
     db,u,s,m = ctx
     data = purchase(buyerAddress='地址', buyerPhone='电话', buyerLegalRepresentative='甲', sellerAddress='乙址', sellerPhone='乙电话', sellerLegalRepresentative='乙', purchasePurpose='用途', issueDate='2026-01-01', validFrom='2026-01-01', validTo='2027-01-01')
     row = s.save_cert(db,u,'purchase',data)
     assert row['buyerAddress'] == '地址' and row['attachmentUrls'] is None
     assert row['totalQty'] == Decimal('100.123456')
-    assert row['status'] == 'draft'
-    assert db.query(m.PlCertQtyLog).count() == 0
+    assert row['status'] == 'active'
+    assert row['openingConfirmed'] is True
+    assert row['usedQty'] == Decimal('0')
+    assert db.query(m.PlCertQtyLog).filter_by(del_flag=0, event_type='opening').count() == 1
     db.rollback()
     assert db.query(m.PlPurchaseCert).count() == 0
 
@@ -56,9 +73,8 @@ def test_manual_optional_image_complete_fields(ctx):
 def test_opening_confirmation_remains_visible_after_reverification_required(ctx):
     db, user, svc, models = ctx
     row = svc.save_cert(db, user, 'purchase', purchase())
-    assert row['openingConfirmed'] is False
-    row = svc.confirm_cert(db, user, 'purchase', row['id'], {
-        'operationId': 'op-ui', 'usedQty': '0', 'singleUseConsumed': False, 'reason': '全新证'})
+    assert row['openingConfirmed'] is True
+    assert row['status'] == 'active'
     changed = svc.save_cert(db, user, 'purchase', {'buyerAddress': '更正地址'}, row['id'])
     assert changed['verifiedTime'] is None
     assert changed['openingConfirmed'] is True
@@ -89,10 +105,12 @@ def test_reference_delta_omitted_empty_and_invalid_prewrite(ctx):
 def test_confirmation_and_history_protection(ctx):
     db,u,s,m=ctx
     row=s.save_cert(db,u,'purchase',purchase())
+    assert row['status']=='active' and row['openingConfirmed'] is True
     with pytest.raises((HTTPException,ValidationError)): s.confirm_cert(db,u,'purchase',row['id'],{'usedQty':'0','singleUseConsumed':True,'reason':'核对','operationId':'op'})
     row=s.confirm_cert(db,u,'purchase',row['id'],{'usedQty':'0','singleUseConsumed':True,'reason':'核对','evidenceUrls':'https://minio/evidence','operationId':'op'})
     assert row['openingSingleUseConsumed']==1 and row['status']=='active'
-    assert db.query(m.PlCertQtyLog).one().event_type=='opening'
+    assert db.query(m.PlCertQtyLog).filter_by(del_flag=0).one().event_type=='opening'
+    # 建档自动期初后身份已锁定；手工覆盖期初后同样不可改核定量/证号。
     for data in ({'quantityUnit':'千克'},{'certNo':'new'},{'totalQty':'90'}):
         with pytest.raises(HTTPException): s.save_cert(db,u,'purchase',data,row['id'])
     with pytest.raises(HTTPException): s.delete_cert(db,u,'purchase',row['id'])
@@ -151,8 +169,9 @@ def test_opening_retry_preserves_balance_and_operation_identity(ctx):
     db.query(m.PlPurchaseCert).one().used_qty=Decimal('5')
     retry=s.confirm_cert(db,u,'purchase',row['id'],opening())
     assert retry['usedQty']==Decimal('5')
-    assert db.query(m.PlCertQtyLog).count()==1
-    assert db.query(m.PlCertOperation).count()==1
+    assert db.query(m.PlCertQtyLog).filter_by(del_flag=0).count()==1
+    # 自动期初操作号 + 手工期初操作号；重放不新增手工操作
+    assert db.query(m.PlCertOperation).filter_by(operation_id='opening').count()==1
     with pytest.raises(HTTPException): s.confirm_cert(db,u,'purchase',row['id'],opening(usedQty='7'))
     with pytest.raises(HTTPException): s.confirm_cert(db,u,'purchase',row['id'],opening(operationId='second'))
 
@@ -209,10 +228,11 @@ def test_reverify_after_correction_keeps_opening_and_balances(ctx):
     assert corrected['status']=='suspended' and corrected['verifiedBy'] is None
     result=s.verify_cert(db,u,'purchase',row['id'],{'operationId':'verify-1','reason':'地址更正核对','evidenceUrls':'https://minio/evidence'})
     assert result['status']=='active' and result['usedQty']==Decimal('5')
-    assert db.query(m.PlCertQtyLog).count()==1
-    assert db.query(m.PlCertOperation).count()==2
+    assert db.query(m.PlCertQtyLog).filter_by(del_flag=0).count()==1
+    # 自动期初 + 手工期初 + 复核
+    assert db.query(m.PlCertOperation).count()==3
     result=s.verify_cert(db,u,'purchase',row['id'],{'operationId':'verify-1','reason':'地址更正核对','evidenceUrls':'https://minio/evidence'})
-    assert db.query(m.PlCertOperation).count()==2
+    assert db.query(m.PlCertOperation).count()==3
 
 
 def test_actual_scope_denies_user_without_roles(ctx):
@@ -261,7 +281,7 @@ def test_historical_opening_accepts_written_source_without_image(ctx):
     row=s.save_cert(db,u,'purchase',purchase())
     result=s.confirm_cert(db,u,'purchase',row['id'],{'operationId':'written-source','usedQty':'5','singleUseConsumed':True,'reason':'历史核对','sourceNote':'2026年10月纸质台账第3页，经张三核实'})
     assert result['usedQty']==Decimal('5') and result['status']=='active'
-    assert db.query(m.PlCertQtyLog).one().event_snapshot['sourceNote'].startswith('2026')
+    assert db.query(m.PlCertQtyLog).filter_by(del_flag=0).one().event_snapshot['sourceNote'].startswith('2026')
 
 
 def test_unknown_use_text_is_preserved_and_requires_confirmation(ctx):
@@ -423,7 +443,7 @@ def test_opening_verify_stale_operation_snapshot_uses_unique_insert_arbitration(
     result=fn(db,u,'purchase',p['id'],body)
     assert missed and result['id']==p['id']
     assert db.query(m.PlCertOperation).count()==count
-    assert db.query(m.PlCertQtyLog).count()==1
+    assert db.query(m.PlCertQtyLog).filter_by(del_flag=0).count()==1
 
 
 def test_opening_and_verify_allow_blank_reason(ctx):
@@ -431,7 +451,7 @@ def test_opening_and_verify_allow_blank_reason(ctx):
     p=s.save_cert(db,u,'purchase',purchase())
     row=s.confirm_cert(db,u,'purchase',p['id'],{**opening(),'reason':' '})
     assert row['status']=='active'
-    assert db.query(m.PlCertQtyLog).one().reason==''
+    assert db.query(m.PlCertQtyLog).filter_by(del_flag=0).one().reason==''
     s.save_cert(db,u,'purchase',{'buyerAddress':'更正地址'},p['id'])
     verified=s.verify_cert(db,u,'purchase',p['id'],{'operationId':'verify-blank','reason':''})
     assert verified['status']=='active' and verified['verifiedTime']
@@ -440,10 +460,12 @@ def test_opening_and_verify_allow_blank_reason(ctx):
 def test_rejected_opening_rolls_back_claimed_operation(ctx):
     db,u,s,m=ctx
     p=s.save_cert(db,u,'purchase',purchase())
+    auto_ops = db.query(m.PlCertOperation).count()
+    auto_logs = db.query(m.PlCertQtyLog).filter_by(del_flag=0).count()
     with pytest.raises(HTTPException):
         s.confirm_cert(db,u,'purchase',p['id'],{**opening(),'usedQty':'999'})
-    assert db.query(m.PlCertOperation).count()==0
-    assert db.query(m.PlCertQtyLog).count()==0
+    assert db.query(m.PlCertOperation).count()==auto_ops
+    assert db.query(m.PlCertQtyLog).filter_by(del_flag=0).count()==auto_logs
 
 
 def test_replay_reads_current_certificate_and_opening_not_old_snapshot(ctx,monkeypatch):
@@ -472,7 +494,7 @@ def test_closed_certificate_keeps_history_and_cannot_be_reopened(ctx):
     used = confirmed['usedQty']
     closed = s.save_cert(db, u, 'purchase', {'status': 'closed'}, row['id'])
     assert closed['status'] == 'closed' and closed['usedQty'] == used
-    assert db.query(m.PlCertQtyLog).count() == 1
+    assert db.query(m.PlCertQtyLog).filter_by(del_flag=0).count() == 1
     for action, payload in (
         ('confirm', {**opening(), 'operationId': 'reopen-opening'}),
         ('verify', {'operationId': 'reopen-verify', 'reason': '重新启用'}),

backend/tests/test_pl_quota_cert_bindings.py

@@ -64,6 +64,44 @@ def test_create_with_bindings_and_bad_create_has_no_pending_row(case):
     assert result['id'] not in ('g1', 'g2')
 
 
+def test_create_with_empty_contract_bindings_auto_enables(case):
+    """前端保存常带 contractBindings:[]；必须与无合同集合路径一样自动启用并落库。"""
+    c = case
+    body = dict(
+        certNo='G-AUTO',
+        productNameOnCert='硫酸',
+        totalQty='300',
+        quantityUnit='吨',
+        validUseText='多次有效',
+        buyerName='买方',
+        sellerName='卖方',
+        contractBindings=[],
+    )
+    result = save(c, body, None)
+    assert result['status'] == 'active'
+    assert result['openingConfirmed'] is True
+    assert result['verifiedBy'] == 'u'
+    row = c.db.query(c.m.PlPurchaseCert).filter_by(id=result['id']).one()
+    assert row.status == 'active'
+    assert row.verified_by == 'u'
+    assert c.db.query(c.m.PlCertQtyLog).filter_by(
+        cert_id=result['id'], event_type='opening', del_flag=0,
+    ).count() == 1
+
+
+def test_get_cert_read_does_not_fake_enable_draft(case):
+    """只读详情/列表不得在未提交事务里把 draft 伪装成 active。"""
+    from app.services import pl_quota_cert_service as certs
+
+    c = case
+    row = c.db.query(c.m.PlPurchaseCert).filter_by(id='g1').one()
+    assert row.status == 'draft'
+    detail = certs.get_cert(c.db, c.user, 'purchase', 'g1')
+    assert detail['status'] == 'draft'
+    assert c.db.query(c.m.PlPurchaseCert).filter_by(id='g1').one().status == 'draft'
+    assert c.db.query(c.m.PlCertQtyLog).filter_by(cert_id='g1', del_flag=0).count() == 0
+
+
 def test_hidden_contract_cannot_be_cleared_but_face_can_change(case, monkeypatch):
     from app.services import pl_quota_cert_bindings as svc
     from app.models.crm_models import CrmContract

backend/tests/test_pl_quota_usage.py

@@ -245,7 +245,7 @@ def test_opening_consumed_and_expiry_vehicle_block(case):
     p.opening_single_use_consumed=0; p.valid_to=date(2000,1,1); c.db.flush()
     with pytest.raises(ValueError): run(c)
     p.valid_to=date(2099,12,31); c.db.flush()
-    with pytest.raises(ValueError, match='车号不在运输证范围'): run(c,vehicle_no='unknown')
+    with pytest.raises(ValueError, match='车牌号「unknown」不在运输证车号范围内'): run(c,vehicle_no='unknown')
 
 
 def test_reserve_uses_transport_cert_that_lists_the_vehicle(case):
@@ -262,9 +262,19 @@ def test_reserve_uses_transport_cert_that_lists_the_vehicle(case):
     assert c.db.get(c.m.PlPurchaseCert, 'p2').reserved_qty == 0
     transport_usages = c.db.query(c.m.PlPlanCertUsage).filter_by(cert_type='transport', del_flag=0).all()
     assert [u.cert_id for u in transport_usages if u.usage_status == 'reserved'] == ['t']
-    with pytest.raises(ValueError, match='车号不在运输证范围'):
+    with pytest.raises(ValueError, match='车牌号「NONE」不在运输证车号范围内'):
         run(c, operation_id='none', vehicle_no='NONE', purchase_cert_ids=['p', 'p2'], transport_cert_ids=['t', 't2'])
 
+
+def test_reserve_accepts_vehicle_no_with_trailing_spaces(case):
+    """计划车牌尾部空格不得误判为不在运输证车号范围。"""
+    c = case
+    # 证上车号无空格；计划侧常带尾部空格（与 PO20261115096 现场一致）
+    assert c.db.get(c.m.PlTransportCert, 't').vehicle_numbers_json == ['CAR']
+    run(c, vehicle_no='CAR ')
+    assert c.db.get(c.m.PlTransportCert, 't').reserved_qty == 29
+
+
 def test_no_cert_operation_is_still_idempotent(case):
     c=case
     first=run(c,sources=[],purchase_cert_ids=[],transport_cert_ids=[])
@@ -498,3 +508,50 @@ def test_new_current_usage_rejects_without_out_of_order_cert_lock(case,monkeypat
     assert c.db.get(c.m.PlPurchaseCert,'p').reserved_qty==29
     assert all(u.reserved_qty==29 for u in c.db.query(c.m.PlPlanCertUsage).all())
     assert c.db.query(c.m.PlCertOperation).count()==1
+
+
+def test_archive_cover_block_message_names_vehicle(monkeypatch):
+    """归档补证被车号挡住时，应点名车牌与运输证，而不是笼统额度不足。"""
+    from types import SimpleNamespace
+    from datetime import date
+    from app.services import pl_quota_usage_service as svc
+
+    transport = SimpleNamespace(
+        id='t1',
+        cert_no='Y-ARCH-1',
+        status='active',
+        verified_by='u',
+        verified_time=date(2026, 1, 1),
+        valid_from=date(2026, 1, 1),
+        valid_to=date(2027, 12, 31),
+        valid_use_text='多次有效',
+        vehicle_numbers_json=['皖A00000'],
+    )
+    r = SimpleNamespace(vehicle_no='皖A9BE91', on_date=date(2026, 11, 15))
+    plan = SimpleNamespace(id='plan1')
+    user = SimpleNamespace(tenant_no='t')
+
+    class OpeningQuery:
+        def first(self):
+            return SimpleNamespace(id='opening')
+
+    db = SimpleNamespace()
+    monkeypatch.setattr(
+        svc,
+        '_query',
+        lambda db, user, model, lock=True, **filters: OpeningQuery(),
+    )
+    monkeypatch.setattr(
+        'app.services.pl_quota_cert_service.ensure_zero_opening',
+        lambda db, user, kind, cert_id, row=None: row or transport,
+    )
+    monkeypatch.setattr(svc, 'usage_policy', lambda text: 'multi')
+    monkeypatch.setattr(svc, 'is_single_use_consumed', lambda *a, **k: False)
+
+    message = svc._archive_cover_block_message(
+        db, user, r, plan, {('transport', 't1'): transport}, [('transport', 't1')],
+    )
+    assert '皖A9BE91' in message
+    assert 'Y-ARCH-1' in message
+    assert '未包含车牌号' in message
+    assert '剩余额度不够覆盖本次实际数量' not in message

web-admin/scripts/cert-ready-default.test.ts

@@ -25,6 +25,20 @@ test('incomplete certificate filter is sent as quotaCertWaivedIn=0', () => {
   assert.deepEqual(params, { quotaCertWaivedIn: '0' })
 })
 
+test('fulfill tab selection column is first among table columns', () => {
+  const web = readFileSync(new URL('../src/views/pl/cert-ready/index.vue', import.meta.url), 'utf8')
+  const selectionIdx = web.indexOf('type="selection"')
+  const columnsIdx = web.indexOf('v-for="column in columns"')
+  const actionsIdx = web.indexOf('label="操作"')
+  assert.ok(selectionIdx > 0, 'selection column missing')
+  assert.ok(columnsIdx > selectionIdx, 'selection must precede business columns')
+  assert.ok(actionsIdx > columnsIdx, 'actions must follow business columns')
+  assert.match(
+    web,
+    /v-if="listMode === 'fulfill'"[\s\S]*?type="selection"[\s\S]*?v-for="column in columns"/
+  )
+})
+
 test('web funnel and miniprogram status dropdown default to incomplete', () => {
   const web = readFileSync(new URL('../src/views/pl/cert-ready/index.vue', import.meta.url), 'utf8')
   const mobile = readFileSync(

web-admin/scripts/customer-cert-quota-panel.test.ts

@@ -65,6 +65,38 @@ test('web customer quota panel uses toQueryParams from useServerColumnFilters',
   assert.match(panel, /watchVisibleColumns/)
 })
 
+test('web customer quota panel computes visibleDefs locally (not from composable)', () => {
+  const panel = readFileSync(
+    new URL('../src/views/pl/contract-quota/CustomerQuotaPanel.vue', import.meta.url),
+    'utf8'
+  )
+  assert.doesNotMatch(
+    panel,
+    /const \{[^}]*\bvisibleDefs\b[^}]*\} = usePersistedListColumns/
+  )
+  assert.match(
+    panel,
+    /const \{ visibleColumns, checkAll, isIndeterminate, resetToDefault \} = usePersistedListColumns/
+  )
+  assert.match(
+    panel,
+    /const visibleDefs = computed\(\(\) =>\s*columns\.filter\(\(column\) => visibleColumns\.value\.includes\(column\.prop\)\)/
+  )
+  assert.match(panel, /v-for="column in visibleDefs"/)
+})
+
+test('h5 customer tab renders customer fields without persisted column defs', () => {
+  const tab = readFileSync(
+    new URL('../src/views/m/pl/plan-cert/CustomerTab.vue', import.meta.url),
+    'utf8'
+  )
+  assert.doesNotMatch(tab, /usePersistedListColumns/)
+  assert.doesNotMatch(tab, /visibleDefs/)
+  assert.match(tab, /row\.customerName/)
+  assert.match(tab, /row\.customerNo/)
+  assert.match(tab, /row\.certCount/)
+})
+
 test('h5 customer quota tab does not call buildQuery', () => {
   const tab = readFileSync(
     new URL('../src/views/m/pl/plan-cert/CustomerTab.vue', import.meta.url),
@@ -73,3 +105,22 @@ test('h5 customer quota tab does not call buildQuery', () => {
   assert.doesNotMatch(tab, /\bbuildQuery\b/)
   assert.match(tab, /quotaStatisticsApi\.customers/)
 })
+
+test('web customer quota panel wires viewport height like StatisticsPanel', () => {
+  const panel = readFileSync(
+    new URL('../src/views/pl/contract-quota/CustomerQuotaPanel.vue', import.meta.url),
+    'utf8'
+  )
+  assert.match(panel, /useElementSize\(tableWrapRef\)/)
+  assert.match(
+    panel,
+    /const expansionStyle = computed\(\(\) => \(\{[\s\S]*width: tableViewportWidth/
+  )
+  assert.match(panel, /tableEl: tableWrapRef/)
+  assert.match(panel, /pagerEl: pagerRef/)
+  assert.doesNotMatch(panel, /\bwrapRef\s*:/)
+  assert.doesNotMatch(
+    panel,
+    /const \{[^}]*\bexpansionStyle\b[^}]*\} = useListTableViewportHeight/
+  )
+})

web-admin/scripts/pl-freight-calc-reset.test.ts

@@ -0,0 +1,91 @@
+import assert from 'node:assert/strict'
+import { existsSync, readFileSync } from 'node:fs'
+import test from 'node:test'
+import ts from 'typescript'
+import * as vue from 'vue'
+
+function loadFreightCalc() {
+  const path = new URL('../src/composables/usePlFreightCalc.ts', import.meta.url)
+  assert.ok(existsSync(path), 'usePlFreightCalc must exist')
+  const source = readFileSync(path, 'utf8')
+  const ast = ts.createSourceFile('usePlFreightCalc.ts', source, ts.ScriptTarget.Latest, true)
+  const code = ast.statements
+    .filter((node) => !ts.isImportDeclaration(node))
+    .map((node) => node.getText(ast))
+    .join('\n')
+  const compiled = ts.transpileModule(code, {
+    compilerOptions: { target: ts.ScriptTarget.ES2022, module: ts.ModuleKind.CommonJS }
+  }).outputText
+  const module = { exports: {} as Record<string, unknown> }
+  const requireStub = () => {
+    throw new Error('unexpected require in freight calc test harness')
+  }
+  // eslint-disable-next-line no-new-func
+  const factory = new Function(
+    'exports',
+    'require',
+    'module',
+    '__filename',
+    '__dirname',
+    'computed',
+    'ref',
+    'watch',
+    compiled + '\nmodule.exports = { usePlFreightCalc, sumProductActualQuantities };'
+  )
+  factory(
+    module.exports,
+    requireStub,
+    module,
+    'usePlFreightCalc.ts',
+    '.',
+    vue.computed,
+    vue.ref,
+    vue.watch
+  )
+  return module.exports as {
+    usePlFreightCalc: typeof import('../src/composables/usePlFreightCalc').usePlFreightCalc
+  }
+}
+
+test('resetFreightCalc re-sums transport qty from existing product actuals', async () => {
+  const { usePlFreightCalc } = loadFreightCalc()
+  const products = vue.ref([{ actualQty: 0.04, unit: '吨' }, { actualQty: 0.04, unit: '吨' }])
+  const hasFreight = vue.ref(1)
+  const freightType = vue.ref('by_weight')
+
+  const scope = vue.effectScope()
+  const calc = scope.run(() => usePlFreightCalc({ products, hasFreight, freightType }))
+  assert.ok(calc)
+  await vue.nextTick()
+  assert.equal(calc!.transportQty.value, 0.08)
+
+  calc!.resetFreightCalc()
+  await vue.nextTick()
+  assert.equal(calc!.transportQtyTouched.value, false)
+  assert.equal(
+    calc!.transportQty.value,
+    0.08,
+    'reset after products already loaded must not leave transport qty empty'
+  )
+  scope.stop()
+})
+
+test('resetFreightCalc restores product sum after manual transport qty edit', async () => {
+  const { usePlFreightCalc } = loadFreightCalc()
+  const products = vue.ref([{ actualQty: 0.04, unit: '吨' }])
+  const hasFreight = vue.ref(1)
+  const freightType = vue.ref('by_weight')
+
+  const scope = vue.effectScope()
+  const calc = scope.run(() => usePlFreightCalc({ products, hasFreight, freightType }))
+  assert.ok(calc)
+  await vue.nextTick()
+
+  calc!.transportQty.value = 9.9
+  calc!.markTransportQtyTouched()
+  calc!.resetFreightCalc()
+  await vue.nextTick()
+  assert.equal(calc!.transportQtyTouched.value, false)
+  assert.equal(calc!.transportQty.value, 0.04)
+  scope.stop()
+})

web-admin/src/composables/usePlFreightCalc.ts

@@ -86,11 +86,13 @@ export function usePlFreightCalc(options: {
   )
 
   function resetFreightCalc() {
-    transportQty.value = null
     transportQtyTouched.value = false
     freightAmount.value = null
     freightAmountTouched.value = false
     freightPrice.value = null
+    // 产品已加载时 watch 不会因 reset 重跑；须立即按当前实际数量回填，
+    // 否则归档弹窗再次打开会误报「运输量须大于0」。
+    transportQty.value = defaultTransportQty.value || null
   }
 
   function markTransportQtyTouched() {

web-admin/src/views/m/pl/plan-cert/ArchiveTab.vue

@@ -60,12 +60,12 @@
           >编辑</button
         >
         <button
-          v-if="canEdit && row.status !== 'void' && row.status !== 'closed'"
+          v-if="canEdit && row.openingConfirmed && row.status !== 'void' && row.status !== 'closed'"
           class="mp-wv-btn"
           type="button"
           @click="confirmRow = row"
         >
-          {{ row.openingConfirmed ? '复核' : '期初确认' }}
+          复核
         </button>
       </div>
     </article>

web-admin/src/views/m/pl/plan-cert/CertConfirmSheet.vue

@@ -8,37 +8,21 @@
     :close-on-click-overlay="!saving"
     @update:show="onShow"
   >
-    <van-nav-bar
-      :title="row?.openingConfirmed ? '复核证件' : '期初核对'"
-      left-text="关闭"
-      @click-left="!saving && emit('close')"
-    />
+    <van-nav-bar title="复核证件" left-text="关闭" @click-left="!saving && emit('close')" />
     <div v-if="row" class="mp-cert-confirm-sheet__body mp-cert-form">
-      <p class="plan-empty-hint">{{ row.certNo }} · {{ row.openingConfirmed ? '复核已启用的证件' : '核对历史用量后启用' }}</p>
-      <template v-if="!row.openingConfirmed">
-        <van-field v-model="opening.usedQty" label="历史已用量" label-align="top" type="number" :disabled="saving" />
-        <van-cell title="一次有效证已有使用记录" center>
-          <template #right-icon>
-            <van-switch v-model="opening.singleUseConsumed" :disabled="saving" size="20px" />
-          </template>
-        </van-cell>
-        <van-field v-model="opening.sourceNote" label="历史来源说明" label-align="top" type="textarea" rows="2" :disabled="saving" />
-        <p class="plan-empty-hint">核对凭据</p>
-        <AttachmentUploadField v-model="opening.evidenceUrls" folder="pl/quota-cert" :disabled="saving" hint="" />
-      </template>
+      <p class="plan-empty-hint">{{ row.certNo }} · 复核已启用的证件</p>
       <van-field v-model="opening.reason" label="核对说明" label-align="top" type="textarea" rows="2" :disabled="saving" />
     </div>
     <footer>
       <button class="mp-wv-btn" type="button" :disabled="saving" @click="emit('close')">取消</button>
       <button class="mp-wv-btn mp-wv-btn--primary" type="button" :disabled="saving" @click="submit">
-        {{ saving ? '正在提交…' : '确认启用' }}
+        {{ saving ? '正在提交…' : '确认复核' }}
       </button>
     </footer>
   </van-popup>
 </template>
 <script setup lang="ts">
   import { ElMessage } from 'element-plus'
-  import AttachmentUploadField from '@/components/business/attachment-upload-field/index.vue'
   import { quotaCertApi, type CertType, type QuotaCert } from '@/api/pl/quota-cert'
   import { randomOperationId } from '@/utils/random-id'
   import './cert-mobile.scss'
@@ -76,19 +60,15 @@
     opening.reason = opening.reason.trim()
     saving.value = true
     try {
-      if (props.row.openingConfirmed) {
-        await quotaCertApi.verify(kind, props.row.id, {
-          operationId: opening.operationId,
-          reason: opening.reason
-        })
-      } else {
-        await quotaCertApi.confirm(kind, props.row.id, { ...opening })
-      }
-      ElMessage.success(props.row.openingConfirmed ? '证件已复核' : '期初已确认')
+      await quotaCertApi.verify(kind, props.row.id, {
+        operationId: opening.operationId,
+        reason: opening.reason
+      })
+      ElMessage.success('证件已复核')
       emit('saved')
       emit('close')
     } catch {
-      ElMessage.error('核对没有提交，请重试')
+      ElMessage.error('复核没有提交，请重试')
     } finally {
       saving.value = false
     }

web-admin/src/views/pl/cert-ready/index.vue

@@ -50,6 +50,14 @@
             class="list-table"
             @selection-change="onSelectionChange"
           >
+            <el-table-column
+              v-if="listMode === 'fulfill'"
+              type="selection"
+              width="48"
+              align="center"
+              header-align="center"
+              :reserve-selection="false"
+            />
             <el-table-column
               v-for="column in columns"
               :key="column.prop"
@@ -93,12 +101,6 @@
                 <span v-else>{{ row[column.prop] || '—' }}</span>
               </template>
             </el-table-column>
-            <el-table-column
-              v-if="listMode === 'fulfill'"
-              type="selection"
-              width="48"
-              :reserve-selection="false"
-            />
             <el-table-column label="操作" min-width="260" fixed="right">
               <template #default="{ row }">
                 <template v-if="listMode === 'fulfill'">

web-admin/src/views/pl/contract-quota/CustomerQuotaPanel.vue

@@ -183,16 +183,28 @@
     toQueryParams,
     watchVisibleColumns
   } = useServerColumnFilters(columns, { onChange: search })
-  const { visibleColumns, checkAll, isIndeterminate, visibleDefs, resetToDefault } =
-    usePersistedListColumns('pl-customer-quota-columns-v1', columns)
+  const { visibleColumns, checkAll, isIndeterminate, resetToDefault } = usePersistedListColumns(
+    'pl-customer-quota-columns-v1',
+    columns
+  )
+  // usePersistedListColumns 不返回 visibleDefs；须本地按可见列过滤，否则只剩展开列
+  const visibleDefs = computed(() =>
+    columns.filter((column) => visibleColumns.value.includes(column.prop))
+  )
   watchVisibleColumns(visibleColumns)
-  const tableWrapRef = ref<HTMLElement>()
-  const pagerRef = ref<HTMLElement>()
+  const tableWrapRef = ref<HTMLElement | null>(null)
+  const pagerRef = ref<HTMLElement | null>(null)
   const tableRef = ref()
-  const { tableHeight, expansionStyle } = useListTableViewportHeight({
-    wrapRef: tableWrapRef,
-    pagerRef,
-    tableRef
+  // 展开区只锁宽度，高度随证件明细撑开；表体高度须走 tableEl/pagerEl，否则卡在 240 下限裁切展开
+  const { width: tableViewportWidth } = useElementSize(tableWrapRef)
+  const expansionStyle = computed(() => ({
+    width: tableViewportWidth.value > 0 ? `${tableViewportWidth.value}px` : undefined
+  }))
+  const { tableHeight, recalc: recalcTableHeight } = useListTableViewportHeight({
+    tableEl: tableWrapRef,
+    pagerEl: pagerRef,
+    tableRef,
+    enabled: computed(() => !props.embedded)
   })
 
   async function init() {
@@ -258,6 +270,7 @@
       ? expanded.some((item: any) => item?.id === row.id)
       : Boolean(expanded)
     if (open) void loadDetail(row)
+    else nextTick(() => recalcTableHeight())
   }
 
   async function loadDetail(row: { id: string }) {
@@ -267,6 +280,12 @@
       details[row.id] = { loading: false, ready: true }
     } catch {
       details[row.id] = { loading: false, error: true }
+    } finally {
+      nextTick(() => {
+        recalcTableHeight()
+        const inst = tableRef.value
+        if (inst && typeof inst.doLayout === 'function') inst.doLayout()
+      })
     }
   }
 
@@ -291,6 +310,9 @@
   }
 
   .statistics-expansion {
-    padding: 8px 12px 16px;
+    box-sizing: border-box;
+    max-width: 100%;
+    padding: 12px 16px 16px;
+    text-align: left;
   }
 </style>

web-admin/src/views/pl/quota-cert/index.vue

@@ -93,11 +93,10 @@
                   ><el-icon><Edit /></el-icon>编辑</el-button
                 >
                 <el-button
-                  v-if="canEdit && row.status !== 'void' && row.status !== 'closed'"
+                  v-if="canEdit && row.openingConfirmed && row.status !== 'void' && row.status !== 'closed'"
                   size="small"
                   @click="openConfirm(row)"
-                  ><el-icon><Check /></el-icon
-                  >{{ row.openingConfirmed ? '复核' : '期初确认' }}</el-button
+                  ><el-icon><Check /></el-icon>复核</el-button
                 >
                 <el-button size="small" @click="view(row)"
                   ><el-icon><View /></el-icon>查看</el-button
@@ -591,35 +590,12 @@
     </SystemViewDialog>
     <el-dialog
       v-model="confirmVisible"
-      :title="confirmRow?.openingConfirmed ? '复核证件' : '期初核对'"
+      title="复核证件"
       width="640px"
       align-center
       class="dialog-form-layout quota-cert-confirm-dialog"
     >
       <el-form label-position="top">
-        <template v-if="!confirmRow?.openingConfirmed">
-          <div class="form-grid-2">
-            <el-form-item>
-              <template #label>
-                历史已用量（证面单位）<span v-if="confirmUnit" class="form-label-tip">{{
-                  confirmUnit
-                }}</span>
-              </template>
-              <el-input v-model="opening.usedQty">
-                <template v-if="confirmUnit" #append>{{ confirmUnit }}</template>
-              </el-input>
-            </el-form-item>
-            <el-form-item label="一次有效证是否已有使用记录">
-              <el-switch v-model="opening.singleUseConsumed" />
-            </el-form-item>
-          </div>
-          <el-form-item label="历史来源说明">
-            <el-input v-model="opening.sourceNote" type="textarea" />
-          </el-form-item>
-          <el-form-item label="核对凭据">
-            <AttachmentUploadField v-model="opening.evidenceUrls" folder="pl/quota-cert" />
-          </el-form-item>
-        </template>
         <el-form-item label="核对说明"
           ><el-input v-model="opening.reason" type="textarea"
         /></el-form-item>
@@ -628,7 +604,7 @@
         ><el-button @click="confirmVisible = false"
           ><el-icon><Close /></el-icon>取消</el-button
         ><el-button type="primary" :loading="saving" @click="confirm"
-          ><el-icon><Check /></el-icon>确认启用</el-button
+          ><el-icon><Check /></el-icon>确认复核</el-button
         ></template
       >
     </el-dialog>
@@ -1818,12 +1794,10 @@
     saving.value = true
     try {
       const targetKind = rowCertType(confirmRow.value)
-      if (confirmRow.value.openingConfirmed)
-        await quotaCertApi.verify(targetKind, confirmRow.value.id, {
-          operationId: opening.operationId,
-          reason: opening.reason
-        })
-      else await quotaCertApi.confirm(targetKind, confirmRow.value.id, opening)
+      await quotaCertApi.verify(targetKind, confirmRow.value.id, {
+        operationId: opening.operationId,
+        reason: opening.reason
+      })
       confirmVisible.value = false
       await load()
     } finally {