"""Reports API endpoints.""" from datetime import date, datetime from typing import Annotated from fastapi import APIRouter, Depends, Query from sqlalchemy import and_, case, func from sqlalchemy.orm import Session from app.core.config import settings from app.core.deps import get_current_admin, get_db from app.models.booking import Booking from app.models.space import Space from app.models.user import User from app.schemas.reports import ( ApprovalRateReport, SpaceUsageItem, SpaceUsageReport, TopUserItem, TopUsersReport, ) router = APIRouter() def calculate_hours_expr() -> any: """Get database-specific expression for calculating hours between datetimes.""" if "sqlite" in settings.database_url: # SQLite: Use julianday function # Returns difference in days, multiply by 24 to get hours return ( func.julianday(Booking.end_datetime) - func.julianday(Booking.start_datetime) ) * 24 else: # PostgreSQL: Use EXTRACT(EPOCH) return func.extract("epoch", Booking.end_datetime - Booking.start_datetime) / 3600 @router.get("/admin/reports/usage", response_model=SpaceUsageReport) def get_usage_report( start_date: date | None = Query(None), end_date: date | None = Query(None), space_id: int | None = Query(None), db: Annotated[Session, Depends(get_db)] = None, current_admin: Annotated[User, Depends(get_current_admin)] = None, ) -> SpaceUsageReport: """Get booking usage report by space.""" query = ( db.query( Booking.space_id, Space.name.label("space_name"), func.count(Booking.id).label("total_bookings"), func.sum(case((Booking.status == "approved", 1), else_=0)).label( "approved_bookings" ), func.sum(case((Booking.status == "pending", 1), else_=0)).label( "pending_bookings" ), func.sum(case((Booking.status == "rejected", 1), else_=0)).label( "rejected_bookings" ), func.sum(case((Booking.status == "canceled", 1), else_=0)).label( "canceled_bookings" ), func.sum(calculate_hours_expr()).label("total_hours"), ) .join(Space) .group_by(Booking.space_id, Space.name) ) # Apply filters filters = [] if start_date: filters.append( Booking.start_datetime >= datetime.combine(start_date, datetime.min.time()) ) if end_date: filters.append( Booking.start_datetime <= datetime.combine(end_date, datetime.max.time()) ) if space_id: filters.append(Booking.space_id == space_id) if filters: query = query.filter(and_(*filters)) results = query.all() items = [ SpaceUsageItem( space_id=r.space_id, space_name=r.space_name, total_bookings=r.total_bookings, approved_bookings=r.approved_bookings or 0, pending_bookings=r.pending_bookings or 0, rejected_bookings=r.rejected_bookings or 0, canceled_bookings=r.canceled_bookings or 0, total_hours=round(float(r.total_hours or 0), 2), ) for r in results ] return SpaceUsageReport( items=items, total_bookings=sum(item.total_bookings for item in items), date_range={"start": start_date, "end": end_date}, ) @router.get("/admin/reports/top-users", response_model=TopUsersReport) def get_top_users_report( start_date: date | None = Query(None), end_date: date | None = Query(None), limit: int = Query(10, ge=1, le=100), db: Annotated[Session, Depends(get_db)] = None, current_admin: Annotated[User, Depends(get_current_admin)] = None, ) -> TopUsersReport: """Get top users by booking count.""" query = ( db.query( Booking.user_id, User.full_name.label("user_name"), User.email.label("user_email"), func.count(Booking.id).label("total_bookings"), func.sum(case((Booking.status == "approved", 1), else_=0)).label( "approved_bookings" ), func.sum(calculate_hours_expr()).label("total_hours"), ) .join(User, Booking.user_id == User.id) .group_by(Booking.user_id, User.full_name, User.email) ) # Apply date filters if start_date: query = query.filter( Booking.start_datetime >= datetime.combine(start_date, datetime.min.time()) ) if end_date: query = query.filter( Booking.start_datetime <= datetime.combine(end_date, datetime.max.time()) ) # Order by total bookings desc query = query.order_by(func.count(Booking.id).desc()).limit(limit) results = query.all() items = [ TopUserItem( user_id=r.user_id, user_name=r.user_name, user_email=r.user_email, total_bookings=r.total_bookings, approved_bookings=r.approved_bookings or 0, total_hours=round(float(r.total_hours or 0), 2), ) for r in results ] return TopUsersReport( items=items, date_range={"start": start_date, "end": end_date}, ) @router.get("/admin/reports/approval-rate", response_model=ApprovalRateReport) def get_approval_rate_report( start_date: date | None = Query(None), end_date: date | None = Query(None), db: Annotated[Session, Depends(get_db)] = None, current_admin: Annotated[User, Depends(get_current_admin)] = None, ) -> ApprovalRateReport: """Get approval/rejection rate report.""" query = db.query( func.count(Booking.id).label("total"), func.sum(case((Booking.status == "approved", 1), else_=0)).label("approved"), func.sum(case((Booking.status == "rejected", 1), else_=0)).label("rejected"), func.sum(case((Booking.status == "pending", 1), else_=0)).label("pending"), func.sum(case((Booking.status == "canceled", 1), else_=0)).label("canceled"), ) # Apply date filters if start_date: query = query.filter( Booking.start_datetime >= datetime.combine(start_date, datetime.min.time()) ) if end_date: query = query.filter( Booking.start_datetime <= datetime.combine(end_date, datetime.max.time()) ) result = query.first() total = result.total or 0 approved = result.approved or 0 rejected = result.rejected or 0 pending = result.pending or 0 canceled = result.canceled or 0 # Calculate rates (exclude pending from denominator) decided = approved + rejected approval_rate = (approved / decided * 100) if decided > 0 else 0 rejection_rate = (rejected / decided * 100) if decided > 0 else 0 return ApprovalRateReport( total_requests=total, approved=approved, rejected=rejected, pending=pending, canceled=canceled, approval_rate=round(approval_rate, 2), rejection_rate=round(rejection_rate, 2), date_range={"start": start_date, "end": end_date}, )