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

from __future__ import annotations

from datetime import date, datetime
from typing import List, Optional, TYPE_CHECKING

from fastapi import HTTPException
from sqlalchemy import and_, exists, func, or_, select
from sqlalchemy.orm import Session

from app.core.utils import generate_uuid
from app.models.pl_models import (
    PlPlan,
    PlPlanContract,
    PlPlanDelivery,
    PlPlanLoad,
    PlPlanProduct,
)
from app.schemas.pl.plan import (
    PlanContractGapRead,
    PlanContractLink,
    PlanDeliveryRead,
    PlanLoadRead,
    PlanProductRead,
)
from app.services.pl.plan_list_filters import build_plan_column_filters
from app.services.pl_data_scope import apply_pl_scope
from app.services.pl_internal_customer import INTERNAL_CUSTOMER_NAMES, is_internal_customer_name
from app.services.pl_plan_service import (
    PLAN_CONTRACT_SIDE_DEMANDER,
    PLAN_CONTRACT_SIDE_SUPPLIER,
    _normalize_contract_links,
    assert_contract_selectable_for_plan,
    load_plan_contracts_grouped,
    replace_plan_contracts,
)
from app.services.pl_plan_stop_service import (
    PLAN_CONTRACT_SIDE_DELIVERY,
    PLAN_CONTRACT_SIDE_LOAD,
    assert_at_most_one_contract,
    load_plan_stops_grouped,
)
from app.services.pl_product_display import load_crm_product_alias_maps, resolve_product_display_names

if TYPE_CHECKING:
    from app.core.db_auth import DbUserContext


def stop_contract_gap(customer_name: Optional[str], contract_count: int) -> bool:
    """段次客户非内置且未关联合同时视为缺口。"""
    return not is_internal_customer_name(customer_name) and contract_count <= 0


def side_is_exempt(name: Optional[str]) -> bool:
    return is_internal_customer_name(name)


def compute_contract_gap_sides(
    *,
    supplier_name: Optional[str],
    customer_name: Optional[str],
    supplier_contract_count: int,
    demander_contract_count: int,
) -> List[str]:
    gaps: List[str] = []
    if stop_contract_gap(supplier_name, supplier_contract_count):
        gaps.append("supplier")
    if stop_contract_gap(customer_name, demander_contract_count):
        gaps.append("demander")
    return gaps


def compute_plan_gap_sides_from_stops(
    load_read: Optional[PlanLoadRead],
    delivery_reads: list[PlanDeliveryRead],
    *,
    supplier_name: Optional[str] = None,
    customer_name: Optional[str] = None,
    supplier_contract_count: int = 0,
    demander_contract_count: int = 0,
) -> list[str]:
    """按段次汇总计划是否仍有合同缺口（supplier=装货侧，demander=送货侧）。"""
    has_load_gap = False
    has_delivery_gap = False

    if load_read is not None:
        has_load_gap = stop_contract_gap(load_read.customer_name, len(load_read.contracts or []))
    elif stop_contract_gap(supplier_name, supplier_contract_count):
        has_load_gap = True

    if delivery_reads:
        has_delivery_gap = any(
            stop_contract_gap(item.customer_name, len(item.contracts or []))
            for item in delivery_reads
        )
    elif stop_contract_gap(customer_name, demander_contract_count):
        has_delivery_gap = True

    gaps: list[str] = []
    if has_load_gap:
        gaps.append("supplier")
    if has_delivery_gap:
        gaps.append("demander")
    return gaps


def _stop_contract_exists(tenant_no: str, plan_id_expr, stop_id_expr, legacy_sides: tuple[str, ...]):
    return exists(
        select(PlPlanContract.id).where(
            PlPlanContract.plan_id == plan_id_expr,
            PlPlanContract.tenant_no == tenant_no,
            PlPlanContract.del_flag == 0,
            or_(
                PlPlanContract.stop_id == stop_id_expr,
                and_(
                    or_(PlPlanContract.stop_id.is_(None), PlPlanContract.stop_id == ""),
                    PlPlanContract.side.in_(legacy_sides),
                ),
            ),
        )
    )


