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

"""运输计划 · 装货/送货段读取与列表摘要。"""
from __future__ import annotations

from collections import defaultdict
from datetime import date, datetime
from decimal import Decimal
from typing import Optional

from fastapi import HTTPException
from sqlalchemy.orm import Session

from app.core.config import settings
from app.services.pl_mobile_stop_meta import load_stop_meta, write_stop_meta
from app.core.utils import generate_uuid
from app.services.pl_mobile_product_identity import resolve_mobile_product_link, is_existing_pending_delivery
from app.models.crm_models import CrmContract, CrmContractProduct
from app.models.pl_models import (
    PlPlan,
    PlPlanContract,
    PlPlanDelivery,
    PlPlanDeliveryProduct,
    PlPlanLoad,
    PlPlanLoadProduct,
    PlTransportAddress,
)
from app.schemas.pl.plan import (
    PlanArchiveBrandQtyItem,
    PlanContractLink,
    PlanDeliveryBody,
    PlanDeliveryRead,
    PlanLoadBody,
    PlanLoadRead,
    PlanStopProductBody,
    PlanStopProductRead,
)
from app.services.pl_brand_archive import load_brand_qtys_by_stop_product_ids
from app.services.pl_internal_customer import is_internal_customer_name
from app.services.pl_plan_stop_rules import normalize_stop_dibang_fields, warn_qty_imbalance
from app.services.pl_plan_service import (
    PLAN_CONTRACT_SIDE_DEMANDER,
    PLAN_CONTRACT_SIDE_SUPPLIER,
    assert_contract_selectable_for_plan,
)
from app.services.pl_product_display import load_crm_product_alias_maps, resolve_product_display_names

PLAN_CONTRACT_SIDE_LOAD = "load"
PLAN_CONTRACT_SIDE_DELIVERY = "delivery"


def _contract_link(row: PlPlanContract) -> PlanContractLink:
    return PlanContractLink(
        contractId=row.contract_id,
        contractNo=row.contract_no,
        stopId=row.stop_id,
    )


def _attach_contract_nos_to_stop_products(
    db: Session,
    tenant_no: str,
    products: list[PlanStopProductRead],
    contracts: list[PlanContractLink],
) -> None:
    """段产品行展示用合同编号：优先段合同关联，其次按 contract_id 查 CRM。"""
    if not products:
        return
    by_id = {
        str(c.contract_id).strip(): (c.contract_no or "").strip()
        for c in contracts or []
        if (c.contract_id or "").strip()
    }
    missing: list[str] = []
    for product in products:
        cid = (product.contract_id or "").strip()
        if not cid:
            continue
        no = by_id.get(cid) or (product.contract_no or "").strip()
        if no:
            product.contract_no = no
            by_id.setdefault(cid, no)
        else:
            missing.append(cid)
    need = [cid for cid in dict.fromkeys(missing) if cid not in by_id or not by_id.get(cid)]
    if not need:
        return
    rows = (
        db.query(CrmContract.id, CrmContract.contract_no)
        .filter(
            CrmContract.tenant_no == tenant_no,
            CrmContract.del_flag == 0,
            CrmContract.id.in_(need),
        )
        .all()
    )
    for cid, cno in rows:
        key = str(cid or "").strip()
        if key and (cno or "").strip():
            by_id[key] = str(cno).strip()
    for product in products:
        cid = (product.contract_id or "").strip()
        if not cid:
            continue
        no = by_id.get(cid)
        if no:
            product.contract_no = no


def _attach_stop_brand_qtys(
    product: PlanStopProductRead,
    brand_qtys_by_stop: dict[str, list[dict]],
) -> PlanStopProductRead:
    raw = brand_qtys_by_stop.get(product.id) or None
    if raw:
        product.brand_qtys = [PlanArchiveBrandQtyItem.model_validate(x) for x in raw]
    else:
        product.brand_qtys = None
    return product


def _resolve_stop_list_product_name(
    *,
    saved_product_name: Optional[str],
    product_id: Optional[str],
    standard_by_id: dict[str, str],
    alias_options_by_id: dict[str, list[str]],
) -> str:
    """列表展示用：优先 CRM 标准名，不展示别名。"""
    display_name, _alias = resolve_product_display_names(
        saved_product_name=saved_product_name,
        product_id=product_id,
        standard_by_id=standard_by_id,
        alias_options_by_id=alias_options_by_id,
    )
    return display_name or (saved_product_name or "")


def _apply_stop_list_product_name(
    product: PlanStopProductRead,
    *,
    standard_by_id: dict[str, str],
    alias_options_by_id: dict[str, list[str]],
) -> PlanStopProductRead:
    product.product_name = _resolve_stop_list_product_name(
        saved_product_name=product.product_name,
        product_id=product.product_id,
        standard_by_id=standard_by_id,
        alias_options_by_id=alias_options_by_id,
    )
    return product


def _load_stop_product(
    row: PlPlanLoadProduct,
    brand_qtys_by_stop: dict[str, list[dict]] | None = None,
    *,
    standard_by_id: dict[str, str] | None = None,
    alias_options_by_id: dict[str, list[str]] | None = None,
) -> PlanStopProductRead:
    product = PlanStopProductRead.model_validate(row)
    if standard_by_id is not None and alias_options_by_id is not None:
        product = _apply_stop_list_product_name(
            product,
            standard_by_id=standard_by_id,
            alias_options_by_id=alias_options_by_id,
        )
    if brand_qtys_by_stop is not None:
        return _attach_stop_brand_qtys(product, brand_qtys_by_stop)
    return product


def _delivery_stop_product(
    row: PlPlanDeliveryProduct,
    brand_qtys_by_stop: dict[str, list[dict]] | None = None,
    *,
    standard_by_id: dict[str, str] | None = None,
    alias_options_by_id: dict[str, list[str]] | None = None,
) -> PlanStopProductRead:
    product = PlanStopProductRead.model_validate(row)
    if standard_by_id is not None and alias_options_by_id is not None:
        product = _apply_stop_list_product_name(
            product,
            standard_by_id=standard_by_id,
            alias_options_by_id=alias_options_by_id,
        )
    if brand_qtys_by_stop is not None:
        return _attach_stop_brand_qtys(product, brand_qtys_by_stop)
    return product


