# main.py import os import io import uuid import re import json import traceback from datetime import datetime, timedelta, timezone import requests from flask import Flask, request, jsonify, Response from flask_cors import CORS import firebase_admin from firebase_admin import credentials, db, storage, auth from pathlib import Path # Import the refactored Sozo business logic from sozo_gen import ( generate_report_draft, generate_single_chart, generate_video_from_project, load_dataframe_safely, deepgram_tts ) import logging import time import threading # ----------------------------------------------------------------------------- # 1. CONFIGURATION & INITIALIZATION # ----------------------------------------------------------------------------- app = Flask(__name__) CORS(app) try: credentials_json_string = os.environ.get("FIREBASE") if not credentials_json_string: raise ValueError("FIREBASE env var not set.") credentials_json = json.loads(credentials_json_string) firebase_db_url = os.environ.get("Firebase_DB") firebase_storage_bucket = os.environ.get("Firebase_Storage") if not firebase_db_url or not firebase_storage_bucket: raise ValueError("Firebase DB/Storage env vars must be set.") cred = credentials.Certificate(credentials_json) firebase_admin.initialize_app(cred, {'databaseURL': firebase_db_url, 'storageBucket': firebase_storage_bucket}) print("Firebase Admin SDK initialized successfully.") except Exception as e: print(f"FATAL: Error initializing Firebase: {e}") bucket = storage.bucket() logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) RESEND_API_KEY = os.environ.get("RESEND_API_KEY") # ----------------------------------------------------------------------------- # 2. HELPER FUNCTIONS # ----------------------------------------------------------------------------- def verify_token(token): try: return auth.verify_id_token(token)['uid'] except Exception: return None def verify_admin(auth_header): if not auth_header or not auth_header.startswith('Bearer '): raise ValueError('Invalid token') token = auth_header.split(' ')[1] uid = verify_token(token) if not uid: raise PermissionError('Invalid user') user_data = db.reference(f'users/{uid}').get() if not user_data or not user_data.get('is_admin', False): raise PermissionError('Admin access required') return uid def is_valid_email(email): """Simple regex for basic email validation.""" regex = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$' return re.match(regex, email) is not None # Global rate limiter for Resend API class ResendRateLimiter: def __init__(self, requests_per_second=1): self.requests_per_second = requests_per_second self.min_interval = 1.0 / requests_per_second self.last_request_time = 0 self.lock = threading.Lock() def wait_if_needed(self): with self.lock: current_time = time.time() time_since_last_request = current_time - self.last_request_time if time_since_last_request < self.min_interval: sleep_time = self.min_interval - time_since_last_request logger.info(f"Rate limiting: waiting {sleep_time:.2f} seconds before sending email") time.sleep(sleep_time) self.last_request_time = time.time() # Global instance - initialize once resend_rate_limiter = ResendRateLimiter(requests_per_second=1) def _send_notification(user_id, user_email, message_content, send_email=False, email_subject=None, email_body=None): """ Internal helper to send notifications. Creates an in-app notification in Firebase and optionally sends an email via Resend. If user_id is None, it will only attempt to send an email. Rate limited to 1 email per second to respect Resend API limits. """ timestamp = datetime.now(timezone.utc).isoformat() # 1. Send In-App Notification (if user_id is provided) if user_id: try: notif_ref = db.reference(f'notifications/{user_id}').push() notif_data = { 'id': notif_ref.key, 'message': message_content, 'created_at': timestamp, 'read': False, 'read_at': None } notif_ref.set(notif_data) logger.info(f"Successfully sent in-app notification to UID {user_id}") except Exception as e: logger.error(f"Failed to send in-app notification to UID {user_id}: {e}") return False # Fail the whole operation if in-app fails for a registered user # 2. Send Email via Resend (if requested) if send_email and user_email: if not RESEND_API_KEY: logger.error("RESEND_API_KEY is not configured. Cannot send email.") return False # Apply rate limiting before making the request resend_rate_limiter.wait_if_needed() # Clean the API key (remove any whitespace) api_key = RESEND_API_KEY.strip() # Debug logging (be careful in production) logger.debug(f"API key format check - starts with 're_': {api_key.startswith('re_')}") logger.debug(f"API key length: {len(api_key)}") logger.debug(f"User email: {user_email}") headers = { "Authorization": f"Bearer {api_key}", "Content-Type": "application/json" } payload = { "from": "Sozo Business Studio ", "to": [user_email], "subject": email_subject, "html": email_body } # Log the request details (excluding sensitive info) logger.debug(f"Request URL: https://api.resend.com/emails") logger.debug(f"Request payload keys: {list(payload.keys())}") logger.debug(f"From address: {payload['from']}") logger.debug(f"To address: {payload['to']}") try: response = requests.post("https://api.resend.com/emails", headers=headers, json=payload) # Log response details for debugging logger.info(f"Resend API response status: {response.status_code}") logger.debug(f"Response headers: {dict(response.headers)}") logger.debug(f"Response content: {response.text}") # Handle rate limiting response if response.status_code == 429: logger.warning(f"Rate limit hit despite internal limiting. Response: {response.text}") # Extract retry-after if available retry_after = response.headers.get('retry-after') if retry_after: logger.info(f"Server requested retry after {retry_after} seconds") time.sleep(float(retry_after)) # Retry once response = requests.post("https://api.resend.com/emails", headers=headers, json=payload) response.raise_for_status() else: return False elif response.status_code == 401: logger.error("=== RESEND 401 UNAUTHORIZED DEBUG ===") logger.error(f"API Key starts correctly: {api_key.startswith('re_')}") logger.error(f"API Key length: {len(api_key)} (should be ~40 chars)") logger.error(f"Authorization header: Bearer {api_key[:10]}...") logger.error("Possible issues:") logger.error("1. API key copied incorrectly (extra chars/spaces)") logger.error("2. API key was regenerated in dashboard but not updated in env") logger.error("3. API key permissions were changed") logger.error("4. Account billing issue") logger.error(f"Full response: {response.text}") return False elif response.status_code == 422: logger.error(f"Resend API validation error (422): {response.text}") return False response.raise_for_status() response_data = response.json() logger.info(f"Successfully sent email to {user_email}. Email ID: {response_data.get('id')}") except requests.exceptions.RequestException as e: logger.error(f"Failed to send email to {user_email} via Resend: {e}") if hasattr(e, 'response') and e.response is not None: logger.error(f"Error response status: {e.response.status_code}") logger.error(f"Error response content: {e.response.text}") return False return True # Alternative: Batch email sending function for multiple emails def send_batch_notifications(notifications_list): """ Send multiple notifications with proper rate limiting. Args: notifications_list: List of dicts with keys: - user_id (optional) - user_email - message_content - send_email (bool) - email_subject (optional) - email_body (optional) Returns: dict: Results with success/failure counts """ results = { 'total': len(notifications_list), 'successful': 0, 'failed': 0, 'errors': [] } logger.info(f"Starting batch notification send for {results['total']} notifications") start_time = time.time() for i, notification in enumerate(notifications_list): try: success = _send_notification( user_id=notification.get('user_id'), user_email=notification.get('user_email'), message_content=notification.get('message_content'), send_email=notification.get('send_email', False), email_subject=notification.get('email_subject'), email_body=notification.get('email_body') ) if success: results['successful'] += 1 else: results['failed'] += 1 results['errors'].append(f"Notification {i+1} failed") except Exception as e: results['failed'] += 1 results['errors'].append(f"Notification {i+1} error: {str(e)}") logger.error(f"Unexpected error processing notification {i+1}: {e}") elapsed_time = time.time() - start_time logger.info(f"Batch notification completed in {elapsed_time:.2f} seconds. " f"Success: {results['successful']}, Failed: {results['failed']}") return results # ----------------------------------------------------------------------------- # 3. AUTHENTICATION & USER MANAGEMENT # ----------------------------------------------------------------------------- @app.route('/api/auth/signup', methods=['POST']) def signup(): try: data = request.get_json() email = data.get('email') password = data.get('password') if not email or not password: return jsonify({'error': 'Email and password are required'}), 400 user = auth.create_user(email=email, password=password) user_ref = db.reference(f'users/{user.uid}') user_data = {'email': email, 'credits': 15, 'is_admin': False, 'created_at': datetime.utcnow().isoformat()} user_ref.set(user_data) return jsonify({'success': True, 'user': {'uid': user.uid, **user_data}}), 201 except Exception as e: return jsonify({'error': str(e)}), 400 @app.route('/api/user/profile', methods=['GET']) def get_user_profile(): try: auth_header = request.headers.get('Authorization', '') if not auth_header.startswith('Bearer '): return jsonify({'error': 'Missing or invalid token'}), 401 token = auth_header.split(' ')[1] uid = verify_token(token) if not uid: return jsonify({'error': 'Invalid or expired token'}), 401 user_data = db.reference(f'users/{uid}').get() if not user_data: return jsonify({'error': 'User not found'}), 404 return jsonify({'uid': uid, **user_data}) except Exception as e: return jsonify({'error': str(e)}), 500 @app.route('/api/auth/google-signin', methods=['POST']) def google_signin(): try: auth_header = request.headers.get('Authorization', '') if not auth_header.startswith('Bearer '): return jsonify({'error': 'Missing or invalid token'}), 401 token = auth_header.split(' ')[1] decoded_token = auth.verify_id_token(token) uid = decoded_token['uid'] email = decoded_token.get('email') user_ref = db.reference(f'users/{uid}') user_data = user_ref.get() if not user_data: user_data = {'email': email, 'credits': 15, 'is_admin': False, 'created_at': datetime.utcnow().isoformat()} user_ref.set(user_data) return jsonify({'success': True, 'user': {'uid': uid, **user_data}}), 200 except Exception as e: return jsonify({'error': str(e)}), 400 # ----------------------------------------------------------------------------- # 3. SOZO BUSINESS STUDIO API ENDPOINTS # ----------------------------------------------------------------------------- @app.route('/api/sozo/projects', methods=['POST']) def create_sozo_project(): logger.info("Endpoint /api/sozo/projects POST: Received request to create new project.") try: auth_header = request.headers.get('Authorization', '') if not auth_header.startswith('Bearer '): logger.warning("Create project failed: Missing or invalid auth header.") return jsonify({'error': 'Missing or invalid token'}), 401 token = auth_header.split(' ')[1] uid = verify_token(token) if not uid: logger.warning("Create project failed: Invalid token.") return jsonify({'error': 'Unauthorized'}), 401 logger.info(f"Token verified for user UID: {uid}") if 'file' not in request.files: logger.warning(f"User {uid}: Create project failed: No file part in request.") return jsonify({'error': 'No file part'}), 400 file = request.files['file'] context = request.form.get('context', '') project_id = uuid.uuid4().hex logger.info(f"User {uid}: Generated new project ID: {project_id}") file_bytes = file.read() ext = Path(file.filename).suffix blob_name = f"sozo_projects/{uid}/{project_id}/data{ext}" logger.info(f"User {uid}: Uploading raw data to storage at {blob_name}") blob = bucket.blob(blob_name) blob.upload_from_string(file_bytes, content_type=file.content_type) logger.info(f"User {uid}: Successfully uploaded raw data for project {project_id}") project_ref = db.reference(f'sozo_projects/{project_id}') project_data = { 'uid': uid, 'id': project_id, 'status': 'uploaded', 'createdAt': datetime.utcnow().isoformat(), 'updatedAt': datetime.utcnow().isoformat(), 'userContext': context, 'originalDataUrl': blob.public_url, 'originalFilename': file.filename } logger.info(f"User {uid}: Saving project metadata to database for project {project_id}") project_ref.set(project_data) df = load_dataframe_safely(io.BytesIO(file_bytes), file.filename) preview_json = df.head().to_json(orient='records') logger.info(f"User {uid}: Project {project_id} created successfully.") return jsonify({ 'success': True, 'project': project_data, 'preview': json.loads(preview_json) }), 201 except Exception as e: logger.error(f"CRITICAL ERROR during project creation: {traceback.format_exc()}") return jsonify({'error': str(e)}), 500 @app.route('/api/sozo/projects', methods=['GET']) def get_sozo_projects(): logger.info("Endpoint /api/sozo/projects GET: Received request to list projects.") try: auth_header = request.headers.get('Authorization', '') if not auth_header.startswith('Bearer '): logger.warning("List projects failed: Missing or invalid auth header.") return jsonify({'error': 'Missing or invalid token'}), 401 token = auth_header.split(' ')[1] uid = verify_token(token) if not uid: logger.warning("List projects failed: Invalid token.") return jsonify({'error': 'Unauthorized'}), 401 logger.info(f"Token verified for user UID: {uid}. Fetching projects.") projects_ref = db.reference('sozo_projects') user_projects = projects_ref.order_by_child('uid').equal_to(uid).get() if not user_projects: logger.info(f"User {uid}: No projects found.") return jsonify([]), 200 # Firebase returns a dictionary, convert it to a list for the client projects_list = [project for project in user_projects.values()] logger.info(f"User {uid}: Found and returning {len(projects_list)} projects.") return jsonify(projects_list), 200 except Exception as e: logger.error(f"CRITICAL ERROR during project list: {traceback.format_exc()}") return jsonify({'error': str(e)}), 500 @app.route('/api/sozo/projects/', methods=['GET']) def get_sozo_project(project_id): logger.info(f"Endpoint /api/sozo/projects/{project_id} GET: Received request for single project.") try: auth_header = request.headers.get('Authorization', '') if not auth_header.startswith('Bearer '): logger.warning(f"Get project {project_id} failed: Missing or invalid auth header.") return jsonify({'error': 'Missing or invalid token'}), 401 token = auth_header.split(' ')[1] uid = verify_token(token) if not uid: logger.warning(f"Get project {project_id} failed: Invalid token.") return jsonify({'error': 'Unauthorized'}), 401 logger.info(f"Token verified for user UID: {uid}. Fetching project {project_id}.") project_ref = db.reference(f'sozo_projects/{project_id}') project_data = project_ref.get() if not project_data: logger.warning(f"User {uid}: Attempted to access non-existent project {project_id}.") return jsonify({'error': 'Project not found'}), 404 if project_data.get('uid') != uid: logger.error(f"User {uid}: UNAUTHORIZED attempt to access project {project_id} owned by {project_data.get('uid')}.") return jsonify({'error': 'Unauthorized to access this project'}), 403 logger.info(f"User {uid}: Successfully fetched project {project_id}.") return jsonify(project_data), 200 except Exception as e: logger.error(f"CRITICAL ERROR during get project {project_id}: {traceback.format_exc()}") return jsonify({'error': str(e)}), 500 @app.route('/api/sozo/projects/', methods=['PUT']) def update_sozo_project(project_id): logger.info(f"Endpoint /api/sozo/projects/{project_id} PUT: Received request to update project.") try: auth_header = request.headers.get('Authorization', '') if not auth_header.startswith('Bearer '): logger.warning(f"Update project {project_id} failed: Missing or invalid auth header.") return jsonify({'error': 'Missing or invalid token'}), 401 token = auth_header.split(' ')[1] uid = verify_token(token) if not uid: logger.warning(f"Update project {project_id} failed: Invalid token.") return jsonify({'error': 'Unauthorized'}), 401 logger.info(f"Token verified for user UID: {uid}. Updating project {project_id}.") project_ref = db.reference(f'sozo_projects/{project_id}') project_data = project_ref.get() if not project_data: logger.warning(f"User {uid}: Attempted to update non-existent project {project_id}.") return jsonify({'error': 'Project not found'}), 404 if project_data.get('uid') != uid: logger.error(f"User {uid}: UNAUTHORIZED attempt to update project {project_id} owned by {project_data.get('uid')}.") return jsonify({'error': 'Unauthorized to update this project'}), 403 logger.info(f"User {uid}: Ownership of project {project_id} verified.") update_data = request.get_json() if not update_data: return jsonify({'error': 'No update data provided'}), 400 # Define fields the user is allowed to update allowed_updates = ['userContext', 'originalFilename'] final_updates = {key: update_data[key] for key in update_data if key in allowed_updates} if not final_updates: logger.warning(f"User {uid}: Update for project {project_id} contained no valid fields.") return jsonify({'error': 'No valid fields to update'}), 400 final_updates['updatedAt'] = datetime.utcnow().isoformat() logger.info(f"User {uid}: Applying updates to project {project_id}: {final_updates}") project_ref.update(final_updates) updated_project = project_ref.get() logger.info(f"User {uid}: Successfully updated project {project_id}.") return jsonify(updated_project), 200 except Exception as e: logger.error(f"CRITICAL ERROR during update project {project_id}: {traceback.format_exc()}") return jsonify({'error': str(e)}), 500 @app.route('/api/sozo/projects/', methods=['DELETE']) def delete_sozo_project(project_id): logger.info(f"Endpoint /api/sozo/projects/{project_id} DELETE: Received request to delete project.") try: auth_header = request.headers.get('Authorization', '') if not auth_header.startswith('Bearer '): logger.warning(f"Delete project {project_id} failed: Missing or invalid auth header.") return jsonify({'error': 'Missing or invalid token'}), 401 token = auth_header.split(' ')[1] uid = verify_token(token) if not uid: logger.warning(f"Delete project {project_id} failed: Invalid token.") return jsonify({'error': 'Unauthorized'}), 401 logger.info(f"Token verified for user UID: {uid}. Deleting project {project_id}.") project_ref = db.reference(f'sozo_projects/{project_id}') project_data = project_ref.get() if not project_data: logger.warning(f"User {uid}: Attempted to delete non-existent project {project_id}.") return jsonify({'error': 'Project not found'}), 404 if project_data.get('uid') != uid: logger.error(f"User {uid}: UNAUTHORIZED attempt to delete project {project_id} owned by {project_data.get('uid')}.") return jsonify({'error': 'Unauthorized to delete this project'}), 403 logger.info(f"User {uid}: Ownership of project {project_id} verified. Proceeding with deletion.") # Delete all associated files from Firebase Storage project_folder_prefix = f"sozo_projects/{uid}/{project_id}/" logger.info(f"User {uid}: Deleting all files from storage folder: {project_folder_prefix}") blobs_to_delete = bucket.list_blobs(prefix=project_folder_prefix) deleted_files_count = 0 for blob in blobs_to_delete: logger.info(f"User {uid}: Deleting file {blob.name} from storage.") blob.delete() deleted_files_count += 1 logger.info(f"User {uid}: Deleted {deleted_files_count} files from storage for project {project_id}.") # Delete project from Realtime Database logger.info(f"User {uid}: Deleting project {project_id} from database.") project_ref.delete() logger.info(f"User {uid}: Successfully deleted project {project_id}.") return jsonify({'success': True, 'message': f'Project {project_id} and all associated files deleted.'}), 200 except Exception as e: logger.error(f"CRITICAL ERROR during delete project {project_id}: {traceback.format_exc()}") return jsonify({'error': str(e)}), 500 @app.route('/api/sozo/projects//generate-report', methods=['POST']) def generate_sozo_report(project_id): logger.info(f"POST /api/sozo/projects/{project_id}/generate-report - Received request") try: token = request.headers.get('Authorization', '').split(' ')[1] uid = verify_token(token) if not uid: return jsonify({'error': 'Unauthorized'}), 401 user_ref = db.reference(f'users/{uid}') user_data = user_ref.get() if not user_data: return jsonify({'error': 'User not found'}), 404 current_credits = user_data.get('credits', 0) if current_credits < 2: return jsonify({'error': 'Insufficient credits. Report generation requires 2 credits.'}), 402 project_ref = db.reference(f'sozo_projects/{project_id}') project_data = project_ref.get() if not project_data or project_data.get('uid') != uid: return jsonify({'error': 'Project not found or unauthorized'}), 404 current_status = project_data.get('status') if current_status in ['generating_report', 'generating_video', 'generating_slides']: logger.warning(f"User {uid} attempted to generate a report for project {project_id} which is already in progress (status: {current_status}).") return jsonify({'error': 'A process is already running for this project.'}), 409 project_ref.update({'status': 'generating_report'}) logger.info(f"Project {project_id} status locked to 'generating_report'.") blob_path = f"sozo_projects/{uid}/{project_id}/data{Path(project_data['originalFilename']).suffix}" blob = bucket.blob(blob_path) file_bytes = blob.download_as_bytes() draft_data = generate_report_draft( io.BytesIO(file_bytes), project_data['originalFilename'], project_data['userContext'], uid, project_id, bucket ) # --- Graceful Fallback Save & Charge Logic --- try: # Happy Path: Try to save everything full_update_data = { 'status': 'draft', 'rawMarkdown': draft_data.get('raw_md'), 'chartUrls': draft_data.get('chartUrls'), 'dataContext': draft_data.get('data_context') } project_ref.update(full_update_data) user_ref.update({'credits': current_credits - 2}) logger.info(f"Project {project_id} successfully updated with full data. User {uid} charged 2 credits.") return jsonify({ 'success': True, 'project': {**project_data, **full_update_data}, 'credits_remaining': current_credits - 2 }), 200 except Exception as save_error: # Fallback Path: Save only the essentials if the full save fails logger.warning(f"Failed to save full project data for {project_id} due to: {save_error}. Saving degraded version.") degraded_update_data = { 'status': 'draft', 'rawMarkdown': draft_data.get('raw_md'), 'chartUrls': draft_data.get('chartUrls'), 'dataContext': None, 'warning': 'Report generated, but context data failed to save. Downstream features may be affected.' } project_ref.update(degraded_update_data) user_ref.update({'credits': current_credits - 2}) logger.info(f"Project {project_id} successfully updated with DEGRADED data. User {uid} charged 2 credits.") return jsonify({ 'success': True, 'project': {**project_data, **degraded_update_data}, 'credits_remaining': current_credits - 2 }), 200 except Exception as e: logger.error(f"CRITICAL error generating report for project {project_id}: {traceback.format_exc()}") db.reference(f'sozo_projects/{project_id}').update({ 'status': 'failed', 'error': str(e) }) return jsonify({'error': str(e)}), 500 def send_video_generation_notification(user_id, user_email, project_name, video_url, send_email=False): """ Send notification when video generation is completed successfully. Args: user_id (str): Firebase user ID user_email (str): User's email address project_name (str): Name of the project video_url (str): URL of the generated video send_email (bool): Whether to send email notification Returns: bool: True if successful, False otherwise """ logger.info(f"=== VIDEO GENERATION NOTIFICATION START ===") logger.info(f"User ID: {user_id}") logger.info(f"User Email: {user_email}") logger.info(f"Project Name: {project_name}") logger.info(f"Video URL: {video_url}") logger.info(f"Send Email: {send_email}") try: # Create the notification message message_content = f"Your video for project '{project_name}' has been generated successfully!" logger.info(f"Created notification message: {message_content}") # Create email subject and body if email notification is requested email_subject = None email_body = None if send_email: logger.info("Preparing email notification content...") email_subject = f"🎥 Video Generation Complete - {project_name}" logger.info(f"Email subject: {email_subject}") email_body = f"""

