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

"""运输计划安排时联动地磅计划与出入库单。"""
from __future__ import annotations

import calendar
import re
from dataclasses import dataclass
from datetime import date, datetime
from decimal import Decimal
from typing import Optional

from sqlalchemy.orm import Session

from app.core.utils import generate_uuid
from app.models.db_models import DbInboundOutbound, DbPurchasePlan, DbWeighingSlip
from app.models.pl_models import PlPlan, PlPlanProduct, PlTransport, PlTransportDbLink
from app.services.db.inventory_id import next_inbound_outbound_id
from app.services.db.mappers import parse_date_str
from app.services.db.plan_helpers import default_execution_status, generate_plan_no
from app.services.db.product_helpers import (
    apply_product_snapshot_to_inventory,
    apply_product_snapshot_to_plan,
    resolve_crm_product,
)


@dataclass
class DibangInventoryLinkResult:
    plan_product_id: str
    product_name: str
    db_purchase_plan_id: str
    db_inventory_id: str
    batch_no: str
    plan_no: str


def resolve_partner_company(plan: PlPlan, plan_type: str) -> str:
    """按运输计划单据类型解析地磅往来单位（不区分 CRM 合同类型）。

    - 入库单 inbound：供方 → party_a，其次 supplier_name
    - 出库单 outbound：需方 → party_b，其次 customer_name
    """
    if plan_type == "inbound":
        return (plan.party_a or plan.supplier_name or "").strip()
    if plan_type == "outbound":
        return (plan.party_b or plan.customer_name or "").strip()
    return ""


def _normalize_to_date(value: date | datetime | str | None) -> Optional[date]:
    if value is None:
        return None
    if isinstance(value, datetime):
        return value.date()
    if isinstance(value, date):
        return value
    if isinstance(value, str):
        return parse_date_str(value)
    return None


def _normalize_plan_date(plan: PlPlan) -> date:
    plan_date = plan.plan_date
    if isinstance(plan_date, datetime):
        return plan_date.date()
    if isinstance(plan_date, date):
        return plan_date
    if isinstance(plan_date, str):
        parsed = parse_date_str(plan_date)
        return parsed or date.today()
    return date.today()


def _resolve_dibang_date(plan: PlPlan) -> date:
    """入库单用送达时间，出库单用取货时间；缺失时回退计划日期。"""
    if plan.plan_type == "inbound":
        resolved = _normalize_to_date(plan.delivery_time)
    elif plan.plan_type == "outbound":
        resolved = _normalize_to_date(plan.pickup_time)
    else:
        resolved = None
    return resolved or _normalize_plan_date(plan)


def _month_range(plan_date: date) -> tuple[date, date]:
    month_start = plan_date.replace(day=1)
    last_day = calendar.monthrange(plan_date.year, plan_date.month)[1]
    month_end = plan_date.replace(day=last_day)
    return month_start, month_end


def _plan_type_to_business_type(plan_type: str) -> Optional[str]:
    if plan_type == "inbound":
        return "入库"
    if plan_type == "outbound":
        return "出库"
    return None


def _product_quantity(product: PlPlanProduct) -> float:
    for value in (product.quantity, product.inbound_estimate_qty, product.outbound_estimate_qty):
        if value is not None:
            return float(value)
    return 0.0


def _apply_plan_quantity_delta(
    db_plan: DbPurchasePlan,
    delta: float,
    *,
    user_id: Optional[str],
) -> None:
    current = float(db_plan.planned_quantity or 0)
    db_plan.planned_quantity = max(0.0, current + float(delta))
    db_plan.update_by = user_id


def _resolve_link_contributed_qty(
    link: PlTransportDbLink,
    inventory: Optional[DbInboundOutbound],
) -> float:
    if link.contributed_qty is not None:
        return float(link.contributed_qty)
    if inventory and inventory.quantity is not None:
        return float(inventory.quantity)
    return 0.0


def _subtract_plan_contribution(
    db: Session,
    *,
    tenant_no: str,
    link: PlTransportDbLink,
    inventory: Optional[DbInboundOutbound],
    user_id: Optional[str],
) -> None:
    if not link.db_purchase_plan_id:
        return
    db_plan = (
        db.query(DbPurchasePlan)
        .filter(
            DbPurchasePlan.id == link.db_purchase_plan_id,
            DbPurchasePlan.tenant_no == tenant_no,
            DbPurchasePlan.del_flag == 0,
        )
        .first()
    )
    if not db_plan:
        return
    qty = _resolve_link_contributed_qty(link, inventory)
    if qty:
        _apply_plan_quantity_delta(db_plan, -qty, user_id=user_id)


def _generate_batch_no(db: Session, tenant_no: str) -> str:
    rows = (
        db.query(DbInboundOutbound.batch_no)
        .filter(
            DbInboundOutbound.tenant_no == tenant_no,
            DbInboundOutbound.del_flag == 0,
            DbInboundOutbound.batch_no.like("CRKDH%"),
        )
        .all()
    )
    max_seq = 0
    pattern = re.compile(r"^CRKDH(\d+)$")
    for (batch_no,) in rows:
        if batch_no:
            match = pattern.match(batch_no)
            if match:
                max_seq = max(max_seq, int(match.group(1)))
    return f"CRKDH{max_seq + 1:07d}"


