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

"""运输计划按段归档（装货/送货段）"""
from __future__ import annotations

from collections import defaultdict
from decimal import Decimal
from typing import Optional

from fastapi import HTTPException
from sqlalchemy.orm import Session

from app.models.pl_models import (
    PlPlan,
    PlPlanDelivery,
    PlPlanDeliveryProduct,
    PlPlanLoad,
    PlPlanLoadProduct,
    PlPlanProduct,
)
from app.schemas.pl.plan import (
    PlanArchiveDeliveryBlock,
    PlanArchiveLoadBlock,
    PlanArchiveStopProductItem,
)
from app.services.pl_brand_archive import persist_stop_product_brand_qtys
from app.services.pl_internal_customer import is_internal_customer_name


def _require_qty(value: Optional[Decimal], label: str) -> Decimal:
    if value is None:
        raise HTTPException(status_code=400, detail=f"{label}不能为空")
    qty = Decimal(str(value))
    if qty < 0:
        raise HTTPException(status_code=400, detail=f"{label}不能为负数")
    return qty


def _apply_stop_product_archive(
    row: PlPlanLoadProduct | PlPlanDeliveryProduct,
    item: PlanArchiveStopProductItem,
    *,
    mode: str,
    label: str,
    user_id: str,
    strict: bool = True,
) -> None:
    if mode == "inbound":
        if item.inbound_qty is None:
            if strict:
                raise HTTPException(status_code=400, detail=f"{label}入库数量不能为空")
            row.inbound_qty = None
            row.inbound_amount = None
            if item.inbound_price is not None:
                row.inbound_price = Decimal(str(item.inbound_price))
        else:
            qty = _require_qty(item.inbound_qty, f"{label}入库数量")
            row.inbound_qty = qty
            if item.inbound_price is not None:
                row.inbound_price = Decimal(str(item.inbound_price))
            price = row.inbound_price
            row.inbound_amount = (qty * Decimal(str(price))) if price is not None else None
    else:
        if item.outbound_qty is None:
            if strict:
                raise HTTPException(status_code=400, detail=f"{label}出库数量不能为空")
            row.outbound_qty = None
            row.outbound_amount = None
            if item.outbound_price is not None:
                row.outbound_price = Decimal(str(item.outbound_price))
        else:
            qty = _require_qty(item.outbound_qty, f"{label}出库数量")
            row.outbound_qty = qty
            if item.outbound_price is not None:
                row.outbound_price = Decimal(str(item.outbound_price))
            price = row.outbound_price
            row.outbound_amount = (qty * Decimal(str(price))) if price is not None else None
    row.update_by = user_id


def _brand_rows_from_item(item: PlanArchiveStopProductItem) -> list[dict]:
    if not item.brand_qtys:
        return []
    rows: list[dict] = []
    for bq in item.brand_qtys:
        row: dict = {
            "brandId": bq.brand_id or "",
            "brandName": bq.brand_name,
            "qty": bq.qty,
        }
        if bq.unit_price is not None:
            row["unitPrice"] = bq.unit_price
        rows.append(row)
    return rows


def _persist_stop_brand_qtys(
    db: Session,
    *,
    tenant_no: str,
    plan: PlPlan,
    stop_type: str,
    stop_row: PlPlanLoadProduct | PlPlanDeliveryProduct,
    item: PlanArchiveStopProductItem,
    user_id: str,
) -> None:
    plan_product = (
        db.query(PlPlanProduct)
        .filter(
            PlPlanProduct.id == stop_row.plan_product_id,
            PlPlanProduct.tenant_no == tenant_no,
            PlPlanProduct.del_flag == 0,
        )
        .first()
    )
    brand_mode = plan_product.brand_mode if plan_product else None
    persist_stop_product_brand_qtys(
        db,
        tenant_no=tenant_no,
        plan_id=plan.id,
        plan_product_id=stop_row.plan_product_id,
        stop_type=stop_type,
        stop_product_id=stop_row.id,
        brand_mode=brand_mode,
        rows=_brand_rows_from_item(item),
        user_id=user_id,
        dept_id=getattr(stop_row, "create_dept_id", "") or "",
    )


