tech_radar/db/database.py

import sqlite3
import json
import os
from typing import List, Optional, Dict, Any
from tech_radar.db.models import JobPosting

class DatabaseManager:
    """Thread-safe SQLite database manager for TechRadar-MCP across all domains & cities."""

    def __init__(self, db_path: str = "tech_radar.db"):
        self.db_path = db_path
        self._init_db()

    def _get_connection(self) -> sqlite3.Connection:
        conn = sqlite3.connect(self.db_path)
        conn.row_factory = sqlite3.Row
        return conn

    def _init_db(self):
        with self._get_connection() as conn:
            cursor = conn.cursor()
            cursor.execute("""
                CREATE TABLE IF NOT EXISTS job_postings (
                    id TEXT PRIMARY KEY,
                    title TEXT NOT NULL,
                    company TEXT NOT NULL,
                    tech_domain TEXT NOT NULL,
                    city TEXT NOT NULL,
                    area TEXT NOT NULL,
                    salary_min_lpa REAL NOT NULL,
                    salary_max_lpa REAL NOT NULL,
                    experience_min_years INTEGER NOT NULL,
                    experience_max_years INTEGER NOT NULL,
                    tech_stack TEXT NOT NULL, -- JSON List
                    requirements TEXT NOT NULL,
                    work_mode TEXT NOT NULL,
                    company_tier TEXT NOT NULL,
                    posted_date TEXT NOT NULL,
                    url TEXT
                )
            """)
            conn.commit()

    def save_job_posting(self, job: JobPosting) -> bool:
        with self._get_connection() as conn:
            cursor = conn.cursor()
            cursor.execute("""
                INSERT OR REPLACE INTO job_postings (
                    id, title, company, tech_domain, city, area, salary_min_lpa, salary_max_lpa,
                    experience_min_years, experience_max_years, tech_stack, requirements,
                    work_mode, company_tier, posted_date, url
                ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
            """, (
                job.id, job.title, job.company, job.tech_domain, job.city, job.area,
                job.salary_min_lpa, job.salary_max_lpa,
                job.experience_min_years, job.experience_max_years,
                json.dumps(job.tech_stack), job.requirements,
                job.work_mode, job.company_tier, job.posted_date, job.url
            ))
            conn.commit()
            return True

    def get_job_by_id(self, job_id: str) -> Optional[JobPosting]:
        with self._get_connection() as conn:
            cursor = conn.cursor()
            cursor.execute("SELECT * FROM job_postings WHERE id = ?", (job_id,))
            row = cursor.fetchone()
            if not row:
                return None
            return self._row_to_job(row)

    def search_jobs(
        self,
        domain: Optional[str] = None,
        city: Optional[str] = None,
        query: Optional[str] = None,
        experience_level: Optional[int] = None,
        min_salary_lpa: Optional[float] = None,
        tech_stack_filter: Optional[List[str]] = None,
        limit: int = 100
    ) -> List[JobPosting]:
        with self._get_connection() as conn:
            cursor = conn.cursor()
            sql = "SELECT * FROM job_postings WHERE 1=1"
            params = []

            if domain and domain.lower() != "all":
                sql += " AND LOWER(tech_domain) = LOWER(?)"
                params.append(domain)

            if city and city.lower() != "all":
                sql += " AND LOWER(city) = LOWER(?)"
                params.append(city)

            if experience_level is not None:
                sql += " AND experience_min_years <= ? AND experience_max_years >= ?"
                params.extend([experience_level, experience_level])

            if min_salary_lpa is not None:
                sql += " AND salary_max_lpa >= ?"
                params.append(min_salary_lpa)

            if query:
                sql += " AND (LOWER(title) LIKE LOWER(?) OR LOWER(company) LIKE LOWER(?) OR LOWER(requirements) LIKE LOWER(?) OR LOWER(area) LIKE LOWER(?))"
                q = f"%{query}%"
                params.extend([q, q, q, q])

            sql += " ORDER BY salary_max_lpa DESC LIMIT ?"
            params.append(limit)

            cursor.execute(sql, params)
            rows = cursor.fetchall()
            jobs = [self._row_to_job(r) for r in rows]

            if tech_stack_filter:
                filter_set = {s.lower() for s in tech_stack_filter}
                jobs = [
                    j for j in jobs
                    if any(ts.lower() in filter_set for ts in j.tech_stack)
                ]

            return jobs

    def get_all_jobs(self) -> List[JobPosting]:
        return self.search_jobs(limit=500)

    def get_market_analytics(self, city: str = "All", domain: str = "All") -> Dict[str, Any]:
        with self._get_connection() as conn:
            cursor = conn.cursor()
            sql = "SELECT COUNT(*), AVG((salary_min_lpa + salary_max_lpa) / 2.0), MIN(salary_min_lpa), MAX(salary_max_lpa) FROM job_postings WHERE 1=1"
            params = []

            if city and city.lower() != "all":
                sql += " AND LOWER(city) = LOWER(?)"
                params.append(city)

            if domain and domain.lower() != "all":
                sql += " AND LOWER(tech_domain) = LOWER(?)"
                params.append(domain)

            cursor.execute(sql, params)
            count, avg_sal, min_sal, max_sal = cursor.fetchone()

            if not count or count == 0:
                return {
                    "city": city,
                    "domain": domain,
                    "total_jobs": 0,
                    "avg_salary_lpa": 0,
                    "salary_range": "N/A",
                    "top_frameworks": [],
                    "top_hubs": [],
                    "top_companies": []
                }

            sql_details = "SELECT tech_stack, area, company FROM job_postings WHERE 1=1"
            params_details = []
            if city and city.lower() != "all":
                sql_details += " AND LOWER(city) = LOWER(?)"
                params_details.append(city)
            if domain and domain.lower() != "all":
                sql_details += " AND LOWER(tech_domain) = LOWER(?)"
                params_details.append(domain)

            cursor.execute(sql_details, params_details)
            rows = cursor.fetchall()

            skill_counts = {}
            hub_counts = {}
            company_set = set()

            for r in rows:
                stacks = json.loads(r["tech_stack"])
                for s in stacks:
                    skill_counts[s] = skill_counts.get(s, 0) + 1
                area = r["area"]
                hub_counts[area] = hub_counts.get(area, 0) + 1
                company_set.add(r["company"])

            sorted_skills = [
                {"skill": k, "count": v, "percentage": round((v / count) * 100, 1)}
                for k, v in sorted(skill_counts.items(), key=lambda x: x[1], reverse=True)[:10]
            ]
            sorted_hubs = [
                {"hub": k, "count": v}
                for k, v in sorted(hub_counts.items(), key=lambda x: x[1], reverse=True)[:5]
            ]

            return {
                "city": city,
                "domain": domain,
                "total_jobs": count,
                "avg_salary_lpa": round(avg_sal or 0, 1),
                "salary_range": f"₹{int(min_sal or 0)}L - ₹{int(max_sal or 0)}L PA",
                "top_frameworks": sorted_skills,
                "top_hubs": sorted_hubs,
                "top_companies": list(company_set)[:10]
            }

    def _row_to_job(self, row: sqlite3.Row) -> JobPosting:
        return JobPosting(
            id=row["id"],
            title=row["title"],
            company=row["company"],
            tech_domain=row["tech_domain"],
            city=row["city"],
            area=row["area"],
            salary_min_lpa=row["salary_min_lpa"],
            salary_max_lpa=row["salary_max_lpa"],
            experience_min_years=row["experience_min_years"],
            experience_max_years=row["experience_max_years"],
            tech_stack=json.loads(row["tech_stack"]),
            requirements=row["requirements"],
            work_mode=row["work_mode"],
            company_tier=row["company_tier"],
            posted_date=row["posted_date"],
            url=row["url"]
        )