def _load_stop_gap_expr(tenant_no: str):
    internal_names = tuple(INTERNAL_CUSTOMER_NAMES)
    customer_not_internal = or_(
        PlPlanLoad.customer_name.is_(None),
        PlPlanLoad.customer_name == "",
        ~PlPlanLoad.customer_name.in_(internal_names),
    )
    return exists(
        select(PlPlanLoad.id).where(
            PlPlanLoad.plan_id == PlPlan.id,
            PlPlanLoad.tenant_no == tenant_no,
            PlPlanLoad.del_flag == 0,
            customer_not_internal,
            ~_stop_contract_exists(
                tenant_no,
                PlPlan.id,
                PlPlanLoad.id,
                (PLAN_CONTRACT_SIDE_LOAD, PLAN_CONTRACT_SIDE_SUPPLIER),
            ),
        )
    )


def _delivery_stop_gap_expr(tenant_no: str):
    internal_names = tuple(INTERNAL_CUSTOMER_NAMES)
    customer_not_internal = or_(
        PlPlanDelivery.customer_name.is_(None),
        PlPlanDelivery.customer_name == "",
        ~PlPlanDelivery.customer_name.in_(internal_names),
    )
    return exists(
        select(PlPlanDelivery.id).where(
            PlPlanDelivery.plan_id == PlPlan.id,
            PlPlanDelivery.tenant_no == tenant_no,
            PlPlanDelivery.del_flag == 0,
            customer_not_internal,
            ~_stop_contract_exists(
                tenant_no,
                PlPlan.id,
                PlPlanDelivery.id,
                (PLAN_CONTRACT_SIDE_DELIVERY, PLAN_CONTRACT_SIDE_DEMANDER),
            ),
        )
    )


def _legacy_supplier_gap_expr(tenant_no: str):
    internal_names = tuple(INTERNAL_CUSTOMER_NAMES)
    has_load = exists(
        select(PlPlanLoad.id).where(
            PlPlanLoad.plan_id == PlPlan.id,
            PlPlanLoad.tenant_no == tenant_no,
            PlPlanLoad.del_flag == 0,
        )
    )
    supplier_not_internal = or_(
        PlPlan.supplier_name.is_(None),
        PlPlan.supplier_name == "",
        ~PlPlan.supplier_name.in_(internal_names),
    )
    return and_(
        ~has_load,
        supplier_not_internal,
        ~_stop_contract_exists(
            tenant_no,
            PlPlan.id,
            PlPlan.id,
            (PLAN_CONTRACT_SIDE_SUPPLIER,),
        ),
    )


def _legacy_demander_gap_expr(tenant_no: str):
    internal_names = tuple(INTERNAL_CUSTOMER_NAMES)
    has_delivery = exists(
        select(PlPlanDelivery.id).where(
            PlPlanDelivery.plan_id == PlPlan.id,
            PlPlanDelivery.tenant_no == tenant_no,
            PlPlanDelivery.del_flag == 0,
        )
    )
    customer_not_internal = or_(
        PlPlan.customer_name.is_(None),
        PlPlan.customer_name == "",
        ~PlPlan.customer_name.in_(internal_names),
    )
    return and_(
        ~has_delivery,
        customer_not_internal,
        ~_stop_contract_exists(
            tenant_no,
            PlPlan.id,
            PlPlan.id,
            (PLAN_CONTRACT_SIDE_DEMANDER,),
        ),
    )


def plan_has_contract_gap_exists(tenant_no: str):
    """列表筛选：任一段次（或旧计划级）存在合同缺口。"""
    return or_(
        _load_stop_gap_expr(tenant_no),
        _delivery_stop_gap_expr(tenant_no),
        _legacy_supplier_gap_expr(tenant_no),
        _legacy_demander_gap_expr(tenant_no),
    )


def _supplier_gap_expr(tenant_no: str):
    return or_(_load_stop_gap_expr(tenant_no), _legacy_supplier_gap_expr(tenant_no))


def _demander_gap_expr(tenant_no: str):
    return or_(_delivery_stop_gap_expr(tenant_no), _legacy_demander_gap_expr(tenant_no))


