Spaces:

ResumeSelectors
/

resume-analyzer

Sleeping

mohit kumar
Initial Resume Analyzer deploy
a4db4ad 5 months ago
28.5 kB
# =============================================================================
# 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")