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

"""运输计划 · 入库单 / 出库单（安排运输落单、现场回传、作废）"""
from __future__ import annotations

import json
import secrets
from dataclasses import dataclass
from datetime import date, datetime
from decimal import Decimal
from typing import Any, Optional, Union

from sqlalchemy.orm import Session

from app.core.config import settings
from app.core.utils import generate_uuid
from app.models.crm_models import CrmProduct, CrmProductBrand
from app.models.db_models import DbInboundOutbound
from app.models.pl_models import (
    PlInboundOrder,
    PlInboundOrderItem,
    PlInboundOrderItemBrandQty,
    PlOutboundOrder,
    PlOutboundOrderItem,
    PlOutboundOrderItemBrandQty,
    PlPlan,
    PlPlanProduct,
    PlPlanProductBrandQty,
    PlTransport,
    PlTransportDbLink,
)
from app.services.db.mappers import parse_date_str
from app.services.pl_dibang_bridge_service import _product_quantity, resolve_partner_company
from app.services.pl_feedback_brand_qty import (
    format_brand_qty_summary,
    persist_io_item_brand_qtys,
    require_brand_qtys_if_needed,
)
from app.services.pl_product_display import apply_product_alias_display, load_crm_product_alias_maps


OrderHeader = Union[PlInboundOrder, PlOutboundOrder]
OrderItem = Union[PlInboundOrderItem, PlOutboundOrderItem]


@dataclass
class IoOrderCreateResult:
    kind: str  # inbound | outbound
    order_id: str
    order_no: str
    site_access_token: str
    site_url: str


def build_io_order_site_url(kind: str, token: str) -> str:
    origin = (settings.PUBLIC_WEB_ORIGIN or "").rstrip("/")
    path = "inbound" if kind == "inbound" else "outbound"
    return f"{origin}/#/public/pl/{path}/{token}"


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, str):
        return parse_date_str(plan_date) or date.today()
    if isinstance(plan_date, date):
        return plan_date
    return date.today()


def _next_order_no(db: Session, tenant_no: str, prefix: str, model: type) -> str:
    day = datetime.now().strftime("%Y%m%d")
    like = f"{prefix}{day}%"
    last = (
        db.query(model.order_no)
        .filter(model.tenant_no == tenant_no, model.order_no.like(like), model.del_flag == 0)
        .order_by(model.order_no.desc())
        .first()
    )
    seq = 1
    if last and last[0] and len(last[0]) >= len(prefix) + 8 + 4:
        try:
            seq = int(last[0][-4:]) + 1
        except ValueError:
            seq = 1
    return f"{prefix}{day}{seq:04d}"


def _dibang_qty_by_plan_product(
    db: Session, tenant_no: str, transport_id: str
) -> dict[str, Decimal]:
    links = (
        db.query(PlTransportDbLink)
        .filter(
            PlTransportDbLink.tenant_no == tenant_no,
            PlTransportDbLink.transport_id == transport_id,
            PlTransportDbLink.del_flag == 0,
        )
        .all()
    )
    if not links:
        return {}
    inv_ids = [x.db_inventory_id for x in links if x.db_inventory_id]
    if not inv_ids:
        return {}
    inv_rows = (
        db.query(DbInboundOutbound)
        .filter(DbInboundOutbound.id.in_(inv_ids), DbInboundOutbound.del_flag == 0)
        .all()
    )
    inv_map = {r.id: r for r in inv_rows}
    out: dict[str, Decimal] = {}
    for link in links:
        inv = inv_map.get(link.db_inventory_id)
        if not inv or inv.quantity is None:
            continue
        out[link.plan_product_id] = Decimal(str(inv.quantity))
    return out


