# ============================================================================= # engine/database.py — All Database Operations # ============================================================================= # PURPOSE: # Single source of truth for all SQLite read/write operations. # Every other file calls this — nothing else touches the DB directly. # # DATABASE: SQLite (single file at data/resume_engine.db) # No setup required. DB file is created automatically on first run. # # TABLES: # resumes — parsed resume records # job_descriptions — parsed JD records # match_results — scored candidate-JD matches # # USAGE: # from engine.database import init_db, insert_resume, get_all_resumes # init_db() # call once on startup # ============================================================================= import os import sys import json import uuid import sqlite3 from datetime import datetime # Add project root to path so config.py is importable sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) import config # ============================================================================= # INTERNAL HELPER — DB Connection # ============================================================================= def _get_connection(db_path: str = None) -> sqlite3.Connection: """ Open a SQLite connection and configure it. - Returns rows as dicts (access by column name, not index) - Enables foreign key enforcement Args: db_path: Path to .db file. Defaults to config.DB_PATH. """ path = db_path or config.DB_PATH os.makedirs(os.path.dirname(path), exist_ok=True) conn = sqlite3.connect(path) conn.row_factory = sqlite3.Row # Rows behave like dicts conn.execute("PRAGMA foreign_keys = ON") return conn # ============================================================================= # INIT — Create Tables # ============================================================================= def init_db(db_path: str = None) -> None: """ Create all three tables if they don't already exist. Safe to call multiple times — uses CREATE TABLE IF NOT EXISTS. Call this once at the start of main.py. Args: db_path: Optional override for DB location (used in tests). """ conn = _get_connection(db_path) with conn: # ------------------------------------------------------------------ # TABLE 1: resumes # One row per uploaded resume file. # ------------------------------------------------------------------ conn.execute(""" CREATE TABLE IF NOT EXISTS resumes ( resume_id TEXT PRIMARY KEY, file_name TEXT NOT NULL, candidate_name TEXT, email TEXT, phone TEXT, age INTEGER DEFAULT 0, location TEXT, linkedin_url TEXT, resume_summary TEXT, resume_text TEXT, skills TEXT, -- JSON list: technical skills soft_skills TEXT, -- JSON list: soft/behavioural skills languages TEXT, -- JSON list: spoken languages experience_years REAL, last_company TEXT, job_titles TEXT, -- JSON list: full-time titles companies TEXT, -- JSON list: full-time companies avg_tenure_months REAL, internships TEXT, -- JSON list: "Company | Role | Duration" internship_count INTEGER DEFAULT 0, education TEXT, -- JSON: {degree,field,institution,graduation_year} graduation_year INTEGER DEFAULT 0, certifications TEXT, -- JSON list projects TEXT, -- JSON list achievements TEXT, -- JSON list hobbies TEXT, -- JSON list recency_flag TEXT, stability_flag INTEGER DEFAULT 0, pros TEXT, cons TEXT, upload_batch TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) """) # ------------------------------------------------------------------ # TABLE 2: job_descriptions # One row per JD uploaded. # ------------------------------------------------------------------ conn.execute(""" CREATE TABLE IF NOT EXISTS job_descriptions ( jd_id TEXT PRIMARY KEY, jd_title TEXT NOT NULL, jd_text TEXT, required_skills TEXT, -- JSON list preferred_skills TEXT, -- JSON list required_experience REAL, education_requirement TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) """) # ------------------------------------------------------------------ # TABLE 3: match_results # One row per resume-JD pair after scoring. # ------------------------------------------------------------------ conn.execute(""" CREATE TABLE IF NOT EXISTS match_results ( match_id TEXT PRIMARY KEY, resume_id TEXT NOT NULL, jd_id TEXT NOT NULL, skill_match_score REAL, experience_score REAL, semantic_score REAL, stability_score REAL, recency_score REAL, education_score REAL, final_score REAL, confidence_tier TEXT, matched_skills TEXT, -- JSON list missing_skills TEXT, -- JSON list standout_skills TEXT, -- JSON list summary TEXT, interview_questions TEXT, -- JSON list of dicts created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (resume_id) REFERENCES resumes(resume_id), FOREIGN KEY (jd_id) REFERENCES job_descriptions(jd_id) ) """) migrations = [ ("age", "INTEGER DEFAULT 0"), ("location", "TEXT"), ("linkedin_url", "TEXT"), ("resume_summary", "TEXT"), ("languages", "TEXT"), ("companies", "TEXT"), ("internships", "TEXT"), ("internship_count", "INTEGER DEFAULT 0"), ("graduation_year", "INTEGER DEFAULT 0"), ("achievements", "TEXT"), ("stability_flag", "INTEGER DEFAULT 0"), ("pros", "TEXT"), ("cons", "TEXT"), ] for col_name, col_type in migrations: try: conn.execute(f"ALTER TABLE resumes ADD COLUMN {col_name} {col_type}") conn.commit() except Exception: pass # Column already exists conn.close() conn.close() print(f"[DB] Initialised — 3 tables ready at: {db_path or config.DB_PATH}") # ============================================================================= # INSERT OPERATIONS # ============================================================================= def insert_resume(resume_data: dict, db_path: str = None) -> str: """ Insert one parsed resume into the resumes table. DEDUPLICATION: If a file with the same file_name already exists, the existing resume_id is returned — no duplicate rows created. This means you can safely re-upload the same folder and it won't create duplicate entries. Args: resume_data: Dict matching the resumes table schema. Keys match column names exactly. db_path: Optional DB path override (used in tests). Returns: resume_id (str): UUID of the inserted or existing record. """ conn = _get_connection(db_path) # If file already in DB, delete the old record so we re-insert fresh data. existing = conn.execute( "SELECT resume_id FROM resumes WHERE file_name = ?", (resume_data.get("file_name"),) ).fetchone() if existing: with conn: old_id = existing["resume_id"] conn.execute("DELETE FROM match_results WHERE resume_id = ?", (old_id,)) conn.execute("DELETE FROM resumes WHERE resume_id = ?", (old_id,)) resume_id = str(uuid.uuid4()) with conn: conn.execute(""" INSERT INTO resumes ( resume_id, file_name, candidate_name, email, phone, age, location, linkedin_url, resume_summary, resume_text, skills, soft_skills, languages, experience_years, last_company, job_titles, companies, avg_tenure_months, internships, internship_count, education, graduation_year, certifications, projects, achievements, hobbies, recency_flag, stability_flag,pros, cons, upload_batch, created_at ) VALUES ( :resume_id, :file_name, :candidate_name, :email, :phone, :age, :location, :linkedin_url, :resume_summary, :resume_text, :skills, :soft_skills, :languages, :experience_years, :last_company, :job_titles, :companies, :avg_tenure_months, :internships, :internship_count, :education, :graduation_year, :certifications, :projects, :achievements, :hobbies, :recency_flag, :stability_flag, :pros, :cons, :upload_batch, :created_at ) """, { "resume_id" : resume_id, "file_name" : resume_data.get("file_name"), "candidate_name" : resume_data.get("candidate_name"), "email" : resume_data.get("email"), "phone" : resume_data.get("phone"), "age" : int(resume_data.get("age") or 0), "location" : resume_data.get("location", ""), "linkedin_url" : resume_data.get("linkedin_url", ""), "resume_summary" : resume_data.get("resume_summary", ""), "resume_text" : resume_data.get("resume_text"), "skills" : resume_data.get("skills", "[]"), "soft_skills" : resume_data.get("soft_skills", "[]"), "languages" : resume_data.get("languages", "[]"), "experience_years" : float(resume_data.get("experience_years") or 0), "last_company" : resume_data.get("last_company", ""), "job_titles" : resume_data.get("job_titles", "[]"), "companies" : resume_data.get("companies", "[]"), "avg_tenure_months" : float(resume_data.get("avg_tenure_months") or 0), "internships" : resume_data.get("internships", "[]"), "internship_count" : int(resume_data.get("internship_count") or 0), "education" : resume_data.get("education", "{}"), "graduation_year" : int(resume_data.get("graduation_year") or 0), "certifications" : resume_data.get("certifications", "[]"), "projects" : resume_data.get("projects", "[]"), "achievements" : resume_data.get("achievements", "[]"), "hobbies" : resume_data.get("hobbies", "[]"), "recency_flag" : resume_data.get("recency_flag", "Unknown"), "stability_flag" : 1 if resume_data.get("stability_flag") else 0, "pros" : resume_data.get("pros", ""), "cons" : resume_data.get("cons", ""), "upload_batch" : resume_data.get("upload_batch", datetime.now().strftime("%Y%m%d_%H%M%S")), "created_at" : datetime.now().isoformat(), }) conn.close() return resume_id def insert_jd(jd_data: dict, db_path: str = None) -> str: """ Insert one job description into the job_descriptions table. DEDUPLICATION: If a JD with the same jd_title already exists, the existing jd_id is returned. Args: jd_data: Dict matching the job_descriptions schema. db_path: Optional DB path override. Returns: jd_id (str): UUID of the inserted or existing record. """ conn = _get_connection(db_path) # If JD already in DB, delete old record so we re-insert fresh data. existing = conn.execute( "SELECT jd_id FROM job_descriptions WHERE jd_title = ?", (jd_data.get("jd_title"),) ).fetchone() if existing: with conn: old_jid = existing["jd_id"] conn.execute("DELETE FROM match_results WHERE jd_id = ?", (old_jid,)) conn.execute("DELETE FROM job_descriptions WHERE jd_id = ?", (old_jid,)) jd_id = str(uuid.uuid4()) with conn: conn.execute(""" INSERT INTO job_descriptions ( jd_id, jd_title, jd_text, required_skills, preferred_skills, required_experience, education_requirement, created_at ) VALUES ( :jd_id, :jd_title, :jd_text, :required_skills, :preferred_skills, :required_experience, :education_requirement, :created_at ) """, { "jd_id" : jd_id, "jd_title" : jd_data.get("jd_title"), "jd_text" : jd_data.get("jd_text"), "required_skills" : jd_data.get("required_skills"), "preferred_skills" : jd_data.get("preferred_skills"), "required_experience" : jd_data.get("required_experience"), "education_requirement" : jd_data.get("education_requirement"), "created_at" : datetime.now().isoformat(), }) conn.close() return jd_id def insert_match_result(match_data: dict, db_path: str = None) -> str: """ Insert one scored match result into the match_results table. Args: match_data: Dict matching the match_results schema. db_path: Optional DB path override. Returns: match_id (str): UUID of the inserted record. """ conn = _get_connection(db_path) match_id = str(uuid.uuid4()) with conn: conn.execute(""" INSERT INTO match_results ( match_id, resume_id, jd_id, skill_match_score, experience_score, semantic_score, stability_score, recency_score, education_score, final_score, confidence_tier, matched_skills, missing_skills, standout_skills, summary, interview_questions, created_at ) VALUES ( :match_id, :resume_id, :jd_id, :skill_match_score, :experience_score, :semantic_score, :stability_score, :recency_score, :education_score, :final_score, :confidence_tier, :matched_skills, :missing_skills, :standout_skills, :summary, :interview_questions, :created_at ) """, { "match_id" : match_id, "resume_id" : match_data.get("resume_id"), "jd_id" : match_data.get("jd_id"), "skill_match_score" : match_data.get("skill_match_score"), "experience_score" : match_data.get("experience_score"), "semantic_score" : match_data.get("semantic_score"), "stability_score" : match_data.get("stability_score"), "recency_score" : match_data.get("recency_score"), "education_score" : match_data.get("education_score"), "final_score" : match_data.get("final_score"), "confidence_tier" : match_data.get("confidence_tier"), "matched_skills" : match_data.get("matched_skills"), "missing_skills" : match_data.get("missing_skills"), "standout_skills" : match_data.get("standout_skills"), "summary" : match_data.get("summary"), "interview_questions": match_data.get("interview_questions"), "created_at" : datetime.now().isoformat(), }) conn.close() return match_id # ============================================================================= # READ OPERATIONS # ============================================================================= def get_all_resumes(db_path: str = None) -> list: """ Fetch all resumes from the database. Returns: List of dicts, one per resume, ordered by created_at DESC (most recently uploaded first). """ conn = _get_connection(db_path) rows = conn.execute( "SELECT * FROM resumes ORDER BY created_at DESC" ).fetchall() conn.close() return [dict(row) for row in rows] def get_resume_by_id(resume_id: str, db_path: str = None) -> dict: """ Fetch a single resume by its UUID. Args: resume_id: UUID string from insert_resume() Returns: Dict of resume data, or None if not found. """ conn = _get_connection(db_path) row = conn.execute( "SELECT * FROM resumes WHERE resume_id = ?", (resume_id,) ).fetchone() conn.close() return dict(row) if row else None def get_all_jds(db_path: str = None) -> list: """ Fetch all job descriptions from the database. Returns: List of dicts ordered by created_at DESC. """ conn = _get_connection(db_path) rows = conn.execute( "SELECT * FROM job_descriptions ORDER BY created_at DESC" ).fetchall() conn.close() return [dict(row) for row in rows] def get_matches_by_jd(jd_id: str, db_path: str = None) -> list: """ Fetch all match results for a given JD, joined with candidate name. Ordered by final_score DESC (highest scoring candidate first). Args: jd_id: UUID of the job description Returns: List of dicts containing match scores + candidate_name. """ conn = _get_connection(db_path) rows = conn.execute(""" SELECT mr.*, r.candidate_name, r.file_name, r.email, r.phone, r.age, r.location, r.resume_summary, r.experience_years, r.last_company, r.job_titles, r.companies, r.avg_tenure_months, r.internships, r.internship_count, r.education, r.graduation_year, r.skills, r.soft_skills, r.languages, r.certifications, r.projects, r.achievements, r.recency_flag, r.stability_flag, r.upload_batch FROM match_results mr JOIN resumes r ON mr.resume_id = r.resume_id WHERE mr.jd_id = ? ORDER BY mr.final_score DESC """, (jd_id,)).fetchall() conn.close() return [dict(row) for row in rows] # ============================================================================= # SYNC OPERATIONS — Keep DB in sync with files on disk # ============================================================================= def sync_resumes_with_filesystem(active_filenames: list, db_path: str = None) -> int: """ Delete resume records from DB whose file_name is NOT in active_filenames. Call this at the start of each pipeline run (after scanning the resumes dir) so that records for deleted files don't pollute results. Args: active_filenames: List of filenames currently on disk (just the basename). db_path: Optional DB path override. Returns: Number of records deleted. """ conn = _get_connection(db_path) existing = conn.execute("SELECT resume_id, file_name FROM resumes").fetchall() stale = [row["resume_id"] for row in existing if row["file_name"] not in active_filenames] deleted = 0 if stale: with conn: for rid in stale: conn.execute("DELETE FROM match_results WHERE resume_id = ?", (rid,)) conn.execute("DELETE FROM resumes WHERE resume_id = ?", (rid,)) deleted += 1 print(f"[DB] Removed {deleted} stale resume record(s) (files deleted from disk)") conn.close() return deleted def sync_jds_with_filesystem(active_jd_titles: list, db_path: str = None) -> int: """ Delete JD records from DB whose jd_title is NOT in active_jd_titles. Args: active_jd_titles: List of JD titles currently parsed from files on disk. db_path: Optional DB path override. Returns: Number of records deleted. """ conn = _get_connection(db_path) existing = conn.execute("SELECT jd_id, jd_title FROM job_descriptions").fetchall() stale = [row["jd_id"] for row in existing if row["jd_title"] not in active_jd_titles] deleted = 0 if stale: with conn: for jid in stale: conn.execute("DELETE FROM match_results WHERE jd_id = ?", (jid,)) conn.execute("DELETE FROM job_descriptions WHERE jd_id = ?", (jid,)) deleted += 1 print(f"[DB] Removed {deleted} stale JD record(s) (files deleted from disk)") conn.close() return deleted # ============================================================================= # EXPORT # ============================================================================= def export_matches_to_csv(jd_id: str, output_path: str = None, db_path: str = None) -> str: """ Export all match results for a JD to a CSV file. Adds a 'rank' column (1 = highest scoring candidate). Args: jd_id: UUID of the job description output_path: Where to save the CSV. Defaults to outputs/match_results/ db_path: Optional DB path override. Returns: Full path to the saved CSV file. """ import pandas as pd matches = get_matches_by_jd(jd_id, db_path) if not matches: print(f"[DB] No matches found for jd_id: {jd_id}") return None df = pd.DataFrame(matches) # Add rank column df.insert(0, "rank", range(1, len(df) + 1)) # Reorder to export columns defined in config (keep any extras at end) ordered = [c for c in config.EXPORT_COLUMNS if c in df.columns] remaining = [c for c in df.columns if c not in ordered] df = df[ordered + remaining] # Build output path if not output_path: os.makedirs(config.OUTPUT_DIR, exist_ok=True) timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") output_path = os.path.join( config.OUTPUT_DIR, f"match_results_{timestamp}.csv" ) df.to_csv(output_path, index=False, quoting=1) print(f"[DB] Exported {len(df)} results → {output_path}") return output_path # ============================================================================= # SELF-TEST # Run this file directly: # python engine/database.py # ============================================================================= if __name__ == "__main__": import tempfile GREEN = "\033[92m" RED = "\033[91m" RESET = "\033[0m" print("\n" + "="*55) print(" DATABASE SELF-TEST") print("="*55) passed = 0 failed = 0 def check(condition, msg): global passed, failed if condition: print(f" {GREEN}✓{RESET} {msg}") passed += 1 else: print(f" {RED}✗{RESET} {msg}") failed += 1 with tempfile.TemporaryDirectory() as tmp: TEST_DB = os.path.join(tmp, "test.db") # 1. init_db try: init_db(TEST_DB) conn = sqlite3.connect(TEST_DB) tables = [r[0] for r in conn.execute( "SELECT name FROM sqlite_master WHERE type='table'" ).fetchall()] conn.close() check("resumes" in tables, "Table 'resumes' created") check("job_descriptions" in tables, "Table 'job_descriptions' created") check("match_results" in tables, "Table 'match_results' created") except Exception as e: print(f" {RED}✗ init_db failed: {e}{RESET}") failed += 3 # 2. insert_resume + deduplication r = { "file_name": "test.pdf", "candidate_name": "Priya Sharma", "email": "priya@test.com", "phone": "9876543210", "resume_text": "Data Analyst with Python and SQL.", "skills": json.dumps(["Python", "SQL"]), "soft_skills": json.dumps(["Communication"]), "experience_years": 4.0, "last_company": "Infosys", "job_titles": json.dumps(["Data Analyst"]), "education": json.dumps({"degree": "B.Tech"}), "certifications": json.dumps([]), "projects": json.dumps([]), "hobbies": json.dumps([]), "avg_tenure_months": 24.0, "recency_flag": "Recent", "upload_batch": "batch_001", } rid1 = insert_resume(r, TEST_DB) rid2 = insert_resume(r, TEST_DB) # duplicate check(rid1 is not None, f"insert_resume() → ID: {rid1[:8]}...") check(rid1 == rid2, "Deduplication: same file returns same ID") # 3. insert_jd j = { "jd_title": "Data Analyst", "jd_text": "Python, SQL required.", "required_skills": json.dumps(["Python", "SQL"]), "preferred_skills": json.dumps(["Tableau"]), "required_experience": 3.0, "education_requirement": "B.Tech", } jid = insert_jd(j, TEST_DB) check(jid is not None, f"insert_jd() → ID: {jid[:8]}...") # 4. insert_match_result m = { "resume_id": rid1, "jd_id": jid, "skill_match_score": 80.0, "experience_score": 75.0, "semantic_score": 0.87, "stability_score": 90.0, "recency_score": 85.0, "education_score": 70.0, "final_score": 81.25, "confidence_tier": "Strong Fit", "matched_skills": json.dumps(["Python", "SQL"]), "missing_skills": json.dumps(["Tableau"]), "standout_skills": json.dumps([]), "summary": "Strong fit. Matched Python and SQL.", "interview_questions": json.dumps([]), } mid = insert_match_result(m, TEST_DB) check(mid is not None, f"insert_match_result() → ID: {mid[:8]}...") # 5. get_all_resumes all_r = get_all_resumes(TEST_DB) check(len(all_r) == 1, f"get_all_resumes() → {len(all_r)} record") check(all_r[0]["candidate_name"] == "Priya Sharma", "Candidate name stored and retrieved correctly") # 6. get_resume_by_id single = get_resume_by_id(rid1, TEST_DB) check(single is not None, "get_resume_by_id() found record") check(get_resume_by_id("fake-id", TEST_DB) is None, "get_resume_by_id() returns None for unknown ID") # 7. get_matches_by_jd matches = get_matches_by_jd(jid, TEST_DB) check(len(matches) == 1, f"get_matches_by_jd() → {len(matches)} match") check(matches[0]["final_score"] == 81.25, f"Final score stored correctly: {matches[0]['final_score']}") print(f"\n Passed: {GREEN}{passed}{RESET} | Failed: {RED}{failed}{RESET}") if failed == 0: print(f"\n {GREEN}✓ All tests passed — database.py is ready.{RESET}\n") else: print(f"\n {RED}Fix the failures above before proceeding.{RESET}\n") print("="*55 + "\n")