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

"""运输计划归档导出：按装货/送货段 × 段内产品展开，与列表展示规则一致。"""
from __future__ import annotations

from datetime import date, datetime
from decimal import Decimal
from typing import Optional, Union

from sqlalchemy.orm import Session

from app.models.pl_models import PlPlan, PlPlanContract, PlPlanProduct, PlTransport
from app.schemas.pl.plan import PlanDeliveryRead, PlanLoadRead, PlanStopProductRead
from app.services.pl.plan_list_filters import customer_name_matches_keywords
from app.services.pl_internal_customer import is_internal_customer_name
from app.services.pl_plan_stop_service import load_plan_stops_grouped


def _unique_contract_nos(*sources: Optional[str]) -> str:
    """按出现顺序去重拼接合同编号。"""
    seen: list[str] = []
    for raw in sources:
        for part in str(raw or "").replace("，", ",").split(","):
            no = part.strip()
            if no and no not in seen:
                seen.append(no)
    return "、".join(seen)


def _load_contract_nos_by_plan(
    db: Session, tenant_no: str, plan_ids: list[str]
) -> dict[str, list[str]]:
    if not plan_ids:
        return {}
    rows = (
        db.query(PlPlanContract)
        .filter(
            PlPlanContract.tenant_no == tenant_no,
            PlPlanContract.del_flag == 0,
            PlPlanContract.plan_id.in_(plan_ids),
        )
        .order_by(PlPlanContract.order_num.asc(), PlPlanContract.create_time.asc())
        .all()
    )
    out: dict[str, list[str]] = {pid: [] for pid in plan_ids}
    for row in rows:
        no = (row.contract_no or "").strip()
        if not no:
            continue
        bucket = out.setdefault(row.plan_id, [])
        if no not in bucket:
            bucket.append(no)
    return out

StopKind = str  # "load" | "delivery"

PLAN_TYPE_LABELS = {
    "direct": "直销单",
    "inbound": "入库单",
    "outbound": "出库单",
}

ARCHIVE_STATUS_LABELS = {
    "archived": "已归档",
    "unarchived": "未归档",
    "draft": "已暂存",
}

FOLLOW_STATUS_LABELS = {
    "wait_notice": "等通知",
    "pending_arrange": "待安排",
    "arranged": "已安排",
    "completed": "已完成",
    "cancelled": "已取消",
    "draft": "待安排",
    "processing": "进行中",
}

FREIGHT_TYPE_LABELS = {
    "by_weight": "按重量",
    "by_trip": "按趟",
}


def resolve_archive_list_io_column(kind: StopKind, customer_name: Optional[str]) -> str:
    """装货/送货段入库出库列方向（与前端 resolveArchiveListIoColumn 一致）。"""
    internal = is_internal_customer_name(customer_name)
    if kind == "load":
        return "outbound" if internal else "inbound"
    return "inbound" if internal else "outbound"


def _format_date_value(value: Optional[Union[date, datetime, str]]) -> str:
    if value is None:
        return ""
    return str(value)[:10]


def _format_plate_display(license_plate: Optional[str], trailer_no: Optional[str]) -> str:
    plate = (license_plate or "").strip()
    trailer = (trailer_no or "").strip()
    if not plate and not trailer:
        return ""
    if plate and trailer:
        return f"{plate}{trailer}"
    return plate or trailer


def _get_stop_archive_actual_qty(product: PlanStopProductRead, kind: StopKind) -> Optional[Decimal]:
    return product.inbound_qty if kind == "load" else product.outbound_qty


def _get_stop_archive_unit_price(product: PlanStopProductRead, kind: StopKind) -> Optional[Decimal]:
    return product.inbound_price if kind == "load" else product.outbound_price


def _get_stop_archive_amount(product: PlanStopProductRead, kind: StopKind) -> Optional[Decimal]:
    stored = product.inbound_amount if kind == "load" else product.outbound_amount
    if stored is not None:
        return stored
    qty = _get_stop_archive_actual_qty(product, kind)
    price = _get_stop_archive_unit_price(product, kind)
    if qty is not None and price is not None:
        return qty * price
    return None


def _stop_kind_label(kind: StopKind, delivery_seq: Optional[int] = None) -> str:
    if kind == "load":
        return "装货"
    return f"送货 #{delivery_seq or 1}"