🎉 Video Generation Complete!

Your video is ready!

Great news! Your video for project "{project_name}" has been successfully generated and is ready for viewing.

What's Next?

  • Click the button above to view your generated video
  • Share your video with your team or clients
  • Download the video for offline use
  • Create more videos with your remaining credits

This video was generated by Sozo Business Studio
Need help? Contact us at support@sozofix.tech

""" logger.info(f"Email body prepared (length: {len(email_body)} chars)") else: logger.info("Email notification not requested, skipping email content preparation") # Use the existing _send_notification function logger.info("Calling _send_notification function...") result = _send_notification( user_id=user_id, user_email=user_email, message_content=message_content, send_email=send_email, email_subject=email_subject, email_body=email_body ) if result: logger.info("✅ Video generation notification sent successfully") else: logger.error("❌ Video generation notification failed") logger.info(f"=== VIDEO GENERATION NOTIFICATION END (Result: {result}) ===") return result except Exception as e: logger.error(f"❌ EXCEPTION in send_video_generation_notification: {e}") logger.error(f"Exception type: {type(e).__name__}") import traceback logger.error(f"Traceback: {traceback.format_exc()}") logger.info(f"=== VIDEO GENERATION NOTIFICATION END (Exception) ===") return False @app.route('/api/sozo/projects//generate-video', methods=['POST']) def generate_sozo_video(project_id): try: token = request.headers.get('Authorization', '').split(' ')[1] uid = verify_token(token) if not uid: return jsonify({'error': 'Unauthorized'}), 401 # Check user credits first user_ref = db.reference(f'users/{uid}') user_data = user_ref.get() if not user_data: return jsonify({'error': 'User not found'}), 404 current_credits = user_data.get('credits', 0) if current_credits < 5: return jsonify({'error': 'Insufficient credits. Video generation requires 5 credits.'}), 402 project_ref = db.reference(f'sozo_projects/{project_id}') project_data = project_ref.get() if not project_data or project_data.get('uid') != uid: return jsonify({'error': 'Project not found or unauthorized'}), 404 data = request.get_json() voice_model = data.get('voice_model', 'aura-2-andromeda-en') # NEW: Get notification preference from request send_email_notification = data.get('send_email_notification', False) project_ref.update({'status': 'generating_video'}) blob_path = f"sozo_projects/{uid}/{project_id}/data{Path(project_data['originalFilename']).suffix}" blob = bucket.blob(blob_path) file_bytes = blob.download_as_bytes() df = load_dataframe_safely(io.BytesIO(file_bytes), project_data['originalFilename']) # Generate the video video_url = generate_video_from_project( df, project_data.get('rawMarkdown', ''), project_data.get('dataContext', {}), uid, project_id, voice_model, bucket ) if not video_url: raise Exception("Video generation failed in core function.") project_ref.update({'status': 'video_complete', 'videoUrl': video_url}) # Deduct credits ONLY after successful generation user_ref.update({'credits': current_credits - 5}) # NEW: Send notification if requested if send_email_notification: logger.info(f"Email notification requested for project {project_id}") project_name = project_data.get('name', 'Unnamed Project') user_email = user_data.get('email') logger.info(f"Project name: {project_name}") logger.info(f"User email available: {user_email is not None}") if user_email: logger.info(f"Sending video generation notification to {user_email}") notification_sent = send_video_generation_notification( user_id=uid, user_email=user_email, project_name=project_name, video_url=video_url, send_email=True ) if not notification_sent: logger.error(f"❌ Failed to send notification for video generation completion to user {uid}") else: logger.info(f"✅ Successfully sent notification for video generation completion to user {uid}") else: logger.warning(f"⚠️ No email found for user {uid}, skipping email notification") else: logger.info("Email notification not requested") notification_sent = False return jsonify({ 'success': True, 'video_url': video_url, 'credits_remaining': current_credits - 5, 'notification_sent': notification_sent if send_email_notification and user_data.get('email') else False }), 200 except Exception as e: db.reference(f'sozo_projects/{project_id}').update({'status': 'failed', 'error': str(e)}) traceback.print_exc() return jsonify({'error': str(e)}), 500 @app.route('/api/sozo/projects//generate-slides', methods=['POST']) def generate_sozo_slides(project_id): logger.info(f"POST /api/sozo/projects/{project_id}/generate-slides - Generating slides") try: token = request.headers.get('Authorization', '').split(' ') uid = verify_token(token) if not uid: return jsonify({'error': 'Unauthorized'}), 401 # Check user credits first user_ref = db.reference(f'users/{uid}') user_data = user_ref.get() if not user_data: return jsonify({'error': 'User not found'}), 404 current_credits = user_data.get('credits', 0) if current_credits < 5: return jsonify({'error': 'Insufficient credits. Slide generation requires 5 credits.'}), 402 project_ref = db.reference(f'sozo_projects/{project_id}') project_data = project_ref.get() if not project_data or project_data.get('uid') != uid: return jsonify({'error': 'Project not found or unauthorized'}), 404 raw_md = project_data.get('rawMarkdown') chart_urls = project_data.get('chartUrls', {}) if not raw_md: return jsonify({'error': 'Report must be generated before slides can be created.'}), 400 # The planner AI needs an LLM instance llm = ChatGoogleGenerativeAI(model="gemini-2.5-flash", google_api_key=os.getenv("GOOGLE_API_KEY"), temperature=0.2) slides_data = generate_slides_from_report( raw_md, chart_urls, uid, project_id, bucket, llm ) if not slides_data: raise Exception("Slide generation failed in core function.") # Save the slides data to the project in Firebase project_ref.update({'slides': slides_data}) # Deduct credits ONLY after successful generation user_ref.update({'credits': current_credits - 5}) logger.info(f"Project {project_id} successfully updated with {len(slides_data)} slides. User {uid} charged 5 credits.") return jsonify({ 'success': True, 'slides': slides_data, 'credits_remaining': current_credits - 5 }), 200 except Exception as e: logger.error(f"CRITICAL error generating slides for project {project_id}: {traceback.format_exc()}") db.reference(f'sozo_projects/{project_id}').update({'status': 'failed_slides', 'error': str(e)}) return jsonify({'error': str(e)}), 500 @app.route('/api/image-proxy', methods=['GET']) def image_proxy(): image_url = request.args.get('url') logger.info(f"[IMAGE PROXY] Received URL: {image_url}") if not image_url: logger.error("[IMAGE PROXY] ERROR: URL parameter is missing") return jsonify({'error': 'URL parameter is missing.'}), 400 try: # Parse Firebase Storage URL # Expected format: https://storage.googleapis.com/bucket-name/path/to/file.ext if 'storage.googleapis.com' not in image_url: logger.error(f"[IMAGE PROXY] ERROR: Invalid Firebase Storage URL: {image_url}") return jsonify({'error': 'Invalid Firebase Storage URL.'}), 400 logger.info(f"[IMAGE PROXY] Parsing URL: {image_url}") # Extract bucket name and blob path from the URL url_parts = image_url.split('storage.googleapis.com/')[1] logger.info(f"[IMAGE PROXY] URL parts after split: {url_parts}") # Remove query parameters if present url_parts = url_parts.split('?')[0] logger.info(f"[IMAGE PROXY] URL parts after removing query params: {url_parts}") # Split into bucket name and blob path path_components = url_parts.split('/', 1) logger.info(f"[IMAGE PROXY] Path components: {path_components}") if len(path_components) < 2: logger.error(f"[IMAGE PROXY] ERROR: Invalid URL format - path_components: {path_components}") return jsonify({'error': 'Invalid URL format.'}), 400 url_bucket_name = path_components[0] blob_path = path_components[1] logger.info(f"[IMAGE PROXY] Extracted bucket name: {url_bucket_name}") logger.info(f"[IMAGE PROXY] Extracted blob path: {blob_path}") # Verify bucket name matches (optional security check) expected_bucket_name = bucket.name logger.info(f"[IMAGE PROXY] Expected bucket name: {expected_bucket_name}") if url_bucket_name != expected_bucket_name: logger.error(f"[IMAGE PROXY] ERROR: Bucket name mismatch - URL: {url_bucket_name}, Expected: {expected_bucket_name}") return jsonify({'error': 'Bucket name mismatch.'}), 403 logger.info(f"[IMAGE PROXY] Creating blob object for path: {blob_path}") # Get the blob blob = bucket.blob(blob_path) logger.info(f"[IMAGE PROXY] Checking if blob exists...") if not blob.exists(): logger.error(f"[IMAGE PROXY] ERROR: Image not found at path: {blob_path}") return jsonify({'error': 'Image not found.'}), 404 logger.info(f"[IMAGE PROXY] Downloading blob...") # Download and return the image image_bytes = blob.download_as_bytes() content_type = blob.content_type or 'application/octet-stream' logger.info(f"[IMAGE PROXY] Successfully downloaded {len(image_bytes)} bytes, content-type: {content_type}") # Add cache headers for better performance response = Response(image_bytes, content_type=content_type) response.headers['Cache-Control'] = 'public, max-age=3600' # Cache for 1 hour return response except IndexError as e: logger.error(f"[IMAGE PROXY] URL parsing IndexError: {e}") logger.error(f"[IMAGE PROXY] URL was: {image_url}") return jsonify({'error': 'Invalid URL format.'}), 400 except Exception as e: # This will catch parsing errors or other unexpected issues. logger.error(f"[IMAGE PROXY] Unexpected error: {e}") logger.error(f"[IMAGE PROXY] Error type: {type(e).__name__}") logger.error(f"[IMAGE PROXY] URL was: {image_url}") import traceback logger.error(f"[IMAGE PROXY] Full traceback: {traceback.format_exc()}") return jsonify({'error': 'Internal server error processing the image request.'}), 500 @app.route('/api/sozo/projects//charts', methods=['POST']) def regenerate_sozo_chart(project_id): logger.info(f"Endpoint /charts POST for project {project_id}") try: token = request.headers.get('Authorization', '').split(' ')[1] uid = verify_token(token) if not uid: return jsonify({'error': 'Unauthorized'}), 401 logger.info(f"Token verified for user {uid} for chart regeneration.") project_ref = db.reference(f'sozo_projects/{project_id}') project_data = project_ref.get() if not project_data or project_data.get('uid') != uid: logger.warning(f"User {uid} failed to regenerate chart: Project {project_id} not found or not owned.") return jsonify({'error': 'Project not found or unauthorized'}), 404 data = request.get_json() description = data.get('description') chart_id_to_replace = data.get('chart_id') if not description or not chart_id_to_replace: return jsonify({'error': 'Chart description and chart_id are required'}), 400 logger.info(f"User {uid}: Regenerating chart '{chart_id_to_replace}' for project {project_id} with new description: '{description}'") blob_path = f"sozo_projects/{uid}/{project_id}/data{Path(project_data['originalFilename']).suffix}" blob = bucket.blob(blob_path) file_bytes = blob.download_as_bytes() df = load_dataframe_safely(io.BytesIO(file_bytes), project_data['originalFilename']) new_chart_spec = generate_single_chart(df, description) logger.info(f"User {uid}: Updating chart spec in database for project {project_id}.") report_content_ref = project_ref.child('report_content') report_content = report_content_ref.get() chart_specs = report_content.get('chart_specs', []) chart_found = False for i, spec in enumerate(chart_specs): if spec.get('id') == chart_id_to_replace: chart_specs[i] = new_chart_spec chart_found = True break if not chart_found: logger.warning(f"User {uid}: Chart with id {chart_id_to_replace} not found in project {project_id}.") return jsonify({'error': f'Chart with id {chart_id_to_replace} not found'}), 404 report_content_ref.child('chart_specs').set(chart_specs) project_ref.update({'updatedAt': datetime.utcnow().isoformat()}) logger.info(f"User {uid}: Successfully regenerated chart {chart_id_to_replace} for project {project_id}.") return jsonify({'success': True, 'new_chart_spec': new_chart_spec}), 200 except Exception as e: logger.error(f"CRITICAL ERROR regenerating chart for {project_id}: {traceback.format_exc()}") return jsonify({'error': str(e)}), 500 @app.route('/api/sozo/projects//update-narration-audio', methods=['POST']) def update_narration_audio(project_id): logger.info(f"Endpoint /update-narration-audio POST for project {project_id}") try: token = request.headers.get('Authorization', '').split(' ')[1] uid = verify_token(token) if not uid: return jsonify({'error': 'Unauthorized'}), 401 logger.info(f"Token verified for user {uid} for narration update.") data = request.get_json() scene_id = data.get('scene_id') narration_text = data.get('narration_text') voice_model = data.get('voice_model', 'aura-2-andromeda-en') if not scene_id or narration_text is None: return jsonify({'error': 'scene_id and narration_text are required'}), 400 logger.info(f"User {uid}: Updating narration for scene {scene_id} in project {project_id}.") audio_bytes = deepgram_tts(narration_text, voice_model) if not audio_bytes: logger.error(f"User {uid}: Deepgram TTS failed for project {project_id}, scene {scene_id}.") return jsonify({'error': 'Failed to generate audio'}), 500 audio_blob_name = f"sozo_projects/{uid}/{project_id}/audio/{scene_id}.mp3" logger.info(f"User {uid}: Uploading new audio to {audio_blob_name}.") audio_blob = bucket.blob(audio_blob_name) audio_blob.upload_from_string(audio_bytes, content_type="audio/mpeg") new_audio_url = audio_blob.public_url logger.info(f"User {uid}: Updating database with new narration and audio URL for project {project_id}.") scene_ref = db.reference(f'sozo_projects/{project_id}/video_script/scenes') scenes = scene_ref.get() scene_found = False if scenes: for i, scene in enumerate(scenes): if scene.get('scene_id') == scene_id: scene_ref.child(str(i)).update({ 'narration': narration_text, 'audio_storage_path': new_audio_url }) scene_found = True break if not scene_found: logger.warning(f"User {uid}: Scene {scene_id} not found in database for project {project_id} during narration update.") return jsonify({'error': 'Scene not found in database'}), 404 project_ref = db.reference(f'sozo_projects/{project_id}') project_ref.update({'updatedAt': datetime.utcnow().isoformat()}) logger.info(f"User {uid}: Successfully updated narration for scene {scene_id} in project {project_id}.") return jsonify({'success': True, 'new_audio_url': new_audio_url}), 200 except Exception as e: logger.error(f"CRITICAL ERROR updating narration for {project_id}: {traceback.format_exc()}") return jsonify({'error': str(e)}), 500 # ----------------------------------------------------------------------------- # 5. UNIVERSAL ENDPOINTS (Waitlist, Feedback, Credits) # ----------------------------------------------------------------------------- @app.route('/join-waitlist', methods=['POST']) def join_waitlist(): try: data = request.get_json() if not data: return jsonify({"status": "error", "message": "Invalid request. JSON payload expected."}), 400 email = data.get('email') if not email or not is_valid_email(email): return jsonify({"status": "error", "message": "A valid email is required."}), 400 email = email.lower() waitlist_ref = db.reference('sozo_waitlist') if waitlist_ref.order_by_child('email').equal_to(email).get(): return jsonify({"status": "success", "message": "You are already on the waitlist!"}), 200 new_entry_ref = waitlist_ref.push() new_entry_ref.set({'email': email, 'timestamp': datetime.utcnow().isoformat() + 'Z'}) return jsonify({"status": "success", "message": "Thank you for joining the waitlist!"}), 201 except Exception as e: traceback.print_exc() return jsonify({"status": "error", "message": "An internal server error occurred."}), 500 @app.route('/api/feedback', methods=['POST']) def submit_feedback(): try: token = request.headers.get('Authorization', '').split(' ')[1] uid = verify_token(token) if not uid: return jsonify({'error': 'Invalid or expired token'}), 401 data = request.get_json() message = data.get('message') if not message: return jsonify({'error': 'message is required'}), 400 user_email = (db.reference(f'users/{uid}').get() or {}).get('email', 'unknown') feedback_ref = db.reference('feedback').push() feedback_ref.set({"user_id": uid, "user_email": user_email, "type": data.get('type', 'general'), "message": message, "created_at": datetime.utcnow().isoformat(), "status": "open"}) return jsonify({"success": True, "feedback_id": feedback_ref.key}), 201 except Exception as e: return jsonify({'error': str(e)}), 500 @app.route('/api/user/request-credits', methods=['POST']) def request_credits(): try: token = request.headers.get('Authorization', '').split(' ')[1] uid = verify_token(token) if not uid: return jsonify({'error': 'Invalid token'}), 401 requested_credits = request.get_json().get('requested_credits') if requested_credits is None: return jsonify({'error': 'requested_credits is required'}), 400 credit_request_ref = db.reference('credit_requests').push() credit_request_ref.set({'user_id': uid, 'requested_credits': requested_credits, 'status': 'pending', 'requested_at': datetime.utcnow().isoformat()}) return jsonify({'success': True, 'request_id': credit_request_ref.key}) except Exception as e: return jsonify({'error': str(e)}), 500 # ----------------------------------------------------------------------------- # 6. ADMIN ENDPOINTS # ----------------------------------------------------------------------------- @app.route('/api/admin/dashboard-stats', methods=['GET']) def get_admin_dashboard_stats(): """A singular endpoint to fetch all key metrics for the admin dashboard.""" try: verify_admin(request.headers.get('Authorization', '')) # Fetch all necessary data in one go all_users = db.reference('users').get() or {} all_projects = db.reference('sozo_projects').get() or {} all_feedback = db.reference('feedback').get() or {} all_credit_requests = db.reference('credit_requests').get() or {} waitlist = db.reference('sozo_waitlist').get() or {} # --- Initialize Stats --- stats = { "user_stats": {"total": 0, "new_24h": 0, "new_7d": 0}, "project_stats": {"total": 0, "new_24h": 0, "failed": 0, "videos_generated": 0}, "action_items": {"open_feedback": 0, "pending_credit_requests": 0}, "growth_stats": {"waitlist_total": 0} } now = datetime.utcnow() one_day_ago = now - timedelta(days=1) seven_days_ago = now - timedelta(days=7) # --- Process Users --- stats["user_stats"]["total"] = len(all_users) for user_data in all_users.values(): created_at_str = user_data.get('created_at') if created_at_str: created_at_dt = datetime.fromisoformat(created_at_str) if created_at_dt > one_day_ago: stats["user_stats"]["new_24h"] += 1 if created_at_dt > seven_days_ago: stats["user_stats"]["new_7d"] += 1 # --- Process Projects --- stats["project_stats"]["total"] = len(all_projects) for project_data in all_projects.values(): created_at_str = project_data.get('createdAt') if created_at_str: created_at_dt = datetime.fromisoformat(created_at_str) if created_at_dt > one_day_ago: stats["project_stats"]["new_24h"] += 1 if project_data.get('status') == 'failed': stats["project_stats"]["failed"] += 1 if project_data.get('status') == 'video_complete': stats["project_stats"]["videos_generated"] += 1 # --- Process Action Items --- stats["action_items"]["open_feedback"] = sum(1 for fb in all_feedback.values() if fb.get('status') == 'open') stats["action_items"]["pending_credit_requests"] = sum(1 for cr in all_credit_requests.values() if cr.get('status') == 'pending') # --- Process Growth --- stats["growth_stats"]["waitlist_total"] = len(waitlist) return jsonify(stats), 200 except Exception as e: traceback.print_exc() return jsonify({'error': str(e)}), 500 @app.route('/api/admin/credit_requests', methods=['GET']) def list_credit_requests(): try: verify_admin(request.headers.get('Authorization', '')) requests_list = [{'id': req_id, **data} for req_id, data in (db.reference('credit_requests').get() or {}).items()] return jsonify({'credit_requests': requests_list}) except Exception as e: return jsonify({'error': str(e)}), 500 @app.route('/api/admin/credit_requests/', methods=['PUT']) def process_credit_request(request_id): try: admin_uid = verify_admin(request.headers.get('Authorization', '')) req_ref = db.reference(f'credit_requests/{request_id}') req_data = req_ref.get() if not req_data: return jsonify({'error': 'Credit request not found'}), 404 decision = request.get_json().get('decision') if decision not in ['approved', 'declined']: return jsonify({'error': 'decision must be "approved" or "declined"'}), 400 if decision == 'approved': user_ref = db.reference(f'users/{req_data["user_id"]}') new_total = (user_ref.get() or {}).get('credits', 0) + float(req_data.get('requested_credits', 0)) user_ref.update({'credits': new_total}) req_ref.update({'status': 'approved', 'processed_by': admin_uid, 'processed_at': datetime.utcnow().isoformat()}) return jsonify({'success': True, 'new_user_credits': new_total}) else: req_ref.update({'status': 'declined', 'processed_by': admin_uid, 'processed_at': datetime.utcnow().isoformat()}) return jsonify({'success': True, 'message': 'Credit request declined'}) except Exception as e: return jsonify({'error': str(e)}), 500 @app.route('/api/admin/users', methods=['GET']) def admin_list_users(): try: verify_admin(request.headers.get('Authorization', '')) all_users = db.reference('users').get() or {} user_list = [{'uid': uid, **data} for uid, data in all_users.items()] return jsonify({'users': user_list}), 200 except Exception as e: return jsonify({'error': str(e)}), 500 @app.route('/api/admin/users//credits', methods=['PUT']) def admin_update_credits(uid): try: verify_admin(request.headers.get('Authorization', '')) add_credits = request.get_json().get('add_credits') if add_credits is None: return jsonify({'error': 'add_credits is required'}), 400 user_ref = db.reference(f'users/{uid}') if not user_ref.get(): return jsonify({'error': 'User not found'}), 404 new_total = user_ref.get().get('credits', 0) + float(add_credits) user_ref.update({'credits': new_total}) return jsonify({'success': True, 'new_total_credits': new_total}) except Exception as e: return jsonify({'error': str(e)}), 500 @app.route('/api/admin/feedback', methods=['GET']) def list_feedback(): """List all feedback submissions for admin review.""" try: verify_admin(request.headers.get('Authorization', '')) # Get all feedback from Firebase all_feedback = db.reference('feedback').get() or {} # Convert to list format with feedback IDs feedback_list = [] for feedback_id, feedback_data in all_feedback.items(): feedback_item = { 'id': feedback_id, **feedback_data } feedback_list.append(feedback_item) # Sort by created_at (most recent first) if timestamp exists feedback_list.sort(key=lambda x: x.get('created_at', ''), reverse=True) return jsonify({'feedback': feedback_list}), 200 except Exception as e: traceback.print_exc() return jsonify({'error': str(e)}), 500 @app.route('/api/admin/feedback/', methods=['PUT']) def update_feedback_status(feedback_id): """Update feedback status (e.g., mark as reviewed, resolved, etc.)""" try: admin_uid = verify_admin(request.headers.get('Authorization', '')) # Check if feedback exists feedback_ref = db.reference(f'feedback/{feedback_id}') feedback_data = feedback_ref.get() if not feedback_data: return jsonify({'error': 'Feedback not found'}), 404 # Get the new status from request request_data = request.get_json() new_status = request_data.get('status') # Validate status valid_statuses = ['open', 'reviewed', 'resolved', 'closed'] if new_status not in valid_statuses: return jsonify({'error': f'Status must be one of: {", ".join(valid_statuses)}'}), 400 # Update feedback with new status and admin info update_data = { 'status': new_status, 'processed_by': admin_uid, 'processed_at': datetime.utcnow().isoformat() } # Add admin notes if provided admin_notes = request_data.get('admin_notes') if admin_notes: update_data['admin_notes'] = admin_notes feedback_ref.update(update_data) return jsonify({ 'success': True, 'message': f'Feedback status updated to {new_status}', 'feedback_id': feedback_id }), 200 except Exception as e: traceback.print_exc() return jsonify({'error': str(e)}), 500 @app.route('/api/admin/feedback/', methods=['GET']) def get_feedback_details(feedback_id): """Get detailed view of a specific feedback item.""" try: verify_admin(request.headers.get('Authorization', '')) # Get feedback details feedback_ref = db.reference(f'feedback/{feedback_id}') feedback_data = feedback_ref.get() if not feedback_data: return jsonify({'error': 'Feedback not found'}), 404 # Add the feedback ID to the response feedback_details = { 'id': feedback_id, **feedback_data } # Optionally get user details if user_id is present if 'user_id' in feedback_data: user_ref = db.reference(f'users/{feedback_data["user_id"]}') user_data = user_ref.get() if user_data: feedback_details['user_details'] = { 'email': user_data.get('email'), 'name': user_data.get('name'), 'created_at': user_data.get('created_at') } return jsonify({'feedback': feedback_details}), 200 except Exception as e: traceback.print_exc() return jsonify({'error': str(e)}), 500 # ----------------------------------------------------------------------------- # 4. NOTIFICATION ENDPOINTS # ----------------------------------------------------------------------------- @app.route('/api/admin/notifications/send', methods=['POST']) def admin_send_notification(): logger.info("Endpoint /api/admin/notifications/send POST: Received request.") try: verify_admin(request.headers.get('Authorization')) data = request.get_json() message_content = data.get('message') target_group = data.get('target_group', 'all') target_users_list = data.get('target_users', []) send_as_email = data.get('send_as_email', False) email_subject = data.get('email_subject') email_body_html = data.get('email_body_html') if not message_content: return jsonify({'error': 'In-app notification message is required'}), 400 if send_as_email and (not email_subject or not email_body_html): return jsonify({'error': 'Email subject and body are required when sending as email.'}), 400 recipients = [] # List of tuples (uid, email) if target_group == 'all': all_users = db.reference('users').get() or {} for uid, user_data in all_users.items(): recipients.append((uid, user_data.get('email'))) elif target_group == 'waitlist': waitlist_users = db.reference('sozo_waitlist').get() or {} for _, user_data in waitlist_users.items(): # For waitlist, UID is None as they are not registered users yet recipients.append((None, user_data.get('email'))) elif target_users_list: all_users = db.reference('users').get() or {} for uid in target_users_list: if uid in all_users: recipients.append((uid, all_users[uid].get('email'))) else: return jsonify({'error': 'Invalid target specified'}), 400 sent_count = 0 for uid_recipient, email_recipient in recipients: if _send_notification( user_id=uid_recipient, user_email=email_recipient, message_content=message_content, send_email=send_as_email, email_subject=email_subject, email_body=email_body_html ): sent_count += 1 return jsonify({'success': True, 'message': f"Notification dispatched for {sent_count} recipient(s)."}), 200 except PermissionError as e: return jsonify({'error': str(e)}), 403 except Exception as e: logger.error(f"CRITICAL ERROR during notification send: {traceback.format_exc()}") return jsonify({'error': str(e)}), 500 @app.route('/api/user/notifications', methods=['GET']) def get_user_notifications(): try: logger.info("Getting user notifications - start") # Get and validate authorization token auth_header = request.headers.get('Authorization', '') logger.info(f"Authorization header present: {bool(auth_header)}") if not auth_header or not auth_header.startswith('Bearer '): logger.warning("Missing or invalid Authorization header format") return jsonify({'error': 'Missing or invalid authorization header'}), 401 token = auth_header.split(' ')[1] logger.info(f"Token extracted, length: {len(token) if token else 0}") # Verify token and get user ID uid = verify_token(token) if not uid: logger.warning(f"Token verification failed for token: {token[:20]}...") return jsonify({'error': 'Unauthorized'}), 401 logger.info(f"User authenticated: {uid}") # Get notifications reference notifications_ref = db.reference(f'notifications/{uid}') logger.info(f"Notifications reference created for path: notifications/{uid}") # Try to get notifications with error handling try: user_notifications = notifications_ref.order_by_child('created_at').get() logger.info(f"Notifications query successful, raw result type: {type(user_notifications)}") logger.info(f"Raw notifications data: {user_notifications}") except Exception as db_error: logger.error(f"Database query failed: {str(db_error)}") # Fallback to unordered query logger.info("Attempting fallback to unordered query") user_notifications = notifications_ref.get() logger.info(f"Fallback query result: {user_notifications}") # Handle empty or None results if not user_notifications: logger.info("No notifications found for user") return jsonify([]), 200 # Convert to list and handle different data structures notifications_list = [] if isinstance(user_notifications, dict): logger.info(f"Processing dict with {len(user_notifications)} items") for key, notification in user_notifications.items(): if isinstance(notification, dict): # Add the key as id if not present notification_copy = notification.copy() if 'id' not in notification_copy: notification_copy['id'] = key notifications_list.append(notification_copy) logger.debug(f"Added notification: {key}") else: logger.warning(f"Unexpected notification format for key {key}: {type(notification)}") else: logger.warning(f"Unexpected notifications data type: {type(user_notifications)}") return jsonify({'error': 'Unexpected data format'}), 500 logger.info(f"Processed {len(notifications_list)} notifications") # Sort notifications by created_at (newest first) try: sorted_notifications = sorted( notifications_list, key=lambda item: item.get('created_at', 0), reverse=True ) logger.info(f"Notifications sorted successfully, count: {len(sorted_notifications)}") # Log first notification for debugging if sorted_notifications: logger.debug(f"First notification: {sorted_notifications[0]}") except Exception as sort_error: logger.error(f"Error sorting notifications: {str(sort_error)}") # Return unsorted if sorting fails sorted_notifications = notifications_list logger.info(f"Returning {len(sorted_notifications)} notifications") return jsonify(sorted_notifications), 200 except Exception as e: logger.error(f"CRITICAL ERROR getting notifications: {traceback.format_exc()}") logger.error(f"Exception type: {type(e).__name__}") logger.error(f"Exception message: {str(e)}") return jsonify({'error': 'Internal server error'}), 500 @app.route('/api/user/notifications//read', methods=['POST']) def mark_notification_read(notification_id): try: token = request.headers.get('Authorization', '').split(' ')[1] uid = verify_token(token) if not uid: return jsonify({'error': 'Unauthorized'}), 401 notif_ref = db.reference(f'notifications/{uid}/{notification_id}') if not notif_ref.get(): return jsonify({'error': 'Notification not found'}), 404 notif_ref.update({'read': True, 'read_at': datetime.now(timezone.utc).isoformat()}) return jsonify({'success': True, 'message': 'Notification marked as read.'}), 200 except Exception as e: logger.error(f"CRITICAL ERROR marking notification read: {traceback.format_exc()}") return jsonify({'error': str(e)}), 500 # ----------------------------------------------------------------------------- # 7. MAIN EXECUTION # ----------------------------------------------------------------------------- if __name__ == '__main__': app.run(debug=True, host="0.0.0.0", port=int(os.environ.get("PORT", 7860)))