fix(pl): 在分页前应用计划用证表头筛选以返回完整匹配结果。

backend/app/api/pl/quota_cert.py

@@ -1,5 +1,6 @@
 """共享证件档案、合同产品适用范围与合法使用证明入口。"""
 from dataclasses import replace
+from datetime import date
 from decimal import Decimal
 from typing import Literal, Optional
 
@@ -64,8 +65,15 @@ class RecognizeBody(InputBody):
 @router.get('/plans')
 def plan_overview(keyword: str = '', status: Optional[str] = None,
                   page: int = Query(1, ge=1), page_size: int = Query(20, alias='pageSize', ge=1, le=100),
+                  plan_no: Optional[str] = Query(None, alias='planNo', max_length=200),
+                  license_plate: Optional[str] = Query(None, alias='licensePlate', max_length=100),
+                  plan_date_from: Optional[date] = Query(None, alias='planDateFrom'),
+                  plan_date_to: Optional[date] = Query(None, alias='planDateTo'),
+                  quota_status_in: Optional[str] = Query(None, alias='quotaStatusIn', max_length=200),
                   db: Session = Depends(get_db), user: DbUserContext = Depends(current_user)):
-    return _wire(overview.list_plans(db, user, keyword, status, page, page_size))
+    return _wire(overview.list_plans(db, user, keyword, status, page, page_size,
+        plan_no=plan_no, license_plate=license_plate, plan_date_from=plan_date_from,
+        plan_date_to=plan_date_to, quota_status_in=quota_status_in))
 
 
 @router.get('/plans/{plan_id}/usage')

backend/app/services/pl_quota_overview.py

@@ -1,4 +1,5 @@
 """计划用证列表：按真实占用/结转状态分组，余额可用性由只读预览实时计算。"""
+from datetime import date
 from fastapi import HTTPException
 from sqlalchemy import case, or_, cast, String
 from app.models.pl_models import PlPlan, PlPlanProduct
@@ -8,9 +9,28 @@ from app.services.pl_data_scope import apply_pl_scope
 STATUSES = ('not_required', 'needs_configuration', 'needs_validation', 'reserved', 'settled', 'exception')
 
 
-def list_plans(db, user, keyword='', status=None, page=1, page_size=20):
+def _filter_date(value):
+    if value in (None, ''):
+        return None
+    if isinstance(value, date):
+        return value
+    try:
+        return date.fromisoformat(value)
+    except (TypeError, ValueError):
+        raise HTTPException(400, '计划日期格式无效') from None
+
+
+def list_plans(db, user, keyword='', status=None, page=1, page_size=20, *,
+               plan_no=None, license_plate=None, plan_date_from=None, plan_date_to=None,
+               quota_status_in=None):
     if status and status not in STATUSES:
         raise HTTPException(400, '未知用证状态')
+    selected_states = [value.strip() for value in (quota_status_in or '').split(',') if value.strip()]
+    if any(value not in STATUSES for value in selected_states):
+        raise HTTPException(400, '未知用证状态')
+    date_from, date_to = _filter_date(plan_date_from), _filter_date(plan_date_to)
+    if date_from and date_to and date_from > date_to:
+        raise HTTPException(400, '计划日期起始不能晚于结束')
     toxic = db.query(PlPlanProduct.id).filter(PlPlanProduct.tenant_no == user.tenant_no,
         PlPlanProduct.plan_id == PlPlan.id, PlPlanProduct.del_flag == 0, PlPlanProduct.is_toxic == 1).exists()
     def usage(*states):
@@ -33,6 +53,17 @@ def list_plans(db, user, keyword='', status=None, page=1, page_size=20):
         query = query.filter(PlPlan.plan_no.contains(keyword))
     if status:
         query = query.filter(state == status)
