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

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

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

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 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,
    qty_delta_override: Optional[Decimal] = None,
    unit_price_override: Optional[Decimal] = None,
) -> None:
    delta = (
        qty_delta_override
        if qty_delta_override is not None
        else 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,
    )
    unit_price = unit_price_override
    if unit_price is None:
        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,
        )


def apply_archive_inventory_for_plan_stops(
    db: Session,
    *,
    tenant_no: str,
    plan: PlPlan,
    user_id: str,
    dept_id: str = "",
) -> None:
    """一车多卸：内部装货出库、内部送货入库。"""
    from app.services.pl_plan_archive_stop_service import iter_stop_archive_inventory_lines

    reverse_plan_inventory_postings(
        db, tenant_no=tenant_no, plan_id=plan.id, user_id=user_id
    )
    product_map = {
        row.id: row
        for row in db.query(PlPlanProduct).filter(
            PlPlanProduct.plan_id == plan.id,
            PlPlanProduct.tenant_no == tenant_no,
            PlPlanProduct.del_flag == 0,
        )
    }
    for plan_product_id, qty, direction, price in iter_stop_archive_inventory_lines(
        db, tenant_no=tenant_no, plan_id=plan.id
    ):
        prod = product_map.get(plan_product_id)
        if not prod:
            continue
        delta = -qty if direction == "outbound" else qty
        apply_brand_stock_change(
            db,
            tenant_no=tenant_no,
            plan=plan,
            plan_product=prod,
            brand_id=EMPTY_BRAND_ID,
            brand_name=None,
            qty=qty,
            user_id=user_id,
            dept_id=dept_id,
            qty_delta_override=delta,
            unit_price_override=Decimal(str(price)) if price is not None else None,
        )