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 Any, Literal, 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,
    PlPlanDelivery,
    PlPlanDeliveryProduct,
    PlPlanLoad,
    PlPlanLoadProduct,
    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,
)
from app.services.pl_plan_stop_rules import normalize_stop_dibang_fields


@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


@dataclass
class StopDibangOption:
    stop_type: Literal["load", "delivery"]
    stop_id: str
    generate_dibang: bool


@dataclass
class _StopDibangProduct:
    plan_product_id: str
    product_id: str
    product_name: str
    model: Optional[str]
    unit: str
    quantity: float


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 _direction_to_business_type(direction: Optional[str]) -> Optional[str]:
    normalized = (direction or "").strip()
    if normalized == "outbound":
        return "出库"
    if normalized == "inbound":
        return "入库"
    return None


def _resolve_stop_business_type(
    stop: PlPlanLoad | PlPlanDelivery,
    stop_type: Literal["load", "delivery"],
) -> Optional[str]:
    _, direction = normalize_stop_dibang_fields(
        stop.customer_name,
        stop_type,
        stop.generate_dibang,
        stop.dibang_direction,
    )
    return _direction_to_business_type(direction)


def resolve_stop_dibang_partner(
    stop_type: Literal["load", "delivery"],
    load: Optional[PlPlanLoad],
    deliveries: list[PlPlanDelivery],
) -> str:
    """装货段地磅往来单位取第一个送货客户；送货段取装货客户。"""
    if stop_type == "load":
        if deliveries:
            return (deliveries[0].customer_name or "").strip()
        return (load.customer_name or "").strip() if load else ""
    return (load.customer_name or "").strip() if load else ""


def _stop_product_quantity(product: Any) -> float:
    if getattr(product, "plan_qty", None) is not None:
        return float(product.plan_qty)
    return 0.0


def plan_has_load_stop(db: Session, tenant_no: str, plan_id: str) -> bool:
    return (
        db.query(PlPlanLoad.id)
        .filter(
            PlPlanLoad.plan_id == plan_id,
            PlPlanLoad.tenant_no == tenant_no,
            PlPlanLoad.del_flag == 0,
        )
        .first()
        is not None
    )


def apply_stop_dibang_options(
    db: Session,
    *,
    tenant_no: str,
    plan_id: str,
    stop_options: list[StopDibangOption],
    user_id: Optional[str],
) -> None:
    for option in stop_options:
        if option.stop_type == "load":
            stop = (
                db.query(PlPlanLoad)
                .filter(
                    PlPlanLoad.id == option.stop_id,
                    PlPlanLoad.plan_id == plan_id,
                    PlPlanLoad.tenant_no == tenant_no,
                    PlPlanLoad.del_flag == 0,
                )
                .first()
            )
        else:
            stop = (
                db.query(PlPlanDelivery)
                .filter(
                    PlPlanDelivery.id == option.stop_id,
                    PlPlanDelivery.plan_id == plan_id,
                    PlPlanDelivery.tenant_no == tenant_no,
                    PlPlanDelivery.del_flag == 0,
                )
                .first()
            )
        if stop:
            gen, direction = normalize_stop_dibang_fields(
                stop.customer_name,
                "load" if option.stop_type == "load" else "delivery",
                1 if option.generate_dibang else 0,
                stop.dibang_direction,
            )
            stop.generate_dibang = gen
            stop.dibang_direction = direction
            stop.update_by = user_id
    db.flush()


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 _inventory_date_editable(db: Session, inventory: DbInboundOutbound) -> bool:
    status = (inventory.status or "").strip()
    if status and status not in _REVOKABLE_INVENTORY_STATUSES:
        return False
    if _inventory_has_weighing_slip(db, inventory.id):
        return False
    return True


def _resolve_link_dibang_date(
    plan: PlPlan,
    link: PlTransportDbLink,
    load: Optional[PlPlanLoad],
    deliveries: list[PlPlanDelivery],
) -> date:
    stop_type = (link.stop_type or "").strip()
    if stop_type == "load":
        return _resolve_stop_dibang_date(
            plan,
            stop_type="load",
            load_time=load.load_time if load else None,
        )
    if stop_type == "delivery":
        delivery = next((row for row in deliveries if row.id == link.stop_id), None)
        if delivery is None and len(deliveries) == 1:
            delivery = deliveries[0]
        return _resolve_stop_dibang_date(
            plan,
            stop_type="delivery",
            delivery_time=delivery.delivery_time if delivery else None,
        )
    return _resolve_dibang_date(plan)


