"""
Baby Cry AI - Mobile PWA Application
With Continuous Learning System
"""
from flask import Flask, request, jsonify, render_template_string, send_from_directory, session, redirect, url_for
from flask_cors import CORS
from werkzeug.security import generate_password_hash, check_password_hash
import os
import sys
import tempfile
import shutil
import numpy as np
import json
from datetime import datetime
from collections import defaultdict
from contextlib import contextmanager
# Load environment variables from .env file if it exists (for local development)
try:
from dotenv import load_dotenv
# Try loading from project root
env_path = os.path.join(os.path.dirname(os.path.dirname(__file__)), '.env')
if os.path.exists(env_path):
load_dotenv(env_path)
print("✅ Loaded environment variables from .env file")
else:
# Try loading from current directory
load_dotenv()
except ImportError:
# python-dotenv not installed, continue without it
pass
except Exception as e:
# If .env loading fails, continue (might be in production)
pass
# Log whether OpenAI key is available (for Cloud analysis mode)
if os.environ.get("OPENAI_API_KEY", "").strip():
print("✅ OPENAI_API_KEY is set (Cloud analysis available)")
else:
print("⚠️ OPENAI_API_KEY not set (Cloud analysis will fail until you add it to .env and restart)")
# Add the src directory to Python path for imports to work from any location
SRC_DIR = os.path.dirname(os.path.abspath(__file__))
if SRC_DIR not in sys.path:
sys.path.insert(0, SRC_DIR)
from models.baseline_model import BaselineModel
from audio_processor import AudioProcessor
from feedback_manager import FeedbackManager
from model_manager import ModelManager
from supabase_storage import SupabaseStorage
try:
from openai_audio import predict_via_openai_audio
except ImportError:
predict_via_openai_audio = None
try:
from models.voc2vec_model import predict_from_audio_file as voc2vec_predict, predict_top_k_from_audio_file as voc2vec_predict_top_k, is_available as voc2vec_available
except ImportError:
try:
from voc2vec_model import predict_from_audio_file as voc2vec_predict, predict_top_k_from_audio_file as voc2vec_predict_top_k, is_available as voc2vec_available
except ImportError:
voc2vec_predict = None
voc2vec_predict_top_k = None
voc2vec_available = lambda **kw: False
# user_analytics and database (SQLite) removed - using Supabase only
# Admin dashboard template will be imported at route definition to avoid circular imports
app = Flask(__name__, static_folder='static')
app.secret_key = os.urandom(24)
# Configure CORS to allow requests from mobile apps and web
CORS(app, resources={
r"/*": {
"origins": "*", # Allow all origins for mobile APK and web
"methods": ["GET", "POST", "OPTIONS"],
"allow_headers": ["Content-Type", "Authorization"]
}
})
# Initialize components
# Use fast_mode=True for faster analysis (skips expensive CQT features)
# This reduces analysis time from ~10s to ~2-3s
audio_processor = AudioProcessor(fast_mode=True)
# Initialize model - Random Forest fallback for 'local' mode (voc2vec is tried first)
# Neural Network is disabled due to poor performance (23.3% accuracy)
model = None
model_type = "unknown"
# Load Random Forest model directly (fallback model)
try:
baseline_model = BaselineModel()
if baseline_model.load_model():
model = baseline_model
model_type = "random_forest"
n_feats = len(baseline_model.expected_feature_names or [])
print(f"✅ Loaded Random Forest Model ({n_feats} features)")
else:
raise Exception("Random Forest model not available")
except Exception as e:
print(f"❌ Random Forest model not available: {e}")
print(" Creating untrained model as fallback")
model = BaselineModel() # Fallback to untrained model
model_type = "random_forest"
# Use environment variables for paths (for cloud deployment)
DATA_DIR = os.environ.get('DATA_DIR', '../data')
MODELS_DIR = os.environ.get('MODELS_DIR', '../models')
FEEDBACK_DIR = os.environ.get('FEEDBACK_DIR', '../feedback_data')
feedback_manager = FeedbackManager(feedback_dir=FEEDBACK_DIR, data_dir=DATA_DIR)
model_manager = ModelManager(models_dir=MODELS_DIR, data_dir=DATA_DIR, feedback_dir=FEEDBACK_DIR)
# Cry/not-cry gate: rejects non-cry audio before classification (Phase 2)
cry_gate = None
try:
import joblib as _joblib
_gate_path = os.path.join(MODELS_DIR, 'cry_detector.pkl')
if os.path.exists(_gate_path):
cry_gate = _joblib.load(_gate_path)
print(f"✅ Loaded cry/not-cry gate (reject threshold {cry_gate.get('reject_threshold')})")
else:
print("⚠️ cry_detector.pkl not found — cry gate disabled")
except Exception as _e:
print(f"⚠️ Cry gate unavailable: {_e}")
cry_gate = None
def cry_gate_probability(features):
"""Return p(cry) for a feature dict, or None if the gate can't run."""
if cry_gate is None or features is None:
return None
try:
row = [features.get(n, 0.0) for n in cry_gate['feature_names']]
X = cry_gate['scaler'].transform(np.array(row).reshape(1, -1))
return float(cry_gate['model'].predict_proba(X)[0][cry_gate['cry_class_index']])
except Exception:
return None
# Initialize Supabase Storage for audio files
supabase_storage = SupabaseStorage()
# Ensure bucket exists on startup
if supabase_storage.enabled:
supabase_storage.ensure_bucket_exists()
# user_analytics removed - all analytics now handled by Supabase
# Stub database for local run without Supabase (predictions/feedback not persisted)
class _LocalStubDb:
@contextmanager
def get_connection(self):
class _FakeCursor:
def execute(self, *a, **k): pass
def fetchone(self): return None
class _FakeConn:
def cursor(self, cursor_factory=None): return _FakeCursor()
def commit(self): pass
def rollback(self): pass
def close(self): pass
yield _FakeConn()
def create_user(self, user_data): pass
def get_user(self, user_id): return None
def get_user_by_email(self, email): return None
def get_user_by_phone(self, phone): return None
def get_user_by_email_or_phone(self, identifier): return None
def get_all_users(self, limit=None, offset=0): return []
def update_user(self, user_id, user_data): return False
def delete_user(self, user_id): return False
def count_users(self): return 0
def ensure_user_exists(self, user_id): pass
def create_prediction(self, prediction_data): pass
def count_predictions(self, user_id=None): return 0
def get_predictions_list(self, limit=None, offset=0, user_id=None): return []
def update_prediction_what_helped(self, prediction_id, user_id, what_helped): return False
def create_feedback(self, feedback_data): pass
def count_feedback(self, is_correct=None): return 0
def get_feedback_list(self, limit=None, offset=0, is_correct=None): return []
def delete_feedback(self, feedback_id): return False
def delete_feedback_by_submission_id(self, submission_id): return False
def delete_feedbacks_by_date_range(self, start_date, end_date): return 0
def delete_feedbacks_by_user_id(self, user_id): return 0
def get_feedback_by_id(self, feedback_id): return None
def create_analytics_event(self, event_data): pass
def get_user_stats(self): return {"total_users": 0, "total_predictions": 0}
def get_feedback_stats(self): return {"total": 0, "correct": 0, "incorrect": 0}
def get_platform_distribution(self): return {}
def get_language_distribution(self): return {}
def get_user_timeline(self, days=30): return []
def get_feedback_timeline(self, days=30): return []
# Initialize database (Supabase PostgreSQL, or stub for local run)
SUPABASE_DB_URL = os.environ.get('SUPABASE_DB_URL')
db = None
DB_INIT_ERROR = None
if not SUPABASE_DB_URL:
db = _LocalStubDb()
DB_INIT_ERROR = 'SUPABASE_DB_URL not set'
print("⚠️ Running without Supabase (SUPABASE_DB_URL not set). Predictions/feedback will not be saved.")
else:
try:
from database_supabase import SupabaseDatabase
print(f"🔌 Attempting to connect to Supabase...")
try:
from urllib.parse import urlparse
parsed = urlparse(SUPABASE_DB_URL)
print(f" Host: {parsed.hostname}:{parsed.port or 5432}")
except Exception:
pass
db = SupabaseDatabase()
print("✅ Connected to Supabase PostgreSQL database")
except Exception as e:
error_details = str(e)
print(f"❌ Failed to connect to Supabase: {error_details}")
print("⚠️ Using local stub (predictions/feedback will not be saved). Fix connection and restart to use Supabase.")
DB_INIT_ERROR = str(e)[:200]
db = _LocalStubDb()
# Store for pending feedback
pending_feedback = {}
# ==================== ADMIN AUTHENTICATION ====================
ADMIN_PASSWORD = os.environ.get('ADMIN_PASSWORD', 'admin123') # Default password, should be changed in production
def admin_required(f):
"""Decorator to require admin authentication"""
from functools import wraps
@wraps(f)
def decorated_function(*args, **kwargs):
if not session.get('admin_authenticated', False):
return jsonify({'error': 'Admin authentication required'}), 403
return f(*args, **kwargs)
return decorated_function
# ==================== TRANSLATION SYSTEM ====================
TRANSLATIONS_DIR = os.path.join(SRC_DIR, 'translations')
translations_cache = {}
def load_translations(lang='ar'):
"""Load translations for a given language"""
if lang in translations_cache:
return translations_cache[lang]
translation_file = os.path.join(TRANSLATIONS_DIR, f'{lang}.json')
if not os.path.exists(translation_file):
# Fallback to Arabic if translation file doesn't exist
if lang != 'ar':
return load_translations('ar')
return {}
try:
with open(translation_file, 'r', encoding='utf-8') as f:
translations = json.load(f)
translations_cache[lang] = translations
return translations
except Exception as e:
print(f"Error loading translations for {lang}: {e}")
return {}
def detect_language():
"""Detect user's preferred language"""
# Priority: query param > session > default (Arabic)
# Always default to Arabic, ignore browser language
lang = request.args.get('lang', '').lower()
if lang in ['ar', 'fr', 'en']:
session['lang'] = lang
return lang
# Check session
if 'lang' in session and session['lang'] in ['ar', 'fr', 'en']:
return session['lang']
# Default to Arabic (ignore browser language)
# Always set session to ensure consistency
session['lang'] = 'ar'
return 'ar'
def t(key, lang=None, **kwargs):
"""Get translation for a key, with optional formatting"""
if lang is None:
lang = detect_language()
translations = load_translations(lang)
# Handle nested keys (e.g., "categories.hunger")
keys = key.split('.')
value = translations
for k in keys:
if isinstance(value, dict):
value = value.get(k, key)
else:
value = key
break
# If translation not found, try English, then return key
if value == key and lang != 'en':
translations_en = load_translations('en')
value = translations_en
for k in keys:
if isinstance(value, dict):
value = value.get(k, key)
else:
value = key
break
# Format with kwargs if provided
if isinstance(value, str) and kwargs:
try:
return value.format(**kwargs)
except:
return value
return value if isinstance(value, str) else key
# Mobile-First PWA HTML Template
HTML_TEMPLATE = """
إطمئن - Itma'In
"""
# Serve static files
@app.route('/static/')
def serve_static(filename):
return send_from_directory('static', filename)
@app.route('/login')
def login_page():
"""Serve the login page"""
lang = detect_language()
translations = load_translations(lang)
# Check if already logged in
if session.get('logged_in'):
return redirect('/')
# Inject translations and language into template
template = HTML_TEMPLATE.replace('{{LANG}}', lang)
template = template.replace('{{DIR}}', 'rtl' if lang == 'ar' else 'ltr')
return render_template_string(template)
@app.route('/')
def home():
"""Serve the main web interface"""
# Check if user is logged in (if login is required)
# For now, we'll allow both logged in and not logged in users
# But we can add authentication check here if needed
# detect_language() now always sets session['lang'] to 'ar' if not set
lang = detect_language()
translations = load_translations(lang)
# Inject translations and language into template
template = HTML_TEMPLATE.replace('{{LANG}}', lang)
template = template.replace('{{DIR}}', 'rtl' if lang == 'ar' else 'ltr')
# Replace translation keys with actual translations
# This will be done in JavaScript for dynamic updates
return render_template_string(template)
@app.route('/api/translations')
def get_translations():
"""Get translations for current language"""
# Check if lang is provided as query parameter (for immediate language switching)
lang_param = request.args.get('lang', '').lower()
if lang_param in ['ar', 'fr', 'en']:
lang = lang_param
# Also update session
session['lang'] = lang
else:
# detect_language() now always sets session['lang'] to 'ar' if not set
lang = detect_language()
translations = load_translations(lang)
return jsonify({
'lang': lang,
'translations': translations,
'rtl': lang == 'ar'
})
@app.route('/api/set-language', methods=['POST'])
def set_language():
"""Set language preference"""
data = request.get_json()
lang = data.get('lang', 'ar')
if lang in ['ar', 'fr', 'en']:
session['lang'] = lang
return jsonify({'success': True, 'lang': lang})
return jsonify({'success': False, 'error': 'Invalid language'}), 400
@app.route('/model-info')
def model_info():
"""Get model information"""
try:
# Try to load model if not ready
model_ready = False
if hasattr(model, 'is_trained'):
model_ready = model.is_trained
elif hasattr(model, 'rf_loaded') or hasattr(model, 'nn_loaded'):
model_ready = getattr(model, 'rf_loaded', False) or getattr(model, 'nn_loaded', False)
else:
model_ready = True
if not model_ready:
if hasattr(model, 'load_model'):
if not model.load_model():
return jsonify({'error': 'Model not trained'})
elif hasattr(model, 'load_models'):
if not model.load_models():
return jsonify({'error': 'Model not trained'})
else:
return jsonify({'error': 'Model not available'})
info = model.get_model_info()
info['model_type'] = model_type
active_version = model_manager.get_active_version()
info['version'] = active_version or 'v1'
return jsonify(info)
except Exception as e:
return jsonify({"status": "Error", "error": str(e), "model_type": model_type})
def cleanup_old_temp_files(max_age_hours=1):
"""
Delete temp audio files older than max_age_hours if no feedback was provided.
This protects user privacy by removing unverified audio files.
"""
import time
temp_dir = os.path.join(feedback_manager.feedback_dir, "temp")
if not os.path.exists(temp_dir):
return 0
current_time = time.time()
max_age_seconds = max_age_hours * 3600
deleted_count = 0
for filename in os.listdir(temp_dir):
file_path = os.path.join(temp_dir, filename)
if os.path.isfile(file_path) and filename.endswith('.wav'):
file_age = current_time - os.path.getmtime(file_path)
# Delete if older than max_age_hours
if file_age > max_age_seconds:
try:
os.remove(file_path)
deleted_count += 1
# Also remove from pending_feedback if exists
audio_id = filename.replace('.wav', '')
if audio_id in pending_feedback:
del pending_feedback[audio_id]
except Exception as e:
print(f"Error deleting temp file {filename}: {e}")
if deleted_count > 0:
print(f"Cleaned up {deleted_count} old temp audio files")
return deleted_count
@app.route('/analyze', methods=['POST'])
def analyze_audio():
"""Analyze uploaded audio file with validation and diagnostics. Only registered users."""
try:
# Require registered user (when using Supabase)
if not isinstance(db, _LocalStubDb):
user_id = (request.headers.get('X-User-ID') or '').strip()
if not user_id:
return jsonify({
'error': 'Please register to use the app. Only registered users can analyze recordings.',
'error_key': 'errors.registration_required',
'requires_registration': True
}), 403
if not db.get_user(user_id):
return jsonify({
'error': 'Please register to use the app. Only registered users can analyze recordings.',
'error_key': 'errors.registration_required',
'requires_registration': True
}), 403
# Cleanup old temp files before processing new request
cleanup_old_temp_files()
# Check for debug mode
debug_mode = request.args.get('debug', 'false').lower() == 'true'
if 'audio' not in request.files:
return jsonify({'error': 'No audio file provided'})
audio_file = request.files['audio']
if audio_file.filename == '':
return jsonify({'error': 'No file selected'})
# Check if model is trained/loaded (different models have different attributes)
model_ready = False
if hasattr(model, 'is_trained'):
model_ready = model.is_trained
elif hasattr(model, 'rf_loaded') or hasattr(model, 'nn_loaded'):
# Ensemble model
model_ready = getattr(model, 'rf_loaded', False) or getattr(model, 'nn_loaded', False)
else:
model_ready = True # Assume ready if no check available
if not model_ready:
if hasattr(model, 'load_model'):
if not model.load_model():
return jsonify({'error': 'Model not trained'})
if not model.load_model():
return jsonify({'error': 'Model not trained'})
elif hasattr(model, 'load_models'):
if not model.load_models():
return jsonify({'error': 'Model not trained'})
else:
return jsonify({'error': 'Model not available'})
audio_id = f"{datetime.now().strftime('%Y%m%d_%H%M%S')}_{os.urandom(4).hex()}"
feedback_temp_dir = os.path.join(feedback_manager.feedback_dir, "temp")
os.makedirs(feedback_temp_dir, exist_ok=True)
temp_path = os.path.join(feedback_temp_dir, f"{audio_id}.wav")
audio_file.save(temp_path)
# Validate audio before analysis
is_valid, validation_message, audio_stats = audio_processor.validate_audio(temp_path)
if not is_valid:
# Clean up invalid file
if os.path.exists(temp_path):
os.remove(temp_path)
# Map validation errors to translation keys
error_key = 'errors.validation_failed'
if 'too short' in validation_message.lower():
duration = audio_stats.get('duration_seconds', 0)
error_key = f"errors.too_short"
elif 'too long' in validation_message.lower():
duration = audio_stats.get('duration_seconds', 0)
error_key = f"errors.too_long"
elif 'too quiet' in validation_message.lower():
level = audio_stats.get('audio_level_db', 0)
error_key = f"errors.too_quiet"
elif 'silence' in validation_message.lower():
ratio = audio_stats.get('silence_ratio', 0) * 100
error_key = f"errors.mostly_silence"
elif 'no meaningful' in validation_message.lower():
error_key = 'errors.no_audio_content'
response = {
'error_key': error_key,
'error': validation_message, # Keep original for fallback
'validation_failed': True,
'stats': audio_stats if debug_mode else None
}
if debug_mode:
response['diagnostics'] = audio_stats
return jsonify(response)
pending_feedback[audio_id] = {
'path': temp_path,
'timestamp': datetime.now().isoformat()
}
# Analysis mode: local (voc2vec or fallback RF) vs cloud (OpenAI)
analysis_mode = (request.form.get('analysis_mode') or 'local').strip().lower()
if analysis_mode not in ('local', 'cloud'):
analysis_mode = 'local'
try:
# Extract features once — used by the cry gate and the RF classifier
features, diagnostics = audio_processor.extract_features_with_diagnostics(temp_path)
# Cry/not-cry gate: reject audio that clearly isn't a baby cry
p_cry = cry_gate_probability(features)
if p_cry is not None and p_cry < cry_gate.get('reject_threshold', 0.30):
if os.path.exists(temp_path):
os.remove(temp_path)
pending_feedback.pop(audio_id, None)
return jsonify({
'not_a_cry': True,
'error_key': 'errors.not_a_cry',
'error': 'This does not sound like a baby cry. Try recording closer to the baby.',
'cry_probability': round(p_cry * 100, 1)
})
top_predictions = None
if analysis_mode == 'cloud':
# Cloud: OpenAI Audio API
if predict_via_openai_audio is None:
if os.path.exists(temp_path):
os.remove(temp_path)
return jsonify({'error': 'OpenAI client not installed. Install the openai package.'})
result = predict_via_openai_audio(temp_path)
prediction, confidence = result[0], result[1]
openai_error = result[2] if len(result) > 2 else None
used_model_type = 'openai_audio'
if prediction is None or confidence is None:
if os.path.exists(temp_path):
os.remove(temp_path)
return jsonify({
'error': openai_error or 'OpenAI API key not configured or request failed. Check OPENAI_API_KEY and try again.'
})
else:
# Local: Random Forest v5 first (CV 0.73), voc2vec as fallback (CV 0.61)
prediction, confidence = None, None
if features is not None:
top_predictions = model.predict_top_k(features, k=2)
if top_predictions:
prediction, confidence = top_predictions[0]
used_model_type = model_type
if prediction is None or confidence is None:
# Fallback: voc2vec embedding classifier
if voc2vec_predict_top_k is not None and voc2vec_available():
classifier_path = os.path.join(MODELS_DIR, 'voc2vec_classifier.pkl')
top_predictions = voc2vec_predict_top_k(temp_path, classifier_path=classifier_path, k=2)
if top_predictions:
prediction, confidence = top_predictions[0]
used_model_type = 'voc2vec'
if prediction is None or confidence is None:
if os.path.exists(temp_path):
os.remove(temp_path)
return jsonify({
'error': (diagnostics or {}).get('message', 'Could not analyze audio'),
'diagnostics': diagnostics if debug_mode else None
})
pending_feedback[audio_id]['prediction'] = prediction
pending_feedback[audio_id]['confidence'] = confidence
# Build response
UNCERTAIN_THRESHOLD = 0.40
response = {
'prediction': prediction,
'confidence': confidence * 100,
'top_predictions': [
{'label': label, 'confidence': round(prob * 100, 1)}
for label, prob in (top_predictions or [(prediction, confidence)])
],
'uncertain': bool(confidence < UNCERTAIN_THRESHOLD),
'timestamp': datetime.now().isoformat(),
'audio_id': audio_id,
'model_type': used_model_type
}
# Track prediction for analytics
try:
# Get user_id from request headers
user_id = request.headers.get('X-User-ID')
if user_id and user_id.strip():
user_id = user_id.strip()
else:
user_id = None
prediction_data = {
'user_id': user_id,
'prediction': prediction,
'confidence': float(confidence * 100),
'audio_id': audio_id,
'model_type': used_model_type,
'timestamp': datetime.now().isoformat()
}
# Save to database (user already validated as registered above)
try:
db.create_prediction(prediction_data)
user_info = f"user: {user_id}" if user_id else "anonymous"
print(f"✅ Prediction saved to database: {prediction} (confidence: {confidence*100:.1f}%, {user_info})")
except Exception as db_error:
print(f"❌ Database error saving prediction: {db_error}")
import traceback
traceback.print_exc()
# Continue even if database save fails
# Data saved to Supabase only
except Exception as e:
print(f"❌ Error tracking prediction: {e}")
import traceback
traceback.print_exc()
# Don't fail the request, but log the error
# Add diagnostics if debug mode (only when we have diagnostics from local path)
if debug_mode and diagnostics is not None:
response['diagnostics'] = {
'duration_seconds': audio_stats.get('duration_seconds'),
'sample_rate': audio_stats.get('sample_rate'),
'audio_level_db': audio_stats.get('audio_level_db'),
'silence_ratio': audio_stats.get('silence_ratio'),
'cry_likelihood': audio_stats.get('cry_likelihood'),
'feature_summary': diagnostics.get('feature_summary', {})
}
return jsonify(response)
except Exception as e:
if os.path.exists(temp_path):
os.remove(temp_path)
if audio_id in pending_feedback:
del pending_feedback[audio_id]
raise e
except Exception as e:
return jsonify({'error': f'Analysis failed: {str(e)}'})
@app.route('/feedback', methods=['POST'])
def receive_feedback():
"""Receive user feedback. Only registered users."""
try:
feedback_data = request.get_json() or {}
# Require registered user (when using Supabase)
if not isinstance(db, _LocalStubDb):
user_id = (feedback_data.get('user_id') or '').strip() if isinstance(feedback_data.get('user_id'), str) else None
if not user_id:
return jsonify({
'error': 'Please register to use the app. Only registered users can submit feedback.',
'error_key': 'errors.registration_required',
'requires_registration': True
}), 403
if not db.get_user(user_id):
return jsonify({
'error': 'Please register to use the app. Only registered users can submit feedback.',
'error_key': 'errors.registration_required',
'requires_registration': True
}), 403
audio_id = feedback_data.get('audio_id')
predicted_label = feedback_data.get('predicted_label')
correct_label = feedback_data.get('correct_label')
is_correct = feedback_data.get('is_correct', False)
what_helped = feedback_data.get('what_helped', [])
if not audio_id or not correct_label:
return jsonify({'error': 'Missing data'})
if audio_id not in pending_feedback:
return jsonify({'error': 'Audio expired'})
pending = pending_feedback[audio_id]
audio_path = pending.get('path')
if not audio_path or not os.path.exists(audio_path):
return jsonify({'error': 'Audio file not found'})
confidence = pending.get('confidence', 0.5)
submission_id, should_retrain = feedback_manager.save_feedback(
audio_path=audio_path,
predicted_label=predicted_label,
correct_label=correct_label,
confidence=confidence,
is_correct=is_correct
)
# Upload audio to Supabase Storage
storage_path = f"verified/{correct_label}/{submission_id}.wav"
audio_url = None
if supabase_storage.enabled:
# Upload to Supabase Storage
audio_url = supabase_storage.upload_audio(audio_path, storage_path)
if audio_url:
print(f"✅ Audio uploaded to Supabase Storage: {audio_url}")
else:
print("⚠️ Failed to upload to Supabase Storage, using local path")
# Fallback to local path if storage upload failed
if not audio_url:
# Get the local file path where audio was saved
file_path_abs = os.path.abspath(os.path.join(feedback_manager.verified_dir, correct_label, f"{submission_id}.wav"))
app_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
# Store path relative to app root as "feedback_data/verified/category/id.wav" for consistent playback
try:
file_path = os.path.relpath(file_path_abs, app_root)
except ValueError:
file_path = os.path.join("feedback_data", "verified", correct_label, f"{submission_id}.wav")
if file_path.startswith(".."):
file_path = os.path.join("feedback_data", "verified", correct_label, f"{submission_id}.wav")
else:
# Use storage URL instead of local path
file_path = audio_url
# Track feedback submission (for analytics)
try:
# Get user_id from request body
user_id = feedback_data.get('user_id')
if user_id and isinstance(user_id, str) and user_id.strip():
user_id = user_id.strip()
else:
user_id = None
feedback_record = {
'audio_id': audio_id,
'user_id': user_id, # Can be None, which is allowed by foreign key
'predicted_label': predicted_label,
'correct_label': correct_label, # This is the correct label - saved to category field too
'is_correct': is_correct,
'confidence': confidence,
'file_path': file_path, # Supabase Storage URL or local path
'submission_id': submission_id, # Submission ID from feedback_manager
'what_helped': ','.join(what_helped) if what_helped else '',
'timestamp': datetime.now().isoformat()
}
# Save to database (user already validated as registered above)
try:
db.create_feedback(feedback_record)
user_info = f"user: {user_id}" if user_id else "anonymous"
print(f"✅ Feedback saved to Supabase: {correct_label} (predicted: {predicted_label}, correct: {is_correct}, confidence: {confidence:.2f}, {user_info})")
print(f" Feedback record: {feedback_record}")
except Exception as db_error:
print(f"❌ Database error saving feedback to Supabase: {db_error}")
import traceback
traceback.print_exc()
print(f" Attempted feedback record: {feedback_record}")
# Return error so frontend knows it failed
return jsonify({'error': f'Failed to save feedback to database: {str(db_error)}'}), 500
# Data saved to Supabase only
except Exception as e:
print(f"❌ Error tracking feedback: {e}")
import traceback
traceback.print_exc()
# Don't fail the request, but log the error
# Audio file has already been copied to verified directory by feedback_manager
# The temp file will be cleaned up by cleanup_old_temp_files() automatically
# Only remove from pending_feedback dict
del pending_feedback[audio_id]
# In-request retraining takes far longer than the request timeout, so it is
# opt-in for local/dev runs only. Production retrains offline via
# scripts/pull_feedback_and_retrain.py, which reads feedback from Supabase.
retrain_triggered = False
if should_retrain and os.environ.get('ENABLE_AUTO_RETRAIN') == '1':
retrain_result = model_manager.check_and_auto_retrain()
if retrain_result and retrain_result.get('success'):
retrain_triggered = True
model.load_model()
return jsonify({
'status': 'Feedback saved',
'submission_id': submission_id,
'retrain_triggered': retrain_triggered
})
except Exception as e:
return jsonify({'error': str(e)})
@app.route('/feedback/stats')
def feedback_stats():
"""Get feedback statistics"""
try:
stats = feedback_manager.get_stats()
return jsonify(stats)
except Exception as e:
return jsonify({'error': str(e)})
@app.route('/model/retrain', methods=['POST'])
def retrain_model():
"""Manually trigger retraining"""
try:
result = model_manager.retrain_with_feedback(include_feedback=True)
if result.get('success'):
model.load_model()
return jsonify(result)
except Exception as e:
return jsonify({'success': False, 'error': str(e)})
@app.route('/api/analytics/error', methods=['POST'])
def analytics_error():
"""Receive error reports from mobile app"""
try:
error_data = request.get_json()
# Log error for debugging (in production, could store in database)
print(f"📊 Error reported: {error_data.get('error', {}).get('message', 'Unknown error')}")
return jsonify({'status': 'received'})
except Exception as e:
return jsonify({'error': str(e)})
@app.route('/model/versions')
def model_versions():
"""Get model versions"""
try:
versions = model_manager.get_versions()
active = model_manager.get_active_version()
return jsonify({
'versions': versions,
'active_version': active
})
except Exception as e:
return jsonify({'error': str(e)})
@app.route('/model/switch/', methods=['POST'])
def switch_model_version(version):
"""Switch model version"""
try:
result = model_manager.switch_version(version)
if result.get('success'):
model.load_model()
return jsonify(result)
except Exception as e:
return jsonify({'success': False, 'error': str(e)})
@app.route('/model/summary')
def model_summary():
"""Get model summary"""
try:
summary = model_manager.get_model_summary()
return jsonify(summary)
except Exception as e:
return jsonify({'error': str(e)})
@app.route('/test-connection', methods=['GET', 'POST'])
def test_connection():
"""Test endpoint to verify connectivity from mobile app"""
return jsonify({
'status': 'success',
'message': 'Connection successful!',
'timestamp': datetime.now().isoformat(),
'method': request.method,
'headers': dict(request.headers),
'origin': request.headers.get('Origin', 'not provided'),
'user_agent': request.headers.get('User-Agent', 'not provided')
})
def _try_reconnect_db():
"""If we're on the stub but Supabase is configured, try to reconnect once.
Lets the app self-heal after a paused Supabase project is restored,
without needing a Space rebuild. Called from /health, which doubles as
the keep-alive ping target.
"""
global db, DB_INIT_ERROR
if not isinstance(db, _LocalStubDb) or not SUPABASE_DB_URL:
return
try:
from database_supabase import SupabaseDatabase
db = SupabaseDatabase()
DB_INIT_ERROR = None
print("✅ Reconnected to Supabase (was on stub)")
except Exception as e:
DB_INIT_ERROR = str(e)[:200]
@app.route('/privacy')
def privacy_policy():
"""Public privacy policy (required for the Play Store listing)."""
return send_from_directory(os.path.join(os.path.dirname(os.path.abspath(__file__)), 'static'), 'privacy.html')
@app.route('/health')
def health_check():
_try_reconnect_db()
db_alive = False
if not isinstance(db, _LocalStubDb):
# Touch the database so the ping also counts as Supabase activity
# (free-tier projects pause after ~7 idle days)
try:
with db.get_connection(retries=1) as conn:
with conn.cursor() as cur:
cur.execute('SELECT 1')
db_alive = True
except Exception as e:
DB_INIT_ERROR_local = str(e)[:200]
return jsonify({
'status': 'healthy',
'database': 'supabase-unreachable',
'db_error': DB_INIT_ERROR_local,
'timestamp': datetime.now().isoformat()
})
return jsonify({
'status': 'healthy',
'database': 'supabase' if db_alive else 'stub',
'db_error': None if db_alive else DB_INIT_ERROR,
'timestamp': datetime.now().isoformat()
})
@app.route('/health-legacy')
def health_check_legacy():
"""Health check endpoint for cloud deployments"""
return jsonify({'status': 'healthy', 'timestamp': datetime.now().isoformat()})
# ==================== ADMIN API ENDPOINTS ====================
@app.route('/api/admin/users/stats')
@admin_required
def admin_users_stats():
"""Get user statistics"""
try:
stats = db.get_user_stats()
return jsonify(stats)
except Exception as e:
return jsonify({'error': str(e)}), 500
@app.route('/api/admin/users', methods=['GET'])
@admin_required
def admin_users_list():
"""Get all users with pagination"""
try:
limit = request.args.get('limit', type=int)
offset = request.args.get('offset', 0, type=int)
users = db.get_all_users(limit=limit, offset=offset)
total = db.count_users()
return jsonify({
'users': users,
'total': total,
'limit': limit,
'offset': offset
})
except Exception as e:
return jsonify({'error': str(e)}), 500
@app.route('/api/admin/users/', methods=['GET'])
@admin_required
def admin_user_get(user_id):
"""Get a specific user"""
try:
user = db.get_user(user_id)
if not user:
return jsonify({'error': 'User not found'}), 404
return jsonify(user)
except Exception as e:
return jsonify({'error': str(e)}), 500
@app.route('/api/admin/users/', methods=['PUT'])
@admin_required
def admin_user_update(user_id):
"""Update a user"""
try:
user_data = request.get_json()
if not user_data:
return jsonify({'error': 'No data provided'}), 400
success = db.update_user(user_id, user_data)
if not success:
return jsonify({'error': 'User not found or no changes made'}), 404
user = db.get_user(user_id)
return jsonify({'success': True, 'user': user})
except Exception as e:
return jsonify({'error': str(e)}), 500
@app.route('/api/admin/users/', methods=['DELETE'])
@admin_required
def admin_user_delete(user_id):
"""Delete a user"""
try:
success = db.delete_user(user_id)
if not success:
return jsonify({'error': 'User not found'}), 404
return jsonify({'success': True})
except Exception as e:
return jsonify({'error': str(e)}), 500
@app.route('/api/admin/users/platform')
@admin_required
def admin_users_platform():
"""Get platform distribution"""
try:
distribution = db.get_platform_distribution()
return jsonify(distribution)
except Exception as e:
return jsonify({'error': str(e)}), 500
@app.route('/api/admin/users/language')
@admin_required
def admin_users_language():
"""Get language distribution"""
try:
distribution = db.get_language_distribution()
return jsonify(distribution)
except Exception as e:
return jsonify({'error': str(e)}), 500
@app.route('/api/admin/users/timeline')
@admin_required
def admin_users_timeline():
"""Get user registration timeline"""
try:
days = int(request.args.get('days', 30))
timeline = db.get_user_timeline(days=days)
return jsonify(timeline)
except Exception as e:
return jsonify({'error': str(e)}), 500
@app.route('/api/admin/model/summary')
@admin_required
def admin_model_summary():
"""Get model summary"""
try:
summary = model_manager.get_model_summary()
return jsonify(summary)
except Exception as e:
return jsonify({'error': str(e)}), 500
@app.route('/api/admin/model/performance')
@admin_required
def admin_model_performance():
"""Get model performance metrics"""
try:
# Get feedback stats from database
feedback_stats = db.get_feedback_stats()
# Get total predictions from database
total_predictions = db.count_predictions()
# Calculate average confidence from recent predictions (using database)
from psycopg2.extras import RealDictCursor
with db.get_connection() as conn:
cursor = conn.cursor(cursor_factory=RealDictCursor)
cursor.execute('''
SELECT AVG(confidence) as avg_confidence
FROM predictions
WHERE timestamp >= NOW() - INTERVAL '30 days'
''')
result = cursor.fetchone()
avg_confidence = float(result['avg_confidence']) if result and result['avg_confidence'] is not None else 0.0
return jsonify({
'average_confidence': avg_confidence,
'total_predictions': total_predictions,
'feedback_stats': feedback_stats
})
except Exception as e:
return jsonify({'error': str(e)}), 500
@app.route('/api/admin/feedback/categories')
@admin_required
def admin_feedback_categories():
"""Get feedback breakdown by category"""
try:
stats = db.get_feedback_stats()
return jsonify({
'category_counts': stats.get('category_counts', {}),
'total': stats.get('total_feedback', 0)
})
except Exception as e:
return jsonify({'error': str(e)}), 500
@app.route('/api/admin/feedback/timeline')
@admin_required
def admin_feedback_timeline():
"""Get feedback submission timeline"""
try:
days = int(request.args.get('days', 30))
timeline = db.get_feedback_timeline(days=days)
return jsonify(timeline)
except Exception as e:
return jsonify({'error': str(e)}), 500
@app.route('/api/admin/feedback/stats')
@admin_required
def admin_feedback_stats():
"""Get comprehensive feedback statistics"""
try:
stats = db.get_feedback_stats()
return jsonify(stats)
except Exception as e:
return jsonify({'error': str(e)}), 500
@app.route('/api/admin/feedback/list')
@admin_required
def admin_feedback_list():
"""Get feedback entries for admin dashboard"""
try:
limit = request.args.get('limit', type=int)
offset = request.args.get('offset', 0, type=int)
is_correct = request.args.get('is_correct')
# Convert is_correct string to boolean if provided
is_correct_bool = None
if is_correct is not None:
is_correct_bool = is_correct.lower() in ['true', '1', 'yes']
feedback_list = db.get_feedback_list(limit=limit, offset=offset, is_correct=is_correct_bool)
return jsonify({
'feedback': feedback_list,
'total': len(feedback_list)
})
except Exception as e:
return jsonify({'error': str(e)}), 500
@app.route('/api/admin/predictions/list')
@admin_required
def admin_predictions_list():
"""Get predictions list for admin dashboard"""
try:
limit = request.args.get('limit', type=int)
offset = request.args.get('offset', 0, type=int)
user_id = request.args.get('user_id')
predictions_list = db.get_predictions_list(limit=limit, offset=offset, user_id=user_id)
return jsonify({
'predictions': predictions_list,
'total': len(predictions_list)
})
except Exception as e:
return jsonify({'error': str(e)}), 500
@app.route('/api/admin/feedback/audio/')
@admin_required
def admin_feedback_audio(file_path):
"""Serve audio file for feedback (admin only)"""
try:
import urllib.parse
# Decode the file path
file_path = urllib.parse.unquote(file_path)
# Check if it's a Supabase Storage URL (starts with https://)
if file_path.startswith('https://'):
# Redirect to Supabase Storage URL
from flask import redirect
return redirect(file_path, code=302)
# Otherwise, it's a local file path - serve from local filesystem
# Security: Only allow files from feedback_data directory
if file_path.startswith('/') and not file_path.startswith('../'):
return jsonify({'error': 'Invalid file path'}), 400
# Get the absolute path of the feedback directory
feedback_dir_abs = os.path.abspath(feedback_manager.feedback_dir)
_app_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
_app_feedback = os.path.normpath(os.path.join(_app_root, 'feedback_data'))
# Handle different path formats and extract relative_path for fallback
relative_path = None
if file_path.startswith('../'):
if '../feedback_data/' in file_path:
relative_path = file_path.replace('../feedback_data/', '')
actual_path = os.path.join(feedback_dir_abs, relative_path)
else:
actual_path = os.path.join(os.path.dirname(feedback_dir_abs), file_path[3:])
elif file_path.startswith('feedback_data/'):
relative_path = file_path.replace('feedback_data/', '')
actual_path = os.path.join(feedback_dir_abs, relative_path)
else:
actual_path = os.path.join(feedback_dir_abs, file_path)
# Normalize path to prevent directory traversal
actual_path = os.path.normpath(actual_path)
# Fallback for old DB paths: if file not found, try under app root feedback_data
# (e.g. old records stored "../feedback_data/verified/..." when cwd differed)
if not os.path.exists(actual_path) and relative_path:
fallback_path = os.path.normpath(os.path.join(_app_feedback, relative_path))
if os.path.exists(fallback_path) and fallback_path.startswith(_app_feedback):
actual_path = fallback_path
print(f"📁 Resolved old path to: {actual_path}")
# Security check: allow feedback_dir_abs or app_root/feedback_data
if not actual_path.startswith(feedback_dir_abs) and not actual_path.startswith(_app_feedback):
print(f"❌ Security check failed: {actual_path} not in {feedback_dir_abs!r} or {_app_feedback!r}")
return jsonify({'error': 'File not in feedback directory'}), 403
print(f"📁 Serving local audio file: {actual_path}")
# Check if file exists
if not os.path.exists(actual_path):
return jsonify({'error': 'Audio file not found'}), 404
# Check if it's a WAV file
if not actual_path.lower().endswith('.wav'):
return jsonify({'error': 'Invalid file type'}), 400
# Serve the file
from flask import send_file
return send_file(actual_path, mimetype='audio/wav', as_attachment=False)
except Exception as e:
print(f"❌ Error serving audio file: {e}")
import traceback
traceback.print_exc()
return jsonify({'error': str(e)}), 500
@app.route('/api/admin/system/health')
@admin_required
def admin_system_health():
"""Get system health status"""
try:
# Check if model is loaded
model_status = 'loaded' if (model and hasattr(model, 'is_trained') and model.is_trained) else 'not_loaded'
# Check if data directories exist
data_ok = os.path.exists(DATA_DIR)
feedback_ok = os.path.exists(FEEDBACK_DIR)
models_ok = os.path.exists(MODELS_DIR)
return jsonify({
'status': 'healthy' if (model_status == 'loaded' and data_ok) else 'degraded',
'model_status': model_status,
'directories': {
'data': data_ok,
'feedback': feedback_ok,
'models': models_ok
},
'timestamp': datetime.now().isoformat()
})
except Exception as e:
return jsonify({'error': str(e)}), 500
@app.route('/api/admin/system/storage')
@admin_required
def admin_system_storage():
"""Get storage usage"""
try:
def get_dir_size(path):
total = 0
if os.path.exists(path):
for dirpath, dirnames, filenames in os.walk(path):
for f in filenames:
fp = os.path.join(dirpath, f)
if os.path.exists(fp):
total += os.path.getsize(fp)
return total
feedback_size = get_dir_size(FEEDBACK_DIR)
models_size = get_dir_size(MODELS_DIR)
data_size = get_dir_size(DATA_DIR)
return jsonify({
'feedback_data': {
'size_bytes': feedback_size,
'size_mb': round(feedback_size / (1024 * 1024), 2)
},
'models': {
'size_bytes': models_size,
'size_mb': round(models_size / (1024 * 1024), 2)
},
'training_data': {
'size_bytes': data_size,
'size_mb': round(data_size / (1024 * 1024), 2)
},
'total_mb': round((feedback_size + models_size + data_size) / (1024 * 1024), 2)
})
except Exception as e:
return jsonify({'error': str(e)}), 500
# ==================== ADMIN ROUTES ====================
@app.route('/admin/login', methods=['GET', 'POST'])
def admin_login():
"""Admin login page and authentication"""
if request.method == 'POST':
data = request.get_json() if request.is_json else request.form
password = data.get('password', '')
if password == ADMIN_PASSWORD:
session['admin_authenticated'] = True
return jsonify({'success': True, 'redirect': '/admin'})
else:
return jsonify({'success': False, 'error': 'Invalid password'}), 401
# GET request - serve login page
lang = detect_language()
translations = load_translations(lang)
admin_translations = translations.get('admin', {})
template = ADMIN_LOGIN_TEMPLATE.replace('{{LANG}}', lang)
template = template.replace('{{DIR}}', 'rtl' if lang == 'ar' else 'ltr')
# Inject translations JSON for JavaScript
translations_json = json.dumps(admin_translations, ensure_ascii=False)
template = template.replace('{{TRANSLATIONS}}', translations_json)
return render_template_string(template)
@app.route('/admin/logout')
def admin_logout():
"""Admin logout"""
session.pop('admin_authenticated', None)
return jsonify({'success': True, 'redirect': '/admin/login'})
@app.route('/admin')
@admin_required
def admin_dashboard():
"""Admin dashboard page"""
from admin_dashboard_template import ADMIN_DASHBOARD_TEMPLATE_STR as ADMIN_DASHBOARD_TEMPLATE
lang = detect_language()
translations = load_translations(lang)
admin_translations = translations.get('admin', {})
template = ADMIN_DASHBOARD_TEMPLATE.replace('{{LANG}}', lang)
template = template.replace('{{DIR}}', 'rtl' if lang == 'ar' else 'ltr')
# Inject translations JSON for JavaScript (empty object if no translations)
translations_json = json.dumps(admin_translations if admin_translations else {}, ensure_ascii=False)
template = template.replace('{{TRANSLATIONS}}', translations_json)
return render_template_string(template)
# ==================== DATA COLLECTION ENDPOINTS ====================
@app.route('/api/users/register', methods=['POST'])
def register_user():
"""Register a new user"""
try:
user_data = request.get_json()
if not user_data:
print("❌ Registration failed: No data provided")
return jsonify({'error': 'No data provided'}), 400
# Validate and sanitize required fields
baby_name = user_data.get('baby_name', '').strip()
if not baby_name:
print("❌ Registration failed: baby_name is required")
return jsonify({'error': 'Baby name is required'}), 400
# Sanitize baby_name (remove excessive whitespace, limit length)
baby_name = ' '.join(baby_name.split())[:100] # Limit to 100 chars
# Extract and validate user_id
user_id = user_data.get('user_id', 'user_' + str(int(datetime.now().timestamp() * 1000)))
user_id = str(user_id).strip() if user_id else None
# Validate user_id format
if not user_id or len(user_id) < 5 or len(user_id) > 100:
print("❌ Registration failed: Invalid user_id format")
return jsonify({'error': 'Invalid user ID format'}), 400
# Validate and sanitize email if provided
email = user_data.get('email')
if email:
email = str(email).strip().lower()
# Basic email validation (contains @ and ., and reasonable length)
if '@' not in email or '.' not in email.split('@')[1] or len(email) > 255:
print(f"❌ Registration failed: Invalid email format: {email}")
return jsonify({'error': 'Invalid email format'}), 400
elif len(email) < 3:
print(f"❌ Registration failed: Email too short: {email}")
return jsonify({'error': 'Invalid email format'}), 400
# Validate and sanitize phone if provided
phone = user_data.get('phone')
if phone:
phone = str(phone).strip()
# Basic phone validation (should contain digits, allow + and spaces)
# Remove common phone formatting characters for validation
phone_digits = ''.join(c for c in phone if c.isdigit())
if len(phone_digits) < 7 or len(phone_digits) > 15 or len(phone) > 25:
print(f"❌ Registration failed: Invalid phone format: {phone}")
return jsonify({'error': 'Invalid phone number format'}), 400
else:
# Keep original format but limit length
phone = phone[:25]
# Require at least email OR phone
if not email and not phone:
print("❌ Registration failed: Email or phone number is required")
return jsonify({'error': 'Email or phone number is required'}), 400
# Sanitize other optional fields
baby_gender = user_data.get('baby_gender')
if baby_gender:
baby_gender = str(baby_gender).strip().lower()
# Only allow valid gender values
if baby_gender not in ['male', 'female', 'm', 'f']:
baby_gender = None
elif baby_gender in ['m', 'f']:
baby_gender = 'male' if baby_gender == 'm' else 'female'
baby_birthday = user_data.get('baby_birthday')
if baby_birthday:
baby_birthday = str(baby_birthday).strip()[:10] # Limit to date format length
platform = str(user_data.get('platform', 'web')).strip().lower()
if platform not in ['web', 'android', 'ios']:
platform = 'web'
language = str(user_data.get('language', 'ar')).strip().lower()
if language not in ['ar', 'en', 'fr']:
language = 'ar'
# Handle password if provided
password = user_data.get('password', '').strip()
password_hash = None
if password:
if len(password) < 6:
return jsonify({'error': 'Password must be at least 6 characters'}), 400
password_hash = generate_password_hash(password)
user_info = {
'user_id': user_id,
'email': email,
'phone': phone,
'password_hash': password_hash,
'baby_name': baby_name,
'baby_gender': baby_gender,
'baby_birthday': baby_birthday,
'platform': platform,
'language': language,
'registration_date': datetime.now().isoformat()
}
# Save to database
try:
db.create_user(user_info)
print(f"✅ User registered successfully: {user_id} (baby: {baby_name}, platform: {user_info['platform']})")
except Exception as db_error:
print(f"❌ Database error during registration: {db_error}")
import traceback
traceback.print_exc()
return jsonify({'error': f'Failed to save user: {str(db_error)}'}), 500
# Data saved to Supabase only
return jsonify({'success': True, 'user_id': user_id})
except Exception as e:
print(f"❌ Registration error: {e}")
import traceback
traceback.print_exc()
return jsonify({'error': f'Registration failed: {str(e)}'}), 500
@app.route('/api/users/login', methods=['POST'])
def login_user():
"""Login user with email/phone and password"""
try:
login_data = request.get_json()
if not login_data:
return jsonify({'error': 'No data provided'}), 400
identifier = login_data.get('email', '').strip() # Can be email or phone
password = login_data.get('password', '')
if not identifier:
return jsonify({'error': 'Email or phone number is required'}), 400
if not password:
return jsonify({'error': 'Password is required'}), 400
# Get user by email or phone
user = db.get_user_by_email_or_phone(identifier)
if not user:
return jsonify({'error': 'Invalid email/phone or password'}), 401
# Check if user has a password_hash (for existing users without password)
if not user.get('password_hash'):
return jsonify({'error': 'Account not set up with password. Please register again.'}), 401
# Verify password
if not check_password_hash(user['password_hash'], password):
return jsonify({'error': 'Invalid email/phone or password'}), 401
# Create session
session['user_id'] = user['user_id']
session['email'] = user.get('email')
session['phone'] = user.get('phone')
session['logged_in'] = True
login_identifier = user.get('email') or user.get('phone') or identifier
print(f"✅ User logged in successfully: {user['user_id']} ({login_identifier})")
return jsonify({
'success': True,
'user_id': user['user_id'],
'email': user['email'],
'phone': user.get('phone'),
'baby_name': user.get('baby_name'),
'baby_birthday': user.get('baby_birthday'),
'baby_gender': user.get('baby_gender')
})
except Exception as e:
print(f"❌ Login error: {e}")
import traceback
traceback.print_exc()
return jsonify({'error': f'Login failed: {str(e)}'}), 500
@app.route('/api/users/logout', methods=['POST'])
def logout_user():
"""Logout user"""
try:
session.clear()
return jsonify({'success': True})
except Exception as e:
return jsonify({'error': str(e)}), 500
@app.route('/api/me/predictions')
def me_predictions_list():
"""Get current user's predictions (cry history). Requires X-User-ID header."""
try:
user_id = (request.headers.get('X-User-ID') or '').strip()
if not user_id:
return jsonify({'error': 'User ID required'}), 401
if not isinstance(db, _LocalStubDb) and not db.get_user(user_id):
return jsonify({'error': 'User not found'}), 401
limit = request.args.get('limit', type=int, default=50)
offset = request.args.get('offset', type=int, default=0)
limit = max(1, min(100, limit)) if limit else 50
offset = max(0, offset) if offset else 0
predictions_list = db.get_predictions_list(limit=limit, offset=offset, user_id=user_id)
total = db.count_predictions(user_id=user_id)
# Serialize for JSON (timestamps)
out = []
for row in predictions_list:
r = dict(row)
ts = r.get('timestamp')
if hasattr(ts, 'isoformat'):
r['timestamp'] = ts.isoformat()
out.append(r)
return jsonify({'predictions': out, 'total': total})
except Exception as e:
return jsonify({'error': str(e)}), 500
@app.route('/api/me/predictions/', methods=['PATCH'])
def me_predictions_update(prediction_id):
"""Update what_helped for a prediction. Requires X-User-ID header."""
try:
user_id = (request.headers.get('X-User-ID') or '').strip()
if not user_id:
return jsonify({'error': 'User ID required'}), 401
data = request.get_json() or {}
what_helped = data.get('what_helped')
if what_helped is not None and not isinstance(what_helped, str):
what_helped = str(what_helped)
updated = db.update_prediction_what_helped(prediction_id, user_id, what_helped)
if not updated:
return jsonify({'error': 'Prediction not found'}), 404
return jsonify({'success': True})
except Exception as e:
return jsonify({'error': str(e)}), 500
@app.route('/api/analytics/track', methods=['POST'])
def track_analytics():
"""Track an analytics event - saved to Supabase"""
try:
data = request.get_json()
event_name = data.get('event_name')
event_data = data.get('event_data', {})
user_id = data.get('user_id')
if not event_name:
return jsonify({'error': 'event_name is required'}), 400
# Save to Supabase analytics_events table
import json as json_lib
analytics_record = {
'event_name': event_name,
'user_id': user_id,
'event_data': event_data,
'timestamp': datetime.now().isoformat()
}
db.create_analytics_event(analytics_record)
return jsonify({'success': True})
except Exception as e:
return jsonify({'error': str(e)}), 500
@app.route('/api/predictions/track', methods=['POST'])
def track_prediction_endpoint():
"""Track a prediction - saved to Supabase"""
try:
prediction_data = request.get_json()
if not prediction_data:
return jsonify({'error': 'No data provided'}), 400
# Save to Supabase predictions table
db.create_prediction(prediction_data)
return jsonify({'success': True})
except Exception as e:
return jsonify({'error': str(e)}), 500
if __name__ == '__main__':
print("🚀 Starting Baby Cry AI (PWA Mode)...")
print("=" * 50)
if model.load_model():
print("✅ Model loaded")
version = model_manager.get_active_version()
if version:
print(f"📦 Version: {version}")
else:
print("⚠️ No model found")
stats = feedback_manager.get_stats()
print(f"📊 Feedback: {stats['total_submissions']} submissions")
# Use PORT env variable (7860 for HF Spaces, 5001 for local)
port = int(os.environ.get('PORT', 5001))
debug = os.environ.get('FLASK_ENV') != 'production'
print(f"\n🌐 Server starting on port {port}...")
print(f"📱 Open: http://localhost:{port}")
app.run(debug=debug, host='0.0.0.0', port=port)