def apply_archive_stop_quantities(
    db: Session,
    *,
    tenant_no: str,
    plan: PlPlan,
    load_block: PlanArchiveLoadBlock,
    delivery_blocks: list[PlanArchiveDeliveryBlock],
    user_id: str,
    strict: bool = True,
) -> None:
    load = (
        db.query(PlPlanLoad)
        .filter(
            PlPlanLoad.id == load_block.stop_id,
            PlPlanLoad.plan_id == plan.id,
            PlPlanLoad.tenant_no == tenant_no,
            PlPlanLoad.del_flag == 0,
        )
        .first()
    )
    if not load:
        raise HTTPException(status_code=400, detail="装货段不存在")

    load_products = {
        row.id: row
        for row in db.query(PlPlanLoadProduct).filter(
            PlPlanLoadProduct.load_id == load.id,
            PlPlanLoadProduct.plan_id == plan.id,
            PlPlanLoadProduct.tenant_no == tenant_no,
            PlPlanLoadProduct.del_flag == 0,
        )
    }
    if not load_products:
        raise HTTPException(status_code=400, detail="装货段无产品明细")
    if strict and len(load_block.products) != len(load_products):
        raise HTTPException(status_code=400, detail="装货段归档产品行不完整")
    for item in load_block.products:
        row = load_products.get(item.id)
        if not row:
            raise HTTPException(status_code=400, detail=f"装货段产品不存在: {item.id}")
        _apply_stop_product_archive(
            row, item, mode="inbound", label="装货段", user_id=user_id, strict=strict
        )
        if item.brand_qtys is not None or strict:
            _persist_stop_brand_qtys(
                db,
                tenant_no=tenant_no,
                plan=plan,
                stop_type="load",
                stop_row=row,
                item=item,
                user_id=user_id,
            )

    deliveries = (
        db.query(PlPlanDelivery)
        .filter(
            PlPlanDelivery.plan_id == plan.id,
            PlPlanDelivery.tenant_no == tenant_no,
            PlPlanDelivery.del_flag == 0,
        )
        .order_by(PlPlanDelivery.seq.asc(), PlPlanDelivery.create_time.asc())
        .all()
    )
    delivery_map = {row.id: row for row in deliveries}
    if strict and len(delivery_blocks) != len(deliveries):
        raise HTTPException(status_code=400, detail="请填写全部送货段归档")

    for block in delivery_blocks:
        delivery = delivery_map.get(block.stop_id)
        if not delivery:
            raise HTTPException(status_code=400, detail="送货段不存在")
        delivery_products = {
            row.id: row
            for row in db.query(PlPlanDeliveryProduct).filter(
                PlPlanDeliveryProduct.delivery_id == delivery.id,
                PlPlanDeliveryProduct.plan_id == plan.id,
                PlPlanDeliveryProduct.tenant_no == tenant_no,
                PlPlanDeliveryProduct.del_flag == 0,
            )
        }
        if not delivery_products:
            raise HTTPException(status_code=400, detail=f"送货段「{delivery.customer_name or ''}」无产品明细")
        if strict and len(block.products) != len(delivery_products):
            raise HTTPException(status_code=400, detail=f"送货段「{delivery.customer_name or ''}」归档产品行不完整")
        label = f"送货段「{delivery.customer_name or delivery.seq}」"
        for item in block.products:
            row = delivery_products.get(item.id)
            if not row:
                raise HTTPException(status_code=400, detail=f"{label}产品不存在: {item.id}")
            _apply_stop_product_archive(
                row, item, mode="outbound", label=label, user_id=user_id, strict=strict
            )
            if item.brand_qtys is not None or strict:
                _persist_stop_brand_qtys(
                    db,
                    tenant_no=tenant_no,
                    plan=plan,
                    stop_type="delivery",
                    stop_row=row,
                    item=item,
                    user_id=user_id,
                )