def build_plan_contract_gap_read(
    plan: PlPlan,
    *,
    supplier_contracts: list[PlanContractLink],
    demander_contracts: list[PlanContractLink],
    products: Optional[list[PlanProductRead]] = None,
    load_read: Optional[PlanLoadRead] = None,
    delivery_reads: Optional[list[PlanDeliveryRead]] = None,
) -> PlanContractGapRead:
    deliveries = list(delivery_reads or [])
    gap_sides = compute_plan_gap_sides_from_stops(
        load_read,
        deliveries,
        supplier_name=plan.supplier_name,
        customer_name=plan.customer_name,
        supplier_contract_count=len(supplier_contracts),
        demander_contract_count=len(demander_contracts),
    )
    return PlanContractGapRead(
        id=plan.id,
        planNo=plan.plan_no,
        planDate=plan.plan_date,
        planType=plan.plan_type,
        followStatus=plan.follow_status,
        supplierId=plan.supplier_id,
        supplierName=plan.supplier_name,
        customerId=plan.customer_id,
        customerName=plan.customer_name,
        gapSides=gap_sides,
        supplierContractExempt=side_is_exempt(plan.supplier_name),
        demanderContractExempt=side_is_exempt(plan.customer_name),
        supplierContracts=supplier_contracts,
        demanderContracts=demander_contracts,
        products=list(products or []),
        load=load_read,
        deliveries=deliveries,
    )


def _load_products_by_plan_ids(
    db: Session,
    plan_ids: list[str],
    *,
    tenant_no: str,
) -> dict[str, list[PlanProductRead]]:
    if not plan_ids:
        return {}
    rows = (
        db.query(PlPlanProduct)
        .filter(
            PlPlanProduct.plan_id.in_(plan_ids),
            PlPlanProduct.del_flag == 0,
        )
        .order_by(PlPlanProduct.order_num.asc(), PlPlanProduct.create_time.asc())
        .all()
    )
    product_ids = {(row.product_id or "").strip() for row in rows if (row.product_id or "").strip()}
    standard_by_id, alias_options_by_id = load_crm_product_alias_maps(
        db, tenant_no=tenant_no, product_ids=product_ids
    )
    out: dict[str, list[PlanProductRead]] = {pid: [] for pid in plan_ids}
    for row in rows:
        item = PlanProductRead.model_validate(row)
        display_name, _alias = resolve_product_display_names(
            saved_product_name=row.product_name,
            product_id=row.product_id,
            standard_by_id=standard_by_id,
            alias_options_by_id=alias_options_by_id,
        )
        item.product_name = display_name or (row.product_name or "")
        out.setdefault(row.plan_id, []).append(item)
    return out


def _parse_csv(value: Optional[str]) -> List[str]:
    if not value:
        return []
    return [item.strip() for item in str(value).split(",") if item.strip()]


