import logging import os import firebase_admin from firebase_admin import credentials, storage, firestore, auth # Initialize logging logger = logging.getLogger("BIT-LOG") class FirebaseManager: """ V261.17: Manages Firebase Storage uploads for resilient audio playback. Replaces local static file serving which causes timeouts and 404s. """ _instance = None _bucket = None _db = None def __new__(cls): if cls._instance is None: cls._instance = super(FirebaseManager, cls).__new__(cls) # Initialization is now handled explicitly in main.py lifespan return cls._instance def initialize(self): """ V5.13.16: Explicit initialization for Firebase Admin SDK. Guarantees initialize_app() runs once at startup. """ if firebase_admin._apps: logger.info("ℹī¸ [FIREBASE] SDK already initialized. Skipping.") self._bucket = storage.bucket() self._db = firestore.client() return try: from config import STORAGE_BUCKET, IS_PRODUCTION, FIREBASE_CREDENTIALS_PATH import json logger.info("🛠ī¸ [FIREBASE] Starting initialization...") cred_dict = None # Mission 1: Try loading from environment variables (checking multiple names for safety) creds_str = os.environ.get("FIREBASE_CREDENTIALS") or os.environ.get("FIREBASE_CREDENTIALS_JSON") if creds_str and len(creds_str.strip()) > 10: try: cred_dict = json.loads(creds_str) logger.info("✅ [FIREBASE] Successfully parsed credentials from Environment Secrets!") except Exception as e: logger.error(f"❌ [FIREBASE] Failed to parse Environment Credentials: {e}") # Fallback to local file only for local development (if no secret is set) if not cred_dict and FIREBASE_CREDENTIALS_PATH: if os.path.exists(FIREBASE_CREDENTIALS_PATH): with open(FIREBASE_CREDENTIALS_PATH, "r", encoding="utf-8") as f: cred_dict = json.load(f) logger.info(f"📂 [FIREBASE] Loading credentials from file: {FIREBASE_CREDENTIALS_PATH}.") else: logger.warning(f"⚠ī¸ [FIREBASE] Credentials file not found at {FIREBASE_CREDENTIALS_PATH}.") if cred_dict: cred = credentials.Certificate(cred_dict) firebase_admin.initialize_app(cred, { 'storageBucket': STORAGE_BUCKET }) logger.info(f"🚀 [FIREBASE] SDK Initialized successfully for {'PROD' if IS_PRODUCTION else 'DEV'}.") else: logger.error("❌ [FIREBASE] CRITICAL ERROR: Firebase credentials not found! Firebase is OFFLINE.") self._bucket = storage.bucket() self._db = firestore.client() logger.info("✨ [FIREBASE] Storage and Firestore clients ready.") except Exception as e: logger.error(f"đŸ”Ĩ [FIREBASE] Initialization failed: {e}") def get_db(self): """V2: Returns the initialized Firestore client.""" if not self._db: # Fallback to lazy init if not explicitly initialized self.initialize() return self._db def verify_token(self, id_token: str): """ V2: Verifies a Firebase Auth ID token. Returns the decoded token dictionary (including 'uid') if valid, else None. """ if not firebase_admin._apps: self.initialize() try: decoded_token = auth.verify_id_token(id_token) return decoded_token except Exception as e: if "expired" in str(e).lower(): logger.warning("⚠ī¸ [FIREBASE] Token expired.") else: logger.error(f"❌ [FIREBASE] Error verifying token: {e}") return None def upload_file(self, local_path: str, destination_blob_name: str) -> str: """ V273.1: Uploads local file and returns a Signed URL (CORS proof). Signed URLs bypass public access restrictions and work reliably in production. """ if not self._bucket: self.initialize() if not self._bucket: logger.error("❌ [FIREBASE] Not initialized. Cannot upload.") return None try: blob = self._bucket.blob(destination_blob_name) blob.upload_from_filename(local_path) # V273.1: Generate Signed URL instead of public_url to fix CORS import datetime signed_url = blob.generate_signed_url( version="v4", expiration=datetime.timedelta(hours=24), method="GET", ) logger.info(f"UPLOAD: [FIREBASE] Signed URL generated for {destination_blob_name}") return signed_url except Exception as e: logger.error(f"ERROR: [FIREBASE] Upload failed for {local_path}: {e}") return None def save_chat_message(self, uid: str, session_id: str, role: str, content: str, image_urls: list = None, metadata: dict = None): """ V318.0: Saves a message to the chat session history in Firestore. Path: users/{uid}/chat_sessions/{session_id}/messages """ db = self.get_db() if not db: logger.error("❌ [FIREBASE] Firestore DB not available for saving chat.") return try: messages_ref = db.collection('users').document(uid).collection('chat_sessions').document(session_id).collection('messages') payload = { "role": role, "content": content, "image_urls": image_urls or [], "metadata": metadata or {}, "timestamp": firestore.SERVER_TIMESTAMP } messages_ref.add(payload) logger.info(f"💾 [FIREBASE] Saved {role} message to session {session_id} for {uid}.") except Exception as e: logger.error(f"❌ [FIREBASE] Failed to save chat message: {e}") def get_chat_history(self, uid: str, session_id: str, limit: int = 10): """ V318.0: Fetches the last N messages from a chat session, ordered by timestamp. """ db = self.get_db() if not db: return [] try: messages_ref = db.collection('users').document(uid).collection('chat_sessions').document(session_id).collection('messages') query = messages_ref.order_by("timestamp", direction=firestore.Query.DESCENDING).limit(limit) # Note: Reversed order results (query is DESC, we want chronological ASC) docs = query.get() history = [] for doc in reversed(list(docs)): history.append(doc.to_dict()) return history except Exception as e: logger.error(f"❌ [FIREBASE] Failed to fetch chat history: {e}") return [] # Singleton accessor firebase_manager = FirebaseManager()