def sync_plan_products_from_stop_archive(
    db: Session,
    *,
    tenant_no: str,
    plan_id: str,
    user_id: str,
) -> None:
    """将段级归档数量汇总回写 pl_plan_product（_row_to_dict 权威来源）。"""
    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,
        )
    }
    inbound_by_plan_product: dict[str, Decimal] = defaultdict(lambda: Decimal("0"))
    outbound_by_plan_product: dict[str, Decimal] = defaultdict(lambda: Decimal("0"))
    inbound_amount_by_plan_product: dict[str, Decimal] = defaultdict(lambda: Decimal("0"))
    outbound_amount_by_plan_product: dict[str, Decimal] = defaultdict(lambda: Decimal("0"))

    load = (
        db.query(PlPlanLoad)
        .filter(
            PlPlanLoad.plan_id == plan_id,
            PlPlanLoad.tenant_no == tenant_no,
            PlPlanLoad.del_flag == 0,
        )
        .first()
    )
    if load:
        load_rows = (
            db.query(PlPlanLoadProduct)
            .filter(
                PlPlanLoadProduct.load_id == load.id,
                PlPlanLoadProduct.plan_id == plan_id,
                PlPlanLoadProduct.tenant_no == tenant_no,
                PlPlanLoadProduct.del_flag == 0,
            )
            .all()
        )
        for row in load_rows:
            if row.plan_product_id not in plan_products:
                continue
            qty = Decimal(str(row.inbound_qty or 0))
            inbound_by_plan_product[row.plan_product_id] += qty
            inbound_amount_by_plan_product[row.plan_product_id] += Decimal(str(row.inbound_amount or 0))

    delivery_rows = (
        db.query(PlPlanDeliveryProduct)
        .filter(
            PlPlanDeliveryProduct.plan_id == plan_id,
            PlPlanDeliveryProduct.tenant_no == tenant_no,
            PlPlanDeliveryProduct.del_flag == 0,
        )
        .all()
    )
    for row in delivery_rows:
        if row.plan_product_id not in plan_products:
            continue
        qty = Decimal(str(row.outbound_qty or 0))
        outbound_by_plan_product[row.plan_product_id] += qty
        outbound_amount_by_plan_product[row.plan_product_id] += Decimal(str(row.outbound_amount or 0))

    for plan_product_id, prod in plan_products.items():
        if plan_product_id in inbound_by_plan_product:
            prod.inbound_qty = inbound_by_plan_product[plan_product_id]
            if inbound_amount_by_plan_product[plan_product_id] > 0:
                prod.inbound_amount = inbound_amount_by_plan_product[plan_product_id]
        if plan_product_id in outbound_by_plan_product:
            prod.outbound_qty = outbound_by_plan_product[plan_product_id]
            if outbound_amount_by_plan_product[plan_product_id] > 0:
                prod.outbound_amount = outbound_amount_by_plan_product[plan_product_id]
        outbound_qty = outbound_by_plan_product.get(plan_product_id, Decimal("0"))
        if outbound_qty > 0:
            prod.actual_qty = outbound_qty
        prod.update_by = user_id


def sum_delivery_outbound_qty(db: Session, *, tenant_no: str, plan_id: str) -> Decimal:
    total = Decimal("0")
    rows = (
        db.query(PlPlanDeliveryProduct)
        .filter(
            PlPlanDeliveryProduct.plan_id == plan_id,
            PlPlanDeliveryProduct.tenant_no == tenant_no,
            PlPlanDeliveryProduct.del_flag == 0,
        )
        .all()
    )
    for row in rows:
        total += Decimal(str(row.outbound_qty or 0))
    return total


def iter_stop_archive_inventory_lines(
    db: Session,
    *,
    tenant_no: str,
    plan_id: str,
):
    """产出内部段库存变动：装货出库、送货入库。"""
    load = (
        db.query(PlPlanLoad)
        .filter(
            PlPlanLoad.plan_id == plan_id,
            PlPlanLoad.tenant_no == tenant_no,
            PlPlanLoad.del_flag == 0,
        )
        .first()
    )
    if load and is_internal_customer_name(load.customer_name):
        rows = (
            db.query(PlPlanLoadProduct)
            .filter(
                PlPlanLoadProduct.load_id == load.id,
                PlPlanLoadProduct.plan_id == plan_id,
                PlPlanLoadProduct.tenant_no == tenant_no,
                PlPlanLoadProduct.del_flag == 0,
            )
            .all()
        )
        for row in rows:
            qty = Decimal(str(row.inbound_qty or 0))
            if qty > 0:
                yield row.plan_product_id, qty, "outbound", row.inbound_price

    deliveries = (
        db.query(PlPlanDelivery)
        .filter(
            PlPlanDelivery.plan_id == plan_id,
            PlPlanDelivery.tenant_no == tenant_no,
            PlPlanDelivery.del_flag == 0,
        )
        .all()
    )
    delivery_ids = [row.id for row in deliveries]
    internal_delivery_ids = {
        row.id for row in deliveries if is_internal_customer_name(row.customer_name)
    }
    if not internal_delivery_ids:
        return
    rows = (
        db.query(PlPlanDeliveryProduct)
        .filter(
            PlPlanDeliveryProduct.delivery_id.in_(list(internal_delivery_ids)),
            PlPlanDeliveryProduct.plan_id == plan_id,
            PlPlanDeliveryProduct.tenant_no == tenant_no,
            PlPlanDeliveryProduct.del_flag == 0,
        )
        .all()
    )
    for row in rows:
        qty = Decimal(str(row.outbound_qty or 0))
        if qty > 0:
            yield row.plan_product_id, qty, "inbound", row.outbound_price