# database.py - PRODUCTION-READY HUGGINGFACE SPACES VERSION import sqlite3 from datetime import datetime, timezone from typing import List, Optional, Dict, Any import json import threading import contextlib import time import os from pathlib import Path from dataclasses import dataclass import logging from functools import wraps # Configure logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) @dataclass class AnalysisResult: """Production data class for analysis results with comprehensive typing""" id: int resume_filename: str jd_filename: str final_score: float verdict: str timestamp: datetime matched_skills: str = "" missing_skills: str = "" hard_match_score: Optional[float] = None semantic_score: Optional[float] = None def __post_init__(self): """Set fallback values after initialization""" if self.hard_match_score is None: self.hard_match_score = self.final_score if self.semantic_score is None: self.semantic_score = self.final_score class DatabaseConfig: """Production HuggingFace Spaces database configuration""" def __init__(self): # Use /tmp directory which is always writable in containers self.tmp_dir = Path("/tmp") self.data_dir = self.tmp_dir / "data" # Create directories safely with enhanced error handling try: self.data_dir.mkdir(parents=True, exist_ok=True) logger.info(f"✅ Data directory created: {self.data_dir}") except Exception as e: logger.warning(f"Could not create data directory: {e}, using /tmp") self.data_dir = self.tmp_dir # Production database configuration self.db_path = os.getenv('DATABASE_PATH', str(self.tmp_dir / 'resume_analysis_v5.db')) self.timeout = float(os.getenv('DATABASE_TIMEOUT', '45.0')) self.max_retries = int(os.getenv('DATABASE_MAX_RETRIES', '4')) self.retry_delay = float(os.getenv('DATABASE_RETRY_DELAY', '0.5')) # Production-safe defaults for HuggingFace Spaces self.enable_wal = False # Disable WAL mode for container safety self.backup_enabled = False # Disable backups in temporary storage logger.info(f"🗄️ Production database configured at: {self.db_path}") config = DatabaseConfig() # Thread-safe database operations db_lock = threading.RLock() def retry_on_db_error(max_retries: int = None): """Production decorator for database operation retry logic""" def decorator(func): @wraps(func) def wrapper(*args, **kwargs): retries = max_retries or config.max_retries last_exception = None for attempt in range(retries + 1): try: return func(*args, **kwargs) except (sqlite3.OperationalError, sqlite3.DatabaseError) as e: last_exception = e error_msg = str(e).lower() # Different retry strategies based on error type if "locked" in error_msg or "busy" in error_msg: retry_delay = config.retry_delay * (2 ** attempt) + 0.1 elif "unable to open" in error_msg: retry_delay = config.retry_delay * (1.5 ** attempt) else: retry_delay = config.retry_delay * (2 ** attempt) if attempt < retries: logger.warning(f"Database operation failed (attempt {attempt + 1}/{retries + 1}): {e}. Retrying in {retry_delay:.1f}s...") time.sleep(retry_delay) else: logger.error(f"Database operation failed after {retries + 1} attempts: {e}") # Graceful fallback for production logger.info(f"Database unavailable, using fallback mode for {func.__name__}") return None return wrapper return decorator @contextlib.contextmanager def get_db_connection(): """Production-grade HuggingFace Spaces database connection""" conn = None try: with db_lock: # Ensure database path exists db_path = Path(config.db_path) db_path.parent.mkdir(parents=True, exist_ok=True) # Production connection with optimized settings conn = sqlite3.connect( str(db_path), timeout=config.timeout, check_same_thread=False, isolation_level=None # Autocommit mode ) # Production-optimized SQLite pragmas conn.execute('PRAGMA journal_mode=DELETE;') # Safe for containers conn.execute('PRAGMA synchronous=NORMAL;') conn.execute('PRAGMA busy_timeout=45000;') conn.execute('PRAGMA foreign_keys=ON;') conn.execute('PRAGMA cache_size=-128000;') conn.execute('PRAGMA temp_store=MEMORY;') # Initialize schema migrate_db_schema(conn) yield conn except sqlite3.OperationalError as e: logger.warning(f"Database connection issue: {e}") yield None except Exception as e: logger.warning(f"Unexpected database error: {e}") yield None finally: if conn: try: conn.commit() conn.close() except Exception as e: logger.debug(f"Connection cleanup note: {e}") def migrate_db_schema(conn: sqlite3.Connection): """Production database schema migration""" if not conn: return try: cursor = conn.cursor() # Enhanced analysis results table cursor.execute(''' CREATE TABLE IF NOT EXISTS analysis_results ( id INTEGER PRIMARY KEY AUTOINCREMENT, resume_filename TEXT NOT NULL, jd_filename TEXT NOT NULL, final_score REAL DEFAULT 0, verdict TEXT DEFAULT 'Unknown', hard_match_score REAL DEFAULT 0, semantic_score REAL DEFAULT 0, technical_depth_score REAL DEFAULT 0, cultural_fit_score REAL DEFAULT 0, growth_potential_score REAL DEFAULT 0, confidence REAL DEFAULT 0, matched_skills TEXT DEFAULT '[]', missing_skills TEXT DEFAULT '[]', bonus_skills TEXT DEFAULT '[]', improvement_suggestions TEXT DEFAULT '[]', quick_wins TEXT DEFAULT '[]', full_result TEXT DEFAULT '{}', processing_time REAL DEFAULT 0, analysis_mode TEXT DEFAULT 'standard', role_title TEXT DEFAULT '', experience_required TEXT DEFAULT '', market_salary_range TEXT DEFAULT '', market_demand TEXT DEFAULT '', created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, version TEXT DEFAULT '5.0.0-production' ) ''') # Enhanced analytics summary table cursor.execute(''' CREATE TABLE IF NOT EXISTS analytics_summary ( id INTEGER PRIMARY KEY DEFAULT 1, total_analyses INTEGER DEFAULT 0, avg_score REAL DEFAULT 0, high_matches INTEGER DEFAULT 0, medium_matches INTEGER DEFAULT 0, low_matches INTEGER DEFAULT 0, exceptional_matches INTEGER DEFAULT 0, success_rate REAL DEFAULT 0, last_updated TIMESTAMP DEFAULT CURRENT_TIMESTAMP, version TEXT DEFAULT '5.0.0-production' ) ''') # Insert default analytics row cursor.execute('INSERT OR IGNORE INTO analytics_summary (id) VALUES (1)') # Production-optimized indexes cursor.execute('CREATE INDEX IF NOT EXISTS idx_final_score ON analysis_results(final_score)') cursor.execute('CREATE INDEX IF NOT EXISTS idx_created_at ON analysis_results(created_at)') cursor.execute('CREATE INDEX IF NOT EXISTS idx_analysis_mode ON analysis_results(analysis_mode)') cursor.execute('CREATE INDEX IF NOT EXISTS idx_verdict ON analysis_results(verdict)') conn.commit() logger.info("✅ Production database schema initialized successfully") except Exception as e: logger.warning(f"Schema migration issue (non-critical): {e}") @retry_on_db_error() def init_database(): """Production database initialization""" try: with get_db_connection() as conn: if conn: # Test database functionality cursor = conn.cursor() cursor.execute('SELECT COUNT(*) FROM analysis_results') count = cursor.fetchone()[0] logger.info(f"✅ Production database initialized with {count} existing records") return True else: logger.warning("⚠️ Database unavailable, running in demo mode") return False except Exception as e: logger.warning(f"Database initialization warning: {e}") return False @retry_on_db_error() def save_analysis_result(analysis_data: dict, resume_filename: str, jd_filename: str) -> bool: """Production analysis result storage with enhanced data extraction""" try: with get_db_connection() as conn: if not conn: logger.info("Database unavailable, analysis not saved") return False cursor = conn.cursor() # Enhanced data extraction for production extracted_data = _extract_analysis_data_enhanced(analysis_data) processing_info = analysis_data.get('processing_info', {}) processing_time = processing_info.get('processing_time', 0) analysis_mode = processing_info.get('analysis_mode', 'enhanced_mock_v2') # Extract enhanced analysis components enhanced = analysis_data.get('enhanced_analysis', {}) job_parsing = enhanced.get('job_parsing', {}) market_insights = enhanced.get('market_insights', {}) relevance_scoring = enhanced.get('relevance_scoring', {}) cursor.execute(''' INSERT INTO analysis_results ( resume_filename, jd_filename, final_score, verdict, hard_match_score, semantic_score, technical_depth_score, cultural_fit_score, growth_potential_score, confidence, matched_skills, missing_skills, bonus_skills, improvement_suggestions, quick_wins, full_result, processing_time, analysis_mode, role_title, experience_required, market_salary_range, market_demand, created_at, updated_at, version ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, datetime('now'), datetime('now'), ?) ''', ( str(resume_filename), str(jd_filename), extracted_data['final_score'], extracted_data['verdict'], extracted_data['hard_match_score'], extracted_data['semantic_score'], relevance_scoring.get('technical_depth_score', extracted_data['final_score']), relevance_scoring.get('cultural_fit_score', extracted_data['final_score']), relevance_scoring.get('growth_potential_score', extracted_data['final_score']), relevance_scoring.get('confidence', 75.0), json.dumps(extracted_data['matched_skills']), json.dumps(extracted_data['missing_skills']), json.dumps(relevance_scoring.get('matched_good_to_have', [])), json.dumps(relevance_scoring.get('improvement_suggestions', [])), json.dumps(relevance_scoring.get('quick_wins', [])), json.dumps(analysis_data), processing_time, analysis_mode, job_parsing.get('role_title', ''), job_parsing.get('experience_required', ''), market_insights.get('salary_range_estimate', ''), market_insights.get('market_demand', ''), '5.0.0-production' )) conn.commit() # Update analytics _update_analytics_async(conn) logger.info(f"✅ Analysis saved: {resume_filename} - Score: {extracted_data['final_score']}") return True except Exception as e: logger.info(f"Analysis save failed: {e}") return False def _extract_analysis_data_enhanced(analysis_data: dict) -> Dict[str, Any]: """Enhanced data extraction for production with comprehensive error handling""" default_data = { 'final_score': 0.0, 'verdict': 'Analysis Completed', 'hard_match_score': 0.0, 'semantic_score': 0.0, 'matched_skills': [], 'missing_skills': [] } try: # Enhanced analysis format (primary) if 'enhanced_analysis' in analysis_data and 'relevance_scoring' in analysis_data['enhanced_analysis']: scoring = analysis_data['enhanced_analysis']['relevance_scoring'] return { 'final_score': float(scoring.get('overall_score', 0)), 'verdict': str(scoring.get('fit_verdict', 'Unknown')), 'hard_match_score': float(scoring.get('skill_match_score', 0)), 'semantic_score': float(scoring.get('experience_match_score', 0)), 'matched_skills': list(scoring.get('matched_must_have', [])), 'missing_skills': list(scoring.get('missing_must_have', [])) } # Standard analysis format (fallback) elif 'relevance_analysis' in analysis_data: relevance = analysis_data['relevance_analysis'] output = analysis_data.get('output_generation', {}) return { 'final_score': float(relevance.get('step_3_scoring_verdict', {}).get('final_score', 0)), 'verdict': str(output.get('verdict', 'Unknown')), 'hard_match_score': float(relevance.get('step_1_hard_match', {}).get('coverage_score', 0)), 'semantic_score': float(relevance.get('step_2_semantic_match', {}).get('experience_alignment_score', 0)), 'matched_skills': list(relevance.get('step_1_hard_match', {}).get('matched_skills', [])), 'missing_skills': list(output.get('missing_skills', [])) } # Mock data format (production demo) elif 'mock_data' in analysis_data: if 'enhanced_analysis' in analysis_data: scoring = analysis_data['enhanced_analysis'].get('relevance_scoring', {}) return { 'final_score': float(scoring.get('overall_score', 75)), 'verdict': str(scoring.get('fit_verdict', 'Good Match')), 'hard_match_score': float(scoring.get('skill_match_score', 70)), 'semantic_score': float(scoring.get('experience_match_score', 80)), 'matched_skills': list(scoring.get('matched_must_have', [])), 'missing_skills': list(scoring.get('missing_must_have', [])) } return default_data except Exception as e: logger.warning(f"Error extracting analysis data, using defaults: {e}") return default_data def _update_analytics_async(conn: sqlite3.Connection): """Update analytics in production-safe way""" try: if conn: update_analytics_summary_internal(conn) except Exception as e: logger.debug(f"Analytics update skipped: {e}") @retry_on_db_error() def get_analysis_history(limit: int = 50, offset: int = 0) -> List[AnalysisResult]: """Get analysis history with production error handling""" try: with get_db_connection() as conn: if not conn: logger.info("Database unavailable, returning empty history") return [] cursor = conn.cursor() cursor.execute(''' SELECT id, resume_filename, jd_filename, final_score, verdict, created_at, matched_skills, missing_skills, hard_match_score, semantic_score FROM analysis_results ORDER BY created_at DESC LIMIT ? OFFSET ? ''', (limit, offset)) results = [] for row in cursor.fetchall(): try: # Handle timestamp safely timestamp = _parse_timestamp(row[5]) if row[5] else datetime.now(timezone.utc) result = AnalysisResult( id=row[0], resume_filename=str(row[1] or 'Unknown'), jd_filename=str(row[2] or 'Unknown'), final_score=float(row[3] or 0), verdict=str(row[4] or 'Unknown'), timestamp=timestamp, matched_skills=row[6] or '[]', missing_skills=row[7] or '[]', hard_match_score=float(row[8] or row[3] or 0), semantic_score=float(row[9] or row[3] or 0) ) results.append(result) except Exception as row_error: logger.debug(f"Skipping malformed row: {row_error}") continue logger.info(f"Retrieved {len(results)} analysis results") return results except Exception as e: logger.info(f"History retrieval failed: {e}") return [] def _parse_timestamp(timestamp_str: str) -> datetime: """Parse timestamp with multiple format support""" if not timestamp_str: return datetime.now(timezone.utc) formats = [ '%Y-%m-%d %H:%M:%S', '%Y-%m-%d %H:%M:%S.%f', '%Y-%m-%dT%H:%M:%S', '%Y-%m-%dT%H:%M:%S.%f', '%Y-%m-%dT%H:%M:%S.%fZ' ] for fmt in formats: try: return datetime.strptime(str(timestamp_str), fmt) except ValueError: continue return datetime.now(timezone.utc) @retry_on_db_error() def get_analytics_summary() -> Dict[str, Any]: """Get analytics summary with production fallbacks""" try: with get_db_connection() as conn: if not conn: return _get_default_analytics() cursor = conn.cursor() cursor.execute(''' SELECT COUNT(*) as total_analyses, COALESCE(AVG(final_score), 0) as avg_score, COUNT(CASE WHEN final_score >= 88 THEN 1 END) as exceptional_matches, COUNT(CASE WHEN final_score >= 80 AND final_score < 88 THEN 1 END) as high_matches, COUNT(CASE WHEN final_score >= 60 AND final_score < 80 THEN 1 END) as medium_matches, COUNT(CASE WHEN final_score < 60 AND final_score > 0 THEN 1 END) as low_matches FROM analysis_results ''') result = cursor.fetchone() total_analyses = result[0] or 0 avg_score = round(float(result[1] or 0), 1) exceptional_matches = result[2] or 0 high_matches = result[3] or 0 medium_matches = result[4] or 0 low_matches = result[5] or 0 # Calculate success rate success_rate = 0.0 if total_analyses > 0: success_rate = round(((exceptional_matches + high_matches + medium_matches) / total_analyses) * 100, 1) return { 'total_analyses': total_analyses, 'avg_score': avg_score, 'exceptional_matches': exceptional_matches, 'high_matches': high_matches, 'medium_matches': medium_matches, 'low_matches': low_matches, 'success_rate': success_rate, 'generated_at': datetime.now(timezone.utc).isoformat(), 'database_available': True, 'storage_location': '/tmp (temporary)', 'version': '5.0.0-production' } except Exception as e: logger.info(f"Analytics unavailable: {e}") return _get_default_analytics() def _get_default_analytics(): """Default analytics for production demo mode""" return { 'total_analyses': 0, 'avg_score': 0.0, 'exceptional_matches': 0, 'high_matches': 0, 'medium_matches': 0, 'low_matches': 0, 'success_rate': 0.0, 'generated_at': datetime.now(timezone.utc).isoformat(), 'database_available': False, 'note': 'Database unavailable - running in production demo mode', 'version': '5.0.0-production' } @retry_on_db_error() def delete_analysis_result(analysis_id: int) -> bool: """Delete analysis result with production error handling""" try: with get_db_connection() as conn: if not conn: logger.info("Database unavailable, cannot delete") return False cursor = conn.cursor() cursor.execute('SELECT id FROM analysis_results WHERE id = ?', (analysis_id,)) if not cursor.fetchone(): logger.info(f"Analysis result {analysis_id} not found") return False cursor.execute('DELETE FROM analysis_results WHERE id = ?', (analysis_id,)) conn.commit() _update_analytics_async(conn) logger.info(f"Analysis result {analysis_id} deleted") return True except Exception as e: logger.info(f"Delete failed: {e}") return False @retry_on_db_error() def clear_all_analysis_history() -> Dict[str, Any]: """Clear all analysis history with production safety""" try: with get_db_connection() as conn: if not conn: return {"success": False, "error": "Database unavailable", "deleted_count": 0} cursor = conn.cursor() cursor.execute('SELECT COUNT(*) FROM analysis_results') total_count = cursor.fetchone()[0] if total_count == 0: return {"success": True, "deleted_count": 0, "message": "No records to delete"} cursor.execute('DELETE FROM analysis_results') cursor.execute(''' UPDATE analytics_summary SET total_analyses = 0, avg_score = 0, high_matches = 0, medium_matches = 0, low_matches = 0, exceptional_matches = 0, last_updated = datetime('now') WHERE id = 1 ''') conn.commit() logger.info(f"Cleared {total_count} analysis records") return { "success": True, "deleted_count": total_count, "message": f"Successfully deleted {total_count} records" } except Exception as e: logger.info(f"Clear history failed: {e}") return {"success": False, "error": str(e), "deleted_count": 0} @retry_on_db_error() def get_analysis_result_by_id(analysis_id: int) -> Dict[str, Any]: """Get single analysis result with production error handling""" try: with get_db_connection() as conn: if not conn: return {"success": False, "error": "Database unavailable"} cursor = conn.cursor() cursor.execute(''' SELECT id, resume_filename, jd_filename, final_score, verdict, created_at, matched_skills, missing_skills, full_result, hard_match_score, semantic_score, processing_time, analysis_mode, role_title, experience_required, market_salary_range, market_demand FROM analysis_results WHERE id = ? ''', (analysis_id,)) row = cursor.fetchone() if not row: return {"success": False, "error": "Analysis not found"} # Parse data safely try: full_result = json.loads(row[8]) if row[8] else {} matched_skills = json.loads(row[6]) if row[6] else [] missing_skills = json.loads(row[7]) if row[7] else [] except json.JSONDecodeError: full_result = {} matched_skills = [] missing_skills = [] # Handle timestamp timestamp = _parse_timestamp(row[5]) if row[5] else datetime.now(timezone.utc) result = { "success": True, "analysis": { "id": row[0], "resume_filename": row[1], "jd_filename": row[2], "final_score": row[3], "verdict": row[4], "timestamp": timestamp.isoformat(), "matched_skills": matched_skills, "missing_skills": missing_skills, "hard_match_score": row[9] or row[3], "semantic_score": row[10] or row[3], "processing_time": row[11] or 0, "analysis_mode": row[12] or 'standard', "role_title": row[13] or '', "experience_required": row[14] or '', "market_salary_range": row[15] or '', "market_demand": row[16] or '', "full_result": full_result, "version": "5.0.0-production" } } return result except Exception as e: logger.info(f"Get analysis by ID failed: {e}") return {"success": False, "error": str(e)} def update_analytics_summary_internal(conn: sqlite3.Connection): """Internal analytics update for production""" if not conn: return try: cursor = conn.cursor() cursor.execute(''' SELECT COUNT(*) as total, COALESCE(AVG(final_score), 0) as avg_score, COUNT(CASE WHEN final_score >= 88 THEN 1 END) as exceptional, COUNT(CASE WHEN final_score >= 80 AND final_score < 88 THEN 1 END) as high, COUNT(CASE WHEN final_score >= 60 AND final_score < 80 THEN 1 END) as medium, COUNT(CASE WHEN final_score < 60 AND final_score > 0 THEN 1 END) as low FROM analysis_results ''') result = cursor.fetchone() total, avg_score, exceptional, high, medium, low = result # Calculate success rate success_rate = 0.0 if total > 0: success_rate = round(((exceptional + high + medium) / total) * 100, 1) cursor.execute(''' UPDATE analytics_summary SET total_analyses = ?, avg_score = ?, exceptional_matches = ?, high_matches = ?, medium_matches = ?, low_matches = ?, success_rate = ?, last_updated = datetime('now') WHERE id = 1 ''', (total, round(avg_score, 1), exceptional, high, medium, low, success_rate)) conn.commit() except Exception as e: logger.debug(f"Analytics update skipped: {e}") def get_recent_analyses(limit: int = 10) -> List[Dict[str, Any]]: """Get recent analyses for production dashboard""" try: results = get_analysis_history(limit) return [ { "id": result.id, "resume": result.resume_filename, "job_description": result.jd_filename, "score": result.final_score, "verdict": result.verdict, "date": result.timestamp.strftime("%Y-%m-%d %H:%M") if hasattr(result.timestamp, 'strftime') else str(result.timestamp), "matched_skills": result.matched_skills, "missing_skills": result.missing_skills, "hard_match_score": result.hard_match_score, "semantic_score": result.semantic_score } for result in results ] except Exception as e: logger.info(f"Recent analyses unavailable: {e}") return [] def get_database_stats() -> Dict[str, Any]: """Get database statistics for production monitoring""" try: with get_db_connection() as conn: if not conn: return { "database_available": False, "error": "Database unavailable", "database_path": config.db_path, "storage_location": "/tmp (temporary)", "version": "5.0.0-production" } cursor = conn.cursor() cursor.execute("SELECT COUNT(*) FROM analysis_results") analysis_count = cursor.fetchone()[0] db_size = 0 try: db_path = Path(config.db_path) if db_path.exists(): db_size = db_path.stat().st_size except: pass cursor.execute("SELECT MIN(created_at), MAX(created_at) FROM analysis_results") date_range = cursor.fetchone() return { "database_path": config.db_path, "database_size_bytes": db_size, "database_size_mb": round(db_size / (1024 * 1024), 2), "analysis_results_count": analysis_count, "earliest_record": date_range[0], "latest_record": date_range[1], "wal_enabled": False, "backup_enabled": False, "database_available": True, "storage_location": "/tmp (temporary)", "version": "5.0.0-production" } except Exception as e: logger.info(f"Database stats unavailable: {e}") return { "database_available": False, "error": str(e), "database_path": config.db_path, "storage_location": "/tmp (temporary)", "version": "5.0.0-production" } # Production initialization try: init_database() logger.info("🚀 Production database module initialized for HuggingFace Spaces") except Exception as e: logger.info(f"Database running in production demo mode: {e}") if __name__ == "__main__": # Test database functionality logger.info("Testing production database...") if init_database(): logger.info("✅ Database test successful") else: logger.info("⚠️ Database running in demo mode")