shanxia2.0/backend/app/api/db/plans.py

from typing import Any, List, Optional

from datetime import date, datetime

from fastapi import APIRouter, Depends, HTTPException, Query, Request, status
from pydantic import BaseModel, Field
from sqlalchemy import func, or_
from sqlalchemy.orm import Session

from app.core.database import get_db
from app.core.db_auth import DbUserContext, get_db_current_user
from app.core.delete_guard import assert_row_deletable_by_creator, user_info_from_context
from app.core.tenant_context import get_tenant_no
from app.core.utils import generate_uuid
from app.services.db_data_scope import apply_db_scope, assert_db_row_in_scope
from app.models.db_models import DbInboundOutbound, DbPurchasePlan, DbWeighingSlip
from app.schemas.db.inventory import InboundOutboundRead
from app.schemas.db.plans import (
    PurchasePlanCreate,
    PurchasePlanRead,
    PurchasePlanUpdate,
    PurchasePlanWithFillLink,
)
from app.schemas.db.weighing import WeighingSlipRead
from app.services.db.mappers import (
    inventory_to_read,
    load_user_display_name_map,
    parse_date_str,
    parse_datetime_str,
    plan_to_read,
    plan_to_read_with_fill_link,
    weighing_to_read,
)
from app.services.db.plan_list_filters import build_plan_column_filters
from app.services.db.product_helpers import (
    apply_product_snapshot_to_plan,
    resolve_crm_product,
)
from app.utils.excel_export import (
    EXPORT_MAX_ROWS,
    ExportColumn,
    ExportValidationError,
    export_rows_to_xlsx_response,
    filter_export_columns,
)

router = APIRouter(prefix="/plans", tags=["plans"])

STATUS_THIRD_PARTY_FILL = "待三方填写"

_PLAN_DATE_FIELDS = {"plan_date", "start_date", "end_date"}
_PLAN_REMARK_FIELD = "remarks"


def _base_query(db: Session, tenant_no: str):
    return db.query(DbPurchasePlan).filter(
        DbPurchasePlan.tenant_no == tenant_no,
        DbPurchasePlan.del_flag == 0,
    )


def _merge_query(request: Request, **known) -> dict:
    out = {k: v for k, v in known.items() if v is not None and v != ""}
    for k, v in request.query_params.multi_items():
        if k in ("page", "pageSize"):
            continue
        if k not in out and v not in (None, ""):
            out[k] = v
    return out


def _param_get(params: dict, *keys: str) -> Any:
    for key in keys:
        if key in params and params[key] not in (None, ""):
            return params[key]
    return None


def _filters_from_params(params: dict) -> list:
    return build_plan_column_filters(
        plan_no=_param_get(params, "planNo", "plan_no"),
        customer_name=_param_get(params, "customerName", "customer_name"),
        product=_param_get(params, "product"),
        specification=_param_get(params, "specification"),
        business_type=_param_get(params, "businessType", "business_type"),
        business_type_in=_param_get(params, "businessTypeIn", "business_type_in"),
        execution_status=_param_get(params, "executionStatus", "execution_status"),
        execution_status_in=_param_get(params, "executionStatusIn", "execution_status_in"),
        planned_quantity_op=_param_get(params, "plannedQuantityOp", "planned_quantity_op"),
        planned_quantity=_param_get(params, "plannedQuantity", "planned_quantity"),
        planned_quantity_min=_param_get(params, "plannedQuantityMin", "planned_quantity_min"),
        planned_quantity_max=_param_get(params, "plannedQuantityMax", "planned_quantity_max"),
        executed_quantity_op=_param_get(params, "executedQuantityOp", "executed_quantity_op"),
        executed_quantity=_param_get(params, "executedQuantity", "executed_quantity"),
        executed_quantity_min=_param_get(params, "executedQuantityMin", "executed_quantity_min"),
        executed_quantity_max=_param_get(params, "executedQuantityMax", "executed_quantity_max"),
        created_from=_param_get(params, "createdFrom", "created_from"),
        created_to=_param_get(params, "createdTo", "created_to"),
        start_date_from=_param_get(params, "startDateFrom", "start_date_from"),
        start_date_to=_param_get(params, "startDateTo", "start_date_to"),
    )


def _body_to_params(body: "PlanExportBody") -> dict:
    data = body.model_dump(by_alias=True, exclude_none=True)
    data.pop("ids", None)
    data.pop("columns", None)
    extra = getattr(body, "model_extra", None) or {}
    for key, value in extra.items():
        if key not in data and value not in (None, ""):
            data[key] = value
    return data