def load_stop_contract_rows_grouped(
    db: Session,
    tenant_no: str,
    plan_ids: list[str],
) -> dict[str, list[PlPlanContract]]:
    if not plan_ids:
        return {}
    rows = (
        db.query(PlPlanContract)
        .filter(
            PlPlanContract.tenant_no == tenant_no,
            PlPlanContract.plan_id.in_(plan_ids),
            PlPlanContract.del_flag == 0,
        )
        .order_by(PlPlanContract.order_num.asc(), PlPlanContract.create_time.asc())
        .all()
    )
    grouped: dict[str, list[PlPlanContract]] = defaultdict(list)
    for row in rows:
        grouped[row.plan_id].append(row)
    return grouped


def _contracts_for_stop(
    contract_rows: list[PlPlanContract],
    stop_id: str,
    legacy_sides: tuple[str, ...],
) -> list[PlanContractLink]:
    by_stop = [_contract_link(r) for r in contract_rows if (r.stop_id or "") == stop_id]
    if by_stop:
        return by_stop
    return [
        _contract_link(r)
        for r in contract_rows
        if not (r.stop_id or "").strip() and r.side in legacy_sides
    ]


def _to_decimal(value) -> Optional[Decimal]:
    if value is None or value == "":
        return None
    try:
        return Decimal(str(value))
    except Exception:
        return None


def resolve_stop_address_from_library(
    db: Session,
    tenant_no: str,
    address: Optional[str],
    lng,
    lat,
) -> Optional[str]:
    """段 address 为空时，按经纬度从运输地址库补全 full_address。"""
    text = (address or "").strip()
    if text:
        return text
    lat_d = _to_decimal(lat)
    lng_d = _to_decimal(lng)
    if lat_d is None and lng_d is None:
        return text or None

    query = db.query(PlTransportAddress).filter(
        PlTransportAddress.tenant_no == tenant_no,
        PlTransportAddress.del_flag == 0,
        PlTransportAddress.status == "1",
    )
    if lat_d is not None:
        query = query.filter(PlTransportAddress.lat == lat_d)
    if lng_d is not None:
        query = query.filter(PlTransportAddress.lng == lng_d)
    rows = query.order_by(PlTransportAddress.create_time.desc()).all()
    if not rows:
        return text or None
    if len(rows) == 1:
        full = (rows[0].full_address or "").strip()
        return full or None
    if lng_d is not None:
        for row in rows:
            if row.lng == lng_d:
                full = (row.full_address or "").strip()
                if full:
                    return full
    full = (rows[0].full_address or "").strip()
    return full or None


def enrich_stop_addresses_from_library(
    db: Session,
    tenant_no: str,
    load: PlanLoadBody,
    deliveries: list[PlanDeliveryBody],
) -> None:
    resolved_load = resolve_stop_address_from_library(
        db, tenant_no, load.address, load.lng, load.lat
    )
    if resolved_load:
        load.address = resolved_load
    for delivery in deliveries:
        resolved = resolve_stop_address_from_library(
            db, tenant_no, delivery.address, delivery.lng, delivery.lat
        )
        if resolved:
            delivery.address = resolved


