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

"""运输计划是否已经开始，以及强制作废时冲销下游单据。

已开始：已过磅、地磅出入库已执行、现场已回传，或入库/出库实际数量被改过。
未开始的计划仍可调整和普通作废。已开始的计划普通用户不能再调整或作废。
"""
from __future__ import annotations

from dataclasses import dataclass, field
from datetime import datetime
from typing import Optional

from sqlalchemy import or_
from sqlalchemy.orm import Session

from app.models.db_models import DbInboundOutbound, DbPurchasePlan, DbWeighingSlip
from app.models.pl_models import (
    PlInboundOrder,
    PlInboundOrderItem,
    PlOutboundOrder,
    PlOutboundOrderItem,
    PlTransport,
    PlTransportDbLink,
    PlTransportSiteLink,
    PlTransportSiteStopFeedback,
)
from app.services.pl_dibang_bridge_service import (
    _REVOKABLE_INVENTORY_STATUSES,
    _subtract_plan_contribution,
)
from app.services.pl_io_order_service import void_io_orders_for_plan

LOCK_WEIGHED = "已过磅"
LOCK_INVENTORY = "地磅出入库已执行"
LOCK_SITE = "现场已回传"
LOCK_MANUAL = "入库或出库实际数量已修改"


def inventory_status_is_executed(status: Optional[str]) -> bool:
    text = (status or "").strip()
    if not text:
        return False
    return text not in _REVOKABLE_INVENTORY_STATUSES


def pick_execution_lock_reason(
    *,
    weighed: bool,
    inventory_executed: bool,
    site_feedback: bool,
    manual_qty: bool,
) -> Optional[str]:
    if weighed:
        return LOCK_WEIGHED
    if inventory_executed:
        return LOCK_INVENTORY
    if site_feedback:
        return LOCK_SITE
    if manual_qty:
        return LOCK_MANUAL
    return None


def execution_lock_message(reason: str, action: str) -> str:
    if action == "arrange":
        return f"该计划{reason}，不能再调整，请联系管理员处理"
    return f"该计划{reason}，不能作废，请联系管理员处理"


@dataclass
class _Signals:
    weighed: set[str] = field(default_factory=set)
    inventory_executed: set[str] = field(default_factory=set)
    site_feedback: set[str] = field(default_factory=set)
    manual_qty: set[str] = field(default_factory=set)


def load_plan_execution_locks(
    db: Session, tenant_no: str, plan_ids: list[str]
) -> dict[str, Optional[str]]:
    ids = [item for item in dict.fromkeys(plan_ids) if item]
    result: dict[str, Optional[str]] = {item: None for item in ids}
    if not ids:
        return result
    signals = _collect_signals(db, tenant_no, ids)
    for plan_id in ids:
        result[plan_id] = pick_execution_lock_reason(
            weighed=plan_id in signals.weighed,
            inventory_executed=plan_id in signals.inventory_executed,
            site_feedback=plan_id in signals.site_feedback,
            manual_qty=plan_id in signals.manual_qty,
        )
    return result


def plan_execution_lock_reason(
    db: Session, tenant_no: str, plan_id: str
) -> Optional[str]:
    return load_plan_execution_locks(db, tenant_no, [plan_id]).get(plan_id)


@dataclass
class ForceVoidImpact:
    weighing_slips: list[str] = field(default_factory=list)
    inventories: list[str] = field(default_factory=list)
    io_orders: list[str] = field(default_factory=list)

    def notice(self) -> str:
        lines = [
            "强制作废会把这趟已开始的记录全部冲销：地磅单作废，地磅月计划占用量退回，"
            "入库单和出库单作废，合同占用量退回。单据和回传仍保留可查。"
        ]
        if self.weighing_slips:
            lines.append("地磅单：" + "、".join(self.weighing_slips))
        if self.inventories:
            lines.append("地磅出入库：" + "、".join(self.inventories))
        if self.io_orders:
            lines.append("入库/出库单：" + "、".join(self.io_orders))
        lines.append("请填写作废原因。此操作不能恢复。")
        return "\n".join(lines)


def collect_force_void_impact(
    db: Session, *, tenant_no: str, plan_id: str
) -> ForceVoidImpact:
    impact = ForceVoidImpact()
    inventories = _live_inventories(db, tenant_no, plan_id)
    inv_ids = [row.id for row in inventories]
    impact.inventories = [_doc_label(row.batch_no, row.id) for row in inventories]
    if inv_ids:
        slips = (
            db.query(DbWeighingSlip)
            .filter(
                DbWeighingSlip.tenant_no == tenant_no,
                DbWeighingSlip.del_flag == 0,
                DbWeighingSlip.is_voided == 0,
                DbWeighingSlip.inbound_outbound_id.in_(inv_ids),
            )
            .all()
        )
        impact.weighing_slips = [_doc_label(row.batch_no, row.id) for row in slips]
    for model in (PlInboundOrder, PlOutboundOrder):
        rows = (
            db.query(model)
            .filter(
                model.tenant_no == tenant_no,
                model.del_flag == 0,
                model.plan_id == plan_id,
                model.status != "voided",
            )
            .all()
        )
        impact.io_orders.extend(_doc_label(row.order_no, row.id) for row in rows)
    return impact


