JhHardwareWRS_BackPoint/app/routers/usage_stats.py

from datetime import date
from math import ceil

from fastapi import APIRouter, Depends, HTTPException, Query, Response, status
from sqlalchemy import and_, select
from sqlalchemy.orm import Session, selectinload

from app.database import get_db
from app.deps import require_roles
from app.models import (
    Equipment,
    Personnel,
    ProductionReport,
    ProductionReportAllocation,
    ProductionReportItem,
    ReportStatus,
    Role,
)
from app.schemas import (
    PageResponse,
    UsageStatsDailyRow,
    UsageStatsDetailOut,
    UsageStatsReportRow,
    UsageStatsRow,
)
from app.services.attendance_points import accessible_point_names, require_attendance_point_access
from app.services.common import round2
from app.services.report_lifecycle import purge_expired_voided_reports
from app.services.usage_stats import UsageStatRow, build_usage_stats_from_allocations
from app.services.usage_stats_export import export_usage_stats_rows

router = APIRouter(prefix="/api/usage-stats", tags=["usage-stats"])


def _clean(value: str | None) -> str:
    return str(value or "").strip()


def _requested_points(db: Session, user: Personnel, attendance_point_name: str | None) -> list[str]:
    requested_name = _clean(attendance_point_name)
    if requested_name:
        try:
            return [require_attendance_point_access(db, user, requested_name)]
        except ValueError as exc:
            raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc)) from exc
        except PermissionError as exc:
            raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=str(exc)) from exc
    return accessible_point_names(db, user)


def _equipment_type_map(db: Session, point_names: list[str]) -> dict[tuple[str, str], str]:
    if not point_names:
        return {}
    rows = db.scalars(select(Equipment).where(Equipment.attendance_point_name.in_(point_names))).all()
    return {(_clean(row.attendance_point_name), _clean(row.device_no)): row.device_type for row in rows}


def _approved_allocations_query(
    point_names: list[str],
    start_date: date | None,
    end_date: date | None,
):
    query = (
        select(ProductionReportAllocation)
        .join(ProductionReport, ProductionReport.id == ProductionReportAllocation.report_id)
        .join(
            ProductionReportItem,
            and_(
                ProductionReportItem.id == ProductionReportAllocation.report_item_id,
                ProductionReportItem.report_id == ProductionReportAllocation.report_id,
            ),
        )
        .options(
            selectinload(ProductionReportAllocation.report).selectinload(ProductionReport.employee),
            selectinload(ProductionReportAllocation.item),
        )
        .where(
            ProductionReport.status == ReportStatus.approved,
            ProductionReport.is_voided.is_(False),
            ProductionReport.attendance_point_name.in_(point_names),
            ProductionReportAllocation.attendance_point_name.in_(point_names),
            ProductionReportItem.attendance_point_name.in_(point_names),
        )
    )
    if start_date:
        query = query.where(ProductionReportAllocation.allocation_date >= start_date)
    if end_date:
        query = query.where(ProductionReportAllocation.allocation_date <= end_date)
    return query.order_by(
        ProductionReportAllocation.allocation_date.desc(),
        ProductionReportAllocation.report_id.desc(),
        ProductionReportAllocation.id.desc(),
    )


def _sort_usage_rows(rows: list[UsageStatRow], sort_by: str) -> list[UsageStatRow]:
    if sort_by == "count":
        return sorted(rows, key=lambda row: (-row.report_count, -row.value, row.name))
    return sorted(rows, key=lambda row: (-row.value, -row.report_count, row.name))


def _usage_row_search_text(row: UsageStatRow) -> str:
    values = [
        row.id,
        row.category,
        row.attendance_point_name,
        row.name,
        row.object_type,
        row.metric_kind,
        row.product_name,
        row.process_name,
        row.stamping_method,
        row.value,
        row.report_count,
        *row.tags,
    ]
    return " ".join(str(value or "") for value in values).lower()


def _filter_usage_rows(rows: list[UsageStatRow], keyword: str | None) -> list[UsageStatRow]:
    text = _clean(keyword).lower()
    if not text:
        return rows
    return [row for row in rows if text in _usage_row_search_text(row)]


def _to_schema_row(row: UsageStatRow) -> UsageStatsRow:
    return UsageStatsRow(
        id=row.id,
        category=row.category,
        attendance_point_name=row.attendance_point_name,
        name=row.name,
        object_type=row.object_type,
        metric_kind=row.metric_kind,
        value=row.value,
        report_count=row.report_count,
        product_name=row.product_name,
        process_name=row.process_name,
        stamping_method=row.stamping_method,
        tags=list(row.tags),
    )