def create_io_order_for_transport(
    db: Session,
    *,
    tenant_no: str,
    plan: PlPlan,
    products: list[PlPlanProduct],
    transport: PlTransport,
    user_id: Optional[str],
    dept_id: str,
    after_dibang: bool = False,
) -> Optional[IoOrderCreateResult]:
    """安排运输后创建入库/出库单。直销返回 None。"""
    if plan.plan_type not in ("inbound", "outbound"):
        return None

    kind = plan.plan_type
    header_model = PlInboundOrder if kind == "inbound" else PlOutboundOrder
    item_model = PlInboundOrderItem if kind == "inbound" else PlOutboundOrderItem
    prefix = "RK" if kind == "inbound" else "CK"

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

    token = secrets.token_urlsafe(32)[:64]
    order_no = _next_order_no(db, tenant_no, prefix, header_model)
    plan_date = _normalize_plan_date(plan)
    license_plate = (transport.license_plate or plan.license_plate or "").strip()
    dibang_map = _dibang_qty_by_plan_product(db, tenant_no, transport.id) if after_dibang else {}

    header = header_model(
        id=generate_uuid(),
        order_no=order_no,
        plan_id=plan.id,
        transport_id=transport.id,
        status="in_progress",
        partner_company=partner,
        vehicle_no=license_plate or None,
        plan_no=(plan.plan_no or "").strip() or None,
        plan_date=plan_date,
        site_access_token=token,
        create_by=user_id,
        create_dept_id=dept_id or "",
        tenant_no=tenant_no,
    )
    db.add(header)

    for idx, product in enumerate(products):
        plan_qty = Decimal(str(_product_quantity(product)))
        actual = dibang_map.get(product.id)
        qty_source = "dibang" if actual is not None else "plan"
        if actual is None:
            actual = plan_qty
        db.add(
            item_model(
                id=generate_uuid(),
                order_id=header.id,
                plan_product_id=product.id,
                product_id=product.product_id,
                product_name=product.product_name,
                specification=product.model,
                unit=product.unit,
                plan_qty=plan_qty,
                actual_qty=actual,
                qty_source=qty_source,
                order_num=idx + 1,
                create_by=user_id,
                create_dept_id=dept_id or "",
                tenant_no=tenant_no,
            )
        )

    db.flush()
    return IoOrderCreateResult(
        kind=kind,
        order_id=header.id,
        order_no=order_no,
        site_access_token=token,
        site_url=build_io_order_site_url(kind, token),
    )


def apply_dibang_prefill_to_order(
    db: Session,
    *,
    tenant_no: str,
    plan: PlPlan,
    transport: PlTransport,
) -> None:
    """地磅单生成后，用数量预填尚未被现场改过的明细 actual_qty。"""
    if plan.plan_type not in ("inbound", "outbound"):
        return
    header_model = PlInboundOrder if plan.plan_type == "inbound" else PlOutboundOrder
    item_model = PlInboundOrderItem if plan.plan_type == "inbound" else PlOutboundOrderItem
    header = (
        db.query(header_model)
        .filter(
            header_model.tenant_no == tenant_no,
            header_model.plan_id == plan.id,
            header_model.transport_id == transport.id,
            header_model.status != "voided",
            header_model.del_flag == 0,
        )
        .order_by(header_model.create_time.desc())
        .first()
    )
    if not header:
        return
    dibang_map = _dibang_qty_by_plan_product(db, tenant_no, transport.id)
    if not dibang_map:
        return
    items = (
        db.query(item_model)
        .filter(item_model.order_id == header.id, item_model.del_flag == 0)
        .all()
    )
    for item in items:
        qty = dibang_map.get(item.plan_product_id)
        if qty is None:
            continue
        if item.qty_source == "manual":
            continue
        item.actual_qty = qty
        item.qty_source = "dibang"
    db.flush()


def void_io_orders_for_plan(
    db: Session,
    *,
    tenant_no: str,
    plan_id: str,
    user_id: Optional[str],
) -> None:
    for model in (PlInboundOrder, PlOutboundOrder):
        rows = (
            db.query(model)
            .filter(
                model.tenant_no == tenant_no,
                model.plan_id == plan_id,
                model.status != "voided",
                model.del_flag == 0,
            )
            .all()
        )
        for row in rows:
            row.status = "voided"
            row.update_by = user_id
    db.flush()


def get_active_io_order_for_plan(
    db: Session, tenant_no: str, plan: PlPlan
) -> Optional[tuple[str, OrderHeader]]:
    if plan.plan_type == "inbound":
        row = (
            db.query(PlInboundOrder)
            .filter(
                PlInboundOrder.tenant_no == tenant_no,
                PlInboundOrder.plan_id == plan.id,
                PlInboundOrder.del_flag == 0,
            )
            .order_by(PlInboundOrder.create_time.desc())
            .first()
        )
        return ("inbound", row) if row else None
    if plan.plan_type == "outbound":
        row = (
            db.query(PlOutboundOrder)
            .filter(
                PlOutboundOrder.tenant_no == tenant_no,
                PlOutboundOrder.plan_id == plan.id,
                PlOutboundOrder.del_flag == 0,
            )
            .order_by(PlOutboundOrder.create_time.desc())
            .first()
        )
        return ("outbound", row) if row else None
    return None


def parse_feedback_images(raw: Optional[str]) -> list[str]:
    if not raw:
        return []
    try:
        parsed = json.loads(raw)
        if isinstance(parsed, list):
            return [str(item) for item in parsed if item]
    except (TypeError, json.JSONDecodeError):
        pass
    return [part.strip() for part in raw.split(",") if part.strip()]