def _ensure_monthly_db_plan_for_inventory(
    db: Session,
    *,
    tenant_no: str,
    plan: PlPlan,
    inventory: DbInboundOutbound,
    dibang_date: date,
    contributed_qty: float,
    user_id: Optional[str],
    dept_id: str,
) -> DbPurchasePlan:
    business_type = (inventory.business_type or "").strip()
    partner = (inventory.partner_company or "").strip()
    product_id = (inventory.product_id or "").strip()
    if not business_type or not product_id:
        raise ValueError("出入库单缺少业务类型或产品，无法改挂地磅月计划")

    existing = find_monthly_db_plan(
        db,
        tenant_no,
        dibang_date,
        business_type,
        partner,
        product_id,
        inventory.specification,
    )
    if existing:
        if contributed_qty:
            _apply_plan_quantity_delta(existing, contributed_qty, user_id=user_id)
        return existing

    month_start, month_end = _month_range(dibang_date)
    product = PlPlanProduct(
        id=generate_uuid(),
        product_id=product_id,
        product_name=inventory.product or "",
        model=inventory.specification,
        unit="吨",
        quantity=Decimal(str(contributed_qty or inventory.quantity or 0)),
    )
    return _create_db_plan(
        db,
        tenant_no,
        plan,
        product,
        business_type,
        partner,
        dibang_date,
        month_start,
        month_end,
        user_id,
        dept_id,
        contributed_qty or float(inventory.quantity or 0),
    )


def sync_linked_dibang_dates_for_plan(
    db: Session,
    *,
    tenant_no: str,
    plan: PlPlan,
    user_id: Optional[str],
    dept_id: str = "",
) -> list[str]:
    """计划保存后按当前段次/计划日期回写可编辑出入库单日期；跨月则改挂月计划。

    不可回写时不抛错，返回 warnings。
    """
    links = (
        db.query(PlTransportDbLink)
        .filter(
            PlTransportDbLink.tenant_no == tenant_no,
            PlTransportDbLink.plan_id == plan.id,
            PlTransportDbLink.del_flag == 0,
        )
        .all()
    )
    if not links:
        return []

    load = (
        db.query(PlPlanLoad)
        .filter(
            PlPlanLoad.plan_id == plan.id,
            PlPlanLoad.tenant_no == tenant_no,
            PlPlanLoad.del_flag == 0,
        )
        .first()
    )
    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()
    )

    warnings: 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:
            continue

        new_date = _resolve_link_dibang_date(plan, link, load, deliveries)
        old_date = _normalize_to_date(inventory.date)
        if old_date == new_date:
            continue

        label = inventory.batch_no or inventory.id
        if not _inventory_date_editable(db, inventory):
            status = (inventory.status or "").strip() or "未知状态"
            if _inventory_has_weighing_slip(db, inventory.id):
                reason = "已关联地磅单"
            else:
                reason = f"状态：{status}"
            warnings.append(
                f"出入库单 {label}（{reason}）未同步日期至 {new_date.isoformat()}，请在地磅管理中手工处理"
            )
            continue

        same_month = (
            old_date is not None
            and old_date.year == new_date.year
            and old_date.month == new_date.month
        )
        inventory.date = new_date
        inventory.update_by = user_id

        if same_month:
            continue

        qty = _resolve_link_contributed_qty(link, inventory)
        _subtract_plan_contribution(
            db,
            tenant_no=tenant_no,
            link=link,
            inventory=inventory,
            user_id=user_id,
        )
        try:
            new_db_plan = _ensure_monthly_db_plan_for_inventory(
                db,
                tenant_no=tenant_no,
                plan=plan,
                inventory=inventory,
                dibang_date=new_date,
                contributed_qty=qty,
                user_id=user_id,
                dept_id=dept_id or "",
            )
        except ValueError as exc:
            warnings.append(f"出入库单 {label} 跨月改挂失败：{exc}")
            continue

        link.db_purchase_plan_id = new_db_plan.id
        link.update_by = user_id
        inventory.purchase_plan_id = new_db_plan.id
        inventory.plan_no = (new_db_plan.plan_no or "").strip() or inventory.plan_no

    db.flush()
    return warnings


def _sync_stop_inventory_links(
    db: Session,
    *,
    tenant_no: str,
    plan: PlPlan,
    transport: PlTransport,
    stop_type: str,
    stop_id: str,
    partner: str,
    business_type: str,
    dibang_date: date,
    stop_products: list[_StopDibangProduct],
    user_id: Optional[str],
    dept_id: str,
) -> list[DibangInventoryLinkResult]:
    if not partner:
        raise ValueError("无法解析往来单位，请检查段客户信息")

    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 stop_products:
        if not product.product_id:
            raise ValueError(f"产品【{product.product_name}】未关联 CRM 产品，无法生成地磅出入库单")

        contributed_qty = product.quantity
        if contributed_qty <= 0:
            continue

        plan_product = PlPlanProduct(
            id=product.plan_product_id,
            product_id=product.product_id,
            product_name=product.product_name,
            model=product.model,
            unit=product.unit,
            quantity=Decimal(str(contributed_qty)),
        )

        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,
                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,
            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.plan_product_id,
            stop_type=stop_type,
            stop_id=stop_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.plan_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 _resolve_stop_dibang_date(
    plan: PlPlan,
    *,
    stop_type: str,
    load_time: date | datetime | str | None = None,
    delivery_time: date | datetime | str | None = None,
) -> date:
    if stop_type == "load":
        resolved = _normalize_to_date(load_time)
        if resolved is None:
            resolved = _normalize_to_date(plan.pickup_time)
    else:
        resolved = _normalize_to_date(delivery_time)
        if resolved is None:
            resolved = _normalize_to_date(plan.delivery_time)
    return resolved or _normalize_plan_date(plan)