+    # 表头条件必须先于统计与分页，不能只过滤已加载的一页。
+    if plan_no:
+        query = query.filter(PlPlan.plan_no.contains(plan_no, autoescape=True))
+    if license_plate:
+        query = query.filter(PlPlan.license_plate.contains(license_plate, autoescape=True))
+    if date_from:
+        query = query.filter(PlPlan.plan_date >= date_from)
+    if date_to:
+        query = query.filter(PlPlan.plan_date <= date_to)
+    if selected_states:
+        query = query.filter(state.in_(selected_states))
     total = query.count()
     rows = query.order_by(PlPlan.create_time.desc(), PlPlan.id).offset((page-1)*page_size).limit(page_size).all()
     return {'total': total, 'list': [{'id': p.id, 'planNo': p.plan_no, 'planDate': p.plan_date,

backend/tests/test_pl_quota_api.py

@@ -78,6 +78,29 @@ def test_decimal_wire_values_are_strings_without_float_roundtrip(api):
     assert _wire({'totalQty': Decimal('999999999999.999999')})['totalQty'] == '999999999999.999999'
 
 
+def test_plan_overview_accepts_web_funnel_aliases_and_validates_dates(api, monkeypatch):
+    from datetime import date
+    from app.services import pl_quota_overview
+    client, app, auth, permissions = api
+    app.dependency_overrides[auth] = lambda: {'role': 'super_admin', 'user_id': 'u', 'tenant_no': 't'}
+    received = []
+    def capture(*args, **kwargs):
+        received.append((args, kwargs))
+        return {'total': 0, 'list': []}
+    monkeypatch.setattr(pl_quota_overview, 'list_plans', capture)
+    response = client.get('/api/transport/quota-certs/plans', params={
+        'keyword': 'PLAN', 'page': 2, 'pageSize': 10, 'planNo': '001', 'licensePlate': '皖A',
+        'planDateFrom': '2026-11-01', 'planDateTo': '2026-11-03', 'quotaStatusIn': 'reserved,settled'})
+    assert response.status_code == 200
+    args, filters = received.pop()
+    assert args[2:] == ('PLAN', None, 2, 10)
+    assert filters == {'plan_no': '001', 'license_plate': '皖A',
+        'plan_date_from': date(2026, 11, 1), 'plan_date_to': date(2026, 11, 3),
+        'quota_status_in': 'reserved,settled'}
+    assert client.get('/api/transport/quota-certs/plans', params={'planDateFrom': 'wrong'}).status_code == 422
+    assert received == []
+
+
 def test_preview_returns_structured_insufficiency_without_claiming_availability(api, monkeypatch):
     from app.services.pl_quota_allocation import InsufficientQuota
     from app.services import pl_quota_planning

backend/tests/test_pl_quota_overview_filters.py

@@ -0,0 +1,113 @@
+"""列表漏斗在服务端分页前生效，使用内存数据库。"""
+from datetime import date, datetime, timedelta
+from types import SimpleNamespace
+
+import pytest
+from fastapi import HTTPException
+from sqlalchemy import create_engine
+from sqlalchemy.orm import Session
+
+from app.models.pl_models import PlPlan, PlPlanProduct
+from app.models.pl_quota_models import PlPlanCertUsage
+from app.services.pl_quota_overview import list_plans
+
+
+@pytest.fixture
+def case():
+    engine = create_engine('sqlite://')
+    for model in (PlPlan, PlPlanProduct, PlPlanCertUsage):
+        model.__table__.create(engine)
+    with Session(engine) as db:
+        user = SimpleNamespace(id='u', tenant_no='t', to_user_info=lambda: {'role': 'super_admin', 'user_id': 'u'})
+        for n in range(25):
+            db.add(PlPlan(id=f'p{n:02}', plan_no=f'PLAN-{n:02}', create_date=date(2026, 11, 1),
+                plan_date=date(2026, 11, 1) + timedelta(days=n % 3), plan_type='direct',
+                belongs_company='shanxia', tenant_no='t', license_plate='皖A12345' if n % 2 == 0 else '皖B67890',
+                create_time=datetime(2026, 11, 1) + timedelta(minutes=n)))
+            if n % 2 == 0:
+                db.add(PlPlanProduct(id=f'pp{n}', plan_id=f'p{n:02}', product_id='sku',
+                    product_no='SKU', product_name='硫酸', is_toxic=1, tenant_no='t'))
+        for key, values in [('foreign', {'tenant_no': 'other'}), ('deleted', {'del_flag': 1}),
+                            ('cancelled', {'follow_status': 'cancelled'})]:
+            db.add(PlPlan(id=key, plan_no=f'PLAN-hidden-{key}', create_date=date(2026, 11, 1),
+                plan_date=date(2026, 11, 1), plan_type='direct', belongs_company='shanxia',
+                **{'tenant_no': 't', **values}))
+        db.commit()
+        yield db, user
+    engine.dispose()
+
+
+def test_funnels_filter_before_pagination_and_count(case):
+    db, user = case
+    first = list_plans(db, user)
+    assert first['total'] == 25 and len(first['list']) == 20
+    assert 'p00' not in {r['id'] for r in first['list']}
+    result = list_plans(db, user, plan_no='PLAN-00', license_plate='A123',
+        plan_date_from=date(2026, 11, 1), plan_date_to=date(2026, 11, 1), quota_status_in='needs_configuration')
+    assert result['total'] == 1 and result['list'][0]['id'] == 'p00'
+    assert list_plans(db, user, plan_no='PLAN-00', page=2)['list'] == []
+
+
+def test_combined_keyword_and_funnels_intersect_and_date_end_is_inclusive(case):
+    db, user = case
+    result = list_plans(db, user, keyword='PLAN-0', license_plate='皖A',
+        plan_date_from='2026-11-02', plan_date_to='2026-11-03',
+        quota_status_in='not_required,needs_configuration')
+    assert {r['id'] for r in result['list']} == {'p02', 'p04', 'p08'}
+    assert result['total'] == 3
+    assert list_plans(db, user, status='not_required', quota_status_in='needs_configuration')['total'] == 0
+
+
+def test_empty_filters_preserve_existing_query_and_scope(case):
+    db, user = case
+    result = list_plans(db, user, plan_no='', license_plate='', quota_status_in='')
+    assert result['total'] == 25
+    assert not {r['id'] for r in result['list']} & {'foreign', 'deleted', 'cancelled'}
+
+
+def test_funnel_status_uses_computed_ledger_state(case):
+    db, user = case
+    db.add(PlPlanCertUsage(id='u', tenant_no='t', plan_id='p00', plan_no_snapshot='PLAN-00',
+        cert_type='purchase', cert_id='g', cert_no_snapshot='G', cert_revision=1, quantity_unit='吨', usage_status='reserved'))
+    db.commit()
+    assert list_plans(db, user, quota_status_in='reserved,settled')['total'] == 1
+    db.get(PlPlanCertUsage, 'u').confirmation_snapshot = {'settlementError': '失败'}
+    db.commit()
+    assert list_plans(db, user, quota_status_in='reserved,settled')['total'] == 0
+    assert list_plans(db, user, quota_status_in='exception')['list'][0]['id'] == 'p00'
+
+
+@pytest.mark.parametrize('filters', [
+    {'quota_status_in': 'reserved,unknown'},
+    {'plan_date_from': 'invalid'},
+    {'plan_date_from': '2026-11-03', 'plan_date_to': '2026-11-01'},
+])
+def test_invalid_filters_rejected(case, filters):
+    db, user = case
+    with pytest.raises(HTTPException) as error:
+        list_plans(db, user, **filters)
+    assert error.value.status_code == 400
+
+
+def test_new_text_filters_treat_sql_wildcards_literally(case):
+    db, user = case
+    assert list_plans(db, user, plan_no='%')['total'] == 0
+    assert list_plans(db, user, license_plate='_')['total'] == 0
+
+
+def test_funnels_preserve_union_of_role_data_scopes(case, monkeypatch):
+    from app.core import data_scope
+    db, _ = case
+    user = SimpleNamespace(id='u', tenant_no='t', to_user_info=lambda: {
+        'role': 'user', 'user_id': 'u', 'dept_id': 'dept'})
+    monkeypatch.setattr(data_scope, '_resolve_user_role_scopes', lambda *args: [
+        {'data_scope': '4', 'details': []},
+        {'data_scope': '5', 'details': [SimpleNamespace(scope_type='user', scope_value='colleague')]},
+    ])
+    db.get(PlPlan, 'p00').create_by = 'u'
+    db.get(PlPlan, 'p01').create_by = 'colleague'
+    db.get(PlPlan, 'p02').create_by = 'outside'
+    db.commit()
+    result = list_plans(db, user, plan_no='PLAN-0', quota_status_in='not_required,needs_configuration')
+    assert result['total'] == 2
+    assert {r['id'] for r in result['list']} == {'p00', 'p01'}