Spaces:
Runtime error
Runtime error
| import math | |
| import re | |
| from typing import List, Tuple, Dict, Any | |
| from tech_radar.db.models import JobPosting | |
| class SemanticVectorStore: | |
| """ | |
| Lightweight, high-performance Vector & Semantic Search Engine for TechRadar-MCP across all software domains. | |
| """ | |
| def __init__(self): | |
| self.jobs: List[JobPosting] = [] | |
| self.doc_vectors: List[Dict[str, float]] = [] | |
| self.idf: Dict[str, float] = {} | |
| def _tokenize(self, text: str) -> List[str]: | |
| words = re.findall(r'\b[a-zA-Z0-9+#\.-]+\b', text.lower()) | |
| return [w for w in words if len(w) > 1] | |
| def index_jobs(self, jobs: List[JobPosting]): | |
| self.jobs = jobs | |
| self.doc_vectors = [] | |
| doc_count = len(jobs) | |
| doc_freq = {} | |
| raw_docs = [] | |
| for job in jobs: | |
| text = f"{job.title} {job.company} {job.tech_domain} {job.city} {job.area} {' '.join(job.tech_stack)} {job.requirements}" | |
| tokens = self._tokenize(text) | |
| raw_docs.append(tokens) | |
| unique_tokens = set(tokens) | |
| for token in unique_tokens: | |
| doc_freq[token] = doc_freq.get(token, 0) + 1 | |
| self.idf = { | |
| token: math.log((doc_count + 1) / (freq + 1)) + 1.0 | |
| for token, freq in doc_freq.items() | |
| } | |
| for tokens in raw_docs: | |
| tf = {} | |
| for t in tokens: | |
| tf[t] = tf.get(t, 0) + 1 | |
| length = len(tokens) or 1 | |
| vec = { | |
| term: (freq / length) * self.idf.get(term, 1.0) | |
| for term, freq in tf.items() | |
| } | |
| self.doc_vectors.append(vec) | |
| def _cosine_similarity(self, vec1: Dict[str, float], vec2: Dict[str, float]) -> float: | |
| intersection = set(vec1.keys()) & set(vec2.keys()) | |
| numerator = sum(vec1[x] * vec2[x] for x in intersection) | |
| sum1 = sum(val ** 2 for val in vec1.values()) | |
| sum2 = sum(val ** 2 for val in vec2.values()) | |
| denominator = math.sqrt(sum1) * math.sqrt(sum2) | |
| if not denominator: | |
| return 0.0 | |
| return float(numerator / denominator) | |
| def search_semantic( | |
| self, | |
| query: str, | |
| domain: str = None, | |
| city: str = None, | |
| top_k: int = 15 | |
| ) -> List[Tuple[JobPosting, float]]: | |
| if not self.doc_vectors or not self.jobs: | |
| return [] | |
| q_tokens = self._tokenize(query) | |
| tf = {} | |
| for t in q_tokens: | |
| tf[t] = tf.get(t, 0) + 1 | |
| q_length = len(q_tokens) or 1 | |
| q_vec = { | |
| term: (freq / q_length) * self.idf.get(term, 1.0) | |
| for term, freq in tf.items() | |
| } | |
| results = [] | |
| for idx, job in enumerate(self.jobs): | |
| if city and city.lower() != "all" and job.city.lower() != city.lower(): | |
| continue | |
| if domain and domain.lower() != "all" and job.tech_domain.lower() != domain.lower(): | |
| continue | |
| sim = self._cosine_similarity(q_vec, self.doc_vectors[idx]) | |
| tech_match_bonus = sum( | |
| 0.15 for t in q_tokens | |
| if any(t.lower() == stack.lower() for stack in job.tech_stack) | |
| ) | |
| final_score = round(min(1.0, sim + tech_match_bonus), 3) | |
| if final_score > 0.01: | |
| results.append((job, final_score)) | |
| results.sort(key=lambda x: x[1], reverse=True) | |
| return results[:top_k] | |
| def match_resume_to_jd(self, resume_text: str, job: JobPosting) -> float: | |
| r_tokens = self._tokenize(resume_text) | |
| jd_text = f"{job.title} {job.tech_domain} {' '.join(job.tech_stack)} {job.requirements}" | |
| jd_tokens = self._tokenize(jd_text) | |
| r_set = set(r_tokens) | |
| jd_set = set(jd_tokens) | |
| tech_matches = [t for t in job.tech_stack if any(t.lower() == r.lower() for r in r_set)] | |
| tech_ratio = len(tech_matches) / (len(job.tech_stack) or 1) | |
| common_vocab = r_set & jd_set | |
| vocab_ratio = len(common_vocab) / (len(jd_set) or 1) | |
| overall_score = (tech_ratio * 0.6) + (vocab_ratio * 0.4) | |
| return round(min(99.0, overall_score * 100.0), 1) | |