def apply_force_void_downstream(
    db: Session,
    *,
    tenant_no: str,
    plan_id: str,
    transport_id: Optional[str],
    user_id: str,
    reason: str,
) -> None:
    """冲销已开始的下游：作废地磅单、退回月计划量、作废出入库单。不提交事务。"""
    now = datetime.now()
    links = _live_links(db, tenant_no, plan_id)
    inventories = _live_inventories(db, tenant_no, plan_id)
    inv_by_id = {row.id: row for row in inventories}
    purchase_plan_ids: set[str] = set()
    inv_ids = list(inv_by_id)
    slips = []
    if inv_ids:
        slips = (
            db.query(DbWeighingSlip)
            .filter(
                DbWeighingSlip.tenant_no == tenant_no,
                DbWeighingSlip.del_flag == 0,
                DbWeighingSlip.is_voided == 0,
                DbWeighingSlip.inbound_outbound_id.in_(inv_ids),
            )
            .all()
        )
    for slip in slips:
        if slip.purchase_plan_id:
            purchase_plan_ids.add(slip.purchase_plan_id)
        slip.is_voided = 1
        slip.void_reason = reason[:500]
        slip.void_time = now
        slip.void_by = user_id
        slip.update_by = user_id
    for link in links:
        if link.db_purchase_plan_id:
            purchase_plan_ids.add(link.db_purchase_plan_id)
        inventory = inv_by_id.get(link.db_inventory_id)
        _subtract_plan_contribution(
            db,
            tenant_no=tenant_no,
            link=link,
            inventory=inventory,
            user_id=user_id,
        )
        if inventory is not None:
            inventory.del_flag = 1
            inventory.update_by = user_id
        link.del_flag = 1
        link.update_by = user_id
    db.flush()
    _refresh_purchase_plan_executed_qty(
        db, tenant_no=tenant_no, purchase_plan_ids=purchase_plan_ids, user_id=user_id
    )
    void_io_orders_for_plan(db, tenant_no=tenant_no, plan_id=plan_id, user_id=user_id)
    if transport_id:
        transport = (
            db.query(PlTransport)
            .filter(
                PlTransport.id == transport_id,
                PlTransport.tenant_no == tenant_no,
                PlTransport.del_flag == 0,
            )
            .first()
        )
        if transport is not None:
            transport.transport_status = "cancelled"
            transport.update_by = user_id


def _doc_label(primary: Optional[str], fallback: str) -> str:
    text = (primary or "").strip()
    return text or fallback


def _live_links(db: Session, tenant_no: str, plan_id: str) -> list[PlTransportDbLink]:
    return (
        db.query(PlTransportDbLink)
        .filter(
            PlTransportDbLink.tenant_no == tenant_no,
            PlTransportDbLink.del_flag == 0,
            PlTransportDbLink.plan_id == plan_id,
        )
        .all()
    )


def _live_inventories(db: Session, tenant_no: str, plan_id: str) -> list[DbInboundOutbound]:
    links = _live_links(db, tenant_no, plan_id)
    inv_ids = [link.db_inventory_id for link in links if link.db_inventory_id]
    if not inv_ids:
        return []
    return (
        db.query(DbInboundOutbound)
        .filter(
            DbInboundOutbound.tenant_no == tenant_no,
            DbInboundOutbound.del_flag == 0,
            DbInboundOutbound.id.in_(inv_ids),
        )
        .all()
    )


