feat(pl): 运输证件五个页签支持按销售单位和购买单位模糊检索。

小程序检索沿用原有关键字，在编号和品名之外匹配证件主体以及计划供方、需方。

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

backend/app/services/pl_quota_cert_service.py

@@ -7,6 +7,7 @@ from uuid import uuid4
 
 from fastapi import HTTPException
 from pydantic import BaseModel
+from sqlalchemy import or_
 from sqlalchemy.exc import IntegrityError
 from app.models.pl_quota_models import (
     PlPurchaseCert, PlTransportCert, PlTransportCertPurchaseRef,
@@ -84,7 +85,13 @@ def list_certs(db, user, cert_type, keyword=None, status=None):
     query = db.query(model).filter(model.tenant_no == user.tenant_no, model.del_flag == 0)
     query = apply_pl_scope(query, model, user, db, user.tenant_no)
     if keyword:
-        query = query.filter(model.cert_no.contains(keyword))
+        sale_name = model.seller_name if cert_type == 'purchase' else model.shipper_name
+        buy_name = model.buyer_name if cert_type == 'purchase' else model.consignee_name
+        query = query.filter(or_(
+            model.cert_no.contains(keyword, autoescape=True),
+            sale_name.contains(keyword, autoescape=True),
+            buy_name.contains(keyword, autoescape=True),
+        ))
     if status:
         query = query.filter(model.status == status)
     return [get_cert(db, user, cert_type, r.id) for r in query.order_by(model.create_time.desc()).all()]

backend/app/services/pl_quota_overview.py

@@ -76,7 +76,11 @@ def list_plans(db, user, keyword='', status=None, page=1, page_size=20, *,
         PlPlan.del_flag == 0, PlPlan.follow_status != 'cancelled', hazard_product)
     query = apply_pl_scope(query, PlPlan, user, db, user.tenant_no)
     if keyword:
-        query = query.filter(PlPlan.plan_no.contains(keyword))
+        query = query.filter(or_(
+            PlPlan.plan_no.contains(keyword, autoescape=True),
+            PlPlan.supplier_name.contains(keyword, autoescape=True),
+            PlPlan.customer_name.contains(keyword, autoescape=True),
+        ))
     if status:
         query = query.filter(state == status)
     # 表头条件必须先于统计与分页，不能只过滤已加载的一页。

backend/app/services/pl_quota_statistics.py

@@ -117,13 +117,15 @@ def _certificates(db, user, keyword=None, cert_type=None, cert_no=None, cert_typ
         if cert_types is not None and kind not in cert_types:
             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)))
-        if cert_no:
-            query = query.filter(model.cert_no.contains(cert_no, autoescape=True))
         seller = model.seller_name if kind == 'purchase' else model.shipper_name
         buyer = model.buyer_name if kind == 'purchase' else model.consignee_name
