feat(pl): Add scoped quota statistics to distinguish shared balances from contract allocations.

backend/app/api/pl/quota_cert.py

@@ -27,6 +27,7 @@ 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
+from app.services import pl_quota_statistics as statistics
 from app.services.pl_quota_allocation import AllocationIncomplete, InsufficientQuota
 from app.services.pl_quota_attachments import read_image
 from app.services.pl_quota_recognition import recognize_quota_cert
@@ -145,6 +146,36 @@ def save_bindings(contract_id: str, body: ContractCertBindingsBody,
     return _write(db, lambda: scopes.save_contract_bindings(db, user, contract_id, body))
 
 
+@router.get('/statistics/contracts')
+def contract_statistics_list(keyword: str = Query('', max_length=200),
+                             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(statistics.list_contracts(db, user, keyword=keyword, page=page, page_size=page_size))
+
+
+@router.get('/statistics/certificates')
+def certificate_statistics_list(keyword: str = Query('', max_length=200),
+                                cert_type: Optional[CertType] = Query(None, alias='certType'),
+                                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(statistics.list_certificates(db, user, keyword=keyword, cert_type=cert_type,
+                                             page=page, page_size=page_size))
+
+
+@router.get('/contracts/{contract_id}/statistics')
+def contract_statistics(contract_id: str, db: Session = Depends(get_db),
+                        user: DbUserContext = Depends(current_user)):
+    return _wire(statistics.contract_statistics(db, user, contract_id))
+
+
+@router.get('/{cert_type}/{cert_id}/statistics')
+def certificate_statistics(cert_type: CertType, cert_id: str, db: Session = Depends(get_db),
+                           user: DbUserContext = Depends(current_user)):
+    return _wire(statistics.certificate_statistics(db, user, cert_type, cert_id))
+
+
 @router.get('/{cert_type}')
 def list_certs(cert_type: CertType, keyword: Optional[str] = None, status: Optional[str] = None,
                db: Session = Depends(get_db), user: DbUserContext = Depends(current_user)):

backend/app/services/pl_quota_statistics.py

@@ -0,0 +1,197 @@
+"""只读额度统计：共享缓存与授权合同分配分开展示，不重建历史余额。"""
+from decimal import Decimal
+from sqlalchemy import and_, or_, literal, func
+from fastapi import HTTPException
+from app.crud.crm.data_scope_helpers import apply_crm_owner_scope
+from app.models.crm_models import CrmContract, CrmContractProduct
+from app.models.pl_models import PlPlan
+from app.models.pl_quota_models import (PlContractCertLink as Link, PlContractCertProduct as Scope,
+    PlPlanCertUsage as Usage, PlPlanCertUsageItem as Item, PlCertQtyLog as Log)
+from app.services.pl_data_scope import apply_pl_scope
+from app.services.pl_quota_cert_service import MODELS, _get_row
+
+ZERO = Decimal('0')
+
+
+def _active(db, user, model):
+    return db.query(model).filter(model.tenant_no == user.tenant_no, model.del_flag == 0)
+
+
+def _contracts(db, user):
+    return apply_crm_owner_scope(_active(db, user, CrmContract), CrmContract,
+                                 user.to_user_info(), db, user.tenant_no)
+
+
+def _certificates(db, user, keyword=None, cert_type=None):
+    queries = []
+    for kind, model in MODELS.items():
+        if cert_type and kind != cert_type:
+            continue
+        query = apply_pl_scope(_active(db, user, model), model, user, db, user.tenant_no)
+        if keyword:
+            query = query.filter(or_(model.cert_no.contains(keyword, autoescape=True),
+                                    model.product_name_on_cert.contains(keyword, autoescape=True)))
+        queries.append(query.with_entities(literal(kind).label('cert_type'), model.id.label('cert_id'),
+            *[getattr(model, field).label(field) for field in ('cert_no', 'product_name_on_cert',
+                'quantity_unit', 'total_qty', 'reserved_qty', 'used_qty', 'status',
+                'valid_from', 'valid_to', 'create_time')]))
+    return (queries[0].union_all(*queries[1:]) if len(queries) > 1 else queries[0]).subquery()
+
+
+def _cert_summary(row):
+    return {'id': row.cert_id, 'certId': row.cert_id, 'certType': row.cert_type, 'certNo': row.cert_no,
+        'productNameOnCert': row.product_name_on_cert, 'quantityUnit': row.quantity_unit,
+        'totalQty': row.total_qty, 'reservedQty': row.reserved_qty, 'usedQty': row.used_qty,
+        'availableQty': row.total_qty - row.reserved_qty - row.used_qty, 'status': row.status,
+        'validFrom': row.valid_from, 'validTo': row.valid_to}
+
+
+def _contract_summary(row):
+    return {'id': row.id, 'contractNo': row.contract_no, 'contractName': row.contract_name,
+            'contractType': row.contract_type}
+
+
+def _key_filter(model, keys):
+    return or_(*[and_(model.cert_type == kind, model.cert_id.in_(
+        [ident for k, ident in keys if k == kind])) for kind in MODELS])
+
+
+def _details(db, user, keys, contract_ids=None):
+    """有限证件/合同范围一次读取维护身份、一次读取授权使用；不连接明细造成乘积。"""
+    if not keys:
+        return {}, []
+    contracts = _contracts(db, user).with_entities(CrmContract.id)
+    if contract_ids is not None:
+        contracts = contracts.filter(CrmContract.id.in_(contract_ids))
+    links = _active(db, user, Link).filter(_key_filter(Link, keys), Link.contract_id.in_(contracts)).all()
+    groups = {}
+    for link in links:
+        groups[(link.cert_type, link.cert_id, link.contract_id)] = {
+            'bindingId': link.id, 'linkStatus': link.link_status, 'scopeRevision': link.scope_revision,
+            'contractReservedQty': ZERO, 'contractUsedQty': ZERO, 'products': []}
+    products = {}
+    if links:
+        rows = db.query(Scope, CrmContractProduct).join(CrmContractProduct,
+            Scope.contract_product_id == CrmContractProduct.id).filter(
+            Scope.tenant_no == user.tenant_no, Scope.del_flag == 0,
+            Scope.contract_cert_link_id.in_([l.id for l in links]),
+            CrmContractProduct.tenant_no == user.tenant_no, CrmContractProduct.del_flag == 0)
+        by_link = {l.id: l for l in links}
+        for scope, product in rows:
+            link = by_link[scope.contract_cert_link_id]
+            if product.contract_id != link.contract_id:
+                continue
+            key = (link.cert_type, link.cert_id, link.contract_id)
+            value = {'id': product.id, 'contractProductId': product.id, 'productId': product.product_id,
+                'productName': product.product_name, 'model': product.model, 'unit': product.unit,
+                'bindingId': link.id, 'scopeId': scope.id, 'scopeStatus': scope.scope_status,
+                'linkStatus': link.link_status, 'reservedQty': ZERO, 'usedQty': ZERO}
+            groups[key]['products'].append(value)
+            products[(*key, product.id)] = value
+    plans = apply_pl_scope(_active(db, user, PlPlan), PlPlan, user, db, user.tenant_no).with_entities(PlPlan.id)
+    rows = db.query(Item, Usage, PlPlan, CrmContractProduct).join(Usage, Item.usage_id == Usage.id).join(
+        PlPlan, Usage.plan_id == PlPlan.id).join(CrmContractProduct, and_(
+            Item.contract_product_id == CrmContractProduct.id, Item.contract_id == CrmContractProduct.contract_id)
+        ).filter(Item.tenant_no == user.tenant_no, Item.del_flag == 0,
+            Usage.tenant_no == user.tenant_no, Usage.del_flag == 0, _key_filter(Usage, keys),
+            Item.contract_id.in_(contracts), PlPlan.id.in_(plans),
+            CrmContractProduct.tenant_no == user.tenant_no, CrmContractProduct.del_flag == 0
+        ).order_by(Item.create_time.desc(), Item.id).all()
+    usage_items = []
+    for item, usage, plan, product in rows:
+        key = (usage.cert_type, usage.cert_id, item.contract_id)
+        group = groups.setdefault(key, {'bindingId': None, 'linkStatus': None, 'scopeRevision': None,
+            'contractReservedQty': ZERO, 'contractUsedQty': ZERO, 'products': []})
+        # 证件单位不可变；异常历史单位不混入当前单位的聚合，明细仍保留追溯。
+        unit = keys[key[:2]]
+        if usage.quantity_unit == unit:
+            group['contractReservedQty'] += item.reserved_qty
+            group['contractUsedQty'] += item.used_qty
+            value = products.get((*key, product.id))
+            if value is None:
+                value = {'id': product.id, 'contractProductId': product.id, 'productId': product.product_id,
+                    'productName': product.product_name, 'model': product.model, 'unit': product.unit,
+                    'bindingId': None, 'scopeId': None, 'scopeStatus': None, 'linkStatus': None,
+                    'reservedQty': ZERO, 'usedQty': ZERO}
+                products[(*key, product.id)] = value
+                group['products'].append(value)
+            value['reservedQty'] += item.reserved_qty
+            value['usedQty'] += item.used_qty
+        usage_items.append({'id': item.id, 'usageId': usage.id, 'certType': usage.cert_type,
+            'certId': usage.cert_id, 'planId': plan.id, 'planNo': plan.plan_no,
+            'stopType': item.source_stop_type, 'stopId': item.source_stop_id,
+            'stopProductId': item.source_stop_product_id, 'planProductId': item.plan_product_id,
+            'contractId': item.contract_id, 'contractProductId': item.contract_product_id,
+            'productName': product.product_name, 'quantityUnit': usage.quantity_unit,
+            'reservedQty': item.reserved_qty, 'usedQty': item.used_qty, 'itemStatus': item.item_status})
+    return groups, usage_items
+
+
+def list_contracts(db, user, keyword=None, page=1, page_size=20):
+    certs = _certificates(db, user)
+    linked = _active(db, user, Link).join(certs, and_(Link.cert_type == certs.c.cert_type,
+        Link.cert_id == certs.c.cert_id)).with_entities(Link.contract_id)
+    query = _contracts(db, user).filter(CrmContract.id.in_(linked))
+    if keyword:
+        query = query.filter(or_(CrmContract.contract_no.contains(keyword, autoescape=True),
+                                 CrmContract.contract_name.contains(keyword, autoescape=True)))
+    total = query.count()
+    rows = query.order_by(CrmContract.create_time.desc(), CrmContract.id).offset((page-1)*page_size).limit(page_size).all()
+    counts = dict(db.query(Link.contract_id, func.count(Link.id)).join(certs,
+        and_(Link.cert_type == certs.c.cert_type, Link.cert_id == certs.c.cert_id)).filter(
+        Link.tenant_no == user.tenant_no, Link.del_flag == 0, Link.contract_id.in_([r.id for r in rows])
+        ).group_by(Link.contract_id).all()) if rows else {}
+    return {'list': [{**_contract_summary(r), 'certCount': counts.get(r.id, 0)} for r in rows],
+            'total': total, 'page': page, 'pageSize': page_size}
+
+
+def list_certificates(db, user, keyword=None, cert_type=None, page=1, page_size=20):
+    certs = _certificates(db, user, keyword, cert_type)
+    query = db.query(certs)
+    total = query.count()
+    rows = query.order_by(certs.c.create_time.desc(), certs.c.cert_type, certs.c.cert_id).offset(
+        (page-1)*page_size).limit(page_size).all()
+    return {'list': [_cert_summary(r) for r in rows], 'total': total, 'page': page, 'pageSize': page_size}
+
+
+def contract_statistics(db, user, contract_id):
+    contract = _contracts(db, user).filter(CrmContract.id == contract_id).first()
+    if contract is None:
+        raise HTTPException(404, '合同不存在或无权访问')
+    certs = _certificates(db, user)
+    linked = _active(db, user, Link).filter(Link.contract_id == contract_id)
+    # 历史使用即使已停止绑定也可追溯，但证件本身仍须有效且授权。
+    used = _active(db, user, Usage).join(Item, Item.usage_id == Usage.id).filter(
+        Item.tenant_no == user.tenant_no, Item.del_flag == 0, Item.contract_id == contract_id)
+    rows = db.query(certs).filter(or_(linked.filter(Link.cert_type == certs.c.cert_type,
+        Link.cert_id == certs.c.cert_id).exists(), used.filter(Usage.cert_type == certs.c.cert_type,
+        Usage.cert_id == certs.c.cert_id).exists())).order_by(certs.c.create_time.desc()).all()
+    keys = {(r.cert_type, r.cert_id): r.quantity_unit for r in rows}
+    groups, _ = _details(db, user, keys, [contract_id])
+    result = []
+    for row in rows:
+        group = groups.get((row.cert_type, row.cert_id, contract_id))
+        if group is not None:
+            result.append({**_cert_summary(row), **group,
+                'productNames': list(dict.fromkeys(p['productName'] for p in group['products']))})
+    return {**_contract_summary(contract), 'certificates': result}
+
+
+def certificate_statistics(db, user, cert_type, cert_id):
+    _get_row(db, user, cert_type, cert_id)
+    certs = _certificates(db, user, cert_type=cert_type)
+    row = db.query(certs).filter(certs.c.cert_id == cert_id).one()
+    groups, items = _details(db, user, {(cert_type, cert_id): row.quantity_unit})
+    ids = [key[2] for key in groups]
+    contracts = _contracts(db, user).filter(CrmContract.id.in_(ids)).order_by(CrmContract.create_time.desc()).all() if ids else []
+    logs = _active(db, user, Log).filter(Log.cert_type == cert_type, Log.cert_id == cert_id,
+        Log.plan_id.is_(None), Log.usage_id.is_(None), Log.quantity_unit == row.quantity_unit)
+    opening = unattributed = ZERO
+    for event, delta in logs.with_entities(Log.event_type, func.sum(Log.used_delta)).group_by(Log.event_type):
+        if event == 'opening':
+            opening += delta
+        else:
+            unattributed += delta
+    return {**_cert_summary(row), 'contracts': [{**_contract_summary(c),
+        **groups[(cert_type, cert_id, c.id)]} for c in contracts], 'usageItems': items,
+        'openingUsedQty': opening, 'unattributedUsedQty': unattributed}

backend/tests/test_pl_quota_api.py

@@ -148,3 +148,26 @@ def test_cert_bindings_api_accepts_collection_and_roundtrips(api):
     assert client.put(detail, json={'contractBindings': [], 'remark': 'edit'}).status_code == 200
     assert client.put(detail, json={'contractBindings': [], 'unknownField': 'bad'}).status_code == 422
     assert client.get(detail).json()['remark'] == 'edit'
+
+
+def test_statistics_routes_are_static_paginated_and_strict(api, monkeypatch):
+    from app.api.pl.quota_cert import router
+    paths = [route.path for route in router.routes]
+    prefix = '/transport/quota-certs'
+    assert prefix + '/statistics/contracts' in paths
+    assert paths.index(prefix + '/statistics/contracts') < paths.index(prefix + '/{cert_type}/{cert_id}')
+    from app.services import pl_quota_statistics as stats
+    client, app, auth, _ = api
+    app.dependency_overrides[auth] = lambda: {'role': 'super_admin', 'user_id': 'u', 'tenant_no': 't'}
+    received = []
+    def capture(*args, **kwargs):
+        received.append(kwargs)
+        return {'list': [], 'total': 0, 'page': 2, 'pageSize': 3}
+    monkeypatch.setattr(stats, 'list_certificates', capture)
+    result = client.get('/api' + prefix + '/statistics/certificates', params={
+        'keyword': '硫酸', 'certType': 'purchase', 'page': 2, 'pageSize': 3})
+    assert result.status_code == 200 and result.json()['pageSize'] == 3
+    assert received == [{'keyword': '硫酸', 'cert_type': 'purchase', 'page': 2, 'page_size': 3}]
+    for params in ({'page': 0}, {'pageSize': 101}, {'certType': 'bad'}, {'keyword': 'x' * 201}):
+        assert client.get('/api' + prefix + '/statistics/certificates', params=params).status_code == 422
+    assert client.get('/api' + prefix + '/contracts/missing/statistics').status_code == 404

backend/tests/test_pl_quota_statistics.py

@@ -0,0 +1,218 @@
+"""统计只读、主对象分页与授权明细聚合回归。"""
+from datetime import date
+from types import SimpleNamespace
+import importlib
+import pytest
+from sqlalchemy import create_engine
+from sqlalchemy.orm import Session
+from app.models.crm_models import CrmContract, CrmContractProduct
+from app.models.pl_models import PlPlan
+from app.models import pl_quota_models as m
+
+
+@pytest.fixture
+def case():
+    engine = create_engine('sqlite://')
+    models = (CrmContract, CrmContractProduct, PlPlan, m.PlPurchaseCert, m.PlTransportCert,
+              m.PlContractCertLink, m.PlContractCertProduct, m.PlPlanCertUsage,
+              m.PlPlanCertUsageItem, m.PlCertQtyLog)
+    for model in models:
+        model.__table__.create(engine)
+    db = Session(engine)
+    user = SimpleNamespace(tenant_no='t', to_user_info=lambda: {'role': 'super_admin', 'user_id': 'u'})
+    for cid in ('a', 'b'):
+        db.add(CrmContract(id=cid, contract_no=cid, contract_name=cid, customer_id='customer',
+                           contract_type='sale', status='approved', tenant_no='t'))
+        db.add(CrmContractProduct(id=cid, contract_id=cid, product_id='chemical', product_name=cid,
+                                 unit='吨', quantity=100, tenant_no='t'))
+        db.add(m.PlContractCertLink(id=cid, contract_id=cid, contract_no_snapshot=cid,
+                                    cert_type='purchase', cert_id='same', tenant_no='t'))
+        db.add(m.PlContractCertProduct(id=cid, contract_cert_link_id=cid,
+                                      contract_product_id=cid, tenant_no='t'))
+    db.add(m.PlPurchaseCert(id='same', cert_no='P', product_name_on_cert='硫酸', total_qty=300,
+        quantity_unit='吨', reserved_qty=180, used_qty=20, valid_use_text='多次有效',
+        buyer_name='买方', seller_name='卖方', tenant_no='t'))
+    db.commit()
+    yield SimpleNamespace(db=db, user=user)
+    db.close()
+    engine.dispose()
+
+
+def service():
+    assert importlib.util.find_spec('app.services.pl_quota_statistics'), '缺少统计服务'
+    return importlib.import_module('app.services.pl_quota_statistics')
+
+
+def add_usage(c, cid, ident, reserved, used=0):
+    c.db.add(PlPlan(id=ident, plan_no=ident, create_date=date.today(), belongs_company='shanxia', plan_type='direct', tenant_no='t'))
+    c.db.add(m.PlPlanCertUsage(id=ident, plan_id=ident, plan_no_snapshot=ident, cert_type='purchase',
+        cert_id='same', cert_no_snapshot='P', cert_revision=1, quantity_unit='吨', tenant_no='t'))
+    c.db.add(m.PlPlanCertUsageItem(id=ident, usage_id=ident, contract_cert_link_id=cid,
+        contract_cert_product_id=cid, contract_scope_revision=1, contract_id=cid,
+        contract_product_id=cid, plan_product_id=ident, source_stop_type='delivery',
+        source_stop_id=ident, source_stop_product_id=ident, allocation_seq=1, source_unit='吨',
+        planned_source_qty=999, reserved_qty=reserved, used_qty=used, tenant_no='t'))
+    c.db.commit()
+
+
+def test_shared_balance_and_contract_allocation_are_distinct_and_read_only(case):
+    s = service()
+    add_usage(case, 'a', 'one', 60, 20)
+    add_usage(case, 'a', 'two', 40)
+    add_usage(case, 'b', 'three', 80)
+    row = s.contract_statistics(case.db, case.user, 'a')['certificates'][0]
+    assert (row['totalQty'], row['reservedQty'], row['usedQty'], row['availableQty']) == (300, 180, 20, 100)
+    assert row['contractReservedQty'] == 100 and row['contractUsedQty'] == 20
+    assert row['products'][0]['reservedQty'] == 100
+    detail = s.certificate_statistics(case.db, case.user, 'purchase', 'same')
+    assert len(detail['usageItems']) == 3
+    assert detail['openingUsedQty'] == 0
+    assert case.db.query(m.PlContractCertLink).count() == 2
+    assert not case.db.new and not case.db.dirty
+
+
+def test_lists_paginate_main_objects_and_filter_keywords(case):
+    s = service()
+    result = s.list_contracts(case.db, case.user, page=1, page_size=1)
+    assert result['total'] == 2 and len(result['list']) == 1 and result['pageSize'] == 1
+    assert s.list_contracts(case.db, case.user, keyword='a')['total'] == 1
+    assert s.list_certificates(case.db, case.user, keyword='硫酸')['total'] == 1
+    other = SimpleNamespace(**{**case.user.__dict__, 'tenant_no': 'other'})
+    assert s.list_contracts(case.db, other)['total'] == 0
+    assert s.list_certificates(case.db, other)['total'] == 0
+
+
+def test_hidden_contracts_plans_and_deleted_products_never_enter_details(case, monkeypatch):
+    s = service()
+    add_usage(case, 'a', 'visible', 30)
+    add_usage(case, 'a', 'hidden-plan', 70)
+    add_usage(case, 'b', 'hidden-contract', 80)
+    monkeypatch.setattr(s, 'apply_crm_owner_scope', lambda q, model, *a: q.filter(model.id == 'a'))
+    monkeypatch.setattr(s, 'apply_pl_scope', lambda q, model, *a: q.filter(model.id != 'hidden-plan'))
+    detail = s.certificate_statistics(case.db, case.user, 'purchase', 'same')
+    assert [x['id'] for x in detail['contracts']] == ['a']
+    assert detail['contracts'][0]['contractReservedQty'] == 30
+    assert [x['planId'] for x in detail['usageItems']] == ['visible']
+    assert detail['openingUsedQty'] == 0 and detail['unattributedUsedQty'] == 0
+    case.db.query(CrmContractProduct).filter_by(id='a').one().del_flag = 1
+    case.db.commit()
+    assert s.certificate_statistics(case.db, case.user, 'purchase', 'same')['usageItems'] == []
+
+
+def test_same_id_certificate_types_units_and_real_opening_are_separate(case):
+    s = service()
+    # 按当前模型构造运输证，验证同 ID 两个额度池不会在分页或汇总中合并。
+    purchase = case.db.query(m.PlPurchaseCert).one()
+    values = {col.name: getattr(purchase, col.name) for col in m.PlTransportCert.__table__.columns
+              if hasattr(purchase, col.name)}
+    values.update(cert_no='T', quantity_unit='kg', total_qty=500, reserved_qty=0, used_qty=0)
+    for col in m.PlTransportCert.__table__.columns:
+        if not col.nullable and col.default is None and col.name not in values:
+            values[col.name] = '运输主体'
+    case.db.add(m.PlTransportCert(**values))
+    for ident, event, amount, plan, tenant, deleted in (
+        ('opening', 'opening', 7, None, 't', 0), ('adjust', 'adjust', 2, None, 't', 0),
+        ('hidden', 'settle', 99, 'hidden-plan', 't', 0),
+        ('other', 'opening', 99, None, 'other', 0), ('deleted', 'opening', 99, None, 't', 1)):
+        case.db.add(m.PlCertQtyLog(id=ident, operation_id=ident, request_hash='hash', cert_type='purchase',
+            cert_id='same', plan_id=plan, event_type=event, quantity_unit='吨', used_delta=amount,
+            before_total_qty=300, after_total_qty=300, before_reserved_qty=0, after_reserved_qty=0,
+            before_used_qty=0, after_used_qty=amount, reason='凭据', tenant_no=tenant, del_flag=deleted))
+    case.db.commit()
+    result = s.list_certificates(case.db, case.user, page_size=1)
+    assert result['total'] == 2 and len(result['list']) == 1
+    assert s.list_certificates(case.db, case.user, cert_type='transport')['list'][0]['quantityUnit'] == 'kg'
+    purchase = s.certificate_statistics(case.db, case.user, 'purchase', 'same')
+    transport = s.certificate_statistics(case.db, case.user, 'transport', 'same')
+    assert purchase['openingUsedQty'] == 7 and purchase['unattributedUsedQty'] == 2
+    assert transport['openingUsedQty'] == 0 and transport['contracts'] == []
+
+
+@pytest.mark.parametrize('model,field,value', [
+    (m.PlPlanCertUsage, 'tenant_no', 'other'), (m.PlPlanCertUsage, 'del_flag', 1),
+    (m.PlPlanCertUsageItem, 'tenant_no', 'other'), (m.PlPlanCertUsageItem, 'del_flag', 1),
+    (PlPlan, 'tenant_no', 'other'), (PlPlan, 'del_flag', 1),
+    (CrmContract, 'del_flag', 1), (CrmContractProduct, 'tenant_no', 'other')])
+def test_invalid_ancestors_exclude_usage(case, model, field, value):
+    s = service()
+    add_usage(case, 'a', 'one', 100)
+    row = case.db.query(model).filter_by(id='a' if model in (CrmContract, CrmContractProduct) else 'one').one()
+    setattr(row, field, value)
+    case.db.commit()
+    detail = s.certificate_statistics(case.db, case.user, 'purchase', 'same')
+    assert detail['usageItems'] == []
+    assert all(c['contractReservedQty'] == 0 for c in detail['contracts'])
+
+
+def test_hidden_certificates_excluded_and_page_queries_do_not_grow_with_rows(case, monkeypatch):
+    from sqlalchemy import event
+    s = service()
+    counts = []
+    for size in (1, 20):
+        statements = []
+        def capture(*args):
+            statements.append(args[2])
+        event.listen(case.db.bind, 'before_cursor_execute', capture)
+        try:
+            s.list_contracts(case.db, case.user, page_size=size)
+        finally:
+            event.remove(case.db.bind, 'before_cursor_execute', capture)
+        counts.append(len(statements))
+        assert all(sql.lstrip().upper().startswith('SELECT') for sql in statements)
+    assert counts[0] == counts[1]
+    monkeypatch.setattr(s, 'apply_pl_scope', lambda q, model, *a: q.filter(model.id != 'same'))
+    assert s.list_contracts(case.db, case.user)['total'] == 0
+    assert s.contract_statistics(case.db, case.user, 'a')['certificates'] == []
+
+
+def test_historical_usage_survives_removed_binding_without_recreation(case):
+    s = service()
+    add_usage(case, 'a', 'one', 0, 20)
+    case.db.query(m.PlContractCertLink).filter_by(id='a').one().del_flag = 1
+    case.db.query(m.PlContractCertProduct).filter_by(id='a').one().del_flag = 1
+    case.db.commit()
+    row = s.contract_statistics(case.db, case.user, 'a')['certificates'][0]
+    assert row['contractUsedQty'] == 20 and row['products'][0]['bindingId'] is None
+    assert case.db.query(m.PlContractCertLink).filter_by(id='a').one().del_flag == 1
+
+
+def test_role_union_uses_existing_crm_owner_and_pl_creator_helpers(case, monkeypatch):
+    from app.core import data_scope
+    s = service()
+    add_usage(case, 'a', 'self-plan', 30)
+    add_usage(case, 'a', 'custom-plan', 40)
+    add_usage(case, 'a', 'hidden-plan', 50)
+    case.db.query(m.PlPurchaseCert).one().create_by = 'u'
+    case.db.query(CrmContract).filter_by(id='a').one().owner_user_id = 'u'
+    case.db.query(CrmContract).filter_by(id='b').one().owner_user_id = 'hidden'
+    for ident, creator in [('self-plan', 'u'), ('custom-plan', 'colleague'), ('hidden-plan', 'hidden')]:
+        case.db.query(PlPlan).filter_by(id=ident).one().create_by = creator
+    case.db.commit()
+    monkeypatch.setattr(data_scope, '_resolve_user_role_scopes', lambda *a: [
+        {'data_scope': '4', 'details': []},
+        {'data_scope': '5', 'details': [SimpleNamespace(scope_type='user', scope_value='colleague')]}])
+    user = SimpleNamespace(tenant_no='t', to_user_info=lambda: {'role': 'staff', 'user_id': 'u', 'dept_id': 'd'})
+    result = s.certificate_statistics(case.db, user, 'purchase', 'same')
+    assert [row['id'] for row in result['contracts']] == ['a']
+    assert result['contracts'][0]['contractReservedQty'] == 70
+    assert {row['planId'] for row in result['usageItems']} == {'self-plan', 'custom-plan'}
+
+
+def test_all_statistics_reads_emit_no_mutations_or_commit(case, monkeypatch):
+    from sqlalchemy import event
+    s = service()
+    statements = []
+    def no_commit():
+        pytest.fail('统计不得提交事务')
+    monkeypatch.setattr(case.db, 'commit', no_commit)
+    def capture(*args):
+        statements.append(args[2])
+    event.listen(case.db.bind, 'before_cursor_execute', capture)
+    try:
+        s.list_contracts(case.db, case.user)
+        s.list_certificates(case.db, case.user)
+        s.contract_statistics(case.db, case.user, 'a')
+        s.certificate_statistics(case.db, case.user, 'purchase', 'same')
+    finally:
+        event.remove(case.db.bind, 'before_cursor_execute', capture)
+    assert all(sql.lstrip().upper().startswith('SELECT') for sql in statements)