def _fetch_plans(
    db: Session, user: DbUserContext, params: dict
) -> list[PurchasePlanWithFillLink]:
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    query = _base_query(db, tenant_no).order_by(DbPurchasePlan.create_time.desc())

    q = _param_get(params, "q")
    if q:
        keyword = str(q).lower()
        query = query.filter(
            or_(
                func.lower(DbPurchasePlan.customer_name).contains(keyword),
                func.lower(DbPurchasePlan.product).contains(keyword),
                func.lower(DbPurchasePlan.plan_no).contains(keyword),
            )
        )

    column_filters = _filters_from_params(params)
    if column_filters:
        query = query.filter(*column_filters)

    query = apply_db_scope(query, DbPurchasePlan, user, db, tenant_no)
    plans = query.all()
    if not plans:
        return []

    plan_ids = [p.id for p in plans]
    fill_links = (
        db.query(DbInboundOutbound)
        .filter(
            DbInboundOutbound.tenant_no == tenant_no,
            DbInboundOutbound.del_flag == 0,
            DbInboundOutbound.purchase_plan_id.in_(plan_ids),
            DbInboundOutbound.status == STATUS_THIRD_PARTY_FILL,
        )
        .all()
    )
    plan_id_to_inv_id = {
        inv.purchase_plan_id: inv.id for inv in fill_links if inv.purchase_plan_id
    }

    plan_nos = [p.plan_no for p in plans if p.plan_no]
    executed_sum_map: dict[tuple[Optional[str], Optional[str]], float] = {}

    if plan_ids or plan_nos:
        weighing_filters = [DbWeighingSlip.is_voided == 0, DbWeighingSlip.del_flag == 0]
        weighing_filters.append(DbWeighingSlip.tenant_no == tenant_no)
        id_filters = []
        if plan_ids:
            id_filters.append(DbWeighingSlip.purchase_plan_id.in_(plan_ids))
        if plan_nos:
            id_filters.append(DbWeighingSlip.plan_no.in_(plan_nos))
        weighing_filters.append(or_(*id_filters))

        weighing_rows = (
            db.query(
                DbWeighingSlip.purchase_plan_id,
                DbWeighingSlip.plan_no,
                func.sum(DbWeighingSlip.net_weight).label("total_executed"),
            )
            .filter(*weighing_filters)
            .group_by(DbWeighingSlip.purchase_plan_id, DbWeighingSlip.plan_no)
            .all()
        )
        for row in weighing_rows:
            key = (row.purchase_plan_id, row.plan_no)
            executed_sum_map[key] = float(row.total_executed or 0)

    def calc_executed_for_plan(plan: DbPurchasePlan) -> float:
        if not executed_sum_map:
            return 0.0
        total = 0.0
        for (pid, pno), executed in executed_sum_map.items():
            if pid is not None and pid == plan.id:
                total += executed
            elif pno and plan.plan_no and pno == plan.plan_no:
                total += executed
        return total

    user_name_map = load_user_display_name_map(db, [p.create_by for p in plans])
    result_list: list[PurchasePlanWithFillLink] = []
    for plan in plans:
        result_list.append(
            plan_to_read_with_fill_link(
                plan,
                executed_quantity=calc_executed_for_plan(plan),
                fill_link_inventory_id=plan_id_to_inv_id.get(plan.id),
                user_name_map=user_name_map,
            )
        )
    return result_list