def load_plan_stops_grouped(
    db: Session,
    tenant_no: str,
    plan_ids: list[str],
) -> dict[str, tuple[Optional[PlanLoadRead], list[PlanDeliveryRead]]]:
    """批量读取装货段与送货段（列表/详情共用）。"""
    if not plan_ids:
        return {}

    loads = (
        db.query(PlPlanLoad)
        .filter(
            PlPlanLoad.tenant_no == tenant_no,
            PlPlanLoad.plan_id.in_(plan_ids),
            PlPlanLoad.del_flag == 0,
        )
        .all()
    )
    load_by_plan = {row.plan_id: row for row in loads}
    load_ids = [row.id for row in loads]

    load_products: dict[str, list[PlPlanLoadProduct]] = defaultdict(list)
    if load_ids:
        for row in (
            db.query(PlPlanLoadProduct)
            .filter(
                PlPlanLoadProduct.tenant_no == tenant_no,
                PlPlanLoadProduct.load_id.in_(load_ids),
                PlPlanLoadProduct.del_flag == 0,
            )
            .order_by(PlPlanLoadProduct.order_num.asc(), PlPlanLoadProduct.create_time.asc())
            .all()
        ):
            load_products[row.load_id].append(row)

    deliveries = (
        db.query(PlPlanDelivery)
        .filter(
            PlPlanDelivery.tenant_no == tenant_no,
            PlPlanDelivery.plan_id.in_(plan_ids),
            PlPlanDelivery.del_flag == 0,
        )
        .order_by(PlPlanDelivery.seq.asc(), PlPlanDelivery.create_time.asc())
        .all()
    )
    deliveries_by_plan: dict[str, list[PlPlanDelivery]] = defaultdict(list)
    delivery_ids: list[str] = []
    for row in deliveries:
        deliveries_by_plan[row.plan_id].append(row)
        delivery_ids.append(row.id)

    delivery_products: dict[str, list[PlPlanDeliveryProduct]] = defaultdict(list)
    if delivery_ids:
        for row in (
            db.query(PlPlanDeliveryProduct)
            .filter(
                PlPlanDeliveryProduct.tenant_no == tenant_no,
                PlPlanDeliveryProduct.delivery_id.in_(delivery_ids),
                PlPlanDeliveryProduct.del_flag == 0,
            )
            .order_by(PlPlanDeliveryProduct.order_num.asc(), PlPlanDeliveryProduct.create_time.asc())
            .all()
        ):
            delivery_products[row.delivery_id].append(row)

    stop_product_ids: list[str] = []
    crm_product_ids: set[str] = set()
    for rows in load_products.values():
        for p in rows:
            if p.id:
                stop_product_ids.append(p.id)
            if (p.product_id or "").strip():
                crm_product_ids.add(p.product_id.strip())
    for rows in delivery_products.values():
        for p in rows:
            if p.id:
                stop_product_ids.append(p.id)
            if (p.product_id or "").strip():
                crm_product_ids.add(p.product_id.strip())
    brand_qtys_by_stop = load_brand_qtys_by_stop_product_ids(
        db, tenant_no=tenant_no, stop_product_ids=stop_product_ids
    )
    standard_by_id, alias_options_by_id = load_crm_product_alias_maps(
        db, tenant_no=tenant_no, product_ids=crm_product_ids
    )

    contracts_by_plan = load_stop_contract_rows_grouped(db, tenant_no, plan_ids)

    result: dict[str, tuple[Optional[PlanLoadRead], list[PlanDeliveryRead]]] = {}
    for plan_id in plan_ids:
        contract_rows = contracts_by_plan.get(plan_id, [])
        load_row = load_by_plan.get(plan_id)
        load_read: Optional[PlanLoadRead] = None
        if load_row:
            load_read = PlanLoadRead.model_validate(load_row)
            resolved_load_address = resolve_stop_address_from_library(
                db,
                tenant_no,
                load_read.address,
                load_read.lng,
                load_read.lat,
            )
            if resolved_load_address:
                load_read.address = resolved_load_address
            load_read.products = [
                _load_stop_product(
                    p,
                    brand_qtys_by_stop,
                    standard_by_id=standard_by_id,
                    alias_options_by_id=alias_options_by_id,
                )
                for p in load_products.get(load_row.id, [])
            ]
            load_read.contracts = _contracts_for_stop(
                contract_rows,
                load_row.id,
                (PLAN_CONTRACT_SIDE_LOAD, PLAN_CONTRACT_SIDE_SUPPLIER),
            )
            _attach_contract_nos_to_stop_products(
                db, tenant_no, load_read.products, load_read.contracts
            )

        delivery_reads: list[PlanDeliveryRead] = []
        plan_deliveries = deliveries_by_plan.get(plan_id, [])
        for idx, delivery_row in enumerate(plan_deliveries):
            delivery_read = PlanDeliveryRead.model_validate(delivery_row)
            resolved_delivery_address = resolve_stop_address_from_library(
                db,
                tenant_no,
                delivery_read.address,
                delivery_read.lng,
                delivery_read.lat,
            )
            if resolved_delivery_address:
                delivery_read.address = resolved_delivery_address
            delivery_read.products = [
                _delivery_stop_product(
                    p,
                    brand_qtys_by_stop,
                    standard_by_id=standard_by_id,
                    alias_options_by_id=alias_options_by_id,
                )
                for p in delivery_products.get(delivery_row.id, [])
            ]
            legacy_sides = (PLAN_CONTRACT_SIDE_DELIVERY, PLAN_CONTRACT_SIDE_DEMANDER)
            # 未迁移合同仅挂到第一个送货段
            if idx == 0:
                delivery_read.contracts = _contracts_for_stop(
                    contract_rows,
                    delivery_row.id,
                    legacy_sides,
                )
            else:
                delivery_read.contracts = _contracts_for_stop(
                    contract_rows,
                    delivery_row.id,
                    (PLAN_CONTRACT_SIDE_DELIVERY,),
                )
            _attach_contract_nos_to_stop_products(
                db, tenant_no, delivery_read.products, delivery_read.contracts
            )
            delivery_reads.append(delivery_read)

        result[plan_id] = (load_read, delivery_reads)

    return result


def resolve_load_customer_name(
    plan: PlPlan,
    load: Optional[PlanLoadRead],
) -> Optional[str]:
    if load and (load.customer_name or "").strip():
        return load.customer_name.strip()
    legacy = (plan.supplier_name or "").strip()
    return legacy or None


def resolve_delivery_customer_summary(
    plan: PlPlan,
    deliveries: list[PlanDeliveryRead],
) -> Optional[str]:
    names: list[str] = []
    seen: set[str] = set()
    for delivery in deliveries:
        name = (delivery.customer_name or "").strip()
        if name and name not in seen:
            seen.add(name)
            names.append(name)

    if names:
        if len(names) <= 2:
            return "、".join(names)
        return f"{names[0]}、{names[1]}等{len(names)}家"

    legacy = (plan.customer_name or plan.destination_name or "").strip()
    return legacy or None


def collect_unique_delivery_date_strings(deliveries) -> list[str]:
    """送货段 delivery_time → 去重升序 YYYY-MM-DD。"""
    days: set[date] = set()
    for row in deliveries or []:
        raw = getattr(row, "delivery_time", None)
        if raw is None or raw == "":
            continue
        if isinstance(raw, datetime):
            days.add(raw.date())
        elif isinstance(raw, date):
            days.add(raw)
        else:
            text = str(raw).strip()
            if not text:
                continue
            try:
                days.add(datetime.strptime(text[:10], "%Y-%m-%d").date())
            except ValueError:
                continue
    return [d.isoformat() for d in sorted(days)]


def attach_plan_stops_to_item(
    plan_item,
    plan: PlPlan,
    stops: tuple[Optional[PlanLoadRead], list[PlanDeliveryRead]],
) -> None:
    load_read, delivery_reads = stops
    plan_item.load = load_read
    plan_item.deliveries = delivery_reads
    plan_item.load_customer_name = resolve_load_customer_name(plan, load_read)
    plan_item.delivery_customer_summary = resolve_delivery_customer_summary(plan, delivery_reads)
    if load_read and (load_read.address or "").strip():
        plan_item.pickup_address = load_read.address
    if delivery_reads and (delivery_reads[0].address or "").strip():
        plan_item.destination_address = delivery_reads[0].address
    plan_item.delivery_times = collect_unique_delivery_date_strings(delivery_reads)


