feat(pl): add stockRole filter for plan list by stop customer

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

backend/app/services/pl/plan_list_filters.py

@@ -10,7 +10,7 @@ from datetime import date, datetime
 from decimal import Decimal, InvalidOperation
 from typing import List, Optional, Union
 
-from sqlalchemy import and_, func, or_
+from sqlalchemy import and_, exists, func, or_, select
 from sqlalchemy.orm import Session
 
 from app.models.pl_models import (
@@ -23,11 +23,13 @@ from app.models.pl_models import (
     PlPlanProduct,
     PlTransport,
 )
+from app.services.pl_internal_customer import INTERNAL_CUSTOMER_NAMES
 from app.services.pl_plan_service import plan_has_incomplete_hazard_cert_exists
 
 DateLike = Union[date, datetime, str, None]
 
 _NUMBER_OPS = frozenset({"between", "eq", "gt", "gte", "lt", "lte"})
+_STOCK_ROLES = frozenset({"inbound", "outbound"})
 
 
 def plan_product_name_exists(db: Session, tenant_no: str, keyword: str):
@@ -321,6 +323,30 @@ def plan_stop_kind_exists(db: Session, tenant_no: str, kind: str):
     )
 
 
+def stock_role_exists_clause(*, tenant_no: str, stock_role: str):
+    """inbound=送货段客户为内部客户；outbound=装货段客户为内部客户。"""
+    role = (stock_role or "").strip().lower()
+    if role not in _STOCK_ROLES:
+        raise ValueError(f"invalid stockRole: {stock_role}")
+    model = PlPlanDelivery if role == "inbound" else PlPlanLoad
+    return exists(
+        select(model.id).where(
+            model.plan_id == PlPlan.id,
+            model.tenant_no == tenant_no,
+            model.del_flag == 0,
+            model.customer_name.in_(tuple(INTERNAL_CUSTOMER_NAMES)),
+        )
+    )
+
+
+def apply_stock_role_filter(query, *, stock_role: Optional[str], tenant_no: str):
+    if stock_role is None or str(stock_role).strip() == "":
+        return query
+    return query.filter(
+        stock_role_exists_clause(tenant_no=tenant_no, stock_role=str(stock_role))
+    )
+
+
 def parse_stop_customer_keywords(raw: Optional[str]) -> List[str]:
     """客户列多值：JSON 数组，或单个纯文本关键字。"""
     if raw is None:

backend/tests/test_pl_plan_stock_role_filter.py

@@ -0,0 +1,56 @@
+"""库存列表 stockRole 过滤单测。"""
+from __future__ import annotations
+
+import pytest
+from sqlalchemy.dialects import mysql
+
+from app.services.pl.plan_list_filters import (
+    apply_stock_role_filter,
+    stock_role_exists_clause,
+)
+from app.services.pl_internal_customer import INTERNAL_CUSTOMER_NAMES
+
+
+def _compile(clause) -> str:
+    return str(
+        clause.compile(dialect=mysql.dialect(), compile_kwargs={"literal_binds": True})
+    ).lower()
+
+
+def test_inbound_sql_mentions_delivery_and_internal_names():
+    clause = stock_role_exists_clause(tenant_no="t1", stock_role="inbound")
+    sql = _compile(clause)
+    assert "pl_plan_delivery" in sql
+    for name in INTERNAL_CUSTOMER_NAMES:
+        assert name.lower() in sql
+
+
+def test_outbound_sql_mentions_load_and_internal_names():
+    clause = stock_role_exists_clause(tenant_no="t1", stock_role="outbound")
+    sql = _compile(clause)
+    assert "pl_plan_load" in sql
+    for name in INTERNAL_CUSTOMER_NAMES:
+        assert name.lower() in sql
+
+
+def test_invalid_stock_role_raises():
+    with pytest.raises(ValueError):
+        stock_role_exists_clause(tenant_no="t1", stock_role="weird")
+
+
+def test_apply_stock_role_none_skips_filter():
+    class Dummy:
+        def filter(self, *a, **k):
+            raise AssertionError("should not filter when stock_role empty")
+
+    dummy = Dummy()
+    assert apply_stock_role_filter(dummy, stock_role=None, tenant_no="t1") is dummy
+
+
+def test_apply_stock_role_blank_skips_filter():
+    class Dummy:
+        def filter(self, *a, **k):
+            raise AssertionError("should not filter when stock_role blank")
+
+    dummy = Dummy()
+    assert apply_stock_role_filter(dummy, stock_role="  ", tenant_no="t1") is dummy