import secrets from sqlalchemy import select from sqlalchemy.orm import Session, joinedload from app.core.config import Settings, get_settings from app.core.errors import AppError from app.models.catalog import Event from app.models.orders import Order, OrderItem, Ticket from app.models.enums import TicketStatus from app.models.user import User from app.services.qr import build_qr_payload, qr_png_base64 def generate_qr_token() -> str: return secrets.token_urlsafe(32) def _load_ticket_detail(db: Session, ticket_id: int, user_id: int | None = None) -> Ticket: query = ( select(Ticket) .options( joinedload(Ticket.order_item).joinedload(OrderItem.ticket_type), joinedload(Ticket.order).joinedload(Order.event).joinedload(Event.venue), ) .where(Ticket.id == ticket_id) ) if user_id is not None: query = query.where(Ticket.user_id == user_id) ticket = db.scalar(query) if not ticket: raise AppError("Ticket not found", status_code=404) if ticket.order and ticket.order.status.value != "paid": raise AppError("Ticket not available", status_code=404) return ticket def ticket_to_detail(db: Session, ticket: Ticket) -> dict: if not ticket.qr_token: ticket.qr_token = generate_qr_token() db.commit() db.refresh(ticket) event = ticket.order.event if ticket.order else None venue = event.venue if event else None user = db.get(User, ticket.user_id) ticket_type = ticket.order_item.ticket_type if ticket.order_item else None payload = build_qr_payload(ticket.qr_token) return { "id": ticket.id, "code": ticket.code, "status": ticket.status, "order_id": ticket.order_id, "event_id": ticket.event_id, "event_title": event.title if event else "Event", "event_starts_at": event.starts_at if event else ticket.created_at, "event_ends_at": event.ends_at if event else None, "venue_name": venue.name if venue else "Bengaluru", "venue_area": venue.area if venue else "", "venue_city": venue.city if venue else "Bengaluru", "attendee_name": (user.full_name if user and user.full_name else user.email if user else "Guest"), "attendee_email": user.email if user else "", "ticket_type_name": ticket_type.name if ticket_type else "General", "ticket_type_id": ticket.ticket_type_id, "qr_token": ticket.qr_token, "qr_payload": payload, "qr_image_base64": qr_png_base64(payload), "used_at": ticket.used_at, "created_at": ticket.created_at, } def get_ticket_detail_for_user(db: Session, ticket_id: int, user_id: int) -> dict: ticket = _load_ticket_detail(db, ticket_id, user_id=user_id) return ticket_to_detail(db, ticket) def verify_ticket_at_gate(db: Session, qr_token: str) -> dict: ticket = db.scalar(select(Ticket).where(Ticket.qr_token == qr_token).with_for_update()) if not ticket: raise AppError("Invalid ticket", status_code=404) order = db.get(Order, ticket.order_id) event = db.get(Event, ticket.event_id) if ticket.event_id else None event_title = event.title if event else "Event" if ticket.status == TicketStatus.used: return { "valid": False, "status": ticket.status, "ticket_id": ticket.id, "ticket_code": ticket.code, "event_title": event_title, "message": "Ticket already used — QR cannot be reused", } if ticket.status == TicketStatus.cancelled: return { "valid": False, "status": ticket.status, "ticket_id": ticket.id, "ticket_code": ticket.code, "event_title": event_title, "message": "Ticket cancelled", } if order is None or order.status.value != "paid": return { "valid": False, "status": ticket.status, "ticket_id": ticket.id, "ticket_code": ticket.code, "event_title": event_title, "message": "Order not paid", } from datetime import datetime, timezone ticket.status = TicketStatus.used ticket.used_at = datetime.now(timezone.utc) db.commit() return { "valid": True, "status": TicketStatus.used, "ticket_id": ticket.id, "ticket_code": ticket.code, "event_title": event_title, "message": "Entry verified", } def assert_checkin_authorized(settings: Settings, api_key: str | None) -> None: expected = settings.ticket_verify_api_key if settings.environment.lower() == "development" and not expected: return if not expected or not api_key: raise AppError("Unauthorized", status_code=401) if not secrets.compare_digest(api_key, expected): raise AppError("Unauthorized", status_code=401)