PixVerse_Video_Generator_API / services /firebase_service.py
Nanny7's picture
Initial commit: PixVerse Video Generator API with Firebase auth, MongoDB tracking, and Digital Ocean storage
534df99
import os
import json
import firebase_admin
from firebase_admin import credentials, auth
import config
class FirebaseService:
def __init__(self):
self._initialized = False
def _initialize(self):
if not self._initialized:
try:
if not firebase_admin._apps:
cred = None
# Priority 1: JSON content from environment variable (for Hugging Face Spaces)
if config.FIREBASE_CREDENTIALS_JSON:
try:
cred_dict = json.loads(config.FIREBASE_CREDENTIALS_JSON)
cred = credentials.Certificate(cred_dict)
print("[Firebase] Initialized with credentials from FIREBASE_CREDENTIALS_JSON")
except json.JSONDecodeError as e:
print(f"[Firebase] Error parsing FIREBASE_CREDENTIALS_JSON: {e}")
# Priority 2: Credentials JSON file
if not cred and config.FIREBASE_CREDENTIALS_PATH and os.path.exists(config.FIREBASE_CREDENTIALS_PATH):
cred = credentials.Certificate(config.FIREBASE_CREDENTIALS_PATH)
print(f"[Firebase] Initialized with credentials file: {config.FIREBASE_CREDENTIALS_PATH}")
# Initialize with credentials or fallback to project ID
if cred:
firebase_admin.initialize_app(cred)
elif config.FIREBASE_PROJECT_ID:
firebase_admin.initialize_app(options={
'projectId': config.FIREBASE_PROJECT_ID
})
print(f"[Firebase] Initialized with project ID only: {config.FIREBASE_PROJECT_ID}")
else:
print("[Firebase] No credentials or project ID configured")
return
self._initialized = True
except Exception as e:
print(f"[Firebase] Initialization error: {e}")
def verify_token(self, id_token: str) -> dict:
self._initialize()
# If Firebase not initialized, skip verification (optional mode)
if not self._initialized:
print("[Firebase] Not initialized - skipping token verification (optional mode)")
return {
"uid": "anonymous",
"email": None,
"verified": False,
"skipped": True
}
try:
decoded_token = auth.verify_id_token(id_token)
return {
"uid": decoded_token.get("uid"),
"email": decoded_token.get("email"),
"verified": True
}
except auth.InvalidIdTokenError:
raise Exception("Invalid Firebase token")
except auth.ExpiredIdTokenError:
raise Exception("Firebase token has expired")
except Exception as e:
error_str = str(e).lower()
# If credentials error or connection error, return anonymous (optional mode)
if any(keyword in error_str for keyword in ["credentials", "connection", "remote", "network", "timeout", "aborted"]):
print(f"[Firebase] Connection/credentials error - skipping verification: {str(e)[:100]}")
return {
"uid": "anonymous",
"email": None,
"verified": False,
"skipped": True
}
raise Exception(f"Firebase authentication failed: {str(e)}")
firebase_service = FirebaseService()