feat(pl): add contract-gap SQL filter and response schemas

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

backend/app/schemas/pl/__init__.py

@@ -43,6 +43,9 @@ from app.schemas.pl.plan import (
     PlanWithProducts,
     PlanListQuery,
     PlanListResponse,
+    PlanSideContractsUpdateBody,
+    PlanContractGapRead,
+    PlanContractGapListResponse,
     CreateFromContractRequest,
     PlanCreateFull,
     PlanTransportCreateBlock,
@@ -147,6 +150,9 @@ __all__ = [
     "PlanWithProducts",
     "PlanListQuery",
     "PlanListResponse",
+    "PlanSideContractsUpdateBody",
+    "PlanContractGapRead",
+    "PlanContractGapListResponse",
     "CreateFromContractRequest",
     "PlanCreateFull",
     "PlanTransportCreateBlock",

backend/app/schemas/pl/plan.py

@@ -419,3 +419,30 @@ class PlanSiteFeedbackUpdateBody(BaseModel):
 class PlanListResponse(BaseModel):
     total: int
     list: list[PlanWithProducts]
+
+
+class PlanSideContractsUpdateBody(BaseModel):
+    contracts: list[PlanContractLink] = Field(default_factory=list)
+    model_config = {"populate_by_name": True}
+
+
+class PlanContractGapRead(BaseModel):
+    id: str
+    plan_no: str = Field(..., alias="planNo")
+    plan_date: Optional[date] = Field(None, alias="planDate")
+    follow_status: str = Field(..., alias="followStatus")
+    supplier_id: Optional[str] = Field(None, alias="supplierId")
+    supplier_name: Optional[str] = Field(None, alias="supplierName")
+    customer_id: Optional[str] = Field(None, alias="customerId")
+    customer_name: Optional[str] = Field(None, alias="customerName")
+    gap_sides: list[str] = Field(default_factory=list, alias="gapSides")
+    supplier_contract_exempt: bool = Field(..., alias="supplierContractExempt")
+    demander_contract_exempt: bool = Field(..., alias="demanderContractExempt")
+    supplier_contracts: list[PlanContractLink] = Field(default_factory=list, alias="supplierContracts")
+    demander_contracts: list[PlanContractLink] = Field(default_factory=list, alias="demanderContracts")
+    model_config = {"populate_by_name": True, "from_attributes": True}
+
+
+class PlanContractGapListResponse(BaseModel):
+    total: int
+    list: list[PlanContractGapRead]

backend/app/services/pl_plan_contract_gap.py

@@ -2,7 +2,11 @@ from __future__ import annotations
 
 from typing import List, Optional
 
-from app.services.pl_internal_customer import is_internal_customer_name
+from sqlalchemy import and_, exists, or_, select
+
+from app.models.pl_models import PlPlan, PlPlanContract
+from app.services.pl_internal_customer import INTERNAL_CUSTOMER_NAMES, is_internal_customer_name
+from app.services.pl_plan_service import PLAN_CONTRACT_SIDE_DEMANDER, PLAN_CONTRACT_SIDE_SUPPLIER
 
 
 def compute_contract_gap_sides(
@@ -22,3 +26,38 @@ def compute_contract_gap_sides(
 
 def side_is_exempt(name: Optional[str]) -> bool:
     return is_internal_customer_name(name)
+
+
+def _side_contract_exists(tenant_no: str, side: str):
+    return exists(
+        select(PlPlanContract.id).where(
+            PlPlanContract.plan_id == PlPlan.id,
+            PlPlanContract.tenant_no == tenant_no,
+            PlPlanContract.del_flag == 0,
+            PlPlanContract.side == side,
+        )
+    )
+
+
+def plan_has_contract_gap_exists(tenant_no: str):
+    """列表筛选：计划存在合同缺口（非内置侧缺少有效 pl_plan_contract 关联）。"""
+    internal_names = tuple(INTERNAL_CUSTOMER_NAMES)
+    supplier_not_internal = or_(
+        PlPlan.supplier_name.is_(None),
+        PlPlan.supplier_name == "",
+        ~PlPlan.supplier_name.in_(internal_names),
+    )
+    customer_not_internal = or_(
+        PlPlan.customer_name.is_(None),
+        PlPlan.customer_name == "",
+        ~PlPlan.customer_name.in_(internal_names),
+    )
+    supplier_gap = and_(
+        supplier_not_internal,
+        ~_side_contract_exists(tenant_no, PLAN_CONTRACT_SIDE_SUPPLIER),
+    )
+    demander_gap = and_(
+        customer_not_internal,
+        ~_side_contract_exists(tenant_no, PLAN_CONTRACT_SIDE_DEMANDER),
+    )
+    return or_(supplier_gap, demander_gap)

backend/tests/test_pl_plan_contract_gap.py

@@ -1,5 +1,7 @@
+from sqlalchemy.dialects import mysql
+
 from app.services.pl_internal_customer import INTERNAL_CUSTOMER_NAMES
-from app.services.pl_plan_contract_gap import compute_contract_gap_sides
+from app.services.pl_plan_contract_gap import compute_contract_gap_sides, plan_has_contract_gap_exists
 
 INTERNAL = next(iter(INTERNAL_CUSTOMER_NAMES))
 
@@ -42,3 +44,28 @@ def test_external_with_contracts_no_gap():
         demander_contract_count=2,
     )
     assert sides == []
+
+
+def test_empty_names_treated_as_non_internal_gap_both():
+    sides = compute_contract_gap_sides(
+        supplier_name=None,
+        customer_name="",
+        supplier_contract_count=0,
+        demander_contract_count=0,
+    )
+    assert sides == ["supplier", "demander"]
+
+
+def _compile(clause) -> str:
+    return str(
+        clause.compile(dialect=mysql.dialect(), compile_kwargs={"literal_binds": True})
+    ).lower()
+
+
+def test_plan_has_contract_gap_exists_mentions_contract_and_internal_names():
+    clause = plan_has_contract_gap_exists("shanxia2.0")
+    sql = _compile(clause)
+
+    assert "pl_plan_contract" in sql
+    assert "supplier" in sql
+    assert "demander" in sql