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

"""计划归档按产品+品牌写入库存台账与流水。

直销单 (direct) 首期不记账；入库增加、出库扣减；出库不足硬拦截。
调整归档时先按 plan_id 冲销旧流水再按新量重记。
"""
from decimal import Decimal
from typing import Optional

from fastapi import HTTPException
from sqlalchemy.orm import Session

from app.core.utils import generate_uuid
from app.models.pl_models import (
    PlInventory,
    PlInventoryTransaction,
    PlPlan,
    PlPlanProduct,
    PlPlanProductBrandQty,
)

EMPTY_BRAND_ID = ""


def qty_delta_for_plan(plan_type: str, qty: Decimal) -> Decimal:
    if plan_type == "inbound":
        return Decimal(str(qty))
    if plan_type == "outbound":
        return -Decimal(str(qty))
    return Decimal("0")  # direct: 首期不记账


def assert_outbound_enough(*, stock_qty: Decimal, outbound_qty: Decimal, brand_label: str) -> None:
    if Decimal(str(outbound_qty)) > Decimal(str(stock_qty)):
        raise ValueError(f"{brand_label}库存不足")


def get_or_create_inventory(
    db: Session,
    *,
    tenant_no: str,
    product_id: str,
    product_no: str,
    product_name: str,
    brand_id: Optional[str],
    brand_name: Optional[str],
    unit: Optional[str],
    model: Optional[str],
    user_id: str,
    dept_id: str,
) -> PlInventory:
    brand_id = brand_id or EMPTY_BRAND_ID
    row = (
        db.query(PlInventory)
        .filter(
            PlInventory.tenant_no == tenant_no,
            PlInventory.product_id == product_id,
            PlInventory.brand_id == brand_id,
            PlInventory.del_flag == 0,
        )
        .first()
    )
    if row:
        return row
    row = PlInventory(
        id=generate_uuid(),
        tenant_no=tenant_no,
        product_id=product_id,
        brand_id=brand_id,
        brand_name=brand_name,
        product_no=product_no or "",
        product_name=product_name or "",
        model=model,
        unit=unit or "吨",
        stock_qty=Decimal("0"),
        stock_amount=Decimal("0"),
        create_by=user_id,
        create_dept_id=dept_id or "",
        update_by=user_id,
    )
    db.add(row)
    db.flush()
    return row


def apply_brand_stock_change(
    db: Session,
    *,
    tenant_no: str,
    plan: PlPlan,
    plan_product: PlPlanProduct,
    brand_id: Optional[str],
    brand_name: Optional[str],
    qty: Decimal,
    user_id: str,
    dept_id: str,
) -> None:
    delta = qty_delta_for_plan(plan.plan_type, Decimal(str(qty)))
    if delta == 0:
        return
    inv = get_or_create_inventory(
        db,
        tenant_no=tenant_no,
        product_id=plan_product.product_id,
        product_no=plan_product.product_no,
        product_name=plan_product.product_name,
        brand_id=brand_id or EMPTY_BRAND_ID,
        brand_name=brand_name,
        unit=plan_product.unit,
        model=plan_product.model,
        user_id=user_id,
        dept_id=dept_id,
    )
    if delta < 0:
        try:
            assert_outbound_enough(
                stock_qty=Decimal(str(inv.stock_qty or 0)),
                outbound_qty=-delta,
                brand_label=brand_name or plan_product.product_name or "",
            )
        except ValueError as exc:
            raise HTTPException(status_code=400, detail=str(exc)) from exc
    unit_price = (
        plan_product.inbound_price
        or plan_product.outbound_price
        or plan_product.tax_price
        or Decimal("0")
    )
    unit_price = Decimal(str(unit_price))
    inv.stock_qty = Decimal(str(inv.stock_qty or 0)) + delta
    inv.stock_amount = Decimal(str(inv.stock_qty or 0)) * unit_price
    inv.update_by = user_id
    db.add(
        PlInventoryTransaction(
            id=generate_uuid(),
            tenant_no=tenant_no,
            product_id=plan_product.product_id,
            brand_id=brand_id or EMPTY_BRAND_ID,
            brand_name=brand_name,
            product_name=plan_product.product_name or "",
            transaction_type="inbound" if delta > 0 else "outbound",
            plan_id=plan.id,
            plan_no=plan.plan_no,
            qty=delta,
            unit_price=unit_price,
            amount=abs(delta) * unit_price,
            stock_after=inv.stock_qty,
            create_by=user_id,
            create_dept_id=dept_id or "",
            update_by=user_id,
        )
    )


