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

from __future__ import annotations

from datetime import date

from fastapi import APIRouter, Depends
from sqlalchemy import func, text
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.models.db_models import DbInboundOutbound, DbWeighingSlip
from app.schemas.db.dashboard import (
    DashboardCargoStat,
    DashboardPartnerCargoStat,
    DashboardStats,
)

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


@router.get("/stats", response_model=DashboardStats)
def get_dashboard_stats(
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> DashboardStats:
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    today = date.today()

    today_inbound = float(
        db.query(func.coalesce(func.sum(DbInboundOutbound.quantity), 0))
        .filter(
            DbInboundOutbound.date == today,
            DbInboundOutbound.business_type == "入库",
            DbInboundOutbound.del_flag == 0,
            DbInboundOutbound.tenant_no == tenant_no,
        )
        .scalar()
        or 0
    )

    today_outbound = float(
        db.query(func.coalesce(func.sum(DbInboundOutbound.quantity), 0))
        .filter(
            DbInboundOutbound.date == today,
            DbInboundOutbound.business_type == "出库",
            DbInboundOutbound.del_flag == 0,
            DbInboundOutbound.tenant_no == tenant_no,
        )
        .scalar()
        or 0
    )

    abnormal_orders = int(
        db.query(func.count(DbWeighingSlip.id))
        .filter(
            DbWeighingSlip.is_voided == 0,
            DbWeighingSlip.del_flag == 0,
            DbWeighingSlip.tenant_no == tenant_no,
        )
        .scalar()
        or 0
    )

    return DashboardStats(
        today_inbound=today_inbound,
        today_outbound=today_outbound,
        abnormal_orders=abnormal_orders,
    )


@router.get("/inbound/by-cargo", response_model=list[DashboardCargoStat])
def get_inbound_by_cargo(
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> list[DashboardCargoStat]:
    return _cargo_stats(db, user, business_type="入库")


@router.get("/inbound/by-partner-cargo", response_model=list[DashboardPartnerCargoStat])
def get_inbound_by_partner_cargo(
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> list[DashboardPartnerCargoStat]:
    return _partner_cargo_stats(db, user, business_type="入库")


@router.get("/outbound/by-cargo", response_model=list[DashboardCargoStat])
def get_outbound_by_cargo(
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> list[DashboardCargoStat]:
    return _cargo_stats(db, user, business_type="出库")


@router.get("/outbound/by-partner-cargo", response_model=list[DashboardPartnerCargoStat])
def get_outbound_by_partner_cargo(
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> list[DashboardPartnerCargoStat]:
    return _partner_cargo_stats(db, user, business_type="出库")


def _cargo_stats(db: Session, user: DbUserContext, *, business_type: str) -> list[DashboardCargoStat]:
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    today = date.today()
    query = text(
        """
        SELECT
            COALESCE(io.product, ws.cargo_name, '未填写货物') AS label,
            COALESCE(SUM(io.quantity), 0) AS planned,
            COALESCE(SUM(ws.net_weight) / 1000.0, 0) AS actual
        FROM db_inbound_outbound io
        LEFT JOIN db_weighing_slip ws
            ON ws.inbound_outbound_id = io.id
            AND ws.is_voided = 0
            AND ws.del_flag = 0
            AND ws.tenant_no = io.tenant_no
        WHERE io.date = :today
            AND io.business_type = :business_type
            AND io.del_flag = 0
            AND io.tenant_no = :tenant_no
        GROUP BY COALESCE(io.product, ws.cargo_name, '未填写货物')
        ORDER BY planned DESC
        """
    )
    rows = db.execute(
        query,
        {"today": today, "business_type": business_type, "tenant_no": tenant_no},
    ).all()
    return [
        DashboardCargoStat(label=row.label, planned=float(row.planned), actual=float(row.actual))
        for row in rows
    ]


def _partner_cargo_stats(
    db: Session,
    user: DbUserContext,
    *,
    business_type: str,
) -> list[DashboardPartnerCargoStat]:
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    today = date.today()
    query = text(
        """
        SELECT
            CONCAT(
                COALESCE(io.partner_company, ''),
                ' - ',
                COALESCE(io.product, ws.cargo_name, '未填写货物')
            ) AS label,
            COALESCE(SUM(io.quantity), 0) AS planned,
            COALESCE(SUM(ws.net_weight) / 1000.0, 0) AS actual
        FROM db_inbound_outbound io
        LEFT JOIN db_weighing_slip ws
            ON ws.inbound_outbound_id = io.id
            AND ws.is_voided = 0
            AND ws.del_flag = 0
            AND ws.tenant_no = io.tenant_no
        WHERE io.date = :today
            AND io.business_type = :business_type
            AND io.del_flag = 0
            AND io.tenant_no = :tenant_no
        GROUP BY io.partner_company, io.product, ws.cargo_name
        ORDER BY io.partner_company, planned DESC
        """
    )
    rows = db.execute(
        query,
        {"today": today, "business_type": business_type, "tenant_no": tenant_no},
    ).all()
    return [
        DashboardPartnerCargoStat(
            label=row.label,
            planned=float(row.planned),
            actual=float(row.actual),
        )
        for row in rows
    ]