services/taxonomy_client.py

"""
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"(?<![a-z0-9]){re.escape(alias_norm)}(?![a-z0-9])", text_norm) is not None


# ---------------------------------------------------------------------------
# Skill extraction
# ---------------------------------------------------------------------------

def _extract_skills(
    text: str,
    aliases: Optional[Dict[str, List[str]]] = None,
) -> 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"