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

from typing import Optional

from datetime import date, datetime

from fastapi import APIRouter, Depends, HTTPException, Query, status
from pydantic import BaseModel
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.tenant_context import get_tenant_no
from app.core.utils import generate_uuid
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,
    parse_date_str,
    parse_datetime_str,
    plan_to_read,
    plan_to_read_with_fill_link,
    weighing_to_read,
)
from app.services.db.product_helpers import (
    apply_product_snapshot_to_plan,
    resolve_crm_product,
)

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


@router.get("/", response_model=list[PurchasePlanWithFillLink])
def list_plans(
    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]:
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    query = _base_query(db, tenant_no).order_by(DbPurchasePlan.create_time.desc())

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

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

    if customer_name:
        query = query.filter(
            func.lower(DbPurchasePlan.customer_name).contains(customer_name.lower())
        )

    if product:
        query = query.filter(func.lower(DbPurchasePlan.product).contains(product.lower()))

    if business_type:
        query = query.filter(DbPurchasePlan.business_type == business_type)

    if execution_status:
        query = query.filter(DbPurchasePlan.execution_status == execution_status)

    if created_from:
        query = query.filter(DbPurchasePlan.create_time >= created_from)

    if created_to:
        query = query.filter(DbPurchasePlan.create_time <= f"{created_to} 23:59:59")

    if start_date_from:
        start_from = parse_date_str(start_date_from)
        if start_from:
            query = query.filter(DbPurchasePlan.start_date >= start_from)

    if start_date_to:
        start_to = parse_date_str(start_date_to)
        if start_to:
            query = query.filter(DbPurchasePlan.start_date <= start_to)

    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

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


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)


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)
    return plan_to_read(plan)


@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)
    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)


@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)
    plan.del_flag = 1
    plan.update_by = user.id
    db.commit()


def _get_plan(db: Session, tenant_no: str, plan_id: str) -> 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",
        )
    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)

    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()

    return PlanRelatedData(
        inventories=[inventory_to_read(inv) for inv in inventories],
        weighings=[weighing_to_read(w) 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()
    )

    return ThirdPartyPlanPageData(
        plan=plan_to_read(plan),
        today_inventories=[inventory_to_read(inv) for inv in inventories],
        month_inventories=[inventory_to_read(inv) for inv in month_history],
    )