def list_plan_contract_gaps(
    db: Session,
    *,
    tenant_no: str,
    user: "DbUserContext",
    page: int = 1,
    page_size: int = 20,
    plan_no: Optional[str] = None,
    gap_type: Optional[str] = None,
    customer_keyword: Optional[str] = None,
    plan_date_from: Optional[date] = None,
    plan_date_to: Optional[date] = None,
    plan_type: Optional[str] = None,
    plan_type_in: Optional[str] = None,
    follow_status: Optional[str] = None,
    q: Optional[str] = None,
    ids: Optional[list] = None,
    max_page_size: Optional[int] = None,
    sort_field: Optional[str] = None,
    sort_order: Optional[str] = None,
) -> tuple[int, list[PlanContractGapRead]]:
    query = db.query(PlPlan).filter(
        PlPlan.tenant_no == tenant_no,
        PlPlan.del_flag == 0,
        PlPlan.follow_status != "cancelled",
        plan_has_contract_gap_exists(tenant_no),
    )
    query = apply_pl_scope(query, PlPlan, user, db, tenant_no)

    gap_vals = [v.lower() for v in _parse_csv(gap_type)]
    if len(gap_vals) == 1:
        gap_type_norm = gap_vals[0]
        if gap_type_norm == "supplier":
            query = query.filter(_supplier_gap_expr(tenant_no))
        elif gap_type_norm == "demander":
            query = query.filter(_demander_gap_expr(tenant_no))
        elif gap_type_norm == "both":
            query = query.filter(
                and_(_supplier_gap_expr(tenant_no), _demander_gap_expr(tenant_no))
            )
    elif len(gap_vals) > 1:
        clauses = []
        for gap_type_norm in gap_vals:
            if gap_type_norm == "supplier":
                clauses.append(_supplier_gap_expr(tenant_no))
            elif gap_type_norm == "demander":
                clauses.append(_demander_gap_expr(tenant_no))
            elif gap_type_norm == "both":
                clauses.append(
                    and_(_supplier_gap_expr(tenant_no), _demander_gap_expr(tenant_no))
                )
        if clauses:
            query = query.filter(or_(*clauses))

    follow_statuses = _parse_csv(follow_status)
    if len(follow_statuses) == 1:
        query = query.filter(PlPlan.follow_status == follow_statuses[0])
    elif len(follow_statuses) > 1:
        query = query.filter(PlPlan.follow_status.in_(follow_statuses))

    if customer_keyword:
        kw = customer_keyword.strip()
        if kw:
            kw_lower = kw.lower()
            load_match = exists(
                select(PlPlanLoad.id).where(
                    PlPlanLoad.plan_id == PlPlan.id,
                    PlPlanLoad.tenant_no == tenant_no,
                    PlPlanLoad.del_flag == 0,
                    func.lower(PlPlanLoad.customer_name).contains(kw_lower),
                )
            )
            delivery_match = exists(
                select(PlPlanDelivery.id).where(
                    PlPlanDelivery.plan_id == PlPlan.id,
                    PlPlanDelivery.tenant_no == tenant_no,
                    PlPlanDelivery.del_flag == 0,
                    func.lower(PlPlanDelivery.customer_name).contains(kw_lower),
                )
            )
            query = query.filter(
                or_(
                    func.lower(PlPlan.supplier_name).contains(kw_lower),
                    func.lower(PlPlan.customer_name).contains(kw_lower),
                    load_match,
                    delivery_match,
                )
            )

    column_filters = build_plan_column_filters(
        db=db,
        tenant_no=tenant_no,
        plan_no=plan_no,
        plan_type=plan_type,
        plan_type_in=plan_type_in,
        plan_date_from=plan_date_from,
        plan_date_to=plan_date_to,
    )
    if column_filters:
        query = query.filter(*column_filters)

    if q:
        q_kw = q.strip()
        if q_kw:
            q_lower = q_kw.lower()
            load_match = exists(
                select(PlPlanLoad.id).where(
                    PlPlanLoad.plan_id == PlPlan.id,
                    PlPlanLoad.tenant_no == tenant_no,
                    PlPlanLoad.del_flag == 0,
                    func.lower(PlPlanLoad.customer_name).contains(q_lower),
                )
            )
            delivery_match = exists(
                select(PlPlanDelivery.id).where(
                    PlPlanDelivery.plan_id == PlPlan.id,
                    PlPlanDelivery.tenant_no == tenant_no,
                    PlPlanDelivery.del_flag == 0,
                    func.lower(PlPlanDelivery.customer_name).contains(q_lower),
                )
            )
            query = query.filter(
                or_(
                    PlPlan.plan_no.contains(q_kw),
                    func.lower(PlPlan.supplier_name).contains(q_lower),
                    func.lower(PlPlan.customer_name).contains(q_lower),
                    load_match,
                    delivery_match,
                )
            )

    if ids:
        query = query.filter(PlPlan.id.in_(ids))

    from app.services.pl.list_sort import order_plan

    query = query.order_by(
        order_plan(sort_field, sort_order, default=PlPlan.plan_date.desc()),
        PlPlan.create_time.desc(),
    )
    total = query.count()
    page = max(1, page)
    cap = int(max_page_size) if max_page_size else 500
    page_size = min(max(1, page_size), cap)
    offset = (page - 1) * page_size
    plans = query.offset(offset).limit(page_size).all()

    plan_ids = [plan.id for plan in plans]
    contracts_by_plan = load_plan_contracts_grouped(db, tenant_no, plan_ids)
    products_by_plan = _load_products_by_plan_ids(db, plan_ids, tenant_no=tenant_no)
    stops_by_plan = load_plan_stops_grouped(db, tenant_no, plan_ids)

    items: list[PlanContractGapRead] = []
    for plan in plans:
        supplier_contracts, demander_contracts = contracts_by_plan.get(plan.id, ([], []))
        load_read, delivery_reads = stops_by_plan.get(plan.id, (None, []))
        items.append(
            build_plan_contract_gap_read(
                plan,
                supplier_contracts=supplier_contracts,
                demander_contracts=demander_contracts,
                products=products_by_plan.get(plan.id, []),
                load_read=load_read,
                delivery_reads=delivery_reads,
            )
        )
    return total, items


