from sqlalchemy.orm import Session from typing import Optional import httpx from app.database.models import User from app.utils.password import hash_password, verify_password from app.utils.jwt import create_access_token, create_refresh_token from app.utils.helpers import generate_id try: from google.oauth2 import id_token from google.auth.transport import requests as google_requests GOOGLE_AUTH_AVAILABLE = True except ImportError: GOOGLE_AUTH_AVAILABLE = False print("⚠️ WARNING: google-auth libraries not found. Google Login will fail.") class AuthService: """Service for authentication operations.""" @staticmethod def create_user(db: Session, email: str, password: str, name: Optional[str] = None) -> User: """ Create a new user with email and password. Args: db: Database session email: User email password: Plain text password name: Optional user name Returns: Created User object Raises: ValueError: If email already exists """ # Check if user exists existing_user = db.query(User).filter(User.email == email).first() if existing_user: raise ValueError("Email already registered") # Hash password password_hash = hash_password(password) # Create user import os admin_email = os.environ.get('ADMIN_EMAIL') is_admin_user = 1 if admin_email and email == admin_email else 0 user = User( id=generate_id(), email=email, password_hash=password_hash, name=name, is_admin=is_admin_user ) db.add(user) db.commit() db.refresh(user) if is_admin_user: print(f"✅ User {email} registered and granted ADMIN privileges!") return user @staticmethod def authenticate_user(db: Session, email: str, password: str) -> Optional[User]: """ Authenticate a user with email and password. Args: db: Database session email: User email password: Plain text password Returns: User object if authentication successful, None otherwise """ user = db.query(User).filter(User.email == email).first() if not user or not user.password_hash: return None if not verify_password(password, user.password_hash): return None return user @staticmethod async def verify_google_token(credential: str, client_id: str) -> dict: """ Verify Google OAuth token and extract user info. Args: credential: Google JWT token from frontend client_id: Google OAuth client ID from settings Returns: Dictionary with user info (email, name, google_id, picture) Raises: ValueError: If token is invalid or verification fails """ if not GOOGLE_AUTH_AVAILABLE: raise ValueError("Google Auth libraries are not installed on this server.") try: # Verify the token with Google idinfo = id_token.verify_oauth2_token( credential, google_requests.Request(), client_id ) # Extract user information from the token return { "email": idinfo.get("email"), "name": idinfo.get("name"), "google_id": idinfo.get("sub"), # 'sub' is the Google user ID "picture": idinfo.get("picture") } except Exception as e: print(f"❌ Error verifying Google token: {e}") raise ValueError(f"Invalid Google token: {str(e)}") @staticmethod def create_user_from_google( db: Session, google_id: str, email: str, name: Optional[str] = None ) -> User: """ Create or get user from Google OAuth. Args: db: Database session google_id: Google user ID email: User email name: Optional user name Returns: User object """ # Check if user exists with this Google ID user = db.query(User).filter(User.google_id == google_id).first() if user: return user # Check if user exists with this email user = db.query(User).filter(User.email == email).first() if user: # Link Google account user.google_id = google_id if name and not user.name: user.name = name db.commit() db.refresh(user) return user # Create new user import os admin_email = os.environ.get('ADMIN_EMAIL') is_admin_user = 1 if admin_email and email == admin_email else 0 user = User( id=generate_id(), email=email, google_id=google_id, name=name, is_admin=is_admin_user ) db.add(user) db.commit() db.refresh(user) if is_admin_user: print(f"✅ User {email} registered via Google and granted ADMIN privileges!") return user @staticmethod def get_user_by_id(db: Session, user_id: str) -> Optional[User]: """ Get user by ID. Args: db: Database session user_id: User ID Returns: User object if found, None otherwise """ return db.query(User).filter(User.id == user_id).first() @staticmethod def generate_tokens(user: User) -> dict: """ Generate access and refresh tokens for a user. Args: user: User object Returns: Dictionary with access_token and refresh_token """ access_token = create_access_token(user.id) refresh_token = create_refresh_token(user.id) return { "access_token": access_token, "refresh_token": refresh_token }