feat(backend): add production waste-output list query helpers.

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

backend/app/crud/production_waste_output.py

@@ -0,0 +1,185 @@
+"""生产任务步骤废弃物产出 — 只读汇总列表。"""
+from __future__ import annotations
+
+from typing import Any, Dict, Iterable, List, Optional, Set, Tuple
+
+from sqlalchemy import and_, or_
+from sqlalchemy.orm import Session
+
+from app.models.models import (
+    BizOutputType,
+    BizProductionTask,
+    BizProductionTaskStep,
+    BizProductionTaskStepOutput,
+    BizProductionWorkOrder,
+)
+
+WASTE_OUTPUT_TYPE_CODE = "FQW"
+WASTE_OUTPUT_TYPE_NAME = "废弃物"
+
+
+def pick_waste_output_type_ids(rows: Iterable[Dict[str, Any]]) -> Set[str]:
+    """从产出类型行中选出废弃物类型 id：优先 type_code=FQW，否则兜底 type_name=废弃物。"""
+    rows = list(rows or [])
+    by_code = {
+        str(r.get("id") or "").strip()
+        for r in rows
+        if str(r.get("type_code") or "").strip() == WASTE_OUTPUT_TYPE_CODE
+        and str(r.get("id") or "").strip()
+    }
+    if by_code:
+        return by_code
+    return {
+        str(r.get("id") or "").strip()
+        for r in rows
+        if str(r.get("type_name") or "").strip() == WASTE_OUTPUT_TYPE_NAME
+        and str(r.get("id") or "").strip()
+    }
+
+
+def match_waste_output_keyword(
+    keyword: Optional[str], output_name: Optional[str], output_barcode: Optional[str]
+) -> bool:
+    kw = (keyword or "").strip()
+    if not kw:
+        return True
+    name = output_name or ""
+    barcode = output_barcode or ""
+    return kw in name or kw in barcode
+
+
+def resolve_waste_output_type_ids(db: Session, tenant_no: str) -> Set[str]:
+    rows = (
+        db.query(BizOutputType.id, BizOutputType.type_code, BizOutputType.type_name)
+        .filter(BizOutputType.tenant_no == tenant_no, BizOutputType.del_flag == 0)
+        .all()
+    )
+    return pick_waste_output_type_ids(
+        [{"id": r.id, "type_code": r.type_code, "type_name": r.type_name} for r in rows]
+    )
+
+
+def _fmt_dt(t) -> str:
+    if t is None:
+        return ""
+    if hasattr(t, "strftime"):
+        return t.strftime("%Y-%m-%d %H:%M:%S")
+    return str(t)
+
+
+def _normalize_date_bound(value: str, *, end_of_day: bool = False) -> str:
+    v = (value or "").strip()
+    if not v:
+        return v
+    if len(v) <= 10:
+        return f"{v} 23:59:59" if end_of_day else f"{v} 00:00:00"
+    return v
+
+
+def list_production_waste_outputs(
+    db: Session,
+    *,
+    tenant_no: str,
+    page: int = 1,
+    page_size: int = 10,
+    keyword: Optional[str] = None,
+    create_time_start: Optional[str] = None,
+    create_time_end: Optional[str] = None,
+    user_info=None,
+) -> Tuple[List[Dict[str, Any]], int]:
+    type_ids = resolve_waste_output_type_ids(db, tenant_no)
+    if not type_ids:
+        return [], 0
+
+    q = (
+        db.query(
+            BizProductionTaskStepOutput,
+            BizProductionTaskStep,
+            BizProductionTask,
+            BizProductionWorkOrder,
+        )
+        .join(
+            BizProductionTaskStep,
+            and_(
+                BizProductionTaskStep.id == BizProductionTaskStepOutput.step_id,
+                BizProductionTaskStep.del_flag == 0,
+            ),
+        )
+        .join(
+            BizProductionTask,
+            and_(
+                BizProductionTask.id == BizProductionTaskStep.task_id,
+                BizProductionTask.del_flag == 0,
+                BizProductionTask.tenant_no == tenant_no,
+            ),
+        )
+        .outerjoin(
+            BizProductionWorkOrder,
+            and_(
+                BizProductionWorkOrder.id == BizProductionTask.work_order_id,
+                BizProductionWorkOrder.del_flag == 0,
+            ),
+        )
+        .filter(
+            BizProductionTaskStepOutput.tenant_no == tenant_no,
+            BizProductionTaskStepOutput.del_flag == 0,
+            BizProductionTaskStepOutput.output_type_id.in_(list(type_ids)),
+        )
+    )
+
+    kw = (keyword or "").strip()
+    if kw:
+        like = f"%{kw}%"
+        q = q.filter(
+            or_(
+                BizProductionTaskStepOutput.output_name.like(like),
+                BizProductionTaskStepOutput.output_barcode.like(like),
+            )
+        )
+    if create_time_start and create_time_start.strip():
+        q = q.filter(
+            BizProductionTaskStepOutput.create_time
+            >= _normalize_date_bound(create_time_start.strip())
+        )
+    if create_time_end and create_time_end.strip():
+        q = q.filter(
+            BizProductionTaskStepOutput.create_time
+            <= _normalize_date_bound(create_time_end.strip(), end_of_day=True)
+        )
+
+    if user_info is not None:
+        from app.core.data_scope import apply_data_scope
+
+        q = apply_data_scope(q, BizProductionTask, user_info, db, tenant_no)
+
+    q = q.order_by(BizProductionTaskStepOutput.create_time.desc())
+    total = q.count()
+    page = max(1, int(page or 1))
+    page_size = max(1, min(int(page_size or 10), 100))
+    rows = q.offset((page - 1) * page_size).limit(page_size).all()
+
+    items: List[Dict[str, Any]] = []
+    for out, step, task, wo in rows:
+        items.append(
+            {
+                "id": out.id,
+                "outputName": out.output_name or "",
+                "outputBarcode": out.output_barcode or "",
+                "remark": out.remark or "",
+                "createTime": _fmt_dt(out.create_time),
+                "stepId": out.step_id or "",
+                "processName": (step.process_name if step else None) or "",
+                "taskId": (task.id if task else None) or "",
+                "taskNo": (task.task_no if task else None) or "",
+                "workOrderId": (wo.id if wo else None) or (task.work_order_id if task else "") or "",
+                "workOrderNo": (wo.work_order_no if wo else None) or "",
+                "sampleCode": (wo.sample_code if wo else None) or "",
+            }
+        )
+    return items, total
+
+
+def total_pages(total: int, page_size: int) -> int:
+    if page_size <= 0:
+        return 0
+    return (total + page_size - 1) // page_size