def uses_multi_stop_payload(load: Optional[PlanLoadBody], deliveries: Optional[list[PlanDeliveryBody]]) -> bool:
    return load is not None or deliveries is not None


def _normalize_contract_links(links: Optional[list[PlanContractLink]]) -> list[PlanContractLink]:
    if not links:
        return []
    seen: set[str] = set()
    out: list[PlanContractLink] = []
    for link in links:
        cid = (link.contract_id or "").strip()
        if not cid or cid in seen:
            continue
        seen.add(cid)
        out.append(link)
    return out


def assert_at_most_one_contract(
    links: list[PlanContractLink],
    side_label: str,
) -> list[PlanContractLink]:
    if len(links) > 1:
        raise HTTPException(
            status_code=422,
            detail=f"{side_label}只能选择 1 份",
        )
    return links


def _assert_product_rows_in_contract_allowed(
    products: Optional[list],
    allowed: set[str],
    *,
    row_label: str = "装货产品",
) -> None:
    for prod in products or []:
        pid = str(getattr(prod, "product_id", None) or "").strip()
        name = (getattr(prod, "product_name", None) or pid or "未命名").strip()
        if not pid or pid not in allowed:
            raise HTTPException(
                status_code=422,
                detail=f"{row_label}「{name}」不在所选合同明细中",
            )


def _contract_product_rows(db: Session, tenant_no: str, contract_id: str) -> list:
    return (
        db.query(CrmContractProduct)
        .filter(
            CrmContractProduct.contract_id == contract_id,
            CrmContractProduct.tenant_no == tenant_no,
            CrmContractProduct.del_flag == 0,
        )
        .all()
    )


def _contract_allowed_product_ids(db: Session, tenant_no: str, contract_id: str) -> set[str]:
    rows = _contract_product_rows(db, tenant_no, contract_id)
    return {
        str(r.product_id).strip()
        for r in rows
        if getattr(r, "product_id", None) and str(r.product_id).strip()
    }


def _assert_stop_products_package_for_contract(
    db: Session,
    tenant_no: str,
    contract_id: str,
    stop_products: Optional[list],
) -> None:
    from app.services.pl_contract_plan_qty import (
        validate_stop_products_package_against_contract,
    )

    rows = _contract_product_rows(db, tenant_no, contract_id)
    ok, err = validate_stop_products_package_against_contract(
        contract_products=rows,
        stop_products=stop_products or [],
    )
    if not ok:
        raise HTTPException(status_code=422, detail=err or "包装规格不符合合同明细")


def _union_contract_product_rows(
    db: Session,
    tenant_no: str,
    links: list[PlanContractLink],
) -> list:
    rows: list = []
    seen: set[str] = set()
    for link in links:
        cid = (link.contract_id or "").strip()
        if not cid or cid in seen:
            continue
        seen.add(cid)
        rows.extend(_contract_product_rows(db, tenant_no, cid))
    return rows


def _assert_stop_products_package_for_union(
    stop_products: Optional[list],
    contract_rows: list,
) -> None:
    from app.services.pl_contract_plan_qty import (
        validate_stop_products_package_against_contract,
    )

    ok, err = validate_stop_products_package_against_contract(
        contract_products=contract_rows,
        stop_products=stop_products or [],
    )
    if not ok:
        raise HTTPException(status_code=422, detail=err or "包装规格不符合合同明细")


def _assert_stop_products_bound_to_lines(
    stop_products: Optional[list],
    contract_rows: list,
    selected_contract_ids: set[str],
    *,
    row_label: str,
) -> None:
    from app.services.pl_contract_plan_qty import (
        contract_line_package_specs,
        normalize_package_spec,
    )

    by_id = {
        str(getattr(row, "id", None) or "").strip(): row
        for row in contract_rows
        if str(getattr(row, "id", None) or "").strip()
    }
    if not by_id:
        return
    for prod in stop_products or []:
        cpid = str(getattr(prod, "contract_product_id", None) or "").strip()
        if not cpid:
            continue
        line = by_id.get(cpid)
        if not line:
            name = (getattr(prod, "product_name", None) or cpid).strip()
            raise HTTPException(
                status_code=422,
                detail=f"{row_label}「{name}」不在所选合同明细中",
            )
        line_cid = str(getattr(line, "contract_id", None) or "").strip()
        if line_cid and selected_contract_ids and line_cid not in selected_contract_ids:
            name = (getattr(prod, "product_name", None) or cpid).strip()
            raise HTTPException(
                status_code=422,
                detail=f"{row_label}「{name}」不在所选合同明细中",
            )
        line_pid = str(getattr(line, "product_id", None) or "").strip()
        prod_pid = str(getattr(prod, "product_id", None) or "").strip()
        if line_pid and prod_pid and line_pid != prod_pid:
            name = (getattr(prod, "product_name", None) or prod_pid).strip()
            raise HTTPException(
                status_code=422,
                detail=f"{row_label}「{name}」与绑定合同行产品不一致",
            )
        pkg = normalize_package_spec(getattr(prod, "package_spec", None))
        specs = contract_line_package_specs(line)
        if pkg and specs and pkg not in specs:
            name = (getattr(prod, "product_name", None) or prod_pid).strip()
            raise HTTPException(
                status_code=422,
                detail=f"{row_label}包装规格不在合同明细中：{name}（{pkg}）",
            )


