Spaces:
Running
Running
| """ | |
| Firebase initialization and helper functions. | |
| """ | |
| import firebase_admin | |
| from firebase_admin import credentials, firestore, auth | |
| from pathlib import Path | |
| from src.utils.config import settings | |
| from src.utils.logger import setup_logger | |
| logger = setup_logger(__name__) | |
| _db = None | |
| def get_firebase_db(): | |
| """ | |
| Initialize Firebase Admin SDK and return Firestore client. | |
| """ | |
| global _db | |
| if _db is not None: | |
| return _db | |
| try: | |
| # Priority 1: Check if credentials string is provided (e.g., from HF Secrets) | |
| if settings.firebase_service_account_json: | |
| import json | |
| service_account_info = json.loads(settings.firebase_service_account_json) | |
| cred = credentials.Certificate(service_account_info) | |
| # Explicitly extract project_id from JSON if possible | |
| project_id = service_account_info.get('project_id') or settings.firebase_project_id | |
| firebase_admin.initialize_app(cred, { | |
| 'storageBucket': settings.firebase_storage_bucket, | |
| 'projectId': project_id | |
| }) | |
| _db = firestore.client() | |
| logger.info(f"Firebase initialized successfully for project: {project_id} (using JSON)") | |
| return _db | |
| # Priority 2: Check for service account file | |
| service_account_path = Path(settings.firebase_service_account_path) | |
| if not service_account_path.exists(): | |
| logger.warning(f"Firebase service account file not found at {service_account_path}. Using fallback/mock behavior if applicable.") | |
| return None | |
| cred = credentials.Certificate(str(service_account_path)) | |
| firebase_admin.initialize_app(cred, { | |
| 'storageBucket': settings.firebase_storage_bucket, | |
| 'projectId': settings.firebase_project_id | |
| }) | |
| _db = firestore.client() | |
| logger.info(f"Firebase initialized successfully for project: {settings.firebase_project_id} (using file)") | |
| return _db | |
| except Exception as e: | |
| logger.error(f"Failed to initialize Firebase: {e}") | |
| return None | |
| def verify_token(id_token: str): | |
| """ | |
| Verify a Firebase ID token and return the payload or error. | |
| Returns: {"payload": decoded_token, "error": error_message} | |
| """ | |
| try: | |
| # Check if any app is initialized, if not try to initialize | |
| if not firebase_admin._apps: | |
| logger.warning("Firebase app not initialized. Attempting default initialization...") | |
| get_firebase_db() | |
| if not firebase_admin._apps: | |
| return {"payload": None, "error": "Firebase app could not be initialized."} | |
| decoded_token = auth.verify_id_token(id_token) | |
| if decoded_token is None: | |
| return {"payload": None, "error": "Firebase SDK returned None for token verification."} | |
| logger.info(f"Firebase token successfully verified for UID: {decoded_token.get('uid')}") | |
| return {"payload": decoded_token, "error": None} | |
| except Exception as e: | |
| error_msg = str(e) or repr(e) | |
| logger.error(f"Firebase token verification failed: {error_msg}") | |
| return {"payload": None, "error": error_msg} | |