def reverse_plan_inventory_postings(
    db: Session,
    *,
    tenant_no: str,
    plan_id: str,
    user_id: str,
) -> None:
    """冲销该计划已产生的库存流水（调整归档前调用）。"""
    txns = (
        db.query(PlInventoryTransaction)
        .filter(
            PlInventoryTransaction.tenant_no == tenant_no,
            PlInventoryTransaction.plan_id == plan_id,
            PlInventoryTransaction.del_flag == 0,
        )
        .order_by(PlInventoryTransaction.create_time.asc())
        .all()
    )
    for txn in txns:
        brand_id = txn.brand_id or EMPTY_BRAND_ID
        inv = (
            db.query(PlInventory)
            .filter(
                PlInventory.tenant_no == tenant_no,
                PlInventory.product_id == txn.product_id,
                PlInventory.brand_id == brand_id,
                PlInventory.del_flag == 0,
            )
            .first()
        )
        if inv:
            inv.stock_qty = Decimal(str(inv.stock_qty or 0)) - Decimal(str(txn.qty or 0))
            unit_price = Decimal(str(txn.unit_price or 0))
            inv.stock_amount = Decimal(str(inv.stock_qty or 0)) * unit_price
            inv.update_by = user_id
        txn.del_flag = 1
        txn.update_by = user_id


def apply_archive_inventory_for_plan(
    db: Session,
    *,
    tenant_no: str,
    plan: PlPlan,
    user_id: str,
    dept_id: str = "",
) -> None:
    """归档后按品牌量（或无品牌 actual_qty）落账；先冲销旧流水再重记。"""
    reverse_plan_inventory_postings(
        db, tenant_no=tenant_no, plan_id=plan.id, user_id=user_id
    )
    if qty_delta_for_plan(plan.plan_type, Decimal("1")) == 0:
        return

    products = (
        db.query(PlPlanProduct)
        .filter(
            PlPlanProduct.plan_id == plan.id,
            PlPlanProduct.tenant_no == tenant_no,
            PlPlanProduct.del_flag == 0,
        )
        .order_by(PlPlanProduct.order_num.asc())
        .all()
    )
    for prod in products:
        brand_rows = (
            db.query(PlPlanProductBrandQty)
            .filter(
                PlPlanProductBrandQty.plan_product_id == prod.id,
                PlPlanProductBrandQty.tenant_no == tenant_no,
                PlPlanProductBrandQty.del_flag == 0,
            )
            .order_by(PlPlanProductBrandQty.order_num.asc())
            .all()
        )
        if prod.brand_mode and brand_rows:
            for bq in brand_rows:
                apply_brand_stock_change(
                    db,
                    tenant_no=tenant_no,
                    plan=plan,
                    plan_product=prod,
                    brand_id=bq.brand_id or EMPTY_BRAND_ID,
                    brand_name=bq.brand_name,
                    qty=Decimal(str(bq.qty or 0)),
                    user_id=user_id,
                    dept_id=dept_id,
                )
            continue
        actual = Decimal(str(prod.actual_qty or 0))
        if actual <= 0:
            continue
        apply_brand_stock_change(
            db,
            tenant_no=tenant_no,
            plan=plan,
            plan_product=prod,
            brand_id=EMPTY_BRAND_ID,
            brand_name=None,
            qty=actual,
            user_id=user_id,
            dept_id=dept_id,
        )