def assert_load_products_in_selected_contract(
    db: Session,
    tenant_no: str,
    load: PlanLoadBody,
    plan_products: Optional[list] = None,
) -> None:
    if is_internal_customer_name(load.customer_name):
        return
    links = _normalize_contract_links(load.contracts)
    if not links:
        return
    rows = _union_contract_product_rows(db, tenant_no, links)
    allowed = {
        str(r.product_id).strip()
        for r in rows
        if getattr(r, "product_id", None) and str(r.product_id).strip()
    }
    _assert_product_rows_in_contract_allowed(load.products, allowed, row_label="装货产品")
    _assert_product_rows_in_contract_allowed(plan_products, allowed, row_label="装货产品")
    merged = list(load.products or []) + list(plan_products or [])
    _assert_stop_products_package_for_union(merged, rows)
    selected_ids = {(link.contract_id or "").strip() for link in links if (link.contract_id or "").strip()}
    _assert_stop_products_bound_to_lines(
        load.products, rows, selected_ids, row_label="装货产品"
    )


def assert_delivery_products_in_selected_contract(
    db: Session,
    tenant_no: str,
    delivery: PlanDeliveryBody,
    seq: int,
) -> None:
    if is_internal_customer_name(delivery.customer_name):
        return
    links = _normalize_contract_links(delivery.contracts)
    if not links:
        return
    rows = _union_contract_product_rows(db, tenant_no, links)
    allowed = {
        str(r.product_id).strip()
        for r in rows
        if getattr(r, "product_id", None) and str(r.product_id).strip()
    }
    _assert_product_rows_in_contract_allowed(
        delivery.products,
        allowed,
        row_label=f"送货段#{seq}产品",
    )
    _assert_stop_products_package_for_union(delivery.products, rows)
    selected_ids = {(link.contract_id or "").strip() for link in links if (link.contract_id or "").strip()}
    _assert_stop_products_bound_to_lines(
        delivery.products, rows, selected_ids, row_label=f"送货段#{seq}产品"
    )


def assert_stop_contract_required(
    customer_name: Optional[str],
    contracts: Optional[list[PlanContractLink]],
    side_label: str,
) -> None:
    """非内部客户装/送货段保存时必须至少选一份合同。"""
    if is_internal_customer_name(customer_name):
        return
    if not _normalize_contract_links(contracts):
        raise HTTPException(status_code=422, detail=f"请为{side_label}选择合同")


def _resolve_stop_contracts(
    db: Session,
    tenant_no: str,
    customer_id: Optional[str],
    customer_name: Optional[str],
    contracts: Optional[list[PlanContractLink]],
    side_label: str,
) -> list[PlanContractLink]:
    if is_internal_customer_name(customer_name):
        return []
    links = _normalize_contract_links(contracts)
    if not links:
        return []
    resolved: list[PlanContractLink] = []
    for link in links:
        contract = assert_contract_selectable_for_plan(
            db,
            link.contract_id,
            tenant_no,
            expected_customer_id=customer_id,
            side_label=side_label,
        )
        resolved.append(
            PlanContractLink(
                contractId=contract.id,
                contractNo=contract.contract_no or link.contract_no,
            )
        )
    return resolved


def has_pending_stop_products(db: Session, tenant_no: str, plan_id: str) -> bool:
    # Uses only pre-existing tables, including when the mobile feature is disabled.
    for model in (PlPlanLoadProduct, PlPlanDeliveryProduct):
        if db.query(model.id).filter(model.tenant_no == tenant_no, model.plan_id == plan_id,
                                    model.del_flag == 0, model.plan_product_id == "").first():
            return True
    return False


def assert_no_pending_stop_products(db: Session, tenant_no: str, plan_id: str) -> None:
    if has_pending_stop_products(db, tenant_no, plan_id):
        raise HTTPException(422, '计划存在待关联装货产品，请补齐关联并正式保存后再安排运输')


def validate_mobile_stop_sources(db: Session, tenant_no: str, load: PlanLoadBody,
                                 deliveries: list[PlanDeliveryBody], products: Optional[list]) -> None:
    """Opt-in source validation also runs for incomplete drafts; never trusts client tenant/line ids."""
    from app.models.crm_models import CrmProduct
    if load is None:
        raise HTTPException(422, "请填写装货信息")
    product_ids = {str(p.product_id) for p in (products or []) if p.product_id}
    for stop in [load] + list(deliveries or []):
        product_ids.update(str(p.product_id) for p in stop.products or [] if p.product_id)
        links = _normalize_contract_links(stop.contracts)
        # Even internal customers cannot attach an unverified foreign contract.
        for link in links:
            assert_contract_selectable_for_plan(db, link.contract_id, tenant_no,
                expected_customer_id=stop.customer_id, side_label="移动段合同")
        contract_rows = _union_contract_product_rows(db, tenant_no, links) if links else []
        by_id = {str(row.id): row for row in contract_rows}
        selected = {str(link.contract_id) for link in links}
        for product in stop.products or []:
            cid = (product.contract_id or '').strip()
            cpid = (product.contract_product_id or '').strip()
            if not cid and not cpid:
                continue
            line = by_id.get(cpid)
            if not cid or not cpid or cid not in selected or line is None or str(line.contract_id) != cid:
                raise HTTPException(422, '移动段产品与所选合同明细关联不一致')
            if str(line.product_id) != str(product.product_id):
                raise HTTPException(422, '移动段合同行产品不一致')
        _assert_stop_products_bound_to_lines(stop.products, contract_rows, selected, row_label="移动段产品")
    if product_ids:
        found = db.query(CrmProduct.id).filter(CrmProduct.tenant_no == tenant_no,
            CrmProduct.del_flag == 0, CrmProduct.id.in_(product_ids)).all()
        if {str(row[0]) for row in found} != product_ids:
            raise HTTPException(422, '移动段产品不存在或不属于当前租户')


def validate_mobile_draft_stops(db: Session, tenant_no: str, load: PlanLoadBody,
                                deliveries: Optional[list[PlanDeliveryBody]], products: Optional[list]):
    rows = list(load.products or []) + [p for stop in deliveries or [] for p in stop.products or []]
    quantities = [getattr(p, 'plan_qty', None) for p in rows] + [getattr(p, 'quantity', None) for p in products or []]
    for value in quantities:
        if value is None:
            continue
        qty = Decimal(str(value))
        if not qty.is_finite() or qty < 0:
            raise HTTPException(422, '暂存产品数量须为非负有限数')
    validate_mobile_stop_sources(db, tenant_no, load, list(deliveries or []), products)
    return load, list(deliveries or [])


