82 lines
3 KiB
Python
82 lines
3 KiB
Python
from datetime import datetime, timezone
|
|
|
|
from sqlalchemy import func, select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.models.catalog import Event, TicketType
|
|
from app.models.enums import EventStatus, OrderStatus, TicketStatus
|
|
from app.models.orders import Order, Ticket
|
|
from app.schemas.admin import DashboardMetrics, EventStatsRow
|
|
|
|
|
|
def get_dashboard_metrics(db: Session) -> DashboardMetrics:
|
|
now = datetime.now(timezone.utc)
|
|
|
|
revenue = db.scalar(
|
|
select(func.coalesce(func.sum(Order.total_paise), 0)).where(Order.status == OrderStatus.paid)
|
|
) or 0
|
|
|
|
tickets_sold = db.scalar(select(func.coalesce(func.sum(TicketType.quantity_sold), 0))) or 0
|
|
tickets_total = db.scalar(select(func.coalesce(func.sum(TicketType.quantity_total), 0))) or 0
|
|
tickets_remaining = max(int(tickets_total) - int(tickets_sold), 0)
|
|
|
|
orders_total = db.scalar(select(func.count(Order.id))) or 0
|
|
orders_paid = db.scalar(select(func.count(Order.id)).where(Order.status == OrderStatus.paid)) or 0
|
|
orders_pending = db.scalar(select(func.count(Order.id)).where(Order.status == OrderStatus.pending)) or 0
|
|
|
|
attendance = db.scalar(select(func.count(Ticket.id)).where(Ticket.status == TicketStatus.used)) or 0
|
|
|
|
upcoming_events = (
|
|
db.scalar(
|
|
select(func.count(Event.id)).where(
|
|
Event.status == EventStatus.published,
|
|
Event.starts_at >= now,
|
|
)
|
|
)
|
|
or 0
|
|
)
|
|
|
|
return DashboardMetrics(
|
|
total_revenue_paise=int(revenue),
|
|
tickets_sold=int(tickets_sold),
|
|
tickets_remaining=tickets_remaining,
|
|
orders_total=int(orders_total),
|
|
orders_paid=int(orders_paid),
|
|
orders_pending=int(orders_pending),
|
|
attendance=int(attendance),
|
|
upcoming_events=int(upcoming_events),
|
|
)
|
|
|
|
|
|
def get_event_statistics(db: Session) -> list[EventStatsRow]:
|
|
events = db.scalars(select(Event).order_by(Event.starts_at.desc())).all()
|
|
rows: list[EventStatsRow] = []
|
|
|
|
for event in events:
|
|
tiers = db.scalars(select(TicketType).where(TicketType.event_id == event.id)).all()
|
|
sold = sum(t.quantity_sold for t in tiers)
|
|
total = sum(t.quantity_total for t in tiers)
|
|
revenue = db.scalar(
|
|
select(func.coalesce(func.sum(Order.total_paise), 0)).where(
|
|
Order.event_id == event.id,
|
|
Order.status == OrderStatus.paid,
|
|
)
|
|
) or 0
|
|
attendance = db.scalar(
|
|
select(func.count(Ticket.id)).where(
|
|
Ticket.event_id == event.id,
|
|
Ticket.status == TicketStatus.used,
|
|
)
|
|
) or 0
|
|
rows.append(
|
|
EventStatsRow(
|
|
event_id=event.id,
|
|
event_title=event.title,
|
|
starts_at=event.starts_at,
|
|
tickets_sold=sold,
|
|
tickets_remaining=max(total - sold, 0),
|
|
revenue_paise=int(revenue),
|
|
attendance=int(attendance),
|
|
)
|
|
)
|
|
return rows
|