def _stop_time(kind: StopKind, stop: Union[PlanLoadRead, PlanDeliveryRead]) -> str:
    if kind == "load":
        return _format_date_value(getattr(stop, "load_time", None))
    return _format_date_value(getattr(stop, "delivery_time", None))


def _load_legacy_products_by_plan(
    db: Session, tenant_no: str, plan_ids: list[str]
) -> dict[str, list[PlPlanProduct]]:
    if not plan_ids:
        return {}
    products = (
        db.query(PlPlanProduct)
        .filter(
            PlPlanProduct.tenant_no == tenant_no,
            PlPlanProduct.del_flag == 0,
            PlPlanProduct.plan_id.in_(plan_ids),
        )
        .order_by(PlPlanProduct.order_num)
        .all()
    )
    grouped: dict[str, list[PlPlanProduct]] = {}
    for product in products:
        grouped.setdefault(product.plan_id, []).append(product)
    return grouped


def _plan_common_export_fields(
    plan: PlPlan,
    transport: Optional[PlTransport],
    *,
    contract_nos: Optional[list[str]] = None,
) -> dict:
    freight_type_raw = transport.freight_type if transport else "by_weight"
    freight_type_label = FREIGHT_TYPE_LABELS.get(freight_type_raw, freight_type_raw or "")
    freight_price = transport.freight_price if transport else None
    freight_amount = plan.freight_amount
    if freight_amount is None and transport is not None:
        freight_amount = transport.freight_amount
    plan_date = _format_date_value(plan.plan_date)
    link_nos = contract_nos or []
    return {
        "planId": plan.id or "",
        "planNo": plan.plan_no or "",
        "contractNo": _unique_contract_nos(plan.contract_no, *link_nos),
        "planDate": plan_date,
        "archiveStatus": ARCHIVE_STATUS_LABELS.get(plan.archive_status or "", plan.archive_status or ""),
        "followStatus": FOLLOW_STATUS_LABELS.get(plan.follow_status or "", plan.follow_status or ""),
        "planType": PLAN_TYPE_LABELS.get(plan.plan_type or "", plan.plan_type or ""),
        "licensePlate": _format_plate_display(plan.license_plate, plan.trailer_no),
        "erpEntered": "是" if int(plan.erp_entered or 0) == 1 else "否",
        "freightType": freight_type_label,
        "freightPrice": freight_price,
        "freightAmount": freight_amount,
        "remark": plan.remark or "",
    }


def _empty_product_fields() -> dict:
    return {
        "productName": "",
        "model": "",
        "packageSpec": "",
        "unit": "",
        "planQty": None,
        "actualQty": None,
        "inboundQty": None,
        "outboundQty": None,
        "unitPrice": None,
        "amount": None,
    }


def _product_row_from_stop_product(
    product: PlanStopProductRead,
    kind: StopKind,
    customer_name: Optional[str],
) -> dict:
    actual_qty = _get_stop_archive_actual_qty(product, kind)
    unit_price = _get_stop_archive_unit_price(product, kind)
    amount = _get_stop_archive_amount(product, kind)
    io_col = resolve_archive_list_io_column(kind, customer_name)
    inbound_qty = actual_qty if io_col == "inbound" else None
    outbound_qty = actual_qty if io_col == "outbound" else None
    return {
        "productName": product.product_name or "",
        "model": product.model or "",
        "packageSpec": product.package_spec or "",
        "unit": product.unit or "",
        "planQty": product.plan_qty,
        "actualQty": actual_qty,
        "inboundQty": inbound_qty,
        "outboundQty": outbound_qty,
        "unitPrice": unit_price,
        "amount": amount,
    }


def _append_stop_product_rows(
    rows: list[dict],
    row_no: int,
    plan_common: dict,
    kind: StopKind,
    stop: Union[PlanLoadRead, PlanDeliveryRead],
    delivery_seq: Optional[int],
    products: list[PlanStopProductRead],
) -> int:
    customer_name = stop.customer_name or ""
    stop_common = {
        **plan_common,
        "stopKind": _stop_kind_label(kind, delivery_seq),
        "stopCustomer": customer_name,
        "stopAddress": stop.address or "",
        "stopTime": _stop_time(kind, stop),
    }
    product_rows = products or [None]
    for product in product_rows:
        row_no += 1
        row = {
            **stop_common,
            "rowNo": row_no,
            **_empty_product_fields(),
        }
        if product is not None:
            row.update(_product_row_from_stop_product(product, kind, customer_name))
        rows.append(row)
    return row_no