def find_monthly_db_plan(
    db: Session,
    tenant_no: str,
    plan_date: date,
    business_type: str,
    customer_name: str,
    product_id: str,
    specification: Optional[str],
) -> Optional[DbPurchasePlan]:
    month_start, month_end = _month_range(plan_date)
    customer = (customer_name or "").strip()
    spec = (specification or "").strip()

    query = db.query(DbPurchasePlan).filter(
        DbPurchasePlan.tenant_no == tenant_no,
        DbPurchasePlan.del_flag == 0,
        DbPurchasePlan.business_type == business_type,
        DbPurchasePlan.product_id == product_id,
        DbPurchasePlan.start_date <= month_end,
        DbPurchasePlan.end_date >= month_start,
    )
    if customer:
        query = query.filter(DbPurchasePlan.customer_name == customer)
    if spec:
        query = query.filter(DbPurchasePlan.specification == spec)

    return query.order_by(DbPurchasePlan.create_time.desc()).first()


def _create_db_plan(
    db: Session,
    tenant_no: str,
    plan: PlPlan,
    product: PlPlanProduct,
    business_type: str,
    customer_name: str,
    dibang_date: date,
    month_start: date,
    month_end: date,
    user_id: Optional[str],
    dept_id: str,
    contributed_qty: float,
) -> DbPurchasePlan:
    snapshot = resolve_crm_product(db, tenant_no, product.product_id, require_on_shelf=False)
    db_plan = DbPurchasePlan(
        id=generate_uuid(),
        tenant_no=tenant_no,
        plan_no=generate_plan_no(),
        plan_date=dibang_date,
        start_date=month_start,
        end_date=month_end,
        business_type=business_type,
        planned_quantity=contributed_qty,
        vehicle_no=(plan.license_plate or "").strip(),
        customer_name=customer_name or None,
        remark=f"来源运输计划 {plan.plan_no}",
        execution_status=default_execution_status(business_type),
        executed_at=None,
        last_weighing_slip_id=None,
        executed_quantity=0,
        auto_generate_enabled=0,
        auto_generate_end_month=None,
        auto_generate_last_month=None,
        create_by=user_id,
        create_dept_id=dept_id,
    )
    apply_product_snapshot_to_plan(db_plan, snapshot)
    db.add(db_plan)
    db.flush()
    return db_plan


def _create_db_inventory(
    db: Session,
    tenant_no: str,
    plan: PlPlan,
    product: PlPlanProduct,
    db_plan: DbPurchasePlan,
    business_type: str,
    customer_name: str,
    license_plate: str,
    dibang_date: date,
    user_id: Optional[str],
    dept_id: str,
) -> DbInboundOutbound:
    snapshot = resolve_crm_product(db, tenant_no, product.product_id, require_on_shelf=False)
    status_value = "待入库" if business_type == "入库" else "待出库"
    inventory_plan_no = ((db_plan.plan_no if db_plan else plan.plan_no) or "").strip() or None
    remark = f"运输计划 {plan.plan_no}"
    purchase_plan_id = db_plan.id if db_plan else None
    inventory = DbInboundOutbound(
        id=next_inbound_outbound_id(db, tenant_no),
        tenant_no=tenant_no,
        date=dibang_date,
        partner_company=customer_name,
        business_type=business_type,
        quantity=Decimal(str(_product_quantity(product))),
        plan_no=inventory_plan_no,
        batch_no=_generate_batch_no(db, tenant_no),
        vehicle_no=(license_plate or "").strip(),
        status=status_value,
        remark=remark,
        purchase_plan_id=purchase_plan_id,
        create_by=user_id,
        create_dept_id=dept_id,
    )
    apply_product_snapshot_to_inventory(inventory, snapshot)
    db.add(inventory)
    db.flush()
    return inventory


# 仅这些状态的出入库单可在「再次安排+勾地磅」时软删后重建
_REVOKABLE_INVENTORY_STATUSES = frozenset({"待入库", "待出库", "待三方填写", "待处理"})


def _inventory_has_weighing_slip(db: Session, inventory_id: str) -> bool:
    return (
        db.query(DbWeighingSlip.id)
        .filter(
            DbWeighingSlip.inbound_outbound_id == inventory_id,
            DbWeighingSlip.del_flag == 0,
            DbWeighingSlip.is_voided == 0,
        )
        .first()
        is not None
    )