def validate_plan_stops_write(
    db: Session,
    tenant_no: str,
    load: Optional[PlanLoadBody],
    deliveries: Optional[list[PlanDeliveryBody]],
    products: Optional[list],
) -> tuple[PlanLoadBody, list[PlanDeliveryBody]]:
    if load is None:
        raise HTTPException(status_code=422, detail="请填写装货信息")
    if not (load.customer_id or "").strip() or not (load.customer_name or "").strip():
        raise HTTPException(status_code=422, detail="请选择装货客户")
    if not deliveries:
        raise HTTPException(status_code=422, detail="请至少添加一段送货信息")
    if products is not None and not products:
        raise HTTPException(status_code=422, detail="请至少保留一条产品明细")

    assert_stop_contract_required(load.customer_name, load.contracts, "装货客户")

    load.contracts = _resolve_stop_contracts(
        db,
        tenant_no,
        load.customer_id,
        load.customer_name,
        load.contracts,
        "装货合同",
    )

    resolved_deliveries: list[PlanDeliveryBody] = []
    for idx, delivery in enumerate(deliveries, start=1):
        if not (delivery.customer_id or "").strip() or not (delivery.customer_name or "").strip():
            raise HTTPException(status_code=422, detail=f"送货段#{idx}请选择客户")
        assert_stop_contract_required(delivery.customer_name, delivery.contracts, f"送货段#{idx}")
        delivery.contracts = _resolve_stop_contracts(
            db,
            tenant_no,
            delivery.customer_id,
            delivery.customer_name,
            delivery.contracts,
            f"送货段#{idx}合同",
        )
        resolved_deliveries.append(delivery)

    enrich_stop_addresses_from_library(db, tenant_no, load, resolved_deliveries)
    assert_load_products_in_selected_contract(db, tenant_no, load, plan_products=products)
    for idx, delivery in enumerate(resolved_deliveries, start=1):
        assert_delivery_products_in_selected_contract(db, tenant_no, delivery, idx)
    return load, resolved_deliveries


def collect_stop_qty_warnings(
    products: Optional[list],
    load: PlanLoadBody,
    deliveries: list[PlanDeliveryBody],
) -> list[str]:
    plan_rows = products or []
    load_rows = load.products or []
    delivery_rows: list[PlanStopProductBody] = []
    for delivery in deliveries:
        delivery_rows.extend(delivery.products or [])
    return warn_qty_imbalance(plan_rows, load_rows, delivery_rows)


def sync_plan_legacy_from_stops(
    plan: PlPlan,
    load: PlanLoadBody,
    deliveries: list[PlanDeliveryBody],
) -> None:
    plan.supplier_id = load.customer_id
    plan.supplier_name = load.customer_name
    plan.pickup_address = load.address
    plan.pickup_lng = load.lng
    plan.pickup_lat = load.lat
    plan.pickup_time = load.load_time
    plan.pickup_contact = load.contact
    plan.pickup_notes = load.notes

    if not deliveries:
        # Incomplete mobile drafts keep the load while clearing stale legacy customer fields.
        plan.customer_id = None
        plan.customer_name = None
        plan.destination_id = None
        plan.destination_name = None
        plan.destination_address = None
        plan.destination_lng = None
        plan.destination_lat = None
        plan.delivery_time = None
        plan.delivery_contact = None
        plan.destination_notes = None
        return

    first = deliveries[0]
    plan.customer_id = first.customer_id
    plan.customer_name = first.customer_name
    plan.destination_id = first.customer_id
    plan.destination_name = first.customer_name
    plan.destination_address = first.address
    plan.destination_lng = first.lng
    plan.destination_lat = first.lat
    plan.delivery_time = first.delivery_time
    plan.delivery_contact = first.contact
    plan.destination_notes = first.notes


def _soft_delete_load_products_and_stop_contracts(
    db: Session,
    tenant_no: str,
    plan_id: str,
    user_id: Optional[str],
) -> None:
    """软删装货产品与装/送货段合同；送货段本体由 upsert 保留主键，不在此全量软删。"""
    now = datetime.now()
    load_ids = [
        row.id
        for row in db.query(PlPlanLoad)
        .filter(
            PlPlanLoad.plan_id == plan_id,
            PlPlanLoad.tenant_no == tenant_no,
            PlPlanLoad.del_flag == 0,
        )
        .all()
    ]
    if load_ids:
        db.query(PlPlanLoadProduct).filter(
            PlPlanLoadProduct.load_id.in_(load_ids),
            PlPlanLoadProduct.del_flag == 0,
        ).update({"del_flag": 1, "update_by": user_id, "update_time": now}, synchronize_session=False)

    db.query(PlPlanContract).filter(
        PlPlanContract.plan_id == plan_id,
        PlPlanContract.tenant_no == tenant_no,
        PlPlanContract.del_flag == 0,
        PlPlanContract.side.in_(
            [PLAN_CONTRACT_SIDE_LOAD, PLAN_CONTRACT_SIDE_DELIVERY]
        ),
    ).update({"del_flag": 1, "update_by": user_id, "update_time": now}, synchronize_session=False)


def _soft_delete_delivery_products(
    db: Session,
    delivery_ids: list[str],
    user_id: Optional[str],
    now: datetime,
) -> None:
    if not delivery_ids:
        return
    db.query(PlPlanDeliveryProduct).filter(
        PlPlanDeliveryProduct.delivery_id.in_(delivery_ids),
        PlPlanDeliveryProduct.del_flag == 0,
    ).update({"del_flag": 1, "update_by": user_id, "update_time": now}, synchronize_session=False)