def _to_detail_out(
    row: UsageStatRow,
    *,
    daily_rows: list[UsageStatsDailyRow] | None = None,
    report_rows: list[UsageStatsReportRow] | None = None,
) -> UsageStatsDetailOut:
    return UsageStatsDetailOut(
        id=row.id,
        category=row.category,
        attendance_point_name=row.attendance_point_name,
        name=row.name,
        object_type=row.object_type,
        metric_kind=row.metric_kind,
        value=row.value,
        report_count=row.report_count,
        product_name=row.product_name,
        process_name=row.process_name,
        stamping_method=row.stamping_method,
        tags=list(row.tags),
        daily_rows=daily_rows or [],
        report_rows=report_rows or [],
    )


def _matches_target(
    row: UsageStatRow,
    *,
    id: str,
    category: str,
    attendance_point_name: str,
    name: str,
    product_name: str,
    process_name: str,
    stamping_method: str,
    metric_kind: str,
) -> bool:
    if id:
        return row.id == id
    if row.category != category:
        return False
    if attendance_point_name and row.attendance_point_name != attendance_point_name:
        return False
    if metric_kind and row.metric_kind != metric_kind:
        return False
    if category == "device":
        return row.name == name
    return (
        row.product_name == product_name
        and row.process_name == process_name
        and row.stamping_method == stamping_method
    )


def _build_usage_stats_export_response(device_rows: list[UsageStatRow], mold_rows: list[UsageStatRow]) -> Response:
    content = export_usage_stats_rows(device_rows, mold_rows)
    return Response(
        content=content,
        media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
        headers={"Content-Disposition": 'attachment; filename="usage-stats.xlsx"'},
    )


def _build_usage_stats_detail_from_allocations(
    *,
    allocations: list,
    category: str,
    target_args: dict,
    equipment_type_by_key: dict[tuple[str, str], str],
) -> UsageStatsDetailOut:
    rows = build_usage_stats_from_allocations(
        allocations=allocations,
        category=category,
        equipment_type_by_key=equipment_type_by_key,
    )
    target = next((row for row in rows if _matches_target(row, **target_args)), None)
    if target is None:
        return UsageStatsDetailOut(
            id=target_args["id"],
            category=category,
            attendance_point_name=target_args["attendance_point_name"],
            name=target_args["name"],
            product_name=target_args["product_name"],
            process_name=target_args["process_name"],
            stamping_method=target_args["stamping_method"],
            metric_kind=target_args["metric_kind"],
        )

    grouped_allocations: dict[tuple[date, int], list] = {}
    group_reports: dict[tuple[date, int], object | None] = {}
    for allocation in allocations:
        allocation_date = getattr(allocation, "allocation_date")
        report = getattr(allocation, "report", None)
        report_id = getattr(allocation, "report_id", None)
        if report_id is None and report is not None:
            report_id = getattr(report, "id", None)
        group_key = (allocation_date, report_id if report_id is not None else id(allocation))
        grouped_allocations.setdefault(group_key, []).append(allocation)
        group_reports.setdefault(group_key, report)

    daily_rows: dict[date, UsageStatsDailyRow] = {}
    report_rows: list[UsageStatsReportRow] = []
    for (allocation_date, report_key), group in grouped_allocations.items():
        group_matches = build_usage_stats_from_allocations(
            allocations=group,
            category=category,
            equipment_type_by_key=equipment_type_by_key,
        )
        match = next((row for row in group_matches if _matches_target(row, **target_args)), None)
        if match is None:
            continue

        day = daily_rows.setdefault(
            allocation_date,
            UsageStatsDailyRow(report_date=allocation_date),
        )
        day.value = round2(day.value + match.value)
        day.report_count += match.report_count

        report = group_reports[(allocation_date, report_key)]
        employee = getattr(report, "employee", None) if report is not None else None
        report_id = (
            getattr(report, "id", None)
            if report is not None
            else getattr(group[0], "report_id", None)
        )
        employee_phone = getattr(report, "employee_phone", "") if report is not None else ""
        report_rows.append(
            UsageStatsReportRow(
                report_id=report_id or 0,
                report_date=allocation_date,
                employee_phone=employee_phone,
                employee_name=employee.name if employee else "",
                display_name=target.name,
                value=match.value,
                report_count=match.report_count,
            )
        )

    return _to_detail_out(
        target,
        daily_rows=sorted(daily_rows.values(), key=lambda row: row.report_date),
        report_rows=sorted(
            report_rows,
            key=lambda row: (row.report_date, row.report_id),
            reverse=True,
        ),
    )