def revoke_regeneratable_dibang_for_transport(
    db: Session,
    *,
    tenant_no: str,
    transport_id: str,
    user_id: Optional[str],
) -> None:
    """再次勾选地磅安排前：软删本运输单此前生成的、仍可作废的出入库单及关联。

    - 不删除地磅月计划（可能被其他业务复用）
    - 若出入库单已完成的/已有关联磅单 → 阻断并提示人工处理
    """
    links = (
        db.query(PlTransportDbLink)
        .filter(
            PlTransportDbLink.tenant_no == tenant_no,
            PlTransportDbLink.transport_id == transport_id,
            PlTransportDbLink.del_flag == 0,
        )
        .all()
    )
    if not links:
        return

    blocking: list[str] = []
    for link in links:
        inventory = (
            db.query(DbInboundOutbound)
            .filter(
                DbInboundOutbound.id == link.db_inventory_id,
                DbInboundOutbound.tenant_no == tenant_no,
                DbInboundOutbound.del_flag == 0,
            )
            .first()
        )
        if not inventory:
            link.del_flag = 1
            link.update_by = user_id
            continue

        label = inventory.batch_no or inventory.id
        status = (inventory.status or "").strip()
        if status and status not in _REVOKABLE_INVENTORY_STATUSES:
            blocking.append(f"{label}（状态：{status}）")
            continue
        if _inventory_has_weighing_slip(db, inventory.id):
            blocking.append(f"{label}（已关联地磅单）")

    if blocking:
        raise ValueError(
            "以下出入库单已执行或已关联地磅，无法随运输安排自动重建："
            + "；".join(blocking)
            + "。请在地磅管理中处理后再安排。"
        )

    for link in links:
        inventory = (
            db.query(DbInboundOutbound)
            .filter(
                DbInboundOutbound.id == link.db_inventory_id,
                DbInboundOutbound.tenant_no == tenant_no,
                DbInboundOutbound.del_flag == 0,
            )
            .first()
        )
        if inventory:
            _subtract_plan_contribution(
                db,
                tenant_no=tenant_no,
                link=link,
                inventory=inventory,
                user_id=user_id,
            )
            inventory.del_flag = 1
            inventory.update_by = user_id
        link.del_flag = 1
        link.update_by = user_id

    db.flush()


def _sync_transport_inventory_links(
    db: Session,
    *,
    tenant_no: str,
    plan: PlPlan,
    products: list[PlPlanProduct],
    transport: PlTransport,
    user_id: Optional[str],
    dept_id: str,
) -> list[DibangInventoryLinkResult]:
    business_type = _plan_type_to_business_type(plan.plan_type)
    if not business_type:
        return []

    partner = resolve_partner_company(plan, plan.plan_type)
    if not partner:
        raise ValueError("无法解析往来单位，请检查合同供方/需方信息")

    dibang_date = _resolve_dibang_date(plan)
    month_start, month_end = _month_range(dibang_date)

    license_plate = (transport.license_plate or plan.license_plate or "").strip()

    results: list[DibangInventoryLinkResult] = []

    for product in products:
        if not product.product_id:
            raise ValueError(f"产品【{product.product_name}】未关联 CRM 产品，无法生成地磅出入库单")

        contributed_qty = _product_quantity(product)

        db_plan = find_monthly_db_plan(
            db,
            tenant_no,
            dibang_date,
            business_type,
            partner,
            product.product_id,
            product.model,
        )
        if not db_plan:
            db_plan = _create_db_plan(
                db,
                tenant_no,
                plan,
                product,
                business_type,
                partner,
                dibang_date,
                month_start,
                month_end,
                user_id,
                dept_id,
                contributed_qty,
            )
        else:
            _apply_plan_quantity_delta(db_plan, contributed_qty, user_id=user_id)

        inventory = _create_db_inventory(
            db,
            tenant_no,
            plan,
            product,
            db_plan,
            business_type,
            partner,
            license_plate,
            dibang_date,
            user_id,
            dept_id,
        )

        link = PlTransportDbLink(
            id=generate_uuid(),
            tenant_no=tenant_no,
            transport_id=transport.id,
            plan_id=plan.id,
            plan_product_id=product.id,
            db_purchase_plan_id=db_plan.id if db_plan else "",
            db_inventory_id=inventory.id,
            contributed_qty=Decimal(str(contributed_qty)),
            remark=f"运输计划 {plan.plan_no}",
            create_by=user_id,
            create_dept_id=dept_id,
        )
        db.add(link)

        results.append(
            DibangInventoryLinkResult(
                plan_product_id=product.id,
                product_name=product.product_name or "",
                db_purchase_plan_id=db_plan.id if db_plan else "",
                db_inventory_id=inventory.id,
                batch_no=inventory.batch_no or "",
                plan_no=(db_plan.plan_no if db_plan else plan.plan_no) or "",
            )
        )

    return results


def sync_dibang_for_transport(
    db: Session,
    *,
    tenant_no: str,
    plan: PlPlan,
    products: list[PlPlanProduct],
    transport: PlTransport,
    user_id: Optional[str],
    dept_id: str,
) -> list[DibangInventoryLinkResult]:
    """勾选「生成地磅单」：按产品生成/复用地磅计划，并创建关联出入库单。"""
    return _sync_transport_inventory_links(
        db,
        tenant_no=tenant_no,
        plan=plan,
        products=products,
        transport=transport,
        user_id=user_id,
        dept_id=dept_id,
    )