def _write_stop_contract_rows(
    db: Session,
    *,
    tenant_no: str,
    plan_id: str,
    stop_id: str,
    side: str,
    contracts: list[PlanContractLink],
    user_id: Optional[str],
    dept_id: Optional[str],
) -> None:
    now = datetime.now()
    db.query(PlPlanContract).filter(
        PlPlanContract.plan_id == plan_id,
        PlPlanContract.tenant_no == tenant_no,
        PlPlanContract.del_flag == 0,
        PlPlanContract.stop_id == stop_id,
    ).update(
        {"del_flag": 1, "update_by": user_id, "update_time": now},
        synchronize_session=False,
    )
    order_num = 0
    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,
            )
        )


def _clear_legacy_side_contracts(
    db: Session,
    *,
    tenant_no: str,
    plan_id: str,
    legacy_sides: tuple[str, ...],
    user_id: Optional[str],
) -> None:
    now = datetime.now()
    db.query(PlPlanContract).filter(
        PlPlanContract.plan_id == plan_id,
        PlPlanContract.tenant_no == tenant_no,
        PlPlanContract.del_flag == 0,
        or_(PlPlanContract.stop_id.is_(None), PlPlanContract.stop_id == ""),
        PlPlanContract.side.in_(legacy_sides),
    ).update(
        {"del_flag": 1, "update_by": user_id, "update_time": now},
        synchronize_session=False,
    )


def update_plan_stop_contracts(
    db: Session,
    *,
    tenant_no: str,
    plan: PlPlan,
    stop_type: str,
    stop_id: str,
    contracts: list[PlanContractLink],
    user_id: Optional[str],
    dept_id: Optional[str],
) -> PlanContractGapRead:
    """保存某装货/送货段的合同关联。"""
    if plan.follow_status == "cancelled":
        raise HTTPException(status_code=400, detail="已作废的计划不能修改")

    stop_type_norm = (stop_type or "").strip().lower()
    stop_id_norm = (stop_id or "").strip()
    if stop_type_norm not in ("load", "delivery") or not stop_id_norm:
        raise HTTPException(status_code=422, detail="stopType 必须为 load 或 delivery，且 stopId 不能为空")

    if stop_type_norm == "load":
        stop = (
            db.query(PlPlanLoad)
            .filter(
                PlPlanLoad.id == stop_id_norm,
                PlPlanLoad.plan_id == plan.id,
                PlPlanLoad.tenant_no == tenant_no,
                PlPlanLoad.del_flag == 0,
            )
            .first()
        )
        side = PLAN_CONTRACT_SIDE_LOAD
        legacy_sides = (PLAN_CONTRACT_SIDE_SUPPLIER,)
        side_label = "装货段合同"
    else:
        stop = (
            db.query(PlPlanDelivery)
            .filter(
                PlPlanDelivery.id == stop_id_norm,
                PlPlanDelivery.plan_id == plan.id,
                PlPlanDelivery.tenant_no == tenant_no,
                PlPlanDelivery.del_flag == 0,
            )
            .first()
        )
        side = PLAN_CONTRACT_SIDE_DELIVERY
        legacy_sides = (PLAN_CONTRACT_SIDE_DEMANDER,)
        side_label = "送货段合同"

    if not stop:
        raise HTTPException(status_code=404, detail="段次不存在")

    if is_internal_customer_name(stop.customer_name):
        raise HTTPException(status_code=400, detail="内置客户无需补充合同")

    links = _normalize_contract_links(contracts)
    links = assert_at_most_one_contract(links, side_label)
    expected_customer_id = stop.customer_id
    resolved: list[PlanContractLink] = []
    for link in links:
        contract = assert_contract_selectable_for_plan(
            db,
            link.contract_id,
            tenant_no,
            expected_customer_id=expected_customer_id,
            side_label=side_label,
        )
        resolved.append(
            PlanContractLink(
                contractId=contract.id,
                contractNo=contract.contract_no or link.contract_no,
                stopId=stop_id_norm,
            )
        )

    _clear_legacy_side_contracts(
        db,
        tenant_no=tenant_no,
        plan_id=plan.id,
        legacy_sides=legacy_sides,
        user_id=user_id,
    )
    _write_stop_contract_rows(
        db,
        tenant_no=tenant_no,
        plan_id=plan.id,
        stop_id=stop_id_norm,
        side=side,
        contracts=resolved,
        user_id=user_id,
        dept_id=dept_id,
    )

    supplier_contracts, demander_contracts = load_plan_contracts_grouped(
        db, tenant_no, [plan.id]
    ).get(plan.id, ([], []))
    products = _load_products_by_plan_ids(db, [plan.id], tenant_no=tenant_no).get(plan.id, [])
    load_read, delivery_reads = load_plan_stops_grouped(db, tenant_no, [plan.id]).get(
        plan.id, (None, [])
    )
    return build_plan_contract_gap_read(
        plan,
        supplier_contracts=supplier_contracts,
        demander_contracts=demander_contracts,
        products=products,
        load_read=load_read,
        delivery_reads=delivery_reads,
    )


