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

"""运输计划列表：批量计算「复制给司机 / 现场 / 磅单二维码」按钮显示标志。

语义与查看弹窗 / usePlPlanArrangeActions / build_site_records_for_arranged_plan 对齐：
- 已安排且未取消，且存在司机链接或现场记录或地磅出入库关联时，显示「复制给司机」
- 有现场链接记录（含运输单按段 site_access_token、出入库单现场令牌回退）时显示「复制给现场」
- 有勾选生成地磅单的出入库关联（db_purchase_plan_id 非空）时显示「复制磅单二维码」
"""
from __future__ import annotations

from dataclasses import dataclass
from typing import Iterable

from sqlalchemy.orm import Session

from app.models.pl_models import (
    PlInboundOrder,
    PlOutboundOrder,
    PlPlan,
    PlTransport,
    PlTransportDbLink,
    PlTransportSiteLink,
)
from app.services.pl_dibang_bridge_service import plan_has_load_stop
from app.services.pl_internal_customer import plan_requires_site_link


@dataclass(frozen=True)
class PlanArrangeCopyFlags:
    can_copy_to_driver: bool = False
    can_copy_to_site: bool = False
    can_copy_pound_qrcode: bool = False


def load_plan_arrange_copy_flags(
    db: Session,
    tenant_no: str,
    plans: Iterable[PlPlan],
) -> dict[str, PlanArrangeCopyFlags]:
    plan_list = list(plans)
    empty = PlanArrangeCopyFlags()
    result: dict[str, PlanArrangeCopyFlags] = {plan.id: empty for plan in plan_list}

    candidates = [
        plan
        for plan in plan_list
        if plan.transport_status == "arranged"
        and plan.follow_status != "cancelled"
        and (plan.transport_id or "").strip()
    ]
    if not candidates:
        return result

    plan_ids = [plan.id for plan in candidates]
    transport_ids = list({plan.transport_id for plan in candidates if plan.transport_id})

    transports = (
        db.query(PlTransport)
        .filter(
            PlTransport.id.in_(transport_ids),
            PlTransport.tenant_no == tenant_no,
            PlTransport.del_flag == 0,
        )
        .all()
    )
    transport_by_id = {
        row.id: row for row in transports if row.transport_status != "cancelled"
    }

    site_plan_ids: set[str] = {
        plan_id
        for (plan_id,) in db.query(PlTransportSiteLink.plan_id)
        .filter(
            PlTransportSiteLink.tenant_no == tenant_no,
            PlTransportSiteLink.plan_id.in_(plan_ids),
            PlTransportSiteLink.del_flag == 0,
        )
        .distinct()
        .all()
    }

    inventory_plan_ids: set[str] = set()
    dibang_plan_ids: set[str] = set()
    for plan_id, db_purchase_plan_id in (
        db.query(PlTransportDbLink.plan_id, PlTransportDbLink.db_purchase_plan_id)
        .filter(
            PlTransportDbLink.tenant_no == tenant_no,
            PlTransportDbLink.plan_id.in_(plan_ids),
            PlTransportDbLink.del_flag == 0,
        )
        .all()
    ):
        inventory_plan_ids.add(plan_id)
        if (db_purchase_plan_id or "").strip():
            dibang_plan_ids.add(plan_id)

    # 与 get_plan_arrange_info 一致：无 site_link 时，用未作废出入库单的现场令牌回退
    inbound_ids = [
        plan.id
        for plan in candidates
        if plan.plan_type == "inbound" and plan.id not in site_plan_ids
    ]
    if inbound_ids:
        for plan_id, site_access_token, status in (
            db.query(PlInboundOrder.plan_id, PlInboundOrder.site_access_token, PlInboundOrder.status)
            .filter(
                PlInboundOrder.tenant_no == tenant_no,
                PlInboundOrder.plan_id.in_(inbound_ids),
                PlInboundOrder.del_flag == 0,
            )
            .all()
        ):
            if status != "voided" and (site_access_token or "").strip():
                site_plan_ids.add(plan_id)

    outbound_ids = [
        plan.id
        for plan in candidates
        if plan.plan_type == "outbound" and plan.id not in site_plan_ids
    ]
    if outbound_ids:
        for plan_id, site_access_token, status in (
            db.query(
                PlOutboundOrder.plan_id, PlOutboundOrder.site_access_token, PlOutboundOrder.status
            )
            .filter(
                PlOutboundOrder.tenant_no == tenant_no,
                PlOutboundOrder.plan_id.in_(outbound_ids),
                PlOutboundOrder.del_flag == 0,
            )
            .all()
        ):
            if status != "voided" and (site_access_token or "").strip():
                site_plan_ids.add(plan_id)

    # 与 build_site_records_for_arranged_plan 一致：无 site_link 行时，装货段 + 运输单 site_access_token
    for plan in candidates:
        if plan.id in site_plan_ids:
            continue
        if not plan_requires_site_link(db, tenant_no=tenant_no, plan=plan):
            continue
        transport = transport_by_id.get(plan.transport_id or "")
        if not transport:
            continue
        if not plan_has_load_stop(db, tenant_no, plan.id):
            continue
        if (transport.site_access_token or "").strip():
            site_plan_ids.add(plan.id)

    for plan in candidates:
        transport = transport_by_id.get(plan.transport_id or "")
        if not transport:
            continue
        has_driver = bool((transport.driver_access_token or "").strip())
        has_site = plan.id in site_plan_ids
        has_inventory = plan.id in inventory_plan_ids
        has_dibang = plan.id in dibang_plan_ids
        if not (has_driver or has_site or has_inventory):
            continue
        result[plan.id] = PlanArrangeCopyFlags(
            can_copy_to_driver=True,
            can_copy_to_site=has_site,
            can_copy_pound_qrcode=has_dibang,
        )

    return result