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

from __future__ import annotations

from datetime import date
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.models.pl_models import PlPlan, PlPlanContract, PlPlanProduct
from app.schemas.pl.plan import PlanContractGapRead, PlanContractLink, PlanProductRead
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,
)

if TYPE_CHECKING:
    from app.core.db_auth import DbUserContext


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 not is_internal_customer_name(supplier_name) and supplier_contract_count <= 0:
        gaps.append("supplier")
    if not is_internal_customer_name(customer_name) and demander_contract_count <= 0:
        gaps.append("demander")
    return gaps


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


def _side_contract_exists(tenant_no: str, side: str):
    return exists(
        select(PlPlanContract.id).where(
            PlPlanContract.plan_id == PlPlan.id,
            PlPlanContract.tenant_no == tenant_no,
            PlPlanContract.del_flag == 0,
            PlPlanContract.side == side,
        )
    )


def _supplier_gap_expr(tenant_no: str):
    internal_names = tuple(INTERNAL_CUSTOMER_NAMES)
    supplier_not_internal = or_(
        PlPlan.supplier_name.is_(None),
        PlPlan.supplier_name == "",
        ~PlPlan.supplier_name.in_(internal_names),
    )
    return and_(
        supplier_not_internal,
        ~_side_contract_exists(tenant_no, PLAN_CONTRACT_SIDE_SUPPLIER),
    )


def _demander_gap_expr(tenant_no: str):
    internal_names = tuple(INTERNAL_CUSTOMER_NAMES)
    customer_not_internal = or_(
        PlPlan.customer_name.is_(None),
        PlPlan.customer_name == "",
        ~PlPlan.customer_name.in_(internal_names),
    )
    return and_(
        customer_not_internal,
        ~_side_contract_exists(tenant_no, PLAN_CONTRACT_SIDE_DEMANDER),
    )


def plan_has_contract_gap_exists(tenant_no: str):
    """列表筛选：计划存在合同缺口（非内置侧缺少有效 pl_plan_contract 关联）。"""
    return or_(_supplier_gap_expr(tenant_no), _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,
) -> PlanContractGapRead:
    gap_sides = compute_contract_gap_sides(
        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 []),
    )


def _load_products_by_plan_ids(
    db: Session,
    plan_ids: list[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()
    )
    out: dict[str, list[PlanProductRead]] = {pid: [] for pid in plan_ids}
    for row in rows:
        out.setdefault(row.plan_id, []).append(PlanProductRead.model_validate(row))
    return out


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,
    q: 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)

    if plan_no:
        query = query.filter(PlPlan.plan_no.contains(plan_no))

    gap_type_norm = (gap_type or "").strip().lower()
    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))
        )

    if customer_keyword:
        kw = customer_keyword.strip()
        if kw:
            kw_lower = kw.lower()
            query = query.filter(
                or_(
                    func.lower(PlPlan.supplier_name).contains(kw_lower),
                    func.lower(PlPlan.customer_name).contains(kw_lower),
                )
            )

    if plan_date_from:
        query = query.filter(PlPlan.plan_date >= plan_date_from)

    if plan_date_to:
        query = query.filter(PlPlan.plan_date <= plan_date_to)

    if q:
        q_kw = q.strip()
        if q_kw:
            q_lower = q_kw.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),
                )
            )

    query = query.order_by(PlPlan.plan_date.desc(), PlPlan.create_time.desc())
    total = query.count()
    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)

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


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]).get(plan.id, [])
    return build_plan_contract_gap_read(
        plan,
        supplier_contracts=supplier_contracts,
        demander_contracts=demander_contracts,
        products=products,
    )