From 79f9649610c007e8871bcd6cab6813f5ab2461ec Mon Sep 17 00:00:00 2001 From: zlwu <704541015@qq.com> Date: Tue, 18 Aug 2026 10:54:00 +0800 Subject: [PATCH] feat(backend): add production waste-output list query helpers. Co-authored-by: Cursor --- backend/app/crud/production_waste_output.py | 185 ++++++++++++++++++ backend/tests/test_production_waste_output.py | 55 ++++++ 2 files changed, 240 insertions(+) create mode 100644 backend/app/crud/production_waste_output.py create mode 100644 backend/tests/test_production_waste_output.py diff --git a/backend/app/crud/production_waste_output.py b/backend/app/crud/production_waste_output.py new file mode 100644 index 00000000..fd720439 --- /dev/null +++ b/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 diff --git a/backend/tests/test_production_waste_output.py b/backend/tests/test_production_waste_output.py new file mode 100644 index 00000000..c62ebff4 --- /dev/null +++ b/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()