zkxw/backend/app/crud/qc_material_month_template.py

"""质检物料月初入库模板 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,
    )