shanxia2.0/backend/app/services/pl_feedback_brand_qty.py

"""现场入出库明细与运输回传的品牌数量校验与持久化"""
from decimal import Decimal
from typing import Any, Optional, Type

from sqlalchemy.orm import Session

from app.core.utils import generate_uuid
from app.services.pl_brand_archive import validate_brand_qty_rows


def format_brand_qty_summary(rows: list[dict]) -> Optional[str]:
    parts = []
    for r in rows:
        qty = Decimal(str(r.get("qty") or 0))
        if qty <= 0:
            continue
        name = (r.get("brandName") or r.get("brand_name") or "").strip() or "未命名品牌"
        qty_text = format(qty.normalize(), "f")
        parts.append(f"{name} {qty_text}")
    return " / ".join(parts) if parts else None


def require_brand_qtys_if_needed(
    *,
    brand_mode: Optional[str],
    required_brand_id: Optional[str],
    actual_qty: Decimal,
    rows: list[dict],
    enforce: bool,
) -> None:
    if not brand_mode:
        return
    if not enforce:
        return
    positive = [r for r in rows if Decimal(str(r.get("qty") or 0)) > 0]
    if not positive:
        raise ValueError("有品牌产品须填写至少一个品牌数量")
    validate_brand_qty_rows(
        brand_mode=brand_mode,
        required_brand_id=required_brand_id,
        actual_qty=Decimal(str(actual_qty)),
        rows=rows,
    )


def persist_io_item_brand_qtys(
    db: Session,
    *,
    model: Type[Any],
    tenant_no: str,
    order_id: str,
    order_item_id: str,
    plan_product_id: str,
    brand_mode: Optional[str],
    rows: list[dict],
    user_id: Optional[str] = None,
) -> None:
    from app.services.pl_brand_archive import prepare_brand_qty_rows
    from app.services.pl_child_sync import apply_keyed_rows
    if rows is None:
        return
    existing = db.query(model).filter(model.order_item_id == order_item_id,
        model.tenant_no == tenant_no, model.del_flag == 0).all()
    prepared = prepare_brand_qty_rows(existing, rows,
        dict(tenant_no=tenant_no, order_id=order_id, order_item_id=order_item_id,
             plan_product_id=plan_product_id))
    apply_keyed_rows(db, model, prepared, user_id=user_id)


def _dict_field(item: dict, *keys: str):
    for key in keys:
        if key in item and item[key] is not None:
            return item[key]
    return None


def persist_transport_product_feedback(
    db: Session,
    *,
    tenant_no: str,
    transport,
    plan_id: str,
    products: list[dict],
    user_id: Optional[str] = None,
    enforce_brand: bool = True,
) -> None:
    from fastapi import HTTPException
    from app.models.pl_models import PlPlanProduct, PlTransportProductBrandQty, PlTransportProductFeedback
    from app.services.pl_brand_archive import prepare_brand_qty_rows
    from app.services.pl_child_sync import apply_keyed_rows, prepare_keyed_rows
    if products is None:
        return
    if transport.tenant_no != tenant_no:
        raise HTTPException(422, '运输单不属于当前租户')
    transport_id = transport.id
    existing = db.query(PlTransportProductFeedback).filter(
        PlTransportProductFeedback.transport_id == transport_id,
        PlTransportProductFeedback.plan_id == plan_id,
        PlTransportProductFeedback.tenant_no == tenant_no,
        PlTransportProductFeedback.del_flag == 0).all()
    brands = db.query(PlTransportProductBrandQty).filter(
        PlTransportProductBrandQty.transport_id == transport_id,
        PlTransportProductBrandQty.tenant_no == tenant_no,
        PlTransportProductBrandQty.del_flag == 0).all()
    plan_products = {row.id: row for row in db.query(PlPlanProduct).filter(
        PlPlanProduct.plan_id == plan_id, PlPlanProduct.tenant_no == tenant_no,
        PlPlanProduct.del_flag == 0).all()}
    values, brand_plans = [], {}
    total_qty = Decimal('0')
    for item in products:
        pid = _dict_field(item, 'planProductId', 'plan_product_id')
        product = plan_products.get(pid)
        if product is None:
            raise HTTPException(422, '回传产品不存在、已删除或不属于当前计划')
        qty = Decimal(str(_dict_field(item, 'actualQty', 'actual_qty') or 0))
        if not qty.is_finite() or qty < 0:
            raise HTTPException(422, '回传数量须为非负有限数')
        total_qty += qty
        value = dict(tenant_no=tenant_no, transport_id=transport_id, plan_id=plan_id,
                     plan_product_id=pid, actual_qty=qty)
        if item.get('id'):
            value['id'] = item['id']
        values.append(value)
        brand_rows = _dict_field(item, 'brandQtys', 'brand_qtys')
        if brand_rows is not None:
            require_brand_qtys_if_needed(brand_mode=product.brand_mode, required_brand_id=product.brand_id,
                actual_qty=qty, rows=brand_rows, enforce=enforce_brand)
            brand_plans[pid] = prepare_brand_qty_rows(
                [row for row in brands if row.plan_product_id == pid], brand_rows,
                dict(tenant_no=tenant_no, transport_id=transport_id, plan_product_id=pid))
    prepared = prepare_keyed_rows(existing, values, ('plan_product_id',), '运输产品回传')
    feedbacks = apply_keyed_rows(db, PlTransportProductFeedback, prepared, user_id=user_id)
    for feedback in feedbacks:
        brand_plan = brand_plans.get(feedback.plan_product_id)
        if brand_plan is not None:
            for _, fields in brand_plan[0]:
                fields['feedback_id'] = feedback.id
            apply_keyed_rows(db, PlTransportProductBrandQty, brand_plan, user_id=user_id)
    removed_feedback_ids = {row.id for row in prepared[1]}
    for row in brands:
        if row.feedback_id in removed_feedback_ids:
            row.del_flag = 1
            row.update_by = user_id
    transport.feedback_qty = total_qty


