""" taxonomy_client.py ================== HTTP client that fetches the skill taxonomy from the scrapper service at startup and re-exports the same names that cv_chunker.py and job_matcher.py previously imported from the deleted cv_utils.py. Usage (processor main.py lifespan): from services.taxonomy_client import load_taxonomy load_taxonomy() # blocks until scrapper is reachable; retries with backoff All other modules just import from here as they did from cv_utils: from services.taxonomy_client import ( DEFAULT_SKILL_ALIASES, DEFAULT_CATEGORY_SKILLS, SOFT_SKILL_KEYS, SENIORITY_ORDER, _TECH_LOC_BLACKLIST, _normalize, _has_alias, _extract_skills, _infer_category, _detect_seniority, _extract_years, compute_years_from_experience, refine_seniority_with_years, _extract_cv_title, ) """ from __future__ import annotations import logging import os import re import time from datetime import datetime as _dt from typing import Any, Dict, List, Optional, Set, Tuple import httpx import sys logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # Configuration # --------------------------------------------------------------------------- _SCRAPPER_URL = os.environ.get( "TAXONOMY_SERVICE_URL", "https://eng-musa-job-scrapper.hf.space" ).rstrip("/") _TAXONOMY_ENDPOINT = f"{_SCRAPPER_URL}/taxonomy" # --------------------------------------------------------------------------- # Module-level taxonomy state (populated by load_taxonomy() at startup) # --------------------------------------------------------------------------- DEFAULT_SKILL_ALIASES: Dict[str, List[str]] = {} DEFAULT_CATEGORY_SKILLS: Dict[str, Set[str]] = {} SOFT_SKILL_KEYS: Set[str] = set() SENIORITY_PATTERNS: Dict[str, List[str]] = {} SENIORITY_ORDER: Dict[str, int] = {} _TECH_LOC_BLACKLIST: Set[str] = set() TITLE_ROLE_WORDS: List[str] = [] TITLE_PHRASE_RE = re.compile( r"\b((?:(?:full[\s\-]?stack|front[\s\-]?end|back[\s\-]?end|senior|junior|lead|" r"mid[\s\-]?level|entry[\s\-]?level|chief|head\s+of|staff|principal)?\s+)?(?:[\w\-]+\s){0,3}" r"(?:developer|engineer|designer|analyst|manager|specialist|consultant|architect|" r"programmer|researcher|officer|director|executive|scientist|trainer|accountant|" r"nurse|doctor|pharmacist|therapist|teacher|instructor|lecturer|recruiter|" r"coordinator|supervisor|administrator|technician|advisor|auditor|bookkeeper|" r"physiotherapist|midwife|clinician|representative|salesperson|buyer|dispatcher|" r"actuary|superintendent|operator))\b", re.IGNORECASE, ) # --------------------------------------------------------------------------- # Taxonomy loader (call once at processor startup) # --------------------------------------------------------------------------- def load_taxonomy( *, max_retries: int = 15, initial_delay: float = 3.0, max_delay: float = 60.0, ) -> None: """ Fetch the taxonomy from the scrapper and populate module-level globals. Retries with exponential back-off until the scrapper responds or max_retries is exhausted (in which case RuntimeError is raised). Logs progress at INFO level so it's visible in the processor console. """ global DEFAULT_SKILL_ALIASES, DEFAULT_CATEGORY_SKILLS, SOFT_SKILL_KEYS global SENIORITY_PATTERNS, SENIORITY_ORDER, _TECH_LOC_BLACKLIST delay = initial_delay last_error: Optional[Exception] = None for attempt in range(1, max_retries + 1): try: logger.info( "[Taxonomy] Attempt %d/%d — fetching %s", attempt, max_retries, _TAXONOMY_ENDPOINT, ) with httpx.Client(timeout=30.0) as client: resp = client.get(_TAXONOMY_ENDPOINT) resp.raise_for_status() data: Dict[str, Any] = resp.json() if not data.get("aliases"): raise ValueError("Taxonomy endpoint returned empty aliases") # Populate globals in-place so modules that already imported them see the updates aliases = data.get("aliases") if aliases: DEFAULT_SKILL_ALIASES.clear() DEFAULT_SKILL_ALIASES.update(aliases) cat_skills = data.get("category_skills") if cat_skills: DEFAULT_CATEGORY_SKILLS.clear() DEFAULT_CATEGORY_SKILLS.update({ cat: set(skills) for cat, skills in cat_skills.items() }) soft_keys = data.get("soft_skill_keys") if soft_keys: SOFT_SKILL_KEYS.clear() SOFT_SKILL_KEYS.update(soft_keys) sen_patterns = data.get("seniority_patterns") if sen_patterns: SENIORITY_PATTERNS.clear() SENIORITY_PATTERNS.update(sen_patterns) sen_order = data.get("seniority_order") if sen_order: SENIORITY_ORDER.clear() SENIORITY_ORDER.update(sen_order) tech_loc = data.get("tech_loc_blacklist") if tech_loc: _TECH_LOC_BLACKLIST.clear() _TECH_LOC_BLACKLIST.update(tech_loc) # Update TITLE_ROLE_WORDS from API if provided api_roles = data.get("title_role_words") if api_roles: TITLE_ROLE_WORDS.clear() TITLE_ROLE_WORDS.extend(api_roles) logger.info( "[Taxonomy] ✅ Loaded: %d skills, %d categories", len(DEFAULT_SKILL_ALIASES), len(DEFAULT_CATEGORY_SKILLS), ) return # success except Exception as exc: last_error = exc logger.warning( "[Taxonomy] ⚠️ Attempt %d failed: %s", attempt, exc ) if attempt < max_retries: logger.info( "[Taxonomy] Retrying in %.1f seconds…", delay ) time.sleep(delay) delay = min(delay * 1.5, max_delay) logger.error( "[Taxonomy] ❌ CRITICAL: Failed to load taxonomy after %d attempts. " "Last error: %s. Stopping the processor.", max_retries, last_error ) sys.exit(1) # --------------------------------------------------------------------------- # Pure normalisation helpers (logic only — no taxonomy data needed) # --------------------------------------------------------------------------- def _normalize(text: str) -> str: text = (text or "").lower() text = text.replace("/", " ") text = text.replace("_", " ") text = re.sub(r"[^a-z0-9+.#\s-]", " ", text) return re.sub(r"\s+", " ", text).strip() def _has_alias(text_norm: str, alias: str) -> bool: alias_norm = _normalize(alias) if not alias_norm: return False return re.search(rf"(? Set[str]: if aliases is None: aliases = DEFAULT_SKILL_ALIASES text_norm = _normalize(text) return { canon for canon, ali_list in aliases.items() if any(_has_alias(text_norm, a) for a in ali_list) } # --------------------------------------------------------------------------- # Category inference # --------------------------------------------------------------------------- _CATEGORY_TITLE_KEYWORDS: Dict[str, List[str]] = { "Admin & Office": ["admin", "office", "stock", "inventory", "warehouse", "cashier", "receptionist", "clerk"], "Software Engineering": ["developer", "engineer", "programmer", "frontend", "backend", "full stack", "mobile", "fullstack", "software", "devops"], "Infrastructure & Cloud": ["infrastructure", "cloud", "sre", "network", "system administrator", "sysadmin", "it administrator", "cloud engineer"], "Cybersecurity": ["security", "cybersecurity", "cyber", "infosec", "soc analyst", "penetration tester", "ethical hacker"], "Design & Creative": ["designer", "ui", "ux", "creative", "graphic", "visual", "illustrator", "photographer"], "Marketing & Growth": ["marketing", "seo", "content", "social media", "growth", "digital", "brand"], "Data & Analytics": ["data analyst", "analytics", "machine learning", "ai", "scientist", "bi analyst", "business intelligence"], "Data Engineering": ["data engineer", "etl", "data pipeline", "data architect", "analytics engineer"], "Finance & Accounting": ["accountant", "finance", "accounting", "auditor", "bookkeeper", "treasurer", "payroll", "tax", "financial"], "Human Resources": ["hr", "human resource", "recruitment", "recruiter", "talent", "people", "hris"], "Sales & Business Dev": ["sales", "business development", "account manager", "sales executive", "representative"], "Healthcare": ["nurse", "doctor", "clinical", "medical", "pharmacy", "patient", "health", "therapist", "midwife"], "Legal": ["legal", "lawyer", "attorney", "advocate", "counsel", "paralegal", "solicitor"], "Education & Training": ["teacher", "lecturer", "tutor", "instructor", "educator", "academic", "trainer", "professor"], "Operations & Logistics": ["logistics", "supply chain", "procurement", "operations", "fleet", "warehouse manager", "quality"], "Customer Support": ["customer support", "help desk", "call centre", "contact centre", "customer care"], "Construction & Engineering": ["civil", "structural", "mechanical", "electrical", "quantity surveyor", "site engineer", "construction"], "Hospitality & Property": ["hotel", "hospitality", "property", "real estate", "housekeeping", "f&b", "restaurant"], } def _infer_category( text: str, skills: Set[str], category_skills: Optional[Dict[str, Set[str]]] = None, ) -> str: if category_skills is None: category_skills = DEFAULT_CATEGORY_SKILLS text_norm = _normalize(text) best, best_score = "Other", 0.0 for cat, cat_skills in category_skills.items(): skill_overlap = len(skills & cat_skills) kw_hits = sum( 1 for kw in _CATEGORY_TITLE_KEYWORDS.get(cat, []) if kw in text_norm ) score = skill_overlap * 2.0 + kw_hits * 1.25 if score > best_score: best_score, best = score, cat return best # --------------------------------------------------------------------------- # Years-of-experience extraction # --------------------------------------------------------------------------- def _extract_years(text: str) -> int: text_norm = _normalize(text) patterns = [ r"(\d+)\+?\s*(?:years|year|yrs|yr)\s*(?:of)?\s*(?:experience|exp)", r"experience\s*(?:of)?\s*(\d+)\+?\s*(?:years|year|yrs|yr)", r"(\d+)\+?\s*(?:years|year|yrs|yr)\s+(?:in\s+)?(?:the\s+)?(?:industry|field|software|development|practice|profession)", ] years = [int(m.group(1)) for p in patterns for m in re.finditer(p, text_norm)] return max(years) if years else 0 # --------------------------------------------------------------------------- # Years from parsed experience date ranges # --------------------------------------------------------------------------- _MONTH_SHORT: Dict[str, int] = { "jan": 1, "feb": 2, "mar": 3, "apr": 4, "may": 5, "jun": 6, "jul": 7, "aug": 8, "sep": 9, "oct": 10, "nov": 11, "dec": 12, } def _parse_date_token(token: str) -> Optional[_dt]: token = token.strip().lower() if re.match(r"present|current|now|ongoing|till\s*date|to\s*date", token): return _dt.now() m = re.match(r"([a-z]+)\s+(\d{4})", token) if m: month = _MONTH_SHORT.get(m.group(1)[:3], 1) try: return _dt(int(m.group(2)), month, 1) except ValueError: return None m = re.match(r"(\d{4})", token) if m: return _dt(int(m.group(1)), 6, 1) return None def compute_years_from_experience(entries: List[Dict[str, Any]]) -> int: """Sum date-range durations across all experience entries → total years.""" total_months = 0 for entry in entries: period = (entry.get("period") or "").strip() if not period: continue parts = re.split(r"\s*[-–—]\s*", period, maxsplit=1) if len(parts) != 2: continue start = _parse_date_token(parts[0]) end = _parse_date_token(parts[1]) if start and end and end > start: months = (end.year - start.year) * 12 + (end.month - start.month) total_months += max(0, months) return total_months // 12 # --------------------------------------------------------------------------- # Seniority detection # --------------------------------------------------------------------------- def _detect_seniority(text: str, *, is_cv: bool = False) -> str: """Detect seniority: title zone first, then body (guards against false positives from advice/boilerplate text in job descriptions).""" title_zone = text[:300] title_norm = _normalize(title_zone) text_norm = _normalize(text) order = ["senior", "junior", "intern", "mid"] if is_cv else ["senior", "intern", "junior", "mid"] for level in order: if any(_has_alias(title_norm, term) for term in SENIORITY_PATTERNS[level]): return level for level in order: if level == "senior" and not is_cv: continue # already checked title zone; skip body re-check to avoid boilerplate inflation if any(_has_alias(text_norm, term) for term in SENIORITY_PATTERNS[level]): return level years = _extract_years(text) if years >= 7: return "senior" if years >= 3: return "mid" return "junior" if is_cv else "mid" def refine_seniority_with_years(detected: str, years: int) -> str: if detected == "mid": if years >= 7: return "senior" if years >= 3: return "mid" if 0 < years < 2: return "junior" return detected if detected == "intern" and years >= 2: if years >= 7: return "senior" if years >= 3: return "mid" return "junior" return detected # --------------------------------------------------------------------------- # CV title extractor # --------------------------------------------------------------------------- def _extract_cv_title(cv_text: str, clean_line_fn=None) -> str: def _simple_clean(raw: str) -> str: line = re.sub(r"^#+\s*", "", raw) line = re.sub(r"\*{1,3}|\_{1,2}|`", "", line) return re.sub(r"\s+", " ", line).strip() clean = clean_line_fn or _simple_clean clean_lines = [clean(x) for x in cv_text.splitlines() if x.strip()] _role_wb_re = re.compile( r"\b(" + "|".join(re.escape(w) for w in TITLE_ROLE_WORDS) + r")\b", re.IGNORECASE, ) for line in clean_lines[1:30]: stripped = line.strip("| \t") if ( len(stripped) <= 80 and bool(_role_wb_re.search(stripped)) and "|" not in stripped and not re.search(r"[@http]", stripped) and not re.search(r"\d{4}\s*[-–]\s*(?:\d{4}|present)", stripped, re.IGNORECASE) ): return stripped.title() _SECTION_HDR_RE = re.compile( r"^(career\s*summary|professional\s*summary|executive\s*summary|" r"personal\s*statement|career\s*objective|professional\s*profile|" r"personal\s*profile|about\s*me|work\s*experience|" r"professional\s*experience|technical\s*skills?|core\s*skills?|" r"key\s*skills?|educational?\s*background|academic\s*background|" r"employment\s*history|career\s*history|skills?\s*&?\s*tools?|" r"skills?|summary|profile|experience|education|projects?|awards?|" r"achievements?|contact|links?|certifications?)$", re.IGNORECASE, ) non_header = [ ln for ln in clean_lines[:40] if not _SECTION_HDR_RE.match(re.sub(r"[^a-zA-Z\s&]", " ", ln).strip()) ] head = ". ".join(non_header)[:1000] m = TITLE_PHRASE_RE.search(head) if m: return m.group(1).strip().title() return "Unknown"