def dump_feedback_images(images: list[str]) -> str:
    return json.dumps(images, ensure_ascii=False)


_VALID_SITE_FEEDBACK_STATUSES = frozenset({"in_progress", "completed", "voided"})

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


def _io_item_brand_qty_model(kind: str):
    if kind == "inbound":
        return PlInboundOrderItemBrandQty
    return PlOutboundOrderItemBrandQty


def update_io_order_items_qty_and_brand(
    db: Session,
    *,
    kind: str,
    tenant_no: str,
    order_id: str,
    items: list[OrderItem],
    item_qty: dict[str, Decimal],
    item_brand_qtys: Optional[dict[str, list[dict]]] = None,
    enforce_brand: bool = False,
    user_id: Optional[str] = None,
) -> None:
    """更新明细实际数量，并按需校验/持久化现场品牌数量。"""
    updating = [it for it in items if it.id in item_qty]
    if not updating:
        return

    plan_product_ids = [it.plan_product_id for it in updating if it.plan_product_id]
    products_by_id: dict[str, PlPlanProduct] = {}
    if plan_product_ids:
        rows = (
            db.query(PlPlanProduct)
            .filter(
                PlPlanProduct.id.in_(plan_product_ids),
                PlPlanProduct.tenant_no == tenant_no,
                PlPlanProduct.del_flag == 0,
            )
            .all()
        )
        products_by_id = {r.id: r for r in rows}
        # 明细仍指向已软删计划产品行时，用原行品牌约定（避免 brand_mode 空导致只删不写）
        missing_ids = [pid for pid in plan_product_ids if pid and pid not in products_by_id]
        if missing_ids:
            stale_rows = (
                db.query(PlPlanProduct)
                .filter(
                    PlPlanProduct.id.in_(missing_ids),
                    PlPlanProduct.tenant_no == tenant_no,
                )
                .all()
            )
            for row in stale_rows:
                products_by_id.setdefault(row.id, row)

    brand_model = _io_item_brand_qty_model(kind)

    for item in updating:
        actual_qty = item_qty[item.id]
        plan_product = products_by_id.get(item.plan_product_id or "")
        brand_mode = plan_product.brand_mode if plan_product else None
        required_brand_id = plan_product.brand_id if plan_product else None

        brand_touched = item_brand_qtys is not None and item.id in item_brand_qtys
        if brand_touched:
            rows = item_brand_qtys[item.id]
            # 客户端带了品牌明细时，即使计划行 brand_mode 丢失也按 any 规则校验/落库
            if not brand_mode and rows:
                brand_mode = "any"
        elif enforce_brand:
            rows = []
        else:
            rows = None

        if rows is not None:
            require_brand_qtys_if_needed(
                brand_mode=brand_mode,
                required_brand_id=required_brand_id,
                actual_qty=actual_qty,
                rows=rows,
                enforce=enforce_brand,
            )

        item.actual_qty = actual_qty
        item.qty_source = "manual"
        item.update_by = user_id

        if brand_touched:
            persist_io_item_brand_qtys(
                db,
                model=brand_model,
                tenant_no=tenant_no,
                order_id=order_id,
                order_item_id=item.id,
                plan_product_id=item.plan_product_id,
                brand_mode=brand_mode,
                rows=item_brand_qtys[item.id],
                user_id=user_id,
            )