def _append_legacy_plan_product_rows(
    rows: list[dict],
    row_no: int,
    plan: PlPlan,
    plan_common: dict,
    products: list[PlPlanProduct],
) -> int:
    """无段表数据时回退计划产品级导出（兼容旧数据）。"""
    stop_common = {
        **plan_common,
        "stopKind": "",
        "stopCustomer": plan.customer_name or plan.supplier_name or "",
        "stopAddress": plan.destination_name or "",
        "stopTime": _format_date_value(plan.delivery_time),
    }
    product_rows = products or [None]
    for product in product_rows:
        row_no += 1
        row = {
            **stop_common,
            "rowNo": row_no,
            **_empty_product_fields(),
        }
        if product is not None:
            row.update(
                {
                    "productName": product.product_name or "",
                    "model": product.model or "",
                    "packageSpec": product.package_spec or "",
                    "unit": product.unit or "",
                    "planQty": product.quantity,
                    "actualQty": product.actual_qty,
                    "inboundQty": product.inbound_qty,
                    "outboundQty": product.outbound_qty,
                    "unitPrice": product.inbound_price or product.outbound_price,
                    "amount": product.inbound_amount or product.outbound_amount,
                }
            )
        rows.append(row)
    return row_no


def build_plan_archive_export_rows(
    db: Session,
    tenant_no: str,
    plans: list[PlPlan],
    transports_by_id: dict[str, PlTransport],
    *,
    stop_kind: Optional[str] = None,
    stop_customer_keywords: Optional[list] = None,
) -> list[dict]:
    """构建归档导出数据行：每计划 × 每段次 × 每段产品一行。

    stop_kind 为 load / delivery 时只导出该类段次，与归档列表段次漏斗一致。
    stop_customer_keywords 非空时只导出客户名称包含任一关键字的段次。
    """
    if not plans:
        return []

    include_load = stop_kind != "delivery"
    include_delivery = stop_kind != "load"
    keywords = [str(item).strip() for item in (stop_customer_keywords or []) if str(item).strip()]

    def _keep_customer(name: Optional[str]) -> bool:
        if not keywords:
            return True
        return customer_name_matches_keywords(name, keywords)

    plan_ids = [plan.id for plan in plans]
    stops_by_plan = load_plan_stops_grouped(db, tenant_no, plan_ids)
    legacy_products_by_plan = _load_legacy_products_by_plan(db, tenant_no, plan_ids)
    contract_nos_by_plan = _load_contract_nos_by_plan(db, tenant_no, plan_ids)

    rows: list[dict] = []
    row_no = 0
    for plan in plans:
        transport = transports_by_id.get(plan.transport_id) if plan.transport_id else None
        plan_common = _plan_common_export_fields(
            plan,
            transport,
            contract_nos=contract_nos_by_plan.get(plan.id, []),
        )
        load_read, delivery_reads = stops_by_plan.get(plan.id, (None, []))

        has_stops = load_read is not None or bool(delivery_reads)
        if not has_stops:
            if not include_load:
                continue
            legacy_customer = plan.customer_name or plan.supplier_name or ""
            if not _keep_customer(legacy_customer):
                continue
            row_no = _append_legacy_plan_product_rows(
                rows,
                row_no,
                plan,
                plan_common,
                legacy_products_by_plan.get(plan.id, []),
            )
            continue

        if include_load and load_read is not None and _keep_customer(load_read.customer_name):
            row_no = _append_stop_product_rows(
                rows,
                row_no,
                plan_common,
                "load",
                load_read,
                None,
                load_read.products or [],
            )
        if include_delivery:
            for idx, delivery_read in enumerate(delivery_reads):
                if not _keep_customer(delivery_read.customer_name):
                    continue
                delivery_seq = delivery_read.seq or (idx + 1)
                row_no = _append_stop_product_rows(
                    rows,
                    row_no,
                    plan_common,
                    "delivery",
                    delivery_read,
                    delivery_seq,
                    delivery_read.products or [],
                )

    return rows