import logging from typing import NoReturn, TypeVar from fastapi import HTTPException, status from sqlalchemy.orm import Session ModelT = TypeVar("ModelT") logger = logging.getLogger(__name__) # ── Student-safe canonical messages ───────────────────────────────────────── # Keep these terse and actionable. Never reference provider names, file paths, # internal services, or exception class names. MSG_DOCUMENT_NOT_PROCESSABLE = ( "This document could not be processed. Please try uploading it again." ) MSG_AUDIO_GENERATION_FAILED = "Audio generation failed. Please try again." MSG_VIDEO_RENDER_FAILED = "Video generation failed. Please try again." MSG_VIDEO_VALIDATION_FAILED = ( "Video settings are invalid. Check your scene plan and try again." ) MSG_EXTRACTION_FAILED = ( "Questions could not be extracted from this file. Please try again." ) def student_safe_error( *, http_status: int, student_message: str, error_code: str, debug_exc: BaseException | None = None, ) -> NoReturn: """Log *debug_exc* server-side (WARNING level) and raise an HTTPException whose ``detail`` only contains the student-safe *student_message*. Usage:: except SomeInternalError as exc: student_safe_error( http_status=status.HTTP_422_UNPROCESSABLE_ENTITY, student_message=MSG_AUDIO_GENERATION_FAILED, error_code="AUDIO_GENERATION_FAILED", debug_exc=exc, ) """ if debug_exc is not None: logger.warning( "[%s] %s: %s", error_code, type(debug_exc).__name__, debug_exc, ) raise HTTPException( status_code=http_status, detail={"code": error_code, "message": student_message}, ) def rate_limited_error( message: str = "You have reached your plan limit for this month. Upgrade to continue.", ) -> NoReturn: """Raise HTTP 429 RATE_LIMITED with the standard student-safe error shape. Use this after ``check_usage_limit`` returns ``(False, reason)`` to block the request with a consistent, student-friendly message that never leaks internal details. """ raise HTTPException( status_code=status.HTTP_429_TOO_MANY_REQUESTS, detail={"code": "RATE_LIMITED", "message": message}, ) def get_or_404(db: Session, model: type[ModelT], object_id: str, label: str) -> ModelT: instance = db.get(model, object_id) if instance is None: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"{label} not found", ) return instance def ensure_user_exists(db: Session, user_id: str) -> None: from app.models.user import User if db.get(User, user_id) is not None: return if user_id != "usr_demo_student": raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="User not found", ) db.add( User( id="usr_demo_student", name="Demo Student", email="student@example.com", role="student", class_level="Plus Two", syllabus="Kerala HSE", preferred_language="English", ) ) db.commit()