@router.get("/", response_model=list[PurchasePlanWithFillLink])
def list_plans(
    request: Request,
    q: Optional[str] = Query(default=None, description="模糊搜索往来单位或产品"),
    plan_no: Optional[str] = Query(default=None, alias="planNo", description="计划号"),
    customer_name: Optional[str] = Query(
        default=None, alias="customerName", description="往来单位"
    ),
    product: Optional[str] = Query(default=None, description="产品"),
    business_type: Optional[str] = Query(default=None, alias="businessType", description="业务类型"),
    execution_status: Optional[str] = Query(
        default=None, alias="executionStatus", description="执行状态"
    ),
    created_from: Optional[str] = Query(default=None, alias="createdFrom", description="创建开始日期"),
    created_to: Optional[str] = Query(default=None, alias="createdTo", description="创建结束日期"),
    start_date_from: Optional[str] = Query(
        default=None,
        alias="startDateFrom",
        description="筛选区间开始：与计划有效期（起始～截止）有交集",
    ),
    start_date_to: Optional[str] = Query(
        default=None,
        alias="startDateTo",
        description="筛选区间结束：与计划有效期（起始～截止）有交集",
    ),
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> list[PurchasePlanWithFillLink]:
    params = _merge_query(
        request,
        q=q,
        planNo=plan_no,
        customerName=customer_name,
        product=product,
        businessType=business_type,
        executionStatus=execution_status,
        createdFrom=created_from,
        createdTo=created_to,
        startDateFrom=start_date_from,
        startDateTo=start_date_to,
    )
    return _fetch_plans(db, user, params)


PLAN_EXPORT_COLUMNS = [
    ExportColumn(title="计划号", field="planNo"),
    ExportColumn(title="业务类型", field="businessType"),
    ExportColumn(title="往来单位", field="customerName"),
    ExportColumn(title="计划日期", field="planDate"),
    ExportColumn(title="起始日期", field="startDate"),
    ExportColumn(title="截止日期", field="endDate"),
    ExportColumn(title="产品", field="product"),
    ExportColumn(title="规格", field="specification"),
    ExportColumn(title="车牌号", field="vehicleNo"),
    ExportColumn(title="计划量(吨)", field="plannedQuantity"),
    ExportColumn(title="已执行量(吨)", field="executedQuantity"),
    ExportColumn(title="执行状态", field="executionStatus"),
    ExportColumn(title="备注", field="remarks"),
    ExportColumn(title="自动生成", field="autoGenerate"),
    ExportColumn(title="创建人", field="createByName"),
]


class PlanExportBody(BaseModel):
    q: Optional[str] = None
    plan_no: Optional[str] = Field(default=None, alias="planNo")
    customer_name: Optional[str] = Field(default=None, alias="customerName")
    product: Optional[str] = None
    business_type: Optional[str] = Field(default=None, alias="businessType")
    execution_status: Optional[str] = Field(default=None, alias="executionStatus")
    created_from: Optional[str] = Field(default=None, alias="createdFrom")
    created_to: Optional[str] = Field(default=None, alias="createdTo")
    start_date_from: Optional[str] = Field(default=None, alias="startDateFrom")
    start_date_to: Optional[str] = Field(default=None, alias="startDateTo")
    ids: Optional[List[str]] = None
    columns: Optional[List[str]] = None

    model_config = {"populate_by_name": True, "extra": "allow"}


@router.post("/export")
def export_plans(
    body: PlanExportBody,
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
):
    """导出计划列表（当前筛选全量，或 ids 已选）。"""
    try:
        items = _fetch_plans(db, user, _body_to_params(body))
        if body.ids:
            id_set = set(body.ids)
            items = [x for x in items if x.id in id_set]
        if len(items) > EXPORT_MAX_ROWS:
            raise ExportValidationError(f"导出超过 {EXPORT_MAX_ROWS} 条，请缩小筛选范围")
        rows = []
        for item in items:
            data = item.model_dump(by_alias=True)
            data["autoGenerate"] = "是" if data.get("autoGenerateEnabled") else "否"
            eq = data.get("executedQuantity") or 0
            try:
                data["executedQuantity"] = round(float(eq) / 1000, 2)
            except (TypeError, ValueError):
                data["executedQuantity"] = 0
            if not data.get("createByName"):
                data["createByName"] = data.get("createBy") or ""
            rows.append(data)
        cols = filter_export_columns(PLAN_EXPORT_COLUMNS, body.columns)
        date_str = datetime.now().strftime("%Y%m%d")
        tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
        return export_rows_to_xlsx_response(
            db,
            tenant_no=tenant_no,
            sheet_title="计划列表",
            columns=cols,
            rows=rows,
            filename_stem=f"计划列表_{date_str}",
        )
    except ExportValidationError as e:
        raise HTTPException(status_code=400, detail=str(e)) from e


def _generate_plan_no() -> str:
    from app.services.db.plan_helpers import generate_plan_no

    return generate_plan_no()


@router.post("/", response_model=PurchasePlanRead, status_code=status.HTTP_201_CREATED)
def create_plan(
    payload: PurchasePlanCreate,
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> PurchasePlanRead:
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    plan_no = payload.plan_no if payload.plan_no else _generate_plan_no()

    execution_status = payload.execution_status
    if not execution_status:
        execution_status = "待入库" if payload.business_type == "入库" else "待出库"

    snapshot = resolve_crm_product(db, tenant_no, payload.product_id)

    plan = DbPurchasePlan(
        id=generate_uuid(),
        tenant_no=tenant_no,
        plan_no=plan_no,
        plan_date=parse_date_str(payload.plan_date) or date.today(),
        start_date=parse_date_str(payload.start_date),
        end_date=parse_date_str(payload.end_date),
        business_type=payload.business_type,
        planned_quantity=payload.planned_quantity,
        vehicle_no=payload.vehicle_no or "",
        customer_name=payload.customer_name.strip() if payload.customer_name else None,
        remark=payload.remarks,
        execution_status=execution_status,
        executed_at=None,
        last_weighing_slip_id=None,
        executed_quantity=0,
        auto_generate_enabled=payload.auto_generate_enabled or 0,
        auto_generate_end_month=payload.auto_generate_end_month,
        auto_generate_last_month=None,
        create_by=user.id,
        create_dept_id=user.dept_id,
    )
    apply_product_snapshot_to_plan(plan, snapshot)
    db.add(plan)
    db.commit()
    db.refresh(plan)
    return plan_to_read(plan, db=db)


class AutoGenerateMonthlyResult(BaseModel):
    created_count: int
    skipped_reason: Optional[str] = None
    current_month: str


@router.post("/run-auto-generate", response_model=AutoGenerateMonthlyResult)
def run_auto_generate_monthly(
    force: bool = Query(default=False, description="true 时跳过「仅每月1日」限制，供验收手动触发"),
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> AutoGenerateMonthlyResult:
    """手动触发计划按月自动生成（定时任务同款逻辑）。"""
    get_tenant_no(db, user.to_tenant_user_dict())
    from app.services.db.auto_generate_plans import run_auto_generate_monthly_plans

    result = run_auto_generate_monthly_plans(db, force=force)
    return AutoGenerateMonthlyResult(**result)


@router.get("/{id}", response_model=PurchasePlanRead)
def get_plan(
    id: str,
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> PurchasePlanRead:
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    plan = _get_plan(db, tenant_no, id, user)
    return plan_to_read(plan, db=db)


@router.patch("/{id}", response_model=PurchasePlanRead)
def update_plan(
    id: str,
    payload: PurchasePlanUpdate,
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> PurchasePlanRead:
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    plan = _get_plan(db, tenant_no, id, user)
    payload_dict = payload.model_dump(exclude_unset=True)
    _apply_plan_product_update(db, tenant_no, plan, payload_dict)
    _apply_plan_update(plan, payload_dict)
    plan.update_by = user.id
    db.commit()
    db.refresh(plan)
    return plan_to_read(plan, db=db)


@router.delete("/{id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_plan(
    id: str,
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> None:
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    plan = _get_plan(db, tenant_no, id, user)

    if plan.execution_status in ("已入库", "已出库"):
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="已完成的计划不能删除",
        )

    assert_row_deletable_by_creator(plan, user_info_from_context(user))
    plan.del_flag = 1
    plan.update_by = user.id
    db.commit()


def _get_plan(
    db: Session, tenant_no: str, plan_id: str, user: DbUserContext
) -> DbPurchasePlan:
    plan = (
        _base_query(db, tenant_no)
        .filter(DbPurchasePlan.id == plan_id)
        .one_or_none()
    )
    if plan is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Plan {plan_id} not found",
        )
    assert_db_row_in_scope(db, plan, DbPurchasePlan, user, tenant_no)
    return plan


def _get_plan_by_id_only(db: Session, plan_id: str) -> DbPurchasePlan:
    plan = (
        db.query(DbPurchasePlan)
        .filter(DbPurchasePlan.id == plan_id, DbPurchasePlan.del_flag == 0)
        .one_or_none()
    )
    if plan is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Plan {plan_id} not found",
        )
    return plan


def _apply_plan_product_update(
    db: Session, tenant_no: str, plan: DbPurchasePlan, payload_dict: dict
) -> None:
    if "product" in payload_dict and "product_id" not in payload_dict:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="请从 CRM 重新选择产品",
        )
    product_id = payload_dict.pop("product_id", None)
    payload_dict.pop("product", None)
    payload_dict.pop("specification", None)
    if product_id is not None:
        snapshot = resolve_crm_product(db, tenant_no, product_id)
        apply_product_snapshot_to_plan(plan, snapshot)


def _apply_plan_update(plan: DbPurchasePlan, payload_dict: dict) -> None:
    for key, value in payload_dict.items():
        if key == _PLAN_REMARK_FIELD:
            if isinstance(value, str):
                value = value.strip()
            plan.remark = value
            continue
        if key in _PLAN_DATE_FIELDS:
            if value is None:
                setattr(plan, key, None)
            else:
                setattr(plan, key, parse_date_str(str(value)))
            continue
        if key == "executed_at":
            plan.executed_at = parse_datetime_str(str(value)) if value else None
            continue
        if isinstance(value, str):
            value = value.strip()
        setattr(plan, key, value)


class PlanRelatedData(BaseModel):
    inventories: list[InboundOutboundRead]
    weighings: list[WeighingSlipRead]


@router.get("/{id}/related", response_model=PlanRelatedData)
def get_plan_related(
    id: str,
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> PlanRelatedData:
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    plan = _get_plan(db, tenant_no, id, user)

    inventories = (
        db.query(DbInboundOutbound)
        .filter(
            DbInboundOutbound.tenant_no == tenant_no,
            DbInboundOutbound.del_flag == 0,
            DbInboundOutbound.purchase_plan_id == plan.id,
        )
        .order_by(DbInboundOutbound.date.desc(), DbInboundOutbound.id.desc())
        .all()
    )

    weighing_query = db.query(DbWeighingSlip).filter(
        DbWeighingSlip.tenant_no == tenant_no,
        DbWeighingSlip.del_flag == 0,
        DbWeighingSlip.is_voided == 0,
    )
    if plan.plan_no:
        weighing_query = weighing_query.filter(
            or_(
                DbWeighingSlip.plan_no == plan.plan_no,
                DbWeighingSlip.purchase_plan_id == plan.id,
            )
        )
    else:
        weighing_query = weighing_query.filter(DbWeighingSlip.purchase_plan_id == plan.id)
    weighings = weighing_query.order_by(DbWeighingSlip.create_time.desc()).all()

    user_name_map = load_user_display_name_map(
        db,
        [inv.create_by for inv in inventories] + [w.create_by for w in weighings],
    )
    return PlanRelatedData(
        inventories=[inventory_to_read(inv, user_name_map=user_name_map) for inv in inventories],
        weighings=[weighing_to_read(w, user_name_map=user_name_map) for w in weighings],
    )


class ThirdPartyPlanPageData(BaseModel):
    plan: PurchasePlanRead
    today_inventories: list[InboundOutboundRead]
    month_inventories: list[InboundOutboundRead]


@router.get("/{id}/third-party-page", response_model=ThirdPartyPlanPageData)
def get_third_party_plan_page(
    id: str,
    db: Session = Depends(get_db),
) -> ThirdPartyPlanPageData:
    """三方 H5 填写页初始化数据（无需门户登录，与原系统一致）。"""
    plan = _get_plan_by_id_only(db, id)
    tenant_no = plan.tenant_no

    today = date.today()
    start = plan.start_date
    end = plan.end_date

    if start and today < start:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="计划尚未生效，当前日期不允许填写出入库单",
        )
    if end and today > end:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="计划已过期，当前日期不允许填写出入库单",
        )

    today_str = today.strftime("%Y-%m-%d")
    today_date = parse_date_str(today_str)
    inventories = (
        db.query(DbInboundOutbound)
        .filter(
            DbInboundOutbound.tenant_no == tenant_no,
            DbInboundOutbound.del_flag == 0,
            DbInboundOutbound.purchase_plan_id == plan.id,
            DbInboundOutbound.date == today_date,
        )
        .order_by(DbInboundOutbound.date.desc(), DbInboundOutbound.id.desc())
        .all()
    )

    month_start = today.replace(day=1)
    month_history = (
        db.query(DbInboundOutbound)
        .filter(
            DbInboundOutbound.tenant_no == tenant_no,
            DbInboundOutbound.del_flag == 0,
            DbInboundOutbound.purchase_plan_id == plan.id,
            DbInboundOutbound.date >= month_start,
            DbInboundOutbound.date < today,
        )
        .order_by(DbInboundOutbound.date.desc(), DbInboundOutbound.id.desc())
        .all()
    )

    user_name_map = load_user_display_name_map(
        db,
        [plan.create_by]
        + [inv.create_by for inv in inventories]
        + [inv.create_by for inv in month_history],
    )
    return ThirdPartyPlanPageData(
        plan=plan_to_read(plan, user_name_map=user_name_map),
        today_inventories=[
            inventory_to_read(inv, user_name_map=user_name_map) for inv in inventories
        ],
        month_inventories=[
            inventory_to_read(inv, user_name_map=user_name_map) for inv in month_history
        ],
    )