sbs-API / main.py
rairo's picture
Update main.py
d8b8a06 verified
# 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 <onboarding@sozofix.tech>",
"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/<string:project_id>', 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/<string:project_id>', 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/<string:project_id>', 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/<string:project_id>/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"""
<div style="font-family: Arial, sans-serif; max-width: 600px; margin: 0 auto; padding: 20px; background-color: #f9f9f9;">
<div style="background-color: white; padding: 30px; border-radius: 10px; box-shadow: 0 2px 10px rgba(0,0,0,0.1);">
<div style="text-align: center; margin-bottom: 30px;">
<h1 style="color: #2563eb; margin: 0; font-size: 28px;">🎉 Video Generation Complete!</h1>
</div>
<div style="background-color: #f0f9ff; padding: 20px; border-radius: 8px; border-left: 4px solid #2563eb; margin-bottom: 25px;">
<h2 style="color: #1e40af; margin: 0 0 10px 0; font-size: 20px;">Your video is ready!</h2>
<p style="color: #374151; margin: 0; font-size: 16px;">
Great news! Your video for project <strong>"{project_name}"</strong> has been successfully generated and is ready for viewing.
</p>
</div>
<div style="text-align: center; margin: 30px 0;">
<a href="{video_url}"
style="display: inline-block; background-color: #2563eb; color: white; padding: 15px 30px;
text-decoration: none; border-radius: 8px; font-weight: bold; font-size: 16px;
box-shadow: 0 2px 4px rgba(37, 99, 235, 0.3);">
🎬 Watch Your Video
</a>
</div>
<div style="background-color: #f8fafc; padding: 20px; border-radius: 8px; margin-top: 25px;">
<h3 style="color: #374151; margin: 0 0 15px 0; font-size: 18px;">What's Next?</h3>
<ul style="color: #6b7280; margin: 0; padding-left: 20px; line-height: 1.6;">
<li>Click the button above to view your generated video</li>
<li>Share your video with your team or clients</li>
<li>Download the video for offline use</li>
<li>Create more videos with your remaining credits</li>
</ul>
</div>
<div style="text-align: center; margin-top: 30px; padding-top: 20px; border-top: 1px solid #e5e7eb;">
<p style="color: #9ca3af; font-size: 14px; margin: 0;">
This video was generated by <strong>Sozo Business Studio</strong><br>
Need help? Contact us at <a href="mailto:support@sozofix.tech" style="color: #2563eb;">support@sozofix.tech</a>
</p>
</div>
</div>
</div>
"""
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/<string:project_id>/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/<string:project_id>/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/<string:project_id>/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/<string:project_id>/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/<string:request_id>', 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/<string:uid>/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/<string:feedback_id>', 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/<string:feedback_id>', 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/<string:notification_id>/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)))