def resolve_delivery_id_for_upsert(
    *,
    plan_id: str,
    payload_id: Optional[str],
    existing_by_id: dict,
) -> tuple[str, Optional[object]]:
    pid = (payload_id or "").strip()
    if not pid:
        return generate_uuid(), None
    row = existing_by_id.get(pid)
    if row is None or getattr(row, "plan_id", None) != plan_id:
        raise HTTPException(status_code=422, detail=f"送货段 id 无效或不属于本计划：{pid}")
    return pid, row


def _write_stop_contracts(
    db: Session,
    tenant_no: str,
    plan_id: str,
    stop_id: str,
    side: str,
    contracts: list[PlanContractLink],
    user_id: Optional[str],
    dept_id: Optional[str],
    order_num_start: int,
) -> int:
    now = datetime.now()
    order_num = order_num_start
    for link in contracts:
        order_num += 1
        db.add(
            PlPlanContract(
                id=generate_uuid(),
                plan_id=plan_id,
                side=side,
                stop_id=stop_id,
                contract_id=link.contract_id,
                contract_no=link.contract_no,
                tenant_no=tenant_no,
                order_num=order_num,
                create_by=user_id,
                create_dept_id=dept_id or "",
                create_time=now,
                update_by=user_id,
                update_time=now,
                del_flag=0,
            )
        )
    return order_num


def _resolve_plan_product_id(
    prod: PlanStopProductBody,
    plan_products: list,
    allow_unlinked_draft: bool = False,
) -> str:
    if getattr(prod, 'unlinked_draft', False):
        if allow_unlinked_draft and not prod.plan_product_id and not getattr(prod, 'plan_product_client_key', None):
            return ""
        raise HTTPException(422, '待关联产品仅可暂存，正式保存前请选择关联装货产品')
    mobile_id = resolve_mobile_product_link(prod, plan_products)
    if mobile_id is not None:
        return mobile_id
    explicit = (prod.plan_product_id or "").strip()
    if explicit:
        for row in plan_products:
            if row.id == explicit:
                return explicit
    product_id = (prod.product_id or "").strip()
    if product_id:
        for row in plan_products:
            if (row.product_id or "").strip() == product_id:
                return row.id
    product_no = (prod.product_no or "").strip()
    if product_no:
        for row in plan_products:
            if (row.product_no or "").strip() == product_no:
                return row.id
    raise HTTPException(
        status_code=422,
        detail=f"无法匹配计划产品行：{prod.product_name or explicit or product_id}",
    )


def _resolve_or_create_plan_load(
    db: Session,
    tenant_no: str,
    plan_id: str,
    load: PlanLoadBody,
    user_id: Optional[str],
    dept_id: Optional[str],
    now: datetime,
) -> str:
    """装货段与 plan 1:1；uk_pl_plan_load_plan 仅按 plan_id 唯一，软删后不可再 INSERT，须原地更新。"""
    generate_dibang, dibang_direction = normalize_stop_dibang_fields(
        load.customer_name,
        "load",
        load.generate_dibang,
        load.dibang_direction,
    )

    existing = (
        db.query(PlPlanLoad)
        .filter(
            PlPlanLoad.plan_id == plan_id,
            PlPlanLoad.tenant_no == tenant_no,
        )
        .order_by(PlPlanLoad.del_flag.asc(), PlPlanLoad.update_time.desc())
        .first()
    )
    if existing:
        existing.customer_id = load.customer_id
        existing.customer_name = load.customer_name
        existing.address = load.address
        existing.lng = load.lng
        existing.lat = load.lat
        existing.load_time = load.load_time
        existing.contact = load.contact
        existing.notes = load.notes
        existing.generate_dibang = generate_dibang
        existing.dibang_direction = dibang_direction
        existing.remark = load.remark
        existing.del_flag = 0
        existing.update_by = user_id
        existing.update_time = now
        return existing.id

    load_id = generate_uuid()
    db.add(
        PlPlanLoad(
            id=load_id,
            plan_id=plan_id,
            customer_id=load.customer_id,
            customer_name=load.customer_name,
            address=load.address,
            lng=load.lng,
            lat=load.lat,
            load_time=load.load_time,
            contact=load.contact,
            notes=load.notes,
            generate_dibang=generate_dibang,
            dibang_direction=dibang_direction,
            tenant_no=tenant_no,
            remark=load.remark,
            create_by=user_id,
            create_dept_id=dept_id or "",
            create_time=now,
            update_by=user_id,
            update_time=now,
            del_flag=0,
        )
    )
    return load_id


