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 _product_match_key(product: PlanStopProductRead) -> tuple[str, str]:
    return (
        str(product.product_id or "").strip(),
        str(product.package_spec or "").strip(),
    )


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_pair_side_fields(*, side: str) -> dict:
    """side: load | delivery"""
    prefix = "load" if side == "load" else "delivery"
    return {
        f"{prefix}Customer": "",
        f"{prefix}Address": "",
        f"{prefix}Time": "",
        f"{prefix}PlanQty": None,
        f"{prefix}ActualQty": None,
        f"{prefix}InboundQty": None,
        f"{prefix}OutboundQty": None,
        f"{prefix}UnitPrice": None,
        f"{prefix}Amount": None,
    }


def _empty_product_identity_fields() -> dict:
    return {
        "productName": "",
        "model": "",
        "packageSpec": "",
        "unit": "",
    }


def _side_qty_fields(
    product: PlanStopProductRead,
    kind: StopKind,
    customer_name: Optional[str],
    *,
    side: str,
) -> dict:
    prefix = "load" if side == "load" else "delivery"
    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 {
        f"{prefix}PlanQty": product.plan_qty,
        f"{prefix}ActualQty": actual_qty,
        f"{prefix}InboundQty": inbound_qty,
        f"{prefix}OutboundQty": outbound_qty,
        f"{prefix}UnitPrice": unit_price,
        f"{prefix}Amount": amount,
    }


def _product_identity_from_stop(product: PlanStopProductRead) -> dict:
    return {
        "productName": product.product_name or "",
        "model": product.model or "",
        "packageSpec": product.package_spec or "",
        "unit": product.unit or "",
    }


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 ""
    side = "load" if kind == "load" else "delivery"
    product_rows = products or [None]
    for product in product_rows:
        row_no += 1
        row = {
            **plan_common,
            "rowNo": row_no,
            "deliveryStopKind": _stop_kind_label(kind, delivery_seq) if kind == "delivery" else "",
            **_empty_pair_side_fields(side="load"),
            **_empty_pair_side_fields(side="delivery"),
            **_empty_product_identity_fields(),
        }
        row[f"{side}Customer"] = customer_name
        row[f"{side}Address"] = stop.address or ""
        row[f"{side}Time"] = _stop_time(kind, stop)
        # 兼容旧测试/调用方字段名
        row["stopKind"] = _stop_kind_label(kind, delivery_seq)
        row["stopCustomer"] = customer_name
        row["stopAddress"] = stop.address or ""
        row["stopTime"] = _stop_time(kind, stop)
        if product is not None:
            row.update(_product_identity_from_stop(product))
            side_fields = _side_qty_fields(product, kind, customer_name, side=side)
            row.update(side_fields)
            # 兼容旧单侧字段
            row["planQty"] = side_fields[f"{side}PlanQty"]
            row["actualQty"] = side_fields[f"{side}ActualQty"]
            row["inboundQty"] = side_fields[f"{side}InboundQty"]
            row["outboundQty"] = side_fields[f"{side}OutboundQty"]
            row["unitPrice"] = side_fields[f"{side}UnitPrice"]
            row["amount"] = side_fields[f"{side}Amount"]
        else:
            row.update(
                {
                    "planQty": None,
                    "actualQty": None,
                    "inboundQty": None,
                    "outboundQty": None,
                    "unitPrice": None,
                    "amount": None,
                }
            )
        rows.append(row)
    return row_no


def _append_paired_product_rows(
    rows: list[dict],
    row_no: int,
    plan_common: dict,
    load: PlanLoadRead,
    delivery: PlanDeliveryRead,
    delivery_seq: int,
) -> int:
    """装货×送货按 productId+packageSpec 内连接，一产品一行。"""
    load_customer = load.customer_name or ""
    delivery_customer = delivery.customer_name or ""
    load_by_key: dict[tuple[str, str], PlanStopProductRead] = {}
    for product in load.products or []:
        key = _product_match_key(product)
        if not key[0]:
            continue
        # 同键多条时保留首条
        load_by_key.setdefault(key, product)

    for product_d in delivery.products or []:
        key = _product_match_key(product_d)
        if not key[0]:
            continue
        product_l = load_by_key.get(key)
        if product_l is None:
            continue
        row_no += 1
        load_fields = _side_qty_fields(product_l, "load", load_customer, side="load")
        delivery_fields = _side_qty_fields(
            product_d, "delivery", delivery_customer, side="delivery"
        )
        row = {
            **plan_common,
            "rowNo": row_no,
            "deliveryStopKind": _stop_kind_label("delivery", delivery_seq),
            "loadCustomer": load_customer,
            "loadAddress": load.address or "",
            "loadTime": _stop_time("load", load),
            "deliveryCustomer": delivery_customer,
            "deliveryAddress": delivery.address or "",
            "deliveryTime": _stop_time("delivery", delivery),
            **_product_identity_from_stop(product_l),
            **load_fields,
            **delivery_fields,
        }
        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:
    """无段表数据时回退计划产品级导出（兼容旧数据）。"""
    product_rows = products or [None]
    for product in product_rows:
        row_no += 1
        row = {
            **plan_common,
            "rowNo": row_no,
            "deliveryStopKind": "",
            **_empty_pair_side_fields(side="load"),
            **_empty_pair_side_fields(side="delivery"),
            **_empty_product_identity_fields(),
            "loadCustomer": plan.supplier_name or plan.customer_name or "",
            "loadAddress": "",
            "deliveryCustomer": plan.customer_name or "",
            "deliveryAddress": plan.destination_name or "",
            "deliveryTime": _format_date_value(plan.delivery_time),
            # 兼容旧字段
            "stopKind": "",
            "stopCustomer": plan.customer_name or plan.supplier_name or "",
            "stopAddress": plan.destination_name or "",
            "stopTime": _format_date_value(plan.delivery_time),
            "planQty": None,
            "actualQty": None,
            "inboundQty": None,
            "outboundQty": None,
            "unitPrice": None,
            "amount": None,
        }
        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 "",
                    "loadPlanQty": product.quantity,
                    "loadActualQty": product.actual_qty,
                    "loadInboundQty": product.inbound_qty,
                    "loadOutboundQty": product.outbound_qty,
                    "loadUnitPrice": product.inbound_price or product.outbound_price,
                    "loadAmount": product.inbound_amount or product.outbound_amount,
                    "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"
    pair_mode = include_load and include_delivery
    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 pair_mode:
            if load_read is None:
                continue
            for idx, delivery_read in enumerate(delivery_reads):
                if keywords and not (
                    _keep_customer(load_read.customer_name)
                    or _keep_customer(delivery_read.customer_name)
                ):
                    continue
                delivery_seq = delivery_read.seq or (idx + 1)
                row_no = _append_paired_product_rows(
                    rows,
                    row_no,
                    plan_common,
                    load_read,
                    delivery_read,
                    delivery_seq,
                )
            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