def _load_stop_products(
    db: Session,
    tenant_no: str,
    load_id: str,
) -> list[_StopDibangProduct]:
    rows = (
        db.query(PlPlanLoadProduct)
        .filter(
            PlPlanLoadProduct.load_id == load_id,
            PlPlanLoadProduct.tenant_no == tenant_no,
            PlPlanLoadProduct.del_flag == 0,
        )
        .order_by(PlPlanLoadProduct.order_num.asc(), PlPlanLoadProduct.create_time.asc())
        .all()
    )
    return [
        _StopDibangProduct(
            plan_product_id=row.plan_product_id,
            product_id=row.product_id,
            product_name=row.product_name or "",
            model=row.model,
            unit=row.unit or "吨",
            quantity=_stop_product_quantity(row),
        )
        for row in rows
    ]


def _delivery_stop_products(
    db: Session,
    tenant_no: str,
    delivery_id: str,
) -> list[_StopDibangProduct]:
    rows = (
        db.query(PlPlanDeliveryProduct)
        .filter(
            PlPlanDeliveryProduct.delivery_id == delivery_id,
            PlPlanDeliveryProduct.tenant_no == tenant_no,
            PlPlanDeliveryProduct.del_flag == 0,
        )
        .order_by(PlPlanDeliveryProduct.order_num.asc(), PlPlanDeliveryProduct.create_time.asc())
        .all()
    )
    return [
        _StopDibangProduct(
            plan_product_id=row.plan_product_id,
            product_id=row.product_id,
            product_name=row.product_name or "",
            model=row.model,
            unit=row.unit or "吨",
            quantity=_stop_product_quantity(row),
        )
        for row in rows
    ]


def sync_dibang_for_plan_stops(
    db: Session,
    *,
    tenant_no: str,
    plan: PlPlan,
    transport: PlTransport,
    stop_options: list[StopDibangOption],
    user_id: Optional[str],
    dept_id: str,
) -> list[DibangInventoryLinkResult]:
    """按装货/送货段勾选生成地磅计划与出入库单。"""
    apply_stop_dibang_options(
        db,
        tenant_no=tenant_no,
        plan_id=plan.id,
        stop_options=stop_options,
        user_id=user_id,
    )

    option_map = {(item.stop_type, item.stop_id): item for item in stop_options}
    results: list[DibangInventoryLinkResult] = []

    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()
    )

    load = (
        db.query(PlPlanLoad)
        .filter(
            PlPlanLoad.plan_id == plan.id,
            PlPlanLoad.tenant_no == tenant_no,
            PlPlanLoad.del_flag == 0,
        )
        .first()
    )
    if load:
        option = option_map.get(("load", load.id))
        if option and option.generate_dibang:
            business_type = _resolve_stop_business_type(load, "load")
            if not business_type:
                raise ValueError("装货段未配置地磅方向，无法生成地磅单")
            stop_products = _load_stop_products(db, tenant_no, load.id)
            if not stop_products:
                raise ValueError("装货段无产品明细，无法生成地磅单")
            partner = resolve_stop_dibang_partner("load", load, deliveries)
            if not partner:
                raise ValueError("无法解析装货段地磅往来单位，请先配置送货客户")
            results.extend(
                _sync_stop_inventory_links(
                    db,
                    tenant_no=tenant_no,
                    plan=plan,
                    transport=transport,
                    stop_type="load",
                    stop_id=load.id,
                    partner=partner,
                    business_type=business_type,
                    dibang_date=_resolve_stop_dibang_date(
                        plan,
                        stop_type="load",
                        load_time=load.load_time,
                    ),
                    stop_products=stop_products,
                    user_id=user_id,
                    dept_id=dept_id,
                )
            )

    for delivery in deliveries:
        option = option_map.get(("delivery", delivery.id))
        if not option or not option.generate_dibang:
            continue
        business_type = _resolve_stop_business_type(delivery, "delivery")
        if not business_type:
            raise ValueError(f"送货段【{delivery.customer_name or delivery.seq}】未配置地磅方向，无法生成地磅单")
        stop_products = _delivery_stop_products(db, tenant_no, delivery.id)
        if not stop_products:
            raise ValueError(f"送货段【{delivery.customer_name or delivery.seq}】无产品明细，无法生成地磅单")
        partner = resolve_stop_dibang_partner("delivery", load, deliveries)
        if not partner:
            raise ValueError("无法解析送货段地磅往来单位，请先配置装货客户")
        results.extend(
            _sync_stop_inventory_links(
                db,
                tenant_no=tenant_no,
                plan=plan,
                transport=transport,
                stop_type="delivery",
                stop_id=delivery.id,
                partner=partner,
                business_type=business_type,
                dibang_date=_resolve_stop_dibang_date(
                    plan,
                    stop_type="delivery",
                    delivery_time=delivery.delivery_time,
                ),
                stop_products=stop_products,
                user_id=user_id,
                dept_id=dept_id,
            )
        )

    return results


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,
            stop_type=None,
            stop_id=None,
            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,
    )