def replace_plan_stops(
    db: Session,
    tenant_no: str,
    plan_id: str,
    load: PlanLoadBody,
    deliveries: list[PlanDeliveryBody],
    user_id: Optional[str],
    dept_id: Optional[str],
    plan_products: Optional[list] = None,
    allow_unlinked_draft: bool = False,
    preserve_pending_draft: bool = False,
) -> None:
    meta_enabled = settings.PL_MOBILE_PLAN_PRODUCT_META_ENABLED
    snapshots = load_stop_meta(db, tenant_no, plan_id, meta_enabled)
    used_snapshots = set()
    _soft_delete_load_products_and_stop_contracts(db, tenant_no, plan_id, user_id)
    now = datetime.now()

    load_id = _resolve_or_create_plan_load(
        db,
        tenant_no,
        plan_id,
        load,
        user_id,
        dept_id,
        now,
    )

    plan_product_rows = plan_products or []

    for idx, prod in enumerate(load.products or []):
        plan_product_id = _resolve_plan_product_id(prod, plan_product_rows)
        stop_product_id = generate_uuid()
        db.add(
            PlPlanLoadProduct(
                id=stop_product_id,
                load_id=load_id,
                plan_id=plan_id,
                plan_product_id=plan_product_id,
                product_id=prod.product_id,
                product_no=prod.product_no,
                product_name=prod.product_name,
                model=prod.model,
                unit=prod.unit or "吨",
                package_spec=prod.package_spec,
                delivery_method=prod.delivery_method,
                contract_id=prod.contract_id,
                contract_product_id=prod.contract_product_id,
                plan_qty=prod.plan_qty,
                tenant_no=tenant_no,
                order_num=idx,
                remark=prod.remark,
                create_by=user_id,
                create_dept_id=dept_id or "",
                create_time=now,
                update_by=user_id,
                update_time=now,
                del_flag=0,
            )
        )

        if meta_enabled:
            write_stop_meta(db, snapshots, used_snapshots, tenant_no, plan_id, 'load', load_id,
                            stop_product_id, prod, next((p for p in plan_product_rows if p.id == plan_product_id), None),
                            user_id, dept_id, now)

    contract_order = 0
    contract_order = _write_stop_contracts(
        db,
        tenant_no,
        plan_id,
        load_id,
        PLAN_CONTRACT_SIDE_LOAD,
        load.contracts or [],
        user_id,
        dept_id,
        contract_order,
    )

    existing_deliveries = (
        db.query(PlPlanDelivery)
        .filter(
            PlPlanDelivery.plan_id == plan_id,
            PlPlanDelivery.tenant_no == tenant_no,
        )
        .all()
    )
    existing_by_id = {row.id: row for row in existing_deliveries}
    kept_ids: set[str] = set()

    for delivery in deliveries:
        generate_dibang, dibang_direction = normalize_stop_dibang_fields(
            delivery.customer_name,
            "delivery",
            delivery.generate_dibang,
            delivery.dibang_direction,
        )

        delivery_id, existing = resolve_delivery_id_for_upsert(
            plan_id=plan_id,
            payload_id=delivery.id,
            existing_by_id=existing_by_id,
        )
        kept_ids.add(delivery_id)

        if existing is not None:
            existing.seq = delivery.seq or 1
            existing.customer_id = delivery.customer_id
            existing.customer_name = delivery.customer_name
            existing.address = delivery.address
            existing.lng = delivery.lng
            existing.lat = delivery.lat
            existing.delivery_time = delivery.delivery_time
            existing.contact = delivery.contact
            existing.notes = delivery.notes
            existing.generate_dibang = generate_dibang
            existing.dibang_direction = dibang_direction
            existing.remark = delivery.remark
            existing.del_flag = 0
            existing.update_by = user_id
            existing.update_time = now
        else:
            db.add(
                PlPlanDelivery(
                    id=delivery_id,
                    plan_id=plan_id,
                    seq=delivery.seq or 1,
                    customer_id=delivery.customer_id,
                    customer_name=delivery.customer_name,
                    address=delivery.address,
                    lng=delivery.lng,
                    lat=delivery.lat,
                    delivery_time=delivery.delivery_time,
                    contact=delivery.contact,
                    notes=delivery.notes,
                    generate_dibang=generate_dibang,
                    dibang_direction=dibang_direction,
                    tenant_no=tenant_no,
                    remark=delivery.remark,
                    create_by=user_id,
                    create_dept_id=dept_id or "",
                    create_time=now,
                    update_by=user_id,
                    update_time=now,
                    del_flag=0,
                )
            )

        pending_products = []
        if preserve_pending_draft:
            pending_products = db.query(PlPlanDeliveryProduct).filter(
                PlPlanDeliveryProduct.tenant_no == tenant_no,
                PlPlanDeliveryProduct.plan_id == plan_id,
                PlPlanDeliveryProduct.delivery_id == delivery_id,
                PlPlanDeliveryProduct.del_flag == 0,
                PlPlanDeliveryProduct.plan_product_id == "",
            ).all()
        _soft_delete_delivery_products(db, [delivery_id], user_id, now)

        for idx, prod in enumerate(delivery.products or []):
            plan_product_id = (
                "" if is_existing_pending_delivery(prod, delivery_id, pending_products)
                else _resolve_plan_product_id(prod, plan_product_rows, allow_unlinked_draft)
            )
            stop_product_id = generate_uuid()
            db.add(
                PlPlanDeliveryProduct(
                    id=stop_product_id,
                    delivery_id=delivery_id,
                    plan_id=plan_id,
                    plan_product_id=plan_product_id,
                    product_id=prod.product_id,
                    product_no=prod.product_no,
                    product_name=prod.product_name,
                    model=prod.model,
                    unit=prod.unit or "吨",
                    package_spec=prod.package_spec,
                    delivery_method=prod.delivery_method,
                    contract_id=prod.contract_id,
                    contract_product_id=prod.contract_product_id,
                    plan_qty=prod.plan_qty,
                    tenant_no=tenant_no,
                    order_num=idx,
                    remark=prod.remark,
                    create_by=user_id,
                    create_dept_id=dept_id or "",
                    create_time=now,
                    update_by=user_id,
                    update_time=now,
                    del_flag=0,
                )
            )

            if meta_enabled:
                write_stop_meta(db, snapshots, used_snapshots, tenant_no, plan_id, 'delivery', delivery_id,
                                stop_product_id, prod, next((p for p in plan_product_rows if p.id == plan_product_id), None),
                                user_id, dept_id, now)

        contract_order = _write_stop_contracts(
            db,
            tenant_no,
            plan_id,
            delivery_id,
            PLAN_CONTRACT_SIDE_DELIVERY,
            delivery.contracts or [],
            user_id,
            dept_id,
            contract_order,
        )

    removed_ids = [row.id for row in existing_deliveries if row.id not in kept_ids and row.del_flag == 0]
    if removed_ids:
        _soft_delete_delivery_products(db, removed_ids, user_id, now)
        db.query(PlPlanDelivery).filter(
            PlPlanDelivery.id.in_(removed_ids),
            PlPlanDelivery.tenant_no == tenant_no,
            PlPlanDelivery.del_flag == 0,
        ).update({"del_flag": 1, "update_by": user_id, "update_time": now}, synchronize_session=False)

    for snapshot in snapshots:
        snapshot.del_flag = 1
        snapshot.update_by = user_id
        snapshot.update_time = now