feat(pl): Maintain legal use proofs per stop product while preserving scoped history.

backend/app/api/pl/quota_cert.py

@@ -16,12 +16,13 @@ from app.core.oper_log_decorator import oper_log
 from app.core.tenant_context import get_tenant_no
 from app.schemas.pl.quota_cert import InputBody, CertConfirmBody, CertVerifyBody, CertTotalAdjustBody
 from app.schemas.pl.quota_scope import ContractCertBindingsBody
-from app.schemas.pl.legal_use_document import LegalDocumentsBody
+from app.schemas.pl.legal_use_document import LegalDocumentsBody, StopLegalDocumentsBody
 from app.schemas.pl.quota_planning import PlanQuotaConfiguration, PlanQuotaPreview
 from app.services import pl_quota_cert_service as certs
 from app.services import pl_quota_contract_service as scopes
 from app.services import pl_quota_contract_query as contract_query
 from app.services import pl_legal_use_document as documents
+from app.services import pl_stop_legal_document as stop_documents
 from app.services import pl_quota_planning as planning
 from app.services import pl_quota_adjustment as adjustment
 from app.services import pl_quota_overview as overview
@@ -225,6 +226,28 @@ def list_documents(keyword: Optional[str] = None, status: Optional[str] = None,
     return documents.list_documents(db, user, keyword, status)
 
 
+@legal_router.get('/stops')
+def list_stop_documents(keyword: Optional[str] = None, status: Optional[str] = None,
+                        contract_id: Optional[str] = Query(None, alias='contractId'),
+                        page: int = Query(1, ge=1), page_size: int = Query(20, alias='pageSize', ge=1, le=100),
+                        db: Session = Depends(get_db), user: DbUserContext = Depends(current_user)):
+    return _wire(stop_documents.list_documents(db, user, keyword, status, contract_id, page, page_size))
+
+
+@legal_router.get('/stops/{plan_id}')
+def get_stop_documents(plan_id: str, contract_id: Optional[str] = Query(None, alias='contractId'),
+                       db: Session = Depends(get_db), user: DbUserContext = Depends(current_user)):
+    return _wire(stop_documents.get_documents(db, user, plan_id, contract_id))
+
+
+@legal_router.patch('/stops/{plan_id}')
+@oper_log('逐段合法使用证明', business_type=2, path_param='plan_id')
+def save_stop_documents(plan_id: str, body: StopLegalDocumentsBody,
+                        contract_id: Optional[str] = Query(None, alias='contractId'),
+                        db: Session = Depends(get_db), user: DbUserContext = Depends(current_user)):
+    return _write(db, lambda: stop_documents.save_documents(db, user, plan_id, body, contract_id))
+
+
 @legal_router.get('/{plan_id}')
 def get_documents(plan_id: str, db: Session = Depends(get_db), user: DbUserContext = Depends(current_user)):
     return documents.get_documents(db, user, plan_id)

backend/app/models/pl_quota_models.py

@@ -299,3 +299,27 @@ class PlCertOperation(Base):
     __table_args__ = (
         UniqueConstraint('tenant_no', 'operation_id', name='uk_cert_request'),
     )
+
+
+class PlStopLegalDocument(Base):
+    """逐段产品证明；稳定身份不随附件更新重建。"""
+    __tablename__ = 'pl_stop_legal_document'
+    id = Column(String(64), primary_key=True)
+    plan_id = Column(String(64), nullable=False)
+    stop_type = Column(String(16), nullable=False)
+    stop_product_id = Column(String(64), nullable=False)
+    legal_use_cert_url = Column(Text, nullable=True)
+    uploaded_by = Column(String(64), nullable=True)
+    uploaded_time = Column(DateTime, nullable=True)
+    tenant_no = Column(String(64), nullable=False)
+    order_num = Column(Integer, nullable=True)
+    remark = Column(String(1000), nullable=True)
+    create_by = Column(String(64), nullable=True)
+    create_dept_id = Column(String(64), nullable=False, default='')
+    create_time = Column(DateTime, nullable=False, default=datetime.now)
+    update_by = Column(String(64), nullable=True)
+    update_time = Column(DateTime, nullable=False, default=datetime.now, onupdate=datetime.now)
+    del_flag = Column(SmallInteger, nullable=False, default=0)
+    __table_args__ = (
+        UniqueConstraint('tenant_no', 'plan_id', 'stop_type', 'stop_product_id', name='uk_stop_legal_document'),
+    )

backend/app/schemas/pl/legal_use_document.py

@@ -1,4 +1,4 @@
-from typing import List, Optional
+from typing import List, Optional, Literal
 from pydantic import Field
 from app.schemas.pl.quota_cert import InputBody
 
@@ -10,3 +10,13 @@ class LegalDocumentProduct(InputBody):
 
 class LegalDocumentsBody(InputBody):
     products: Optional[List[LegalDocumentProduct]] = None
+
+
+class StopLegalDocument(InputBody):
+    stop_type: Literal['load', 'delivery']
+    stop_product_id: str = Field(min_length=1, max_length=64)
+    legal_use_cert_url: Optional[str] = Field(None, max_length=2000)
+
+
+class StopLegalDocumentsBody(InputBody):
+    items: Optional[List[StopLegalDocument]] = None

backend/app/services/pl_child_references.py

@@ -15,6 +15,13 @@ def assert_stop_products_removable(db, tenant_no, plan_id, stop_type, stop_id, r
     if not rows:
         return
     ids = {row.id for row in rows}
+    from app.models.pl_quota_models import PlStopLegalDocument
+    if inspect(db.connection()).has_table(PlStopLegalDocument.__tablename__):
+        if _exists(db, PlStopLegalDocument, tenant_no,
+                   PlStopLegalDocument.plan_id == plan_id, PlStopLegalDocument.stop_type == stop_type,
+                   PlStopLegalDocument.stop_product_id.in_(ids),
+                   PlStopLegalDocument.legal_use_cert_url.isnot(None), PlStopLegalDocument.legal_use_cert_url != ''):
+            raise HTTPException(409, '段产品已有合法使用证明，不能移除或改变身份')
     if any(any(getattr(row, field, None) is not None for field in
                ('inbound_qty', 'inbound_price', 'inbound_amount', 'outbound_qty', 'outbound_price', 'outbound_amount'))
            or bool(getattr(row, 'archive_remark', None)) for row in rows):

backend/app/services/pl_plan_stop_service.py

@@ -1361,7 +1361,10 @@ def prepare_plan_stops_write(
                     if matches:
                         row = matches[0]
                         ids.add(row.id)
-                if row and (row.plan_product_id != linked or row.product_id != product.product_id):
+                if row and (row.plan_product_id != linked or row.product_id != product.product_id
+                            or any((getattr(row, field) or '') != (getattr(product, field) or '')
+                                   for field in ('contract_id', 'contract_product_id')
+                                   if field in product.model_fields_set)):
                     assert_stop_products_removable(db, tenant_no, plan_id, kind, stop_id, [row])
                 product_jobs.append((row, product, linked))
             removed = [row for row in old if row.id not in ids]

backend/app/services/pl_stop_legal_document.py

@@ -0,0 +1,185 @@
+"""逐装送段产品维护证明；权限筛选先于汇总、分页，不参与额度或归档。"""
+from datetime import datetime
+from uuid import uuid4
+
+from fastapi import HTTPException
+from pydantic import BaseModel
+from sqlalchemy import and_, or_, func, case
+
+from app.crud.crm.data_scope_helpers import apply_crm_owner_scope
+from app.models.crm_models import CrmContract
+from app.models.pl_models import PlPlan, PlPlanProduct, PlPlanLoad, PlPlanLoadProduct, PlPlanDelivery, PlPlanDeliveryProduct
+from app.models.pl_quota_models import PlStopLegalDocument as Document
+from app.schemas.pl.legal_use_document import StopLegalDocumentsBody
+from app.services.pl_data_scope import apply_pl_scope
+from app.services.pl_legal_use_document import _plan
+from app.services.pl_quota_contract_service import _contract
+from app.services.pl_plan_cert_perm import user_can_upload_plan_cert
+from app.services.pl_quota_attachments import authorize_attachment, storage_base_url
+
+SIDES = (('load', PlPlanLoadProduct, PlPlanLoad, 'load_id'),
+         ('delivery', PlPlanDeliveryProduct, PlPlanDelivery, 'delivery_id'))
+
+
+def _query(db, user, side, model, parent, field, contract_id=None):
+    contracts = db.query(CrmContract.id).filter_by(tenant_no=user.tenant_no, del_flag=0)
+    contracts = apply_crm_owner_scope(contracts, CrmContract, user.to_user_info(), db, user.tenant_no)
+    query = db.query(model, parent, PlPlanProduct, CrmContract, Document).join(
+        parent, and_(getattr(model, field) == parent.id, model.plan_id == parent.plan_id,
+                     parent.tenant_no == user.tenant_no, parent.del_flag == 0)).join(
+        PlPlanProduct, and_(model.plan_product_id == PlPlanProduct.id, model.plan_id == PlPlanProduct.plan_id,
+                           model.product_id == PlPlanProduct.product_id,
+                           PlPlanProduct.tenant_no == user.tenant_no, PlPlanProduct.del_flag == 0,
+                           PlPlanProduct.is_explosive == 1)).outerjoin(
+        CrmContract, and_(model.contract_id == CrmContract.id, CrmContract.tenant_no == user.tenant_no,
+                          CrmContract.del_flag == 0)).outerjoin(
+        Document, and_(Document.plan_id == model.plan_id, Document.stop_type == side,
+                       Document.stop_product_id == model.id, Document.tenant_no == user.tenant_no,
+                       Document.del_flag == 0)).filter(
+        model.tenant_no == user.tenant_no, model.del_flag == 0,
+        or_(model.contract_id.is_(None), model.contract_id == '', model.contract_id.in_(contracts)))
+    if contract_id:
+        query = query.filter(model.contract_id == contract_id)
+    return query
+
+
+def _unlinked_legacy(db, user):
+    """无有效段的旧附件只在全局计划范围展示，不猜测合同归属。"""
+    query = db.query(PlPlanProduct).filter_by(tenant_no=user.tenant_no, del_flag=0, is_explosive=1).filter(
+        PlPlanProduct.legal_use_cert_url.isnot(None), PlPlanProduct.legal_use_cert_url != '')
+    for _, model, parent, field in SIDES:
+        linked = db.query(model.id).join(parent, getattr(model, field) == parent.id).filter(
+            model.plan_product_id == PlPlanProduct.id, model.plan_id == PlPlanProduct.plan_id,
+            model.tenant_no == user.tenant_no, model.del_flag == 0,
+            parent.tenant_no == user.tenant_no, parent.del_flag == 0, parent.plan_id == model.plan_id)
+        query = query.filter(~linked.exists())
+    return query
+
+
+def _legacy_row(product):
+    return {'id': product.id, 'productId': product.product_id, 'productName': product.product_name,
+            'model': product.model, 'legalUseCertUrl': product.legal_use_cert_url}
+
+
+def _details(db, user, plans, contract_id=None, lock=False):
+    result = {p.id: {'id': p.id, 'planNo': p.plan_no, 'planDate': p.plan_date,
+                     'licensePlate': p.license_plate, 'transportStatus': p.transport_status,
+                     'archiveStatus': p.archive_status, 'items': [], 'legacyProducts': []} for p in plans}
+    legacy = {p.id: {} for p in plans}
+    for side, model, parent, field in SIDES:
+        query = _query(db, user, side, model, parent, field, contract_id).filter(model.plan_id.in_(result))
+        if lock:
+            query = query.with_for_update().populate_existing()
+        for row, stop, product, contract, document in query.order_by(parent.order_num, parent.id, model.order_num, model.id).all():
+            url = document.legal_use_cert_url if document else None
+            result[row.plan_id]['items'].append({
+                'stopType': side, 'stopId': stop.id, 'stopOrder': getattr(stop, 'seq', None) or stop.order_num or 1,
+                'stopProductId': row.id, 'productId': row.product_id, 'productName': row.product_name,
+                'model': row.model, 'unit': row.unit, 'contractId': row.contract_id,
+                'contractNo': contract.contract_no if contract else None, 'contractProductId': row.contract_product_id,
+                'partyName': stop.customer_name, 'address': stop.address, 'legalUseCertUrl': url,
+                'documentStatus': 'uploaded' if url else 'pending',
+                'uploadedBy': document.uploaded_by if document else None,
+                'uploadedTime': document.uploaded_time if document else None})
+            if product.legal_use_cert_url:
+                legacy[row.plan_id][product.id] = _legacy_row(product)
+    if not contract_id:
+        for product in _unlinked_legacy(db, user).filter(PlPlanProduct.plan_id.in_(result)).all():
+            legacy[product.plan_id][product.id] = _legacy_row(product)
+    for pid, value in result.items():
+        value['requiredCount'] = len(value['items'])
+        value['uploadedCount'] = sum(r['documentStatus'] == 'uploaded' for r in value['items'])
+        value['legacyProducts'] = list(legacy[pid].values())
+        value['documentStatus'] = ('not_required' if not value['requiredCount'] else
+                                   'uploaded' if value['uploadedCount'] == value['requiredCount'] else 'pending')
+    return result
+
+
+def get_documents(db, user, plan_id, contract_id=None):
+    plan = _plan(db, user, plan_id)
+    if contract_id:
+        _contract(db, user, contract_id)
+    return _details(db, user, [plan], contract_id)[plan_id]
+
+
+def list_documents(db, user, keyword=None, status=None, contract_id=None, page=1, page_size=20):
+    if status not in (None, 'pending', 'uploaded') or page < 1 or not 1 <= page_size <= 100:
+        raise HTTPException(400, '无效资料状态或分页参数')
+    if contract_id:
+        _contract(db, user, contract_id)
+    queries = [_query(db, user, side, model, parent, field, contract_id).with_entities(
+        model.plan_id.label('plan_id'), case((and_(Document.legal_use_cert_url.isnot(None),
+        Document.legal_use_cert_url != ''), 1), else_=0).label('uploaded'))
+        for side, model, parent, field in SIDES]
+    sources = queries[0].union_all(queries[1]).subquery()
+    counts = db.query(sources.c.plan_id, func.count().label('required'),
+                      func.sum(sources.c.uploaded).label('uploaded')).group_by(sources.c.plan_id).subquery()
+    query = db.query(PlPlan).outerjoin(counts, counts.c.plan_id == PlPlan.id).filter(
+        PlPlan.tenant_no == user.tenant_no, PlPlan.del_flag == 0)
+    if contract_id:
+        query = query.filter(counts.c.required > 0)
+    else:
+        query = query.filter(or_(counts.c.required > 0, PlPlan.id.in_(
+            _unlinked_legacy(db, user).with_entities(PlPlanProduct.plan_id))))
+    query = apply_pl_scope(query, PlPlan, user, db, user.tenant_no)
+    if keyword:
+        query = query.filter(PlPlan.plan_no.contains(keyword))
+    if status == 'pending':
+        query = query.filter(counts.c.uploaded < counts.c.required)
+    elif status == 'uploaded':
+        query = query.filter(counts.c.uploaded == counts.c.required)
+    total = query.count()
+    plans = query.order_by(PlPlan.create_time.desc(), PlPlan.id).offset((page - 1) * page_size).limit(page_size).all()
+    details = _details(db, user, plans, contract_id) if plans else {}
+    return {'list': [details[p.id] for p in plans], 'total': total, 'page': page, 'pageSize': page_size}
+
+
+def save_documents(db, user, plan_id, data, contract_id=None):
+    if not user_can_upload_plan_cert(db, user):
+        raise HTTPException(403, '无权维护合法使用证明')
+    body = StopLegalDocumentsBody(**(data.model_dump(exclude_unset=True) if isinstance(data, BaseModel) else data))
+    plan = _plan(db, user, plan_id, lock=True)
+    if contract_id:
+        _contract(db, user, contract_id, lock=True)
+    if 'items' in body.model_fields_set and body.items is None:
+        raise HTTPException(400, '段产品集合不可为null')
+    details = _details(db, user, [plan], contract_id, lock=True)[plan_id]
+    valid = {(r['stopType'], r['stopProductId']) for r in details['items']}
+    # 包含软删除行以复用稳定唯一键；清空不会删除记录或创建信息。
+    records = {(r.stop_type, r.stop_product_id): r for r in db.query(Document).filter_by(
+        tenant_no=user.tenant_no, plan_id=plan_id).with_for_update().populate_existing().all()}
+    seen, changes = set(), []
+    for item in body.items or []:
+        key = (item.stop_type, item.stop_product_id)
+        if key in seen or key not in valid:
+            raise HTTPException(400, '段产品重复、不适用或不在可访问计划合同范围内')
+        seen.add(key)
+        if 'legal_use_cert_url' not in item.model_fields_set:
+            continue
+        row = records.get(key)
+        old = row.legal_use_cert_url if row and row.del_flag == 0 else ''
+        urls = list(dict.fromkeys(v.strip() for v in (item.legal_use_cert_url or '').split(',') if v.strip()))
+        for url in urls:
+            if url not in (old or '').split(','):
+                authorize_attachment(db, user, url, storage_base_url())
+        value = ','.join(urls)
+        if value != (old or ''):
+            changes.append((key, row, value))
+    for key, row, value in changes:
+        if row is None:
+            row = Document(id=str(uuid4()), tenant_no=user.tenant_no, plan_id=plan_id,
+                           stop_type=key[0], stop_product_id=key[1], create_by=user.id,
+                           create_dept_id=user.dept_id or '')
+            db.add(row)
+        row.del_flag = 0
+        row.legal_use_cert_url = value
+        row.update_by = user.id
+        row.uploaded_by = user.id if value else None
+        row.uploaded_time = datetime.now() if value else None
+    if changes:
+        from app.services.pl_plan_service import record_plan_change
+        record_plan_change(db, tenant_no=user.tenant_no, plan_id=plan_id, change_type='legal_use_document',
+                           change_content='维护逐段合法使用证明：' + '、'.join(k[0] + ':' + k[1] for k, _, _ in changes),
+                           user_id=user.id, user_name=getattr(user, 'name', None), dept_id=user.dept_id or '')
+    db.flush()
+    return get_documents(db, user, plan_id, contract_id)

backend/tests/test_pl_child_differential_save.py

@@ -317,7 +317,7 @@ def test_update_api_partial_product_edit_retains_plan_total(db,monkeypatch):
     plan=m.PlPlan(id='p',tenant_no='t',plan_no='P',create_date=date(2026,10,27),belongs_company='shanxia',plan_type='inbound',total_amount=120)
     db.add(plan);db.flush()
     monkeypatch.setattr(api,'get_tenant_no',lambda *args:'t')
-    monkeypatch.setattr(api,'_get_plan',lambda *args:plan)
+    monkeypatch.setattr(api,'_get_plan',lambda *args,**kwargs:plan)
     monkeypatch.setattr(api,'collect_hazard_product_ids',lambda *args:set())
     monkeypatch.setattr(api,'resolve_hazard_flags_from_crm',lambda *args,**kwargs:(0,0))
     monkeypatch.setattr(api,'sync_linked_dibang_dates_for_plan',lambda *args,**kwargs:[])

backend/tests/test_pl_stop_legal_document.py

@@ -0,0 +1,214 @@
+from datetime import date
+from types import SimpleNamespace
+
+import pytest
+from fastapi import HTTPException
+from sqlalchemy import create_engine
+from sqlalchemy.orm import Session
+
+
+@pytest.fixture
+def case(monkeypatch):
+    from app.models import pl_models as m
+    from app.models.crm_models import CrmContract, SysOss
+    from app.models.pl_quota_models import PlStopLegalDocument
+    from app.services import pl_stop_legal_document as svc
+    engine = create_engine('sqlite://')
+    for model in (m.PlPlan, m.PlPlanProduct, m.PlPlanLoad, m.PlPlanLoadProduct,
+                  m.PlPlanDelivery, m.PlPlanDeliveryProduct, m.PlPlanChange,
+                  m.PlPlanContract, m.PlPlanProductBrandQty, m.PlTransportDriverStopFeedback,
+                  m.PlTransportSiteStopFeedback, m.PlTransportDbLink, CrmContract, SysOss, PlStopLegalDocument):
+        model.__table__.create(engine)
+    db = Session(engine)
+    user = SimpleNamespace(id='u', name='测试', dept_id='d', tenant_no='t', role='super_admin',
+                           to_user_info=lambda: {'role': 'super_admin', 'user_id': 'u'})
+    db.add(m.PlPlan(id='p', plan_no='P', create_date=date.today(), belongs_company='shanxia',
+                   plan_type='direct', tenant_no='t', transport_status='arranged', archive_status='archived'))
+    db.add(m.PlPlanProduct(id='pp', plan_id='p', product_id='sku', product_no='S', product_name='产品',
+                           is_explosive=1, legal_use_cert_url='legacy.jpg', tenant_no='t'))
+    db.add(CrmContract(id='c', contract_no='C', contract_name='合同', customer_id='customer', contract_type='sale', tenant_no='t'))
+    for side, rid, cid in [('load', 'l', None), ('delivery', 'd1', 'c'), ('delivery', 'd2', None)]:
+        parent, product = (m.PlPlanLoad, m.PlPlanLoadProduct) if side == 'load' else (m.PlPlanDelivery, m.PlPlanDeliveryProduct)
+        db.add(parent(id=rid, plan_id='p', tenant_no='t', customer_name=rid))
+        db.add(product(id=rid+'p', plan_id='p', plan_product_id='pp', product_id='sku', product_no='S',
+                       product_name='产品', tenant_no='t', contract_id=cid, **{side+'_id': rid}))
+    db.add(SysOss(id='o', url='legal/new.jpg', tenant_no='t', create_by='u'))
+    db.commit()
+    monkeypatch.setattr(svc, 'storage_base_url', lambda: 'https://files.example/bucket')
+    yield SimpleNamespace(db=db, user=user, svc=svc, m=m, doc=PlStopLegalDocument)
+    db.close()
+    engine.dispose()
+
+
+def patch(c, items, **kwargs):
+    return c.svc.save_documents(c.db, c.user, 'p', {'items': items}, **kwargs)
+
+
+def item(rid='d1p', url='legal/new.jpg', side='delivery'):
+    return {'stopType': side, 'stopProductId': rid, 'legalUseCertUrl': url}
+
+
+def test_each_stop_counts_independently_and_legacy_is_separate(case):
+    c = case
+    result = patch(c, [item()])
+    assert (result['uploadedCount'], result['requiredCount']) == (1, 3)
+    assert result['legacyProducts'][0]['legalUseCertUrl'] == 'legacy.jpg'
+    scoped = c.svc.get_documents(c.db, c.user, 'p', contract_id='c')
+    assert (scoped['uploadedCount'], scoped['requiredCount']) == (1, 1)
+    assert scoped['items'][0]['contractNo'] == 'C'
+    assert c.db.get(c.m.PlPlan, 'p').archive_status == 'archived'
+
+
+def test_patch_identity_omissions_and_single_clear(case):
+    c = case
+    patch(c, [item()])
+    row = c.db.query(c.doc).one()
+    original = (row.id, row.create_by, row.create_time)
+    for body in ({}, {'items': []}, {'items': [{'stopType': 'delivery', 'stopProductId': 'd1p'}]}):
+        assert c.svc.save_documents(c.db, c.user, 'p', body)['uploadedCount'] == 1
+    patch(c, [item(url='')])
+    assert (row.id, row.create_by, row.create_time) == original
+    assert c.svc.get_documents(c.db, c.user, 'p')['uploadedCount'] == 0
+
+
+@pytest.mark.parametrize('bad', [item(), item('missing'), item('lp'), item(url='https://evil/a'), item(url='../a')])
+def test_all_validation_precedes_write(case, bad):
+    c = case
+    with pytest.raises(HTTPException):
+        patch(c, [item(), bad])
+    assert c.db.query(c.doc).count() == 0
+    assert c.db.query(c.m.PlPlanChange).count() == 0
+
+
+def test_permissions_tenant_parent_and_contract_scope(case, monkeypatch):
+    c = case
+    with pytest.raises(HTTPException):
+        patch(c, [item('d2p')], contract_id='c')
+    other = SimpleNamespace(**{**c.user.__dict__, 'tenant_no': 'other'})
+    with pytest.raises(HTTPException):
+        c.svc.get_documents(c.db, other, 'p')
+    c.db.get(c.m.PlPlanDelivery, 'd1').plan_id = 'other'
+    c.db.flush()
+    with pytest.raises(HTTPException):
+        patch(c, [item()])
+    monkeypatch.setattr(c.svc, 'user_can_upload_plan_cert', lambda *a: False)
+    with pytest.raises(HTTPException) as error:
+        patch(c, [])
+    assert error.value.status_code == 403
+
+
+def test_filtering_precedes_pagination(case):
+    c = case
+    patch(c, [item()])
+    assert c.svc.list_documents(c.db, c.user, status='uploaded')['total'] == 0
+    result = c.svc.list_documents(c.db, c.user, status='uploaded', contract_id='c', page=1, page_size=1)
+    assert result['total'] == 1 and len(result['list']) == 1
+    assert result['pageSize'] == 1
+    assert c.svc.list_documents(c.db, c.user, keyword='missing')['total'] == 0
+
+
+def test_document_protects_stop_product_until_cleared(case):
+    from app.services.pl_child_references import assert_stop_products_removable
+    c = case
+    patch(c, [item()])
+    row = c.db.get(c.m.PlPlanDeliveryProduct, 'd1p')
+    with pytest.raises(HTTPException) as error:
+        assert_stop_products_removable(c.db, 't', 'p', 'delivery', 'd1', [row])
+    assert error.value.status_code == 409
+    patch(c, [item(url='')])
+    assert_stop_products_removable(c.db, 't', 'p', 'delivery', 'd1', [row])
+
+
+def test_static_routes_registered_before_legacy_dynamic_route():
+    from app.api.pl.quota_cert import legal_router
+    routes = [(r.path, r.methods) for r in legal_router.routes]
+    base = '/transport/legal-use-documents'
+    assert (base + '/stops', {'GET'}) in routes
+    assert (base + '/stops/{plan_id}', {'PATCH'}) in routes
+    assert next(i for i, r in enumerate(routes) if r[0] == base+'/stops') < next(i for i, r in enumerate(routes) if r[0] == base+'/{plan_id}')
+
+
+def test_hidden_contract_rows_do_not_leak_and_null_is_rejected(case, monkeypatch):
+    c = case
+    monkeypatch.setattr(c.svc, 'apply_crm_owner_scope', lambda query, model, *a: query.filter(model.id != 'c'))
+    result = c.svc.get_documents(c.db, c.user, 'p')
+    assert result['requiredCount'] == 2
+    assert all(r['contractId'] != 'c' for r in result['items'])
+    assert c.svc.list_documents(c.db, c.user)['list'][0]['requiredCount'] == 2
+    with pytest.raises(HTTPException):
+        patch(c, [item()])
+    with pytest.raises(HTTPException):
+        patch(c, None)
+
+
+@pytest.mark.parametrize('field,value', [('contractId', None), ('contractProductId', 'new')])
+def test_contract_identity_change_protected(case, field, value):
+    from app.schemas.pl.plan import PlanDeliveryBody
+    from app.services.pl_plan_stop_service import prepare_plan_stops_write
+    c = case
+    patch(c, [item()])
+    body = PlanDeliveryBody(id='d1', products=[{'id': 'd1p', 'planProductId': 'pp', 'productId': 'sku',
+                        'productNo': 'S', 'productName': '产品', field: value}])
+    with pytest.raises(HTTPException) as error:
+        prepare_plan_stops_write(c.db, 't', 'p', None, [body], [c.db.get(c.m.PlPlanProduct, 'pp')])
+    assert error.value.status_code == 409
+
+
+def test_legacy_only_plan_is_visible_but_not_uploaded(case):
+    c = case
+    c.db.query(c.m.PlPlanLoadProduct).update({'del_flag': 1})
+    c.db.query(c.m.PlPlanDeliveryProduct).update({'del_flag': 1})
+    c.db.flush()
+    result = c.svc.get_documents(c.db, c.user, 'p')
+    assert result['legacyProducts'][0]['legalUseCertUrl'] == 'legacy.jpg'
+    assert result['requiredCount'] == 0 and result['documentStatus'] == 'not_required'
+    assert c.svc.list_documents(c.db, c.user)['total'] == 1
+    assert c.svc.list_documents(c.db, c.user, status='uploaded')['total'] == 0
+
+
+def test_multiple_plans_filter_then_page_and_route_shape(case):
+    from app.api.pl.quota_cert import list_stop_documents
+    c = case
+    patch(c, [item()])
+    for i in range(3):
+        pid = 'p' + str(i)
+        c.db.add(c.m.PlPlan(id=pid, plan_no='P'+str(i), create_date=date.today(), belongs_company='shanxia',
+                           plan_type='direct', tenant_no='t'))
+        c.db.add(c.m.PlPlanProduct(id=pid+'pp', plan_id=pid, product_id='sku', product_no='S',
+                    product_name='产品', is_explosive=1, tenant_no='t'))
+        c.db.add(c.m.PlPlanDelivery(id=pid+'d', plan_id=pid, tenant_no='t'))
+        c.db.add(c.m.PlPlanDeliveryProduct(id=pid+'dp', delivery_id=pid+'d', plan_id=pid,
+                    plan_product_id=pid+'pp', product_id='sku', product_no='S', product_name='产品', tenant_no='t'))
+    c.db.flush()
+    first = list_stop_documents(keyword=None, status='pending', contract_id=None, page=1, page_size=2, db=c.db, user=c.user)
+    second = c.svc.list_documents(c.db, c.user, status='pending', page=2, page_size=2)
+    assert first['total'] == second['total'] == 4
+    assert len(first['list']) == len(second['list']) == 2
+    assert {r['id'] for r in first['list']}.isdisjoint(r['id'] for r in second['list'])
+    assert {'uploadedCount', 'requiredCount', 'items', 'legacyProducts', 'planNo'} <= first['list'][0].keys()
+
+
+def test_foreign_and_deleted_stop_products_rejected_before_write(case):
+    c = case
+    row = c.db.get(c.m.PlPlanDeliveryProduct, 'd2p')
+    for field, value in [('tenant_no', 'other'), ('del_flag', 1)]:
+        row.tenant_no, row.del_flag = 't', 0
+        setattr(row, field, value)
+        c.db.flush()
+        with pytest.raises(HTTPException):
+            patch(c, [item(), item('d2p')])
+        assert c.db.query(c.doc).count() == 0
+
+
+def test_soft_deleted_document_not_counted_or_protecting_and_reuses_identity(case):
+    from app.services.pl_child_references import assert_stop_products_removable
+    c = case
+    patch(c, [item()])
+    row = c.db.query(c.doc).one()
+    identity = row.id, row.create_time
+    row.del_flag = 1
+    c.db.flush()
+    assert c.svc.get_documents(c.db, c.user, 'p')['uploadedCount'] == 0
+    assert_stop_products_removable(c.db, 't', 'p', 'delivery', 'd1', [c.db.get(c.m.PlPlanDeliveryProduct, 'd1p')])
+    patch(c, [item()])
+    assert (row.id, row.create_time) == identity and row.del_flag == 0

docs/sql/2026-11-01-pl-stop-legal-document.sql

@@ -0,0 +1,21 @@
+-- 逐段产品合法使用证明；仅定义迁移，部署时执行。旧计划产品附件不自动迁移。
+CREATE TABLE IF NOT EXISTS `pl_stop_legal_document` (
+  `id` VARCHAR(64) NOT NULL COMMENT '主键UUID',
+  `plan_id` VARCHAR(64) NOT NULL COMMENT '运输计划ID',
+  `stop_type` VARCHAR(16) NOT NULL COMMENT 'load装货段/delivery送货段',
+  `stop_product_id` VARCHAR(64) NOT NULL COMMENT '对应段产品明细ID',
+  `legal_use_cert_url` TEXT NULL COMMENT 'MinIO附件，多附件逗号分隔',
+  `uploaded_by` VARCHAR(64) NULL COMMENT '最近上传人',
+  `uploaded_time` DATETIME NULL COMMENT '最近上传时间',
+  `tenant_no` VARCHAR(64) NOT NULL COMMENT '租户编号',
+  `order_num` INT NULL COMMENT '排序号',
+  `remark` VARCHAR(1000) NULL COMMENT '备注',
+  `create_by` VARCHAR(64) NULL COMMENT '创建人',
+  `create_dept_id` VARCHAR(64) NOT NULL DEFAULT '' COMMENT '创建部门',
+  `create_time` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
+  `update_by` VARCHAR(64) NULL COMMENT '更新人',
+  `update_time` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
+  `del_flag` SMALLINT NOT NULL DEFAULT 0 COMMENT '0正常1删除',
+  PRIMARY KEY (`id`),
+  UNIQUE KEY `uk_stop_legal_document` (`tenant_no`, `plan_id`, `stop_type`, `stop_product_id`)
+) ENGINE=InnoDB ROW_FORMAT=DYNAMIC DEFAULT CHARSET=utf8mb4 COMMENT='逐段产品合法使用证明';