""" Google OAuth authentication router. Frontend sends the Google ID token, backend verifies it and returns a JWT. """ from fastapi import APIRouter, Depends, HTTPException from pydantic import BaseModel from sqlalchemy.orm import Session from google.oauth2 import id_token from google.auth.transport import requests as google_requests from app.config import settings from app.database import get_db from app.models.user import User, PlanTier from app.auth import create_access_token, get_current_user router = APIRouter(prefix="/api/auth", tags=["auth"]) class GoogleLoginRequest(BaseModel): credential: str # Google ID token from frontend class AuthResponse(BaseModel): access_token: str token_type: str = "bearer" user: "UserOut" class UserOut(BaseModel): id: int email: str name: str picture: str | None plan: str videos_used_this_period: int video_limit: int can_create_video: bool class Config: from_attributes = True # Fix forward ref AuthResponse.model_rebuild() @router.post("/google", response_model=AuthResponse) def google_login(body: GoogleLoginRequest, db: Session = Depends(get_db)): """ Verify Google ID token and create/login user. Returns a JWT access token. """ try: idinfo = id_token.verify_oauth2_token( body.credential, google_requests.Request(), settings.GOOGLE_CLIENT_ID, ) except ValueError as e: print(f"[AUTH ERROR] Google token verification failed: {e}") raise HTTPException(status_code=401, detail=f"Invalid Google token: {e}") google_id = idinfo["sub"] email = idinfo.get("email", "") name = idinfo.get("name", email.split("@")[0]) picture = idinfo.get("picture") if not email: raise HTTPException(status_code=400, detail="Email not provided by Google") # Find or create user user = db.query(User).filter(User.google_id == google_id).first() if not user: # Check if email already exists (shouldn't happen with Google, but safe) user = db.query(User).filter(User.email == email).first() if user: # Link Google ID to existing account user.google_id = google_id user.picture = picture or user.picture else: # Create new user user = User( email=email, name=name, picture=picture, google_id=google_id, plan=PlanTier.FREE, videos_used_this_period=0, ) db.add(user) else: # Update name/picture on login user.name = name user.picture = picture or user.picture # Local testing: override plan if DEFAULT_PLAN is set in .env if settings.DEFAULT_PLAN: override = settings.DEFAULT_PLAN.upper() try: user.plan = PlanTier(override.lower()) except ValueError: pass # ignore invalid values db.commit() db.refresh(user) token = create_access_token(user.id) return AuthResponse( access_token=token, user=UserOut( id=user.id, email=user.email, name=user.name, picture=user.picture, plan=user.plan.value, videos_used_this_period=user.videos_used_this_period, video_limit=user.video_limit, can_create_video=user.can_create_video, ), ) @router.get("/me", response_model=UserOut) def get_me(user: User = Depends(get_current_user)): """Get the current authenticated user.""" return UserOut( id=user.id, email=user.email, name=user.name, picture=user.picture, plan=user.plan.value, videos_used_this_period=user.videos_used_this_period, video_limit=user.video_limit, can_create_video=user.can_create_video, ) @router.post("/logout") def logout_cleanup( user: User = Depends(get_current_user), db: Session = Depends(get_db), ): """ Handle logout. Data is NOT deleted here — the periodic cleanup task handles deletion after 24 hours for free-tier users. """ return {"detail": "Logged out"}