def _collect_signals(db: Session, tenant_no: str, plan_ids: list[str]) -> _Signals:
    signals = _Signals()
    links = (
        db.query(PlTransportDbLink)
        .filter(
            PlTransportDbLink.tenant_no == tenant_no,
            PlTransportDbLink.del_flag == 0,
            PlTransportDbLink.plan_id.in_(plan_ids),
        )
        .all()
    )
    plans_by_inv: dict[str, set[str]] = {}
    for link in links:
        if not link.db_inventory_id:
            continue
        plans_by_inv.setdefault(link.db_inventory_id, set()).add(link.plan_id)
    inv_ids = list(plans_by_inv)
    if inv_ids:
        inventories = (
            db.query(DbInboundOutbound)
            .filter(
                DbInboundOutbound.tenant_no == tenant_no,
                DbInboundOutbound.del_flag == 0,
                DbInboundOutbound.id.in_(inv_ids),
            )
            .all()
        )
        for inventory in inventories:
            owners = plans_by_inv.get(inventory.id, set())
            if inventory_status_is_executed(inventory.status):
                signals.inventory_executed.update(owners)
            if inventory.site_feedback_time is not None:
                signals.site_feedback.update(owners)
        slip_rows = (
            db.query(DbWeighingSlip.inbound_outbound_id)
            .filter(
                DbWeighingSlip.tenant_no == tenant_no,
                DbWeighingSlip.del_flag == 0,
                DbWeighingSlip.is_voided == 0,
                DbWeighingSlip.inbound_outbound_id.in_(inv_ids),
            )
            .all()
        )
        for (inv_id,) in slip_rows:
            signals.weighed.update(plans_by_inv.get(inv_id, set()))

    _collect_io_signals(db, tenant_no, plan_ids, signals)

    site_link_rows = (
        db.query(PlTransportSiteLink.plan_id)
        .filter(
            PlTransportSiteLink.tenant_no == tenant_no,
            PlTransportSiteLink.del_flag == 0,
            PlTransportSiteLink.plan_id.in_(plan_ids),
            PlTransportSiteLink.site_feedback_time.isnot(None),
        )
        .all()
    )
    signals.site_feedback.update(row[0] for row in site_link_rows if row[0])

    stop_rows = (
        db.query(PlTransportSiteStopFeedback.plan_id)
        .filter(
            PlTransportSiteStopFeedback.tenant_no == tenant_no,
            PlTransportSiteStopFeedback.del_flag == 0,
            PlTransportSiteStopFeedback.plan_id.in_(plan_ids),
            PlTransportSiteStopFeedback.feedback_time.isnot(None),
        )
        .all()
    )
    signals.site_feedback.update(row[0] for row in stop_rows if row[0])
    return signals


def _collect_io_signals(
    db: Session, tenant_no: str, plan_ids: list[str], signals: _Signals
) -> None:
    for header_model, item_model in (
        (PlInboundOrder, PlInboundOrderItem),
        (PlOutboundOrder, PlOutboundOrderItem),
    ):
        headers = (
            db.query(header_model)
            .filter(
                header_model.tenant_no == tenant_no,
                header_model.del_flag == 0,
                header_model.plan_id.in_(plan_ids),
                header_model.status != "voided",
            )
            .all()
        )
        if not headers:
            continue
        plan_by_order = {row.id: row.plan_id for row in headers}
        for row in headers:
            if row.site_feedback_time is not None or row.site_first_submitted_at is not None:
                signals.site_feedback.add(row.plan_id)
        manual_rows = (
            db.query(item_model.order_id)
            .filter(
                item_model.tenant_no == tenant_no,
                item_model.del_flag == 0,
                item_model.order_id.in_(list(plan_by_order)),
                item_model.qty_source == "manual",
            )
            .all()
        )
        for (order_id,) in manual_rows:
            plan_id = plan_by_order.get(order_id)
            if plan_id:
                signals.manual_qty.add(plan_id)


def _refresh_purchase_plan_executed_qty(
    db: Session,
    *,
    tenant_no: str,
    purchase_plan_ids: set[str],
    user_id: str,
) -> None:
    for plan_id in purchase_plan_ids:
        plan = (
            db.query(DbPurchasePlan)
            .filter(
                DbPurchasePlan.id == plan_id,
                DbPurchasePlan.tenant_no == tenant_no,
                DbPurchasePlan.del_flag == 0,
            )
            .first()
        )
        if plan is None:
            continue
        slips = (
            db.query(DbWeighingSlip)
            .filter(
                DbWeighingSlip.is_voided == 0,
                DbWeighingSlip.del_flag == 0,
                DbWeighingSlip.tenant_no == tenant_no,
                or_(
                    DbWeighingSlip.plan_no == plan.plan_no,
                    DbWeighingSlip.purchase_plan_id == plan.id,
                ),
            )
            .all()
        )
        total_executed = sum(slip.net_weight or 0 for slip in slips)
        plan.executed_quantity = total_executed
        planned_quantity_kg = (plan.planned_quantity or 0) * 1000
        if total_executed >= planned_quantity_kg and planned_quantity_kg > 0:
            plan.execution_status = "已入库" if plan.business_type == "入库" else "已出库"
        elif total_executed > 0:
            plan.execution_status = "入库中" if plan.business_type == "入库" else "出库中"
        else:
            plan.execution_status = "待入库" if plan.business_type == "入库" else "待出库"
        plan.update_by = user_id