From 3aa8aebc29cb79baf9ff7b97da23161f76aa5934 Mon Sep 17 00:00:00 2001 From: zlwu <704541015@qq.com> Date: Wed, 19 Aug 2026 12:46:25 +0800 Subject: [PATCH] feat(backend): add qc material month template CRUD preview and confirm Co-authored-by: Cursor --- .../app/crud/qc_material_month_template.py | 364 ++++++++++++++++++ .../tests/test_qc_material_month_template.py | 49 +++ 2 files changed, 413 insertions(+) create mode 100644 backend/app/crud/qc_material_month_template.py create mode 100644 backend/tests/test_qc_material_month_template.py diff --git a/backend/app/crud/qc_material_month_template.py b/backend/app/crud/qc_material_month_template.py new file mode 100644 index 00000000..636c7800 --- /dev/null +++ b/backend/app/crud/qc_material_month_template.py @@ -0,0 +1,364 @@ +"""质检物料月初入库模板 CRUD / preview / confirm。""" +from __future__ import annotations + +from datetime import date, datetime +from decimal import Decimal +from typing import Any, Dict, List, Optional + +from sqlalchemy.orm import Session + +from app.core.utils import generate_uuid +from app.crud import qc_material as mat_svc +from app.crud import qc_material_stock_batch as batch_svc +from app.crud.qc_material import _is_enabled_status +from app.crud.qc_material_stock_batch import _fmt_date, _parse_date, _require_decimal, _require_str +from app.crud.qc_material_stock_in_no import allocate_stock_in_nos +from app.models.qc_material import BizQcMaterialMonthTemplate + + +def _require_template_fields(payload: dict) -> dict: + """校验并规范化模板写入字段；缺必填抛 ValueError。""" + if not isinstance(payload, dict): + raise ValueError("模板数据无效") + + material_id = (payload.get("material_id") or payload.get("materialId") or "").strip() or None + material_code = _require_str( + payload.get("material_code") or payload.get("materialCode"), "物料编码" + ) + material_name = _require_str( + payload.get("material_name") or payload.get("materialName"), "物料名称" + ) + catalog_no = _require_str(payload.get("catalog_no") or payload.get("catalogNo"), "货号") + batch_no = _require_str(payload.get("batch_no") or payload.get("batchNo"), "批号") + manufacturer = _require_str(payload.get("manufacturer"), "厂家") + unit_price = _require_decimal(payload.get("unit_price") or payload.get("unitPrice"), "单价") + amount = _require_decimal(payload.get("amount"), "金额") + in_qty_raw = payload.get("in_qty") if payload.get("in_qty") is not None else payload.get("inQty") + in_qty = _require_decimal(in_qty_raw, "入库数量") + if in_qty <= 0: + raise ValueError("入库数量必须大于0") + expire_date = _parse_date( + payload.get("expire_date") or payload.get("expireDate"), label="过期日期" + ) + + spec_model = (payload.get("spec_model") or payload.get("specModel") or "").strip() or None + unit = (payload.get("unit") or "").strip() or None + remark = (payload.get("remark") or "").strip() or None + order_num = payload.get("order_num") + if order_num is None: + order_num = payload.get("orderNum") + if order_num is None: + order_num = 0 + try: + order_num = int(order_num) + except (TypeError, ValueError) as e: + raise ValueError("排序号格式无效") from e + + return { + "material_id": material_id, + "material_code": material_code, + "material_name": material_name, + "spec_model": spec_model, + "unit": unit, + "catalog_no": catalog_no, + "batch_no": batch_no, + "manufacturer": manufacturer, + "unit_price": unit_price, + "amount": amount, + "in_qty": in_qty, + "expire_date": expire_date, + "order_num": order_num, + "remark": remark, + } + + +def _row_to_dict(row: BizQcMaterialMonthTemplate) -> Dict[str, Any]: + return { + "id": row.id, + "materialId": row.material_id or "", + "materialCode": row.material_code or "", + "materialName": row.material_name or "", + "specModel": row.spec_model or "", + "unit": row.unit or "", + "catalogNo": row.catalog_no or "", + "batchNo": row.batch_no or "", + "manufacturer": row.manufacturer or "", + "unitPrice": float(row.unit_price) if row.unit_price is not None else 0, + "amount": float(row.amount) if row.amount is not None else 0, + "inQty": float(row.in_qty) if row.in_qty is not None else 0, + "expireDate": _fmt_date(row.expire_date), + "orderNum": row.order_num if row.order_num is not None else 0, + "remark": row.remark or "", + "createTime": row.create_time.strftime("%Y-%m-%d %H:%M:%S") if row.create_time else "", + "updateTime": row.update_time.strftime("%Y-%m-%d %H:%M:%S") if row.update_time else "", + } + + +def list_templates(db: Session, *, tenant_no: str) -> List[Dict[str, Any]]: + rows = ( + db.query(BizQcMaterialMonthTemplate) + .filter( + BizQcMaterialMonthTemplate.tenant_no == tenant_no, + BizQcMaterialMonthTemplate.del_flag == 0, + ) + .order_by( + BizQcMaterialMonthTemplate.order_num.asc(), + BizQcMaterialMonthTemplate.create_time.asc(), + ) + .all() + ) + return [_row_to_dict(r) for r in rows] + + +def create_template( + db: Session, + *, + data: dict, + tenant_no: str, + user_id: Optional[str], + create_dept_id: str = "", +) -> Dict[str, Any]: + fields = _require_template_fields(data) + if fields["material_id"]: + mat = mat_svc.get_material(db, fields["material_id"], tenant_no=tenant_no) + if not mat: + raise ValueError("关联物资不存在") + now = datetime.now() + row = BizQcMaterialMonthTemplate( + id=generate_uuid(), + material_id=fields["material_id"], + material_code=fields["material_code"], + material_name=fields["material_name"], + spec_model=fields["spec_model"], + unit=fields["unit"], + catalog_no=fields["catalog_no"], + batch_no=fields["batch_no"], + manufacturer=fields["manufacturer"], + unit_price=fields["unit_price"], + amount=fields["amount"], + in_qty=fields["in_qty"], + expire_date=fields["expire_date"], + tenant_no=tenant_no, + order_num=fields["order_num"], + remark=fields["remark"], + create_by=user_id, + create_dept_id=create_dept_id or "", + create_time=now, + update_by=user_id, + update_time=now, + del_flag=0, + ) + db.add(row) + db.commit() + db.refresh(row) + return _row_to_dict(row) + + +def update_template( + db: Session, + row_id: str, + *, + data: dict, + tenant_no: str, + user_id: Optional[str], +) -> Dict[str, Any]: + row = ( + db.query(BizQcMaterialMonthTemplate) + .filter( + BizQcMaterialMonthTemplate.id == row_id, + BizQcMaterialMonthTemplate.tenant_no == tenant_no, + BizQcMaterialMonthTemplate.del_flag == 0, + ) + .first() + ) + if not row: + raise ValueError("模板行不存在") + fields = _require_template_fields(data) + if fields["material_id"]: + mat = mat_svc.get_material(db, fields["material_id"], tenant_no=tenant_no) + if not mat: + raise ValueError("关联物资不存在") + row.material_id = fields["material_id"] + row.material_code = fields["material_code"] + row.material_name = fields["material_name"] + row.spec_model = fields["spec_model"] + row.unit = fields["unit"] + row.catalog_no = fields["catalog_no"] + row.batch_no = fields["batch_no"] + row.manufacturer = fields["manufacturer"] + row.unit_price = fields["unit_price"] + row.amount = fields["amount"] + row.in_qty = fields["in_qty"] + row.expire_date = fields["expire_date"] + row.order_num = fields["order_num"] + row.remark = fields["remark"] + row.update_by = user_id + row.update_time = datetime.now() + db.commit() + db.refresh(row) + return _row_to_dict(row) + + +def soft_delete_template(db: Session, row_id: str, *, tenant_no: str) -> None: + row = ( + db.query(BizQcMaterialMonthTemplate) + .filter( + BizQcMaterialMonthTemplate.id == row_id, + BizQcMaterialMonthTemplate.tenant_no == tenant_no, + BizQcMaterialMonthTemplate.del_flag == 0, + ) + .first() + ) + if not row: + raise ValueError("模板行不存在") + row.del_flag = 1 + row.update_time = datetime.now() + db.commit() + + +def preview_templates( + db: Session, *, tenant_no: str, stock_in_date: date +) -> List[Dict[str, Any]]: + rows = ( + db.query(BizQcMaterialMonthTemplate) + .filter( + BizQcMaterialMonthTemplate.tenant_no == tenant_no, + BizQcMaterialMonthTemplate.del_flag == 0, + ) + .order_by( + BizQcMaterialMonthTemplate.order_num.asc(), + BizQcMaterialMonthTemplate.create_time.asc(), + ) + .all() + ) + if not rows: + raise ValueError("请先维护月初模板") + nos = allocate_stock_in_nos( + db, tenant_no=tenant_no, stock_in_date=stock_in_date, count=len(rows) + ) + out: List[Dict[str, Any]] = [] + for i, row in enumerate(rows): + out.append( + { + "templateId": row.id, + "materialId": row.material_id or "", + "materialCode": row.material_code or "", + "materialName": row.material_name or "", + "specModel": row.spec_model or "", + "unit": row.unit or "", + "stockInNo": nos[i], + "stockInDate": _fmt_date(stock_in_date), + "expireDate": _fmt_date(row.expire_date), + "catalogNo": row.catalog_no or "", + "batchNo": row.batch_no or "", + "manufacturer": row.manufacturer or "", + "unitPrice": float(row.unit_price) if row.unit_price is not None else 0, + "amount": float(row.amount) if row.amount is not None else 0, + "qty": float(row.in_qty) if row.in_qty is not None else 0, + "remark": row.remark or "", + } + ) + return out + + +def ensure_material_id( + db: Session, + *, + tenant_no: str, + user_id: Optional[str], + material_id: Optional[str], + material_code: str, + material_name: str, + spec_model: Optional[str], + unit: Optional[str], +) -> str: + mid = (material_id or "").strip() + if mid: + mat = mat_svc.get_material(db, mid, tenant_no=tenant_no) + if not mat: + raise ValueError("关联物资不存在") + if not _is_enabled_status(mat.status): + raise ValueError(f"物资「{mat.material_name or mid}」已停用") + return mat.id + + code = _require_str(material_code, "物料编码") + name = _require_str(material_name, "物料名称") + existing = mat_svc.get_material_by_code(db, code, tenant_no=tenant_no) + if existing: + if not _is_enabled_status(existing.status): + raise ValueError(f"物资「{existing.material_name or code}」已停用") + return existing.id + + created = mat_svc.create_material( + db, + material_code=code, + material_name=name, + spec_model=spec_model, + unit=unit, + stock_qty=Decimal("0"), + status="1", + tenant_no=tenant_no, + user_id=user_id, + commit=False, + ) + return created.id + + +def confirm_month_import( + db: Session, + *, + tenant_no: str, + user_id: Optional[str], + stock_in_date: date, + items: list, + remark: Optional[str] = None, + operate_time: Optional[datetime] = None, +) -> int: + if not items: + raise ValueError("请至少添加一条入库明细") + + stock_in_items: List[Dict[str, Any]] = [] + for it in items: + if not isinstance(it, dict): + raise ValueError("入库明细格式无效") + mid = ensure_material_id( + db, + tenant_no=tenant_no, + user_id=user_id, + material_id=it.get("material_id") or it.get("materialId"), + material_code=it.get("material_code") or it.get("materialCode") or "", + material_name=it.get("material_name") or it.get("materialName") or "", + spec_model=it.get("spec_model") or it.get("specModel"), + unit=it.get("unit"), + ) + stock_in_items.append( + { + "material_id": mid, + "materialId": mid, + "stock_in_no": it.get("stock_in_no") or it.get("stockInNo"), + "stockInNo": it.get("stock_in_no") or it.get("stockInNo"), + "stock_in_date": _fmt_date(stock_in_date), + "stockInDate": _fmt_date(stock_in_date), + "expire_date": it.get("expire_date") or it.get("expireDate"), + "expireDate": it.get("expire_date") or it.get("expireDate"), + "catalog_no": it.get("catalog_no") or it.get("catalogNo"), + "catalogNo": it.get("catalog_no") or it.get("catalogNo"), + "batch_no": it.get("batch_no") or it.get("batchNo"), + "batchNo": it.get("batch_no") or it.get("batchNo"), + "manufacturer": it.get("manufacturer"), + "unit_price": it.get("unit_price") if it.get("unit_price") is not None else it.get("unitPrice"), + "unitPrice": it.get("unit_price") if it.get("unit_price") is not None else it.get("unitPrice"), + "amount": it.get("amount"), + "qty": it.get("qty"), + "remark": it.get("remark"), + } + ) + + return batch_svc.batch_stock_in( + db, + items=stock_in_items, + remark=remark, + tenant_no=tenant_no, + user_id=user_id, + operate_time=operate_time, + ) diff --git a/backend/tests/test_qc_material_month_template.py b/backend/tests/test_qc_material_month_template.py new file mode 100644 index 00000000..0d013447 --- /dev/null +++ b/backend/tests/test_qc_material_month_template.py @@ -0,0 +1,49 @@ +"""质检物料月初模板：字段校验纯函数单测。""" +import unittest +from datetime import date +from decimal import Decimal + +from app.crud.qc_material_month_template import _require_template_fields + + +class TestRequireTemplateFields(unittest.TestCase): + def _base(self, **overrides): + data = { + "materialCode": "QC-M-001", + "materialName": "试剂A", + "catalogNo": "CAT-1", + "batchNo": "B-1", + "manufacturer": "厂家X", + "unitPrice": 10.5, + "amount": 105, + "inQty": 10, + "expireDate": "2027-01-01", + } + data.update(overrides) + return data + + def test_ok(self): + got = _require_template_fields(self._base()) + self.assertEqual(got["material_code"], "QC-M-001") + self.assertEqual(got["material_name"], "试剂A") + self.assertEqual(got["catalog_no"], "CAT-1") + self.assertEqual(got["in_qty"], Decimal("10")) + self.assertEqual(got["expire_date"], date(2027, 1, 1)) + + def test_missing_catalog(self): + with self.assertRaises(ValueError) as ctx: + _require_template_fields(self._base(catalogNo="")) + self.assertIn("货号", str(ctx.exception)) + + def test_qty_not_positive(self): + with self.assertRaises(ValueError) as ctx: + _require_template_fields(self._base(inQty=0)) + self.assertIn("入库数量", str(ctx.exception)) + + def test_missing_code_name(self): + with self.assertRaises(ValueError) as ctx: + _require_template_fields(self._base(materialCode="")) + self.assertIn("物料编码", str(ctx.exception)) + with self.assertRaises(ValueError) as ctx2: + _require_template_fields(self._base(materialName="")) + self.assertIn("物料名称", str(ctx2.exception))