def update_plan_side_contracts(
    db: Session,
    *,
    tenant_no: str,
    plan: PlPlan,
    side: str,
    contracts: list[PlanContractLink],
    user_id: Optional[str],
    dept_id: Optional[str],
) -> PlanContractGapRead:
    """单侧保存计划合同关联（旧计划级接口，兼容保留）。"""
    if plan.follow_status == "cancelled":
        raise HTTPException(status_code=400, detail="已作废的计划不能修改")
    if side not in (PLAN_CONTRACT_SIDE_SUPPLIER, PLAN_CONTRACT_SIDE_DEMANDER):
        raise HTTPException(status_code=422, detail="side 必须为 supplier 或 demander")

    current_s, current_d = load_plan_contracts_grouped(db, tenant_no, [plan.id]).get(
        plan.id, ([], [])
    )
    links = _normalize_contract_links(contracts)
    expected_customer_id = (
        plan.supplier_id if side == PLAN_CONTRACT_SIDE_SUPPLIER else plan.customer_id
    )
    side_label = "供方合同" if side == PLAN_CONTRACT_SIDE_SUPPLIER else "需方合同"
    resolved: list[PlanContractLink] = []
    for link in links:
        contract = assert_contract_selectable_for_plan(
            db,
            link.contract_id,
            tenant_no,
            expected_customer_id=expected_customer_id,
            side_label=side_label,
        )
        resolved.append(
            PlanContractLink(
                contractId=contract.id,
                contractNo=contract.contract_no or link.contract_no,
            )
        )

    supplier_links = resolved if side == PLAN_CONTRACT_SIDE_SUPPLIER else current_s
    demander_links = resolved if side == PLAN_CONTRACT_SIDE_DEMANDER else current_d
    replace_plan_contracts(
        db,
        tenant_no=tenant_no,
        plan_id=plan.id,
        supplier_contracts=supplier_links,
        demander_contracts=demander_links,
        user_id=user_id,
        dept_id=dept_id,
    )

    supplier_contracts, demander_contracts = load_plan_contracts_grouped(
        db, tenant_no, [plan.id]
    ).get(plan.id, ([], []))
    products = _load_products_by_plan_ids(db, [plan.id], tenant_no=tenant_no).get(plan.id, [])
    load_read, delivery_reads = load_plan_stops_grouped(db, tenant_no, [plan.id]).get(
        plan.id, (None, [])
    )
    return build_plan_contract_gap_read(
        plan,
        supplier_contracts=supplier_contracts,
        demander_contracts=demander_contracts,
        products=products,
        load_read=load_read,
        delivery_reads=delivery_reads,
    )