backend/tests/test_production_waste_output.py

@@ -0,0 +1,55 @@
+"""生产废弃物列表：产出类型解析与关键字匹配语义。"""
+import unittest
+
+from app.crud.production_waste_output import (
+    WASTE_OUTPUT_TYPE_CODE,
+    WASTE_OUTPUT_TYPE_NAME,
+    match_waste_output_keyword,
+    pick_waste_output_type_ids,
+)
+
+
+class TestPickWasteOutputTypeIds(unittest.TestCase):
+    def test_prefers_type_code_fqw(self):
+        rows = [
+            {"id": "a", "type_code": "YXW", "type_name": "有效物"},
+            {"id": "b", "type_code": "FQW", "type_name": "废弃物"},
+            {"id": "c", "type_code": "", "type_name": "废弃物"},
+        ]
+        self.assertEqual(pick_waste_output_type_ids(rows), {"b"})
+
+    def test_fallback_type_name_when_no_fqw(self):
+        rows = [
+            {"id": "a", "type_code": "YXW", "type_name": "有效物"},
+            {"id": "c", "type_code": "", "type_name": "废弃物"},
+            {"id": "d", "type_code": None, "type_name": "废弃物"},
+        ]
+        self.assertEqual(pick_waste_output_type_ids(rows), {"c", "d"})
+
+    def test_empty_when_no_match(self):
+        self.assertEqual(pick_waste_output_type_ids([]), set())
+        self.assertEqual(
+            pick_waste_output_type_ids([{"id": "a", "type_code": "YXW", "type_name": "有效物"}]),
+            set(),
+        )
+
+
+class TestMatchWasteOutputKeyword(unittest.TestCase):
+    def test_matches_name_or_barcode(self):
+        self.assertTrue(match_waste_output_keyword("废液", "废液A", "X1"))
+        self.assertTrue(match_waste_output_keyword("X1", "废液", "BX1Y"))
+        self.assertFalse(match_waste_output_keyword("zzz", "废液", "X1"))
+
+    def test_blank_keyword_always_true(self):
+        self.assertTrue(match_waste_output_keyword("", "废液", ""))
+        self.assertTrue(match_waste_output_keyword(None, "", ""))
+
+
+class TestConstants(unittest.TestCase):
+    def test_constants(self):
+        self.assertEqual(WASTE_OUTPUT_TYPE_CODE, "FQW")
+        self.assertEqual(WASTE_OUTPUT_TYPE_NAME, "废弃物")
+
+
+if __name__ == "__main__":
+    unittest.main()