feat(backend): add qc material month template CRUD preview and confirm
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
zlwuandCursor committed
1 parent
1d3f73d3b9
commit
3aa8aebc29
2 files changed
+413
No files matched your search
@@ -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,
|
||||
)
|
||||
@@ -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))
|
||||
Reference in new issue
Block a user