def apply_io_order_stop_site_feedbacks(
    db: Session,
    *,
    kind: str,
    header: OrderHeader,
    items: list[OrderItem],
    stops: list[dict[str, Any]],
    enforce_brand: bool = False,
    user_id: Optional[str] = None,
) -> tuple[dict[str, Decimal], Optional[dict[str, list[dict]]], list[str]]:
    """保存按段现场回传，并汇总为出入库明细数量。"""
    from collections import defaultdict

    from app.services.pl_feedback_brand_qty import require_brand_qtys_if_needed
    from app.services.pl_internal_customer import is_internal_customer_name
    from app.services.pl_transport_stop_feedback_service import (
        load_io_order_site_stop_segments,
        save_site_stop_feedback,
    )

    segments = load_io_order_site_stop_segments(
        db,
        tenant_no=header.tenant_no,
        plan_id=header.plan_id,
        transport_id=header.transport_id,
    )
    segment_map = {(seg.stop_type, seg.stop_id): seg for seg in segments}
    plan_products = (
        db.query(PlPlanProduct)
        .filter(
            PlPlanProduct.plan_id == header.plan_id,
            PlPlanProduct.tenant_no == header.tenant_no,
            PlPlanProduct.del_flag == 0,
        )
        .all()
    )
    plan_products_by_id = {row.id: row for row in plan_products}

    qty_by_plan_product: dict[str, Decimal] = defaultdict(lambda: Decimal("0"))
    brand_by_plan_product: dict[str, dict[str, dict[str, Any]]] = defaultdict(dict)
    merged_images: list[str] = []

    default_business_type = PLAN_TYPE_BUSINESS_LABEL.get(kind, kind)

    for stop_payload in stops:
        stop_type = (stop_payload.get("stopType") or stop_payload.get("stop_type") or "").strip()
        stop_id = (stop_payload.get("stopId") or stop_payload.get("stop_id") or "").strip()
        if stop_type not in ("load", "delivery") or not stop_id:
            raise ValueError("停靠段参数无效")
        segment = segment_map.get((stop_type, stop_id))
        if not segment:
            raise ValueError("停靠段不存在或不可回传")

        needs_product_qty = is_internal_customer_name(segment.customer_name)
        allowed_products = {row.stop_product_id: row for row in segment.products}
        products_payload: list[dict[str, Any]] | None = [] if needs_product_qty else None
        if needs_product_qty:
            for item in stop_payload.get("products") or []:
                stop_product_id = (
                    item.get("stopProductId") or item.get("stop_product_id") or ""
                ).strip()
                stop_row = allowed_products.get(stop_product_id)
                if not stop_row:
                    raise ValueError("产品不属于当前停靠段")
                actual_qty = Decimal(str(item.get("actualQty", item.get("actual_qty") or 0)))
                brand_qtys = item.get("brandQtys") or item.get("brand_qtys") or []
                plan_product = plan_products_by_id.get(stop_row.plan_product_id)
                require_brand_qtys_if_needed(
                    brand_mode=plan_product.brand_mode if plan_product else None,
                    required_brand_id=plan_product.brand_id if plan_product else None,
                    actual_qty=actual_qty,
                    rows=brand_qtys if isinstance(brand_qtys, list) else [],
                    enforce=enforce_brand,
                )
                row_payload: dict[str, Any] = {
                    "stopProductId": stop_product_id,
                    "actualQty": float(actual_qty),
                }
                if brand_qtys:
                    row_payload["brandQtys"] = brand_qtys
                products_payload.append(row_payload)

                pp_id = stop_row.plan_product_id
                if actual_qty > 0:
                    qty_by_plan_product[pp_id] += actual_qty
                if brand_qtys:
                    for brand_row in brand_qtys:
                        if not isinstance(brand_row, dict):
                            continue
                        brand_id = str(brand_row.get("brandId") or brand_row.get("brand_id") or "")
                        qty = Decimal(str(brand_row.get("qty") or 0))
                        if qty <= 0:
                            continue
                        existing = brand_by_plan_product[pp_id].get(brand_id)
                        if existing:
                            existing["qty"] = float(Decimal(str(existing["qty"])) + qty)
                        else:
                            brand_by_plan_product[pp_id][brand_id] = {
                                "brandId": brand_id,
                                "brandName": brand_row.get("brandName") or brand_row.get("brand_name"),
                                "qty": float(qty),
                            }
        elif stop_payload.get("products"):
            raise ValueError("该停靠段无需填写产品数量")

        attachment_urls = stop_payload.get("attachmentUrls") or stop_payload.get("attachment_urls") or []
        merged_images.extend(
            [str(url).strip() for url in attachment_urls if str(url).strip()]
        )
        save_site_stop_feedback(
            db,
            tenant_no=header.tenant_no,
            transport_id=header.transport_id,
            plan_id=header.plan_id,
            stop_type=stop_type,
            stop_id=stop_id,
            business_type=segment.business_type or default_business_type,
            attachment_urls=list(attachment_urls),
            products=products_payload,
            user_id=user_id,
            require_qty_before_attachments=needs_product_qty,
        )

    item_by_plan_product = {it.plan_product_id: it for it in items if it.plan_product_id}
    qty_map: dict[str, Decimal] = {}
    brand_qty_map: dict[str, list[dict]] = {}
    for plan_product_id, qty in qty_by_plan_product.items():
        item = item_by_plan_product.get(plan_product_id)
        if item:
            qty_map[item.id] = qty
    for plan_product_id, brand_rows in brand_by_plan_product.items():
        item = item_by_plan_product.get(plan_product_id)
        if item and brand_rows:
            brand_qty_map[item.id] = list(brand_rows.values())

    return qty_map, brand_qty_map or None, merged_images