@router.get("/summary", response_model=PageResponse)
def usage_stats_summary(
    category: str = Query("device", pattern="^(device|mold)$"),
    start_date: date | None = None,
    end_date: date | None = None,
    attendance_point_name: str = "",
    keyword: str = "",
    sort_by: str = Query("value", pattern="^(value|count)$"),
    page: int = Query(1, ge=1),
    page_size: int = Query(10, ge=1, le=100),
    user: Personnel = Depends(require_roles(Role.admin, Role.manager)),
    db: Session = Depends(get_db),
) -> PageResponse:
    purge_expired_voided_reports(db)
    point_names = _requested_points(db, user, attendance_point_name)
    allocations = db.scalars(_approved_allocations_query(point_names, start_date, end_date)).all()
    rows = _sort_usage_rows(
        _filter_usage_rows(
            build_usage_stats_from_allocations(
                allocations=allocations,
                category=category,
                equipment_type_by_key=_equipment_type_map(db, point_names),
            ),
            keyword,
        ),
        sort_by,
    )
    safe_page_size = min(100, max(1, page_size))
    total = len(rows)
    total_pages = max(1, ceil(total / safe_page_size))
    start = (page - 1) * safe_page_size
    return PageResponse(
        page=page,
        page_size=safe_page_size,
        total=total,
        total_pages=total_pages,
        rows=[_to_schema_row(row) for row in rows[start : start + safe_page_size]],
    )


@router.get("/export")
def export_usage_stats_excel(
    start_date: date | None = None,
    end_date: date | None = None,
    attendance_point_name: str = "",
    keyword: str = "",
    sort_by: str = Query("value", pattern="^(value|count)$"),
    user: Personnel = Depends(require_roles(Role.admin, Role.manager)),
    db: Session = Depends(get_db),
) -> Response:
    purge_expired_voided_reports(db)
    point_names = _requested_points(db, user, attendance_point_name)
    equipment_type_by_key = _equipment_type_map(db, point_names)
    allocations = db.scalars(_approved_allocations_query(point_names, start_date, end_date)).all()
    device_rows = _sort_usage_rows(
        _filter_usage_rows(
            build_usage_stats_from_allocations(
                allocations=allocations,
                category="device",
                equipment_type_by_key=equipment_type_by_key,
            ),
            keyword,
        ),
        sort_by,
    )
    mold_rows = _sort_usage_rows(
        _filter_usage_rows(
            build_usage_stats_from_allocations(
                allocations=allocations,
                category="mold",
                equipment_type_by_key=equipment_type_by_key,
            ),
            keyword,
        ),
        sort_by,
    )
    return _build_usage_stats_export_response(device_rows, mold_rows)


@router.get("/detail", response_model=UsageStatsDetailOut)
def usage_stats_detail(
    category: str = Query("device", pattern="^(device|mold)$"),
    id: str = "",
    metric_kind: str = "",
    start_date: date | None = None,
    end_date: date | None = None,
    attendance_point_name: str = "",
    name: str = "",
    product_name: str = "",
    process_name: str = "",
    stamping_method: str = "",
    user: Personnel = Depends(require_roles(Role.admin, Role.manager)),
    db: Session = Depends(get_db),
) -> UsageStatsDetailOut:
    purge_expired_voided_reports(db)
    point_names = _requested_points(db, user, attendance_point_name)
    equipment_type_by_key = _equipment_type_map(db, point_names)
    allocations = db.scalars(_approved_allocations_query(point_names, start_date, end_date)).all()
    target_args = {
        "id": _clean(id),
        "category": category,
        "attendance_point_name": _clean(attendance_point_name),
        "name": _clean(name),
        "product_name": _clean(product_name),
        "process_name": _clean(process_name),
        "stamping_method": _clean(stamping_method),
        "metric_kind": _clean(metric_kind),
    }
    return _build_usage_stats_detail_from_allocations(
        allocations=allocations,
        category=category,
        target_args=target_args,
        equipment_type_by_key=equipment_type_by_key,
    )