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"] )