+        if keyword:
+            query = query.filter(or_(model.cert_no.contains(keyword, autoescape=True),
+                                    model.product_name_on_cert.contains(keyword, autoescape=True),
+                                    seller.contains(keyword, autoescape=True),
+                                    buyer.contains(keyword, autoescape=True)))
+        if cert_no:
+            query = query.filter(model.cert_no.contains(cert_no, 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',
@@ -238,6 +240,21 @@ def _details(db, user, keys, contract_ids=None):
     return groups, usage_items
 
 
+def _contract_ids_by_party_keyword(db, user, keyword):
+    """合同列表按关联证件的销售单位、购买单位模糊命中。"""
+    selected = []
+    for kind, model in MODELS.items():
+        seller = model.seller_name if kind == 'purchase' else model.shipper_name
+        buyer = model.buyer_name if kind == 'purchase' else model.consignee_name
+        cert_ids = apply_pl_scope(_active(db, user, model), model, user, db, user.tenant_no).filter(
+            or_(seller.contains(keyword, autoescape=True), buyer.contains(keyword, autoescape=True))
+        ).with_entities(model.id)
+        selected.append(db.query(Link.contract_id.label('contract_id')).filter(
+            Link.tenant_no == user.tenant_no, Link.del_flag == 0,
+            Link.cert_type == kind, Link.cert_id.in_(cert_ids)))
+    return selected[0].union(*selected[1:])
+
+
 def list_contracts(db, user, keyword=None, page=1, page_size=20, contract_no=None, contract_name=None,
                    contract_type_in=None, cert_count_op=None, cert_count=None, cert_count_min=None,
                    cert_count_max=None):
@@ -249,7 +266,8 @@ def list_contracts(db, user, keyword=None, page=1, page_size=20, contract_no=Non
     query = _contracts(db, user).join(counts, counts.c.contract_id == CrmContract.id)
     if keyword:
         query = query.filter(or_(CrmContract.contract_no.contains(keyword, autoescape=True),
-                                 CrmContract.contract_name.contains(keyword, autoescape=True)))
+                                 CrmContract.contract_name.contains(keyword, autoescape=True),
+                                 CrmContract.id.in_(_contract_ids_by_party_keyword(db, user, keyword))))
     if _text(contract_no):
         query = query.filter(CrmContract.contract_no.contains(_text(contract_no), autoescape=True))
     if _text(contract_name):

backend/app/services/pl_quota_waiver.py

@@ -214,7 +214,11 @@ def list_ready_plans(db, user, keyword="", waived=None, page=1, page_size=20, pl
     )
     query = apply_pl_scope(query, PlPlan, user, db, user.tenant_no)
     if keyword:
-        query = query.filter(PlPlan.plan_no.contains(keyword, autoescape=True))
+        query = query.filter(or_(
+            PlPlan.plan_no.contains(keyword, autoescape=True),
+            PlPlan.supplier_name.contains(keyword, autoescape=True),
+            PlPlan.customer_name.contains(keyword, autoescape=True),
+        ))
     for column, value in (
         (PlPlan.plan_no, plan_no),
         (PlPlan.license_plate, license_plate),

backend/tests/test_pl_quota_cert_archive.py

@@ -179,6 +179,18 @@ def test_readonly_cache_and_tenant_cannot_be_written(ctx):
         with pytest.raises(ValidationError): s.save_cert(db,u,'purchase',{**purchase(),field:value})
 
 
+def test_list_keyword_matches_sale_and_buy_party(ctx):
+    db, u, s, m = ctx
+    purchase_row = s.save_cert(db, u, 'purchase', purchase())
+    transport_row = s.save_cert(db, u, 'transport', transport(consigneeName='收货单位甲'))
+    assert [row['id'] for row in s.list_certs(db, u, 'purchase', keyword='销方')] == [purchase_row['id']]
+    assert [row['id'] for row in s.list_certs(db, u, 'purchase', keyword='购方')] == [purchase_row['id']]
+    assert s.list_certs(db, u, 'purchase', keyword='货主') == []
+    assert [row['id'] for row in s.list_certs(db, u, 'transport', keyword='货主')] == [transport_row['id']]
+    assert [row['id'] for row in s.list_certs(db, u, 'transport', keyword='收货')] == [transport_row['id']]
+    assert s.list_certs(db, u, 'transport', keyword='%') == []
+
+
 def test_transport_all_fields_and_scope_filtered_list(ctx):
     db,u,s,m=ctx
     data=transport(shipperAddress='地址',shipperPhone='电话',shipperLegalRepresentative='甲',packagingText='槽罐',carrierName='运方',carrierAddressPhone='住址/电话',transportMode='汽车',routeText='路线',consigneeName='收方',applicantName='申请方')

backend/tests/test_pl_quota_overview_filters.py

@@ -70,6 +70,18 @@ def test_funnels_filter_before_pagination_and_count(case):
         list_plans(db, user, follow_status_in='unknown')
 
 
+def test_keyword_matches_supplier_and_customer(case):
+    db, user = case
+    plan = db.get(PlPlan, 'p00')
+    plan.supplier_name = '山峡供方'
+    plan.customer_name = '川沃需方'
+    db.commit()
+    assert {row['id'] for row in list_plans(db, user, keyword='山峡')['list']} == {'p00'}
+    assert list_plans(db, user, keyword='川沃')['total'] == 1
+    assert list_plans(db, user, keyword='不存在单位')['total'] == 0
+    assert list_plans(db, user, keyword='%')['total'] == 0
+
+
 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',

backend/tests/test_pl_quota_statistics.py

@@ -72,6 +72,23 @@ def test_shared_balance_and_contract_allocation_are_distinct_and_read_only(case)
     assert not case.db.new and not case.db.dirty
 
 
+def test_keyword_matches_sale_and_buy_party_on_certs_and_linked_contracts(case):
+    s = service()
+    case.db.add(m.PlTransportCert(
+        id='ship', cert_no='T-party', product_name_on_cert='盐酸', total_qty=1, quantity_unit='吨',
+        valid_use_text='一次有效', shipper_name='托运销售', consignee_name='收货购买',
+        vehicle_numbers_json=['皖A'], tenant_no='t'))
+    case.db.commit()
+    assert s.list_certificates(case.db, case.user, keyword='买方')['total'] == 1
+    assert s.list_certificates(case.db, case.user, keyword='卖方')['total'] == 1
+    assert s.list_certificates(case.db, case.user, keyword='托运')['total'] == 1
+    assert s.list_certificates(case.db, case.user, keyword='收货购买')['total'] == 1
+    assert s.list_certificates(case.db, case.user, keyword='无关单位')['total'] == 0
+    assert s.list_certificates(case.db, case.user, keyword='%')['total'] == 0
+    assert s.list_contracts(case.db, case.user, keyword='买方')['total'] == 2
+    assert s.list_contracts(case.db, case.user, keyword='托运销售')['total'] == 0
+
+
 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)

backend/tests/test_pl_quota_waiver.py

@@ -4,6 +4,37 @@ from types import SimpleNamespace
 from fastapi import HTTPException
 
 
+def test_ready_keyword_matches_supplier_and_customer(monkeypatch):
+    from datetime import date
+
+    from sqlalchemy import create_engine, true
+    from sqlalchemy.orm import Session
+
+    from app.models.pl_models import PlPlan
+    from app.services import pl_quota_waiver as svc
+
+    monkeypatch.setattr(svc, "_toxic_exists", lambda *args, **kwargs: true())
+    monkeypatch.setattr(svc, "apply_pl_scope", lambda query, *args, **kwargs: query)
+    monkeypatch.setattr(svc, "_product_names", lambda *args, **kwargs: {})
+    engine = create_engine("sqlite://")
+    PlPlan.__table__.create(engine)
+    db = Session(engine)
+    user = SimpleNamespace(tenant_no="t")
+    db.add(PlPlan(id="sale", plan_no="PO-S", create_date=date(2026, 11, 1), plan_type="direct",
+                  belongs_company="shanxia", tenant_no="t", supplier_name="销售单位甲", customer_name="其他需方"))
+    db.add(PlPlan(id="buy", plan_no="PO-B", create_date=date(2026, 11, 2), plan_type="direct",
+                  belongs_company="shanxia", tenant_no="t", supplier_name="其他供方", customer_name="购买单位乙"))
+    db.commit()
+    try:
+        assert [row["id"] for row in svc.list_ready_plans(db, user, keyword="销售单位")["list"]] == ["sale"]
+        assert [row["id"] for row in svc.list_ready_plans(db, user, keyword="购买单位")["list"]] == ["buy"]
+        assert svc.list_ready_plans(db, user, keyword="PO-S")["total"] == 1
+        assert svc.list_ready_plans(db, user, keyword="%")["total"] == 0
+    finally:
+        db.close()
+        engine.dispose()
+
+
 def test_ready_list_orders_by_plan_date_ascending(monkeypatch):
     from app.services import pl_quota_waiver as svc
 

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

@@ -361,10 +361,10 @@
   ]
   const placeholder = computed(() =>
     tab.value === 'contract'
-      ? '合同编号 / 名称'
+      ? '合同编号 / 名称 / 销售或购买单位'
       : tab.value === 'plan' || tab.value === 'ready'
-        ? '计划编号'
-        : '证件编号 / 品名'
+        ? '计划编号 / 供方 / 需方'
+        : '证件编号 / 品名 / 销售或购买单位'
   )
 
   function onTab(name: string | number) {