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.db_models import DbInboundOutbound
from app.models.pl_models import (
    PlInboundOrder,
    PlInboundOrderItem,
    PlOutboundOrder,
    PlOutboundOrderItem,
    PlPlan,
    PlPlanProduct,
    PlTransport,
    PlTransportDbLink,
)
from app.services.db.mappers import parse_date_str
from app.services.pl_dibang_bridge_service import _product_quantity, resolve_partner_company


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"})


def apply_site_feedback(
    db: Session,
    *,
    header: OrderHeader,
    items: list[OrderItem],
    item_qty: dict[str, Decimal],
    remark: Optional[str],
    images: list[str],
    status: str,
    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()
    for item in items:
        if item.id not in item_qty:
            continue
        item.actual_qty = item_qty[item.id]
        item.qty_source = "manual"
        item.update_by = 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"

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


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("pickupNotes", None)
        data.setdefault("destinationName", None)
        data.setdefault("destinationAddress", None)
        data.setdefault("destinationLng", None)
        data.setdefault("destinationLat", None)
        data.setdefault("deliveryTime", 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["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["destinationNotes"] = plan.destination_notes


def merge_plan_product_into_item(
    item: dict,
    product: Optional[PlPlanProduct],
    *,
    unit_map: dict[str, str],
) -> None:
    if not product:
        item.setdefault("productNo", None)
        item.setdefault("packageSpec", None)
        item.setdefault("remark", None)
        item["unitLabel"] = resolve_unit_label(unit_map, item.get("unit"))
        return
    item["productNo"] = product.product_no
    item["packageSpec"] = product.package_spec
    item["remark"] = (product.remark or "").strip() or None
    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)


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,
) -> 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")
        for it in raw_items
        if it.get("planProductId")
    ]
    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}

    for it in raw_items:
        pp = products_by_id.get(it.get("planProductId") or "")
        merge_plan_product_into_item(it, pp, unit_map=unit_map)
        if public_unit_as_label:
            it["unit"] = it.get("unitLabel") or resolve_unit_label(unit_map, it.get("unit"))

    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,
        )
    return data