def load_transport_driver_product_feedbacks(
    db: Session,
    *,
    tenant_no: str,
    transport_id: str,
) -> list[dict]:
    """管理端读运输单时附司机按产品回传及品牌摘要。"""
    from app.models.pl_models import (
        PlPlanProduct,
        PlTransportProductBrandQty,
        PlTransportProductFeedback,
    )

    feedback_rows = (
        db.query(PlTransportProductFeedback)
        .filter(
            PlTransportProductFeedback.transport_id == transport_id,
            PlTransportProductFeedback.tenant_no == tenant_no,
            PlTransportProductFeedback.del_flag == 0,
        )
        .order_by(PlTransportProductFeedback.order_num.asc())
        .all()
    )
    if not feedback_rows:
        return []

    plan_product_ids = [fb.plan_product_id for fb in feedback_rows if fb.plan_product_id]
    products_by_id: dict[str, PlPlanProduct] = {}
    if plan_product_ids:
        product_rows = (
            db.query(PlPlanProduct)
            .filter(
                PlPlanProduct.id.in_(plan_product_ids),
                PlPlanProduct.tenant_no == tenant_no,
                PlPlanProduct.del_flag == 0,
            )
            .all()
        )
        products_by_id = {row.id: row for row in product_rows}

    brand_qtys_by_plan_product: dict[str, list[dict]] = {}
    feedback_ids = [fb.id for fb in feedback_rows]
    brand_qty_rows = (
        db.query(PlTransportProductBrandQty)
        .filter(
            PlTransportProductBrandQty.transport_id == transport_id,
            PlTransportProductBrandQty.tenant_no == tenant_no,
            PlTransportProductBrandQty.del_flag == 0,
            PlTransportProductBrandQty.feedback_id.in_(feedback_ids),
        )
        .order_by(PlTransportProductBrandQty.order_num.asc())
        .all()
    )
    for row in brand_qty_rows:
        brand_qtys_by_plan_product.setdefault(row.plan_product_id, []).append(
            {
                "brandId": row.brand_id or "",
                "brandName": row.brand_name,
                "qty": row.qty,
            }
        )

    result: list[dict] = []
    for fb in feedback_rows:
        brand_qtys = brand_qtys_by_plan_product.get(fb.plan_product_id, [])
        pp = products_by_id.get(fb.plan_product_id or "")
        result.append(
            {
                "planProductId": fb.plan_product_id,
                "productName": (pp.product_name if pp else None) or "",
                "productNo": (pp.product_no if pp else None) or None,
                "model": (pp.model if pp else None) or None,
                "unit": (pp.unit if pp else None) or None,
                "packageSpec": (pp.package_spec if pp else None) or None,
                "brandMode": (pp.brand_mode if pp else None) or None,
                "brandId": (pp.brand_id if pp else None) or None,
                "brandName": (pp.brand_name if pp else None) or None,
                "actualQty": fb.actual_qty,
                "brandQtys": brand_qtys or None,
                "brandSummary": format_brand_qty_summary(brand_qtys),
            }
        )
    return result