def apply_site_feedback(
    db: Session,
    *,
    kind: str,
    header: OrderHeader,
    items: list[OrderItem],
    item_qty: dict[str, Decimal],
    remark: Optional[str],
    images: list[str],
    status: str,
    item_brand_qtys: Optional[dict[str, list[dict]]] = None,
    enforce_brand: bool = False,
    user_id: Optional[str] = None,
) -> None:
    if header.status == "voided":
        raise ValueError("单据已作废，无法提交")
    if status not in _VALID_SITE_FEEDBACK_STATUSES:
        raise ValueError(f"无效状态: {status}")
    now = datetime.now()
    update_io_order_items_qty_and_brand(
        db,
        kind=kind,
        tenant_no=header.tenant_no,
        order_id=header.id,
        items=items,
        item_qty=item_qty,
        item_brand_qtys=item_brand_qtys,
        enforce_brand=enforce_brand,
        user_id=user_id,
    )
    header.site_feedback_remark = (remark or "").strip() or None
    header.site_feedback_images = dump_feedback_images(images) if images else header.site_feedback_images
    header.site_feedback_time = now
    if not header.site_first_submitted_at:
        header.site_first_submitted_at = now
    header.status = status
    header.update_by = user_id
    db.flush()


CRM_PRODUCT_UNIT = "crm_product_unit"


def resolve_unit_label(unit_map: dict[str, str], unit_code: Optional[str]) -> str:
    code = (unit_code or "").strip()
    if not code:
        return "吨"
    return unit_map.get(code, code)


def merge_plan_logistics_into_order(data: dict, plan: Optional[PlPlan]) -> None:
    if not plan:
        data.setdefault("businessType", None)
        data.setdefault("planRemark", None)
        data.setdefault("pickupAddress", None)
        data.setdefault("pickupLng", None)
        data.setdefault("pickupLat", None)
        data.setdefault("pickupTime", None)
        data.setdefault("pickupContact", None)
        data.setdefault("pickupNotes", None)
        data.setdefault("destinationName", None)
        data.setdefault("destinationAddress", None)
        data.setdefault("destinationLng", None)
        data.setdefault("destinationLat", None)
        data.setdefault("deliveryTime", None)
        data.setdefault("deliveryContact", None)
        data.setdefault("destinationNotes", None)
        return
    data["businessType"] = PLAN_TYPE_BUSINESS_LABEL.get(plan.plan_type, plan.plan_type)
    data["planRemark"] = (plan.remark or "").strip() or None
    data["pickupAddress"] = plan.pickup_address
    data["pickupLng"] = plan.pickup_lng
    data["pickupLat"] = plan.pickup_lat
    data["pickupTime"] = plan.pickup_time
    data["pickupContact"] = plan.pickup_contact
    data["pickupNotes"] = plan.pickup_notes
    data["destinationName"] = plan.destination_name
    data["destinationAddress"] = plan.destination_address or plan.destination_name
    data["destinationLng"] = plan.destination_lng
    data["destinationLat"] = plan.destination_lat
    data["deliveryTime"] = plan.delivery_time
    data["deliveryContact"] = plan.delivery_contact
    data["destinationNotes"] = plan.destination_notes


def merge_plan_product_into_item(
    item: dict,
    product: Optional[PlPlanProduct],
    *,
    unit_map: dict[str, str],
    crm_product_no: Optional[str] = None,
) -> None:
    if not product:
        item.setdefault("productNo", None)
        item.setdefault("categoryName", None)
        item.setdefault("packageSpec", None)
        item.setdefault("deliveryMethod", None)
        item.setdefault("remark", None)
        item["unitLabel"] = resolve_unit_label(unit_map, item.get("unit"))
        return
    # 优先 CRM 产品编码；计划冗余 product_no 可能被误存为 product_id
    product_id = (product.product_id or item.get("productId") or item.get("product_id") or "").strip()
    plan_no = (product.product_no or "").strip()
    resolved_no = (crm_product_no or "").strip()
    if not resolved_no and plan_no and plan_no != product_id:
        resolved_no = plan_no
    # CRM 缺失时仍回退计划 product_no，避免前端编码列空白
    if not resolved_no and plan_no:
        resolved_no = plan_no
    item["productNo"] = resolved_no or None
    item["categoryName"] = product.category_name
    item["packageSpec"] = product.package_spec or item.get("packageSpec")
    item["deliveryMethod"] = product.delivery_method
    item["remark"] = (product.remark or "").strip() or item.get("remark")
    if not item.get("specification") and product.model:
        item["specification"] = product.model
    item["unitLabel"] = resolve_unit_label(unit_map, item.get("unit") or product.unit)
    # 无品牌拆分量时，回退计划产品上的品牌快照
    if product.brand_name and not item.get("brandSummary"):
        item["brandName"] = product.brand_name
    if product.brand_mode:
        item.setdefault("brandMode", product.brand_mode)


