Spaces:
Running
Running
| """ | |
| persistence.py | |
| -------------- | |
| Persistence layer for candidate results, feedback, and analytics. | |
| Features: | |
| - SQLite for local storage (no external DB needed) | |
| - Store results, feedback, hiring outcomes | |
| - Query historical matches | |
| - Analytics on hiring success | |
| Author: SmartHire AI | |
| """ | |
| import json | |
| import logging | |
| import sqlite3 | |
| from datetime import datetime | |
| from pathlib import Path | |
| from typing import Dict, List, Optional, Tuple | |
| logger = logging.getLogger(__name__) | |
| DB_PATH = Path("smarthire_data.db") | |
| class CandidateDatabase: | |
| """ | |
| SQLite-backed candidate and result persistence. | |
| """ | |
| def __init__(self, db_path: Path = DB_PATH): | |
| self.db_path = db_path | |
| self._init_db() | |
| def _init_db(self) -> None: | |
| """Initialize database schema if not exists.""" | |
| conn = sqlite3.connect(self.db_path) | |
| cursor = conn.cursor() | |
| # Candidates table | |
| cursor.execute(""" | |
| CREATE TABLE IF NOT EXISTS candidates ( | |
| id INTEGER PRIMARY KEY, | |
| name TEXT UNIQUE, | |
| email TEXT, | |
| phone TEXT, | |
| resume_text TEXT, | |
| skills TEXT, | |
| created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP | |
| ) | |
| """) | |
| # Job descriptions table | |
| cursor.execute(""" | |
| CREATE TABLE IF NOT EXISTS job_descriptions ( | |
| id INTEGER PRIMARY KEY, | |
| title TEXT, | |
| jd_text TEXT, | |
| skills_required TEXT, | |
| created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP | |
| ) | |
| """) | |
| # Match results table | |
| cursor.execute(""" | |
| CREATE TABLE IF NOT EXISTS match_results ( | |
| id INTEGER PRIMARY KEY, | |
| candidate_id INTEGER, | |
| jd_id INTEGER, | |
| match_score REAL, | |
| semantic_score REAL, | |
| skill_coverage REAL, | |
| recommendation TEXT, | |
| result_json TEXT, | |
| created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, | |
| FOREIGN KEY(candidate_id) REFERENCES candidates(id), | |
| FOREIGN KEY(jd_id) REFERENCES job_descriptions(id) | |
| ) | |
| """) | |
| # Feedback table | |
| cursor.execute(""" | |
| CREATE TABLE IF NOT EXISTS feedback ( | |
| id INTEGER PRIMARY KEY, | |
| match_id INTEGER, | |
| outcome TEXT, | |
| notes TEXT, | |
| rating INTEGER, | |
| created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, | |
| FOREIGN KEY(match_id) REFERENCES match_results(id) | |
| ) | |
| """) | |
| conn.commit() | |
| conn.close() | |
| logger.info(f"Database initialized: {self.db_path}") | |
| def add_candidate(self, name: str, email: str, phone: str, resume_text: str) -> int: | |
| """Add a candidate. Returns candidate_id.""" | |
| conn = sqlite3.connect(self.db_path) | |
| cursor = conn.cursor() | |
| try: | |
| cursor.execute( | |
| "INSERT INTO candidates (name, email, phone, resume_text) VALUES (?, ?, ?, ?)", | |
| (name, email, phone, resume_text), | |
| ) | |
| conn.commit() | |
| candidate_id = cursor.lastrowid | |
| logger.info(f"Candidate added: {name} (ID: {candidate_id})") | |
| return candidate_id | |
| except sqlite3.IntegrityError: | |
| logger.warning(f"Candidate already exists: {name}") | |
| return self.get_candidate_id(name) | |
| finally: | |
| conn.close() | |
| def get_candidate_id(self, name: str) -> Optional[int]: | |
| """Get candidate ID by name.""" | |
| conn = sqlite3.connect(self.db_path) | |
| cursor = conn.cursor() | |
| cursor.execute("SELECT id FROM candidates WHERE name = ?", (name,)) | |
| result = cursor.fetchone() | |
| conn.close() | |
| return result[0] if result else None | |
| def add_job_description(self, title: str, jd_text: str) -> int: | |
| """Add a job description. Returns jd_id.""" | |
| conn = sqlite3.connect(self.db_path) | |
| cursor = conn.cursor() | |
| cursor.execute( | |
| "INSERT INTO job_descriptions (title, jd_text) VALUES (?, ?)", | |
| (title, jd_text), | |
| ) | |
| conn.commit() | |
| jd_id = cursor.lastrowid | |
| conn.close() | |
| return jd_id | |
| def save_match_result( | |
| self, | |
| candidate_name: str, | |
| jd_id: int, | |
| match_score: float, | |
| semantic_score: float, | |
| skill_coverage: float, | |
| recommendation: str, | |
| result_dict: Dict, | |
| ) -> int: | |
| """Save a match result.""" | |
| candidate_id = self.get_candidate_id(candidate_name) | |
| if not candidate_id: | |
| logger.warning(f"Candidate not found: {candidate_name}") | |
| return None | |
| conn = sqlite3.connect(self.db_path) | |
| cursor = conn.cursor() | |
| cursor.execute( | |
| """ | |
| INSERT INTO match_results ( | |
| candidate_id, jd_id, match_score, semantic_score, | |
| skill_coverage, recommendation, result_json | |
| ) VALUES (?, ?, ?, ?, ?, ?, ?) | |
| """, | |
| ( | |
| candidate_id, | |
| jd_id, | |
| match_score, | |
| semantic_score, | |
| skill_coverage, | |
| recommendation, | |
| json.dumps(result_dict), | |
| ), | |
| ) | |
| conn.commit() | |
| result_id = cursor.lastrowid | |
| conn.close() | |
| logger.info(f"Match result saved (ID: {result_id})") | |
| return result_id | |
| def add_feedback(self, match_id: int, outcome: str, rating: int, notes: str = "") -> int: | |
| """Add feedback for a match. outcome: 'hired'|'rejected'|'in_progress'""" | |
| conn = sqlite3.connect(self.db_path) | |
| cursor = conn.cursor() | |
| cursor.execute( | |
| "INSERT INTO feedback (match_id, outcome, rating, notes) VALUES (?, ?, ?, ?)", | |
| (match_id, outcome, rating, notes), | |
| ) | |
| conn.commit() | |
| feedback_id = cursor.lastrowid | |
| conn.close() | |
| logger.info(f"Feedback added for match {match_id}") | |
| return feedback_id | |
| def get_analytics(self) -> Dict: | |
| """Get overall hiring analytics.""" | |
| conn = sqlite3.connect(self.db_path) | |
| cursor = conn.cursor() | |
| # Total matches | |
| cursor.execute("SELECT COUNT(*) FROM match_results") | |
| total_matches = cursor.fetchone()[0] | |
| # Hiring outcomes | |
| cursor.execute( | |
| "SELECT outcome, COUNT(*) FROM feedback GROUP BY outcome" | |
| ) | |
| outcomes = {row[0]: row[1] for row in cursor.fetchall()} | |
| # Average scores by outcome | |
| cursor.execute( | |
| """ | |
| SELECT f.outcome, AVG(m.match_score) | |
| FROM feedback f | |
| JOIN match_results m ON f.match_id = m.id | |
| GROUP BY f.outcome | |
| """ | |
| ) | |
| avg_scores = {row[0]: round(row[1], 2) for row in cursor.fetchall()} | |
| # Success rate (hired / total) | |
| hired = outcomes.get("hired", 0) | |
| total_with_feedback = sum(outcomes.values()) | |
| success_rate = ( | |
| (hired / total_with_feedback * 100) | |
| if total_with_feedback > 0 | |
| else 0.0 | |
| ) | |
| conn.close() | |
| return { | |
| "total_matches": total_matches, | |
| "outcomes": outcomes, | |
| "average_scores_by_outcome": avg_scores, | |
| "hiring_success_rate": round(success_rate, 2), | |
| "total_with_feedback": total_with_feedback, | |
| } | |
| def get_match_history(self, candidate_name: str) -> List[Dict]: | |
| """Get all matches for a candidate.""" | |
| candidate_id = self.get_candidate_id(candidate_name) | |
| if not candidate_id: | |
| return [] | |
| conn = sqlite3.connect(self.db_path) | |
| cursor = conn.cursor() | |
| cursor.execute( | |
| """ | |
| SELECT id, match_score, semantic_score, skill_coverage, | |
| recommendation, created_at | |
| FROM match_results | |
| WHERE candidate_id = ? | |
| ORDER BY created_at DESC | |
| """, | |
| (candidate_id,), | |
| ) | |
| rows = cursor.fetchall() | |
| conn.close() | |
| return [ | |
| { | |
| "id": row[0], | |
| "match_score": row[1], | |
| "semantic_score": row[2], | |
| "skill_coverage": row[3], | |
| "recommendation": row[4], | |
| "created_at": row[5], | |
| } | |
| for row in rows | |
| ] | |