380 lines
13 KiB
Python
380 lines
13 KiB
Python
"""质检物料月初入库模板 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_str
|
|
from app.crud.qc_material_stock_in_no import allocate_stock_in_nos
|
|
from app.models.qc_material import BizQcMaterialMonthTemplate
|
|
|
|
|
|
def _optional_str(v: Any) -> str:
|
|
return str(v or "").strip()
|
|
|
|
|
|
def _optional_decimal(v: Any, *, default: Decimal = Decimal("0")) -> Decimal:
|
|
if v is None or str(v).strip() == "":
|
|
return default
|
|
try:
|
|
d = Decimal(str(v))
|
|
except Exception as e:
|
|
raise ValueError("数值格式无效") from e
|
|
if d < 0:
|
|
raise ValueError("数值不能为负数")
|
|
return d
|
|
|
|
|
|
def _optional_expire_date(v: Any) -> Optional[date]:
|
|
s = str(v or "").strip()
|
|
if not s:
|
|
return None
|
|
return _parse_date(v, label="过期日期")
|
|
|
|
|
|
def _require_template_fields(payload: dict) -> dict:
|
|
"""规范化模板写入字段；模板行字段均可空，确认入库时再校验必填。"""
|
|
if not isinstance(payload, dict):
|
|
raise ValueError("模板数据无效")
|
|
|
|
material_id = (payload.get("material_id") or payload.get("materialId") or "").strip() or None
|
|
material_code = _optional_str(payload.get("material_code") or payload.get("materialCode"))
|
|
material_name = _optional_str(payload.get("material_name") or payload.get("materialName"))
|
|
catalog_no = _optional_str(payload.get("catalog_no") or payload.get("catalogNo"))
|
|
batch_no = _optional_str(payload.get("batch_no") or payload.get("batchNo"))
|
|
manufacturer = _optional_str(payload.get("manufacturer"))
|
|
unit_price = _optional_decimal(payload.get("unit_price") or payload.get("unitPrice"))
|
|
amount = _optional_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 = _optional_decimal(in_qty_raw)
|
|
expire_date = _optional_expire_date(payload.get("expire_date") or payload.get("expireDate"))
|
|
|
|
spec_model = _optional_str(payload.get("spec_model") or payload.get("specModel")) or None
|
|
unit = _optional_str(payload.get("unit")) or None
|
|
remark = _optional_str(payload.get("remark")) 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,
|
|
)
|