def _format_io_product_summary_line(item: dict) -> str:
    name = (item.get("productName") or "").strip()
    if not name:
        return ""
    bits: list[str] = []
    brand = (item.get("brandSummary") or item.get("brandName") or "").strip()
    if brand:
        bits.append(f"品牌 {brand}")
    spec = (item.get("specification") or "").strip()
    if spec:
        bits.append(f"型号/规格 {spec}")
    unit = (item.get("unitLabel") or item.get("unit") or "").strip()
    if unit:
        bits.append(f"单位 {unit}")
    package = (item.get("packageSpec") or "").strip()
    if package:
        bits.append(f"包装规格 {package}")
    delivery = (item.get("deliveryMethod") or "").strip()
    if delivery:
        delivery_labels = {
            "self_pick": "山峡自提",
            "one_ticket": "一票制",
            "customer_pick": "客户自提",
        }
        bits.append(f"送货方式 {delivery_labels.get(delivery, delivery)}")
    return f"{name}，{'，'.join(bits)}" if bits else name


def enrich_io_items_with_brand_feedback(
    db: Session,
    *,
    kind: str,
    tenant_no: str,
    items: list[dict],
    products_by_id: dict[str, PlPlanProduct],
    prefer_site_brand_summary: bool = False,
) -> None:
    """读时附加品牌模式、CRM 品牌选项与现场品牌数量子表。"""
    if not items:
        return

    crm_product_ids: set[str] = set()
    order_item_ids: list[str] = []
    for it in items:
        item_id = (it.get("id") or "").strip()
        if item_id:
            order_item_ids.append(item_id)
        plan_product_id = (it.get("planProductId") or it.get("plan_product_id") or "").strip()
        pp = products_by_id.get(plan_product_id)
        product_id = (
            (pp.product_id if pp else None) or it.get("productId") or it.get("product_id") or ""
        ).strip()
        if product_id:
            crm_product_ids.add(product_id)

    brand_options_by_product: dict[str, list[dict]] = {}
    if crm_product_ids:
        brand_rows = (
            db.query(CrmProductBrand)
            .filter(
                CrmProductBrand.product_id.in_(list(crm_product_ids)),
                CrmProductBrand.tenant_no == tenant_no,
                CrmProductBrand.status == "on",
                CrmProductBrand.del_flag == 0,
            )
            .order_by(CrmProductBrand.order_num.asc(), CrmProductBrand.create_time.asc())
            .all()
        )
        for row in brand_rows:
            brand_options_by_product.setdefault(row.product_id, []).append(
                {"brandId": row.id, "brandName": row.brand_name}
            )

    site_rows_by_item: dict[str, list[dict]] = {}
    if order_item_ids:
        brand_model = _io_item_brand_qty_model(kind)
        site_rows = (
            db.query(brand_model)
            .filter(
                brand_model.order_item_id.in_(order_item_ids),
                brand_model.tenant_no == tenant_no,
                brand_model.del_flag == 0,
            )
            .order_by(brand_model.order_num.asc())
            .all()
        )
        for row in site_rows:
            site_rows_by_item.setdefault(row.order_item_id, []).append(
                {
                    "brandId": row.brand_id or "",
                    "brandName": row.brand_name,
                    "qty": row.qty,
                }
            )

    for it in items:
        plan_product_id = (it.get("planProductId") or it.get("plan_product_id") or "").strip()
        pp = products_by_id.get(plan_product_id)
        product_id = (
            (pp.product_id if pp else None) or it.get("productId") or it.get("product_id") or ""
        ).strip()

        if pp:
            it["brandMode"] = pp.brand_mode
            it["brandId"] = pp.brand_id
            if pp.brand_name:
                it["brandName"] = pp.brand_name
        else:
            it.setdefault("brandMode", None)
            it.setdefault("brandId", None)

        options = brand_options_by_product.get(product_id, [])
        it["brandOptions"] = options if options else None

        site_qtys = site_rows_by_item.get(it.get("id") or "", [])
        it["brandQtys"] = site_qtys if site_qtys else None

        if prefer_site_brand_summary:
            site_summary = format_brand_qty_summary(site_qtys)
            if site_summary:
                it["brandSummary"] = site_summary


def enrich_io_order_dict(
    db: Session,
    data: dict,
    *,
    tenant_no: str,
    plan_id: Optional[str],
    items: list[Any] | None,
    public_unit_as_label: bool = False,
    kind: str = "inbound",
) -> dict:
    """读时附加计划取送货与产品扩展字段。"""
    from app.utils.excel_export import load_dict_label_maps

    plan = None
    if plan_id:
        plan = (
            db.query(PlPlan)
            .filter(
                PlPlan.id == plan_id,
                PlPlan.tenant_no == tenant_no,
                PlPlan.del_flag == 0,
            )
            .first()
        )
    merge_plan_logistics_into_order(data, plan)

    unit_map = load_dict_label_maps(
        db, [CRM_PRODUCT_UNIT], tenant_no=tenant_no
    ).get(CRM_PRODUCT_UNIT, {})

    raw_items = data.get("items")
    if raw_items is None:
        return data

    product_ids = [
        (it.get("planProductId") or it.get("plan_product_id") or "").strip()
        for it in raw_items
        if (it.get("planProductId") or it.get("plan_product_id") or "").strip()
    ]
    products_by_id: dict[str, PlPlanProduct] = {}
    if product_ids:
        rows = (
            db.query(PlPlanProduct)
            .filter(
                PlPlanProduct.id.in_(product_ids),
                PlPlanProduct.tenant_no == tenant_no,
                PlPlanProduct.del_flag == 0,
            )
            .all()
        )
        products_by_id = {r.id: r for r in rows}
        # 明细仍挂已软删计划产品时，保留原行用于品牌约定与 CRM 产品定位（勿误用其它在用产品）
        missing_ids = [pid for pid in product_ids if pid and pid not in products_by_id]
        if missing_ids:
            stale_rows = (
                db.query(PlPlanProduct)
                .filter(
                    PlPlanProduct.id.in_(missing_ids),
                    PlPlanProduct.tenant_no == tenant_no,
                )
                .all()
            )
            for row in stale_rows:
                products_by_id.setdefault(row.id, row)

    # 计划产品行可能被编辑重建导致 plan_product_id 失效：按计划再拉全量做 product_id / 序号兜底
    plan_products: list[PlPlanProduct] = []
    products_by_crm_id: dict[str, PlPlanProduct] = {}
    if plan_id:
        plan_products = (
            db.query(PlPlanProduct)
            .filter(
                PlPlanProduct.plan_id == plan_id,
                PlPlanProduct.tenant_no == tenant_no,
                PlPlanProduct.del_flag == 0,
            )
            .order_by(PlPlanProduct.order_num.asc(), PlPlanProduct.create_time.asc())
            .all()
        )
        for row in plan_products:
            products_by_id.setdefault(row.id, row)
            crm_key = (row.product_id or "").strip()
            if crm_key and crm_key not in products_by_crm_id:
                products_by_crm_id[crm_key] = row

    crm_ids = {
        (pp.product_id or "").strip()
        for pp in products_by_id.values()
        if (pp.product_id or "").strip()
    }
    for it in raw_items:
        pid = (it.get("productId") or it.get("product_id") or "").strip()
        if pid:
            crm_ids.add(pid)
    crm_no_by_id: dict[str, str] = {}
    if crm_ids:
        crm_rows = (
            db.query(CrmProduct.id, CrmProduct.product_no)
            .filter(
                CrmProduct.id.in_(list(crm_ids)),
                CrmProduct.tenant_no == tenant_no,
                CrmProduct.del_flag == 0,
            )
            .all()
        )
        for row in crm_rows:
            no = (row.product_no or "").strip()
            if no:
                crm_no_by_id[row.id] = no

    standard_by_id, alias_options_by_id = load_crm_product_alias_maps(
        db, tenant_no=tenant_no, product_ids=crm_ids
    )

    for idx, it in enumerate(raw_items):
        plan_product_id = (it.get("planProductId") or it.get("plan_product_id") or "").strip()
        if plan_product_id and not it.get("planProductId"):
            it["planProductId"] = plan_product_id
        pp = products_by_id.get(plan_product_id) if plan_product_id else None
        # 仅在明细未挂 planProductId 时用 CRM/序号兜底；已挂 id 时勿换成其它在用产品（会错带指定品牌）
        if not pp and not plan_product_id:
            crm_key = (it.get("productId") or it.get("product_id") or "").strip()
            if crm_key:
                pp = products_by_crm_id.get(crm_key)
            if not pp and len(plan_products) == len(raw_items):
                pp = plan_products[idx]
        if pp and not it.get("planProductId"):
            it["planProductId"] = pp.id
        product_id = (
            (pp.product_id if pp else None) or it.get("productId") or it.get("product_id") or ""
        ).strip()
        # 优先用计划产品已存名（可能是别名）做拆分；明细行 productName 同步
        saved_name = (pp.product_name if pp else None) or it.get("productName") or it.get(
            "product_name"
        )
        if saved_name:
            it["productName"] = saved_name
        merge_plan_product_into_item(
            it,
            pp,
            unit_map=unit_map,
            crm_product_no=crm_no_by_id.get(product_id),
        )
        apply_product_alias_display(
            it,
            product_id=product_id,
            standard_by_id=standard_by_id,
            alias_options_by_id=alias_options_by_id,
        )
        if public_unit_as_label:
            it["unit"] = it.get("unitLabel") or resolve_unit_label(unit_map, it.get("unit"))

    brand_summary_by_pp: dict[str, str] = {}
    if product_ids:
        brand_rows = (
            db.query(PlPlanProductBrandQty)
            .filter(
                PlPlanProductBrandQty.plan_product_id.in_(product_ids),
                PlPlanProductBrandQty.tenant_no == tenant_no,
                PlPlanProductBrandQty.del_flag == 0,
            )
            .order_by(PlPlanProductBrandQty.order_num.asc())
            .all()
        )
        grouped: dict[str, list[str]] = {}
        for br in brand_rows:
            qty = Decimal(str(br.qty or 0))
            if qty <= 0:
                continue
            name = (br.brand_name or "").strip() or "未命名品牌"
            # 去掉多余尾零，如 6.000 -> 6
            qty_text = format(qty.normalize(), "f")
            grouped.setdefault(br.plan_product_id, []).append(f"{name} {qty_text}")
        brand_summary_by_pp = {k: " / ".join(v) for k, v in grouped.items()}

    for it in raw_items:
        pp_id = it.get("planProductId") or ""
        brand_summary = brand_summary_by_pp.get(pp_id) or None
        if brand_summary:
            it["brandSummary"] = brand_summary
        elif it.get("brandName"):
            it["brandSummary"] = it.get("brandName")
        else:
            it["brandSummary"] = None

    enrich_io_items_with_brand_feedback(
        db,
        kind=kind,
        tenant_no=tenant_no,
        items=raw_items,
        products_by_id=products_by_id,
        prefer_site_brand_summary=public_unit_as_label,
    )

    # 列表摘要：产品名 + 品牌/规格/单位/包装/送货方式
    summary_lines = [_format_io_product_summary_line(it) for it in raw_items]
    data["productSummary"] = "；".join(line for line in summary_lines if line)[:800]

    return data


def order_header_to_dict(
    kind: str,
    row: OrderHeader,
    items: list[Any] | None = None,
    *,
    db: Session | None = None,
    public_unit_as_label: bool = False,
) -> dict:
    token = row.site_access_token or ""
    data = {
        "id": row.id,
        "orderNo": row.order_no,
        "planId": row.plan_id,
        "transportId": row.transport_id,
        "status": row.status,
        "partnerCompany": row.partner_company,
        "vehicleNo": row.vehicle_no,
        "planNo": row.plan_no,
        "planDate": row.plan_date.isoformat() if row.plan_date else None,
        "siteAccessToken": token,
        "siteUrl": build_io_order_site_url(kind, token) if token else None,
        "siteFeedbackRemark": row.site_feedback_remark,
        "siteFeedbackImages": parse_feedback_images(row.site_feedback_images),
        "siteFeedbackTime": row.site_feedback_time,
        "siteFirstSubmittedAt": row.site_first_submitted_at,
        "remark": row.remark,
        "tenantNo": row.tenant_no,
        "createTime": row.create_time,
        "updateTime": row.update_time,
        "kind": kind,
    }
    if items is not None:
        data["items"] = [
            {
                "id": it.id,
                "orderId": it.order_id,
                "planProductId": it.plan_product_id,
                "productId": it.product_id,
                "productName": it.product_name,
                "specification": it.specification,
                "unit": it.unit,
                "planQty": it.plan_qty,
                "actualQty": it.actual_qty,
                "qtySource": it.qty_source,
            }
            for it in items
        ]
        data["productSummary"] = "、".join(
            (it.product_name or "").strip() for it in items if (it.product_name or "").strip()
        )[:200]
        data["actualQtyTotal"] = float(sum(Decimal(str(it.actual_qty or 0)) for it in items))
    if db is not None:
        enrich_io_order_dict(
            db,
            data,
            tenant_no=row.tenant_no,
            plan_id=row.plan_id,
            items=items,
            public_unit_as_label=public_unit_as_label,
            kind=kind,
        )
    return data