src/utils.py

"""
Shared constants, helpers, and data loaders for the runtime pipeline.
Reference date: 2026-06-10.
"""

import json
import math
import re
from datetime import date
from pathlib import Path
from typing import Iterator

REFERENCE_DATE = date(2026, 6, 10)

# --------------------------------------------------------------------------- #
# Company lists
# --------------------------------------------------------------------------- #

IT_SERVICES: set[str] = {
    "tcs", "tata consultancy", "infosys", "wipro", "accenture", "cognizant",
    "capgemini", "hcl", "hcl technologies", "tech mahindra", "mphasis",
    "hexaware", "ltimindtree", "mindtree", "persistent", "persistent systems",
    "coforge", "birlasoft", "niit technologies", "l&t infotech",
    "lti", "mastech", "kpit", "sonata software",
}

# Fictional companies used as honeypot markers (with founding year)
FICTIONAL_COMPANIES: dict[str, int] = {
    "pied piper": 2014,
    "initech": 1999,
    "wayne enterprises": 1939,
    "stark industries": 1940,
    "dunder mifflin": 1949,
    "umbrella corporation": 1968,
    "soylent": 2010,
    "globex": 1996,
}

# --------------------------------------------------------------------------- #
# JD skill lists (used by feature_engineering.py)
# --------------------------------------------------------------------------- #

JD_CORE_SKILLS: list[str] = [
    "embeddings", "retrieval", "vector", "ranking", "search",
    "sentence-transformers", "sentence_transformers", "faiss", "elasticsearch",
    "pinecone", "weaviate", "ndcg", "evaluation", "a/b", "python",
    "llm", "fine-tuning", "finetuning", "lora", "recommendation",
    "nlp", "ir", "information retrieval",
]

JD_NICE_SKILLS: list[str] = [
    "xgboost", "learning-to-rank", "ltr", "distributed systems",
    "inference optimization", "open source", "hr tech", "recruitment",
    "qdrant", "milvus", "opensearch",
]

# City names mapped for preferred-city matching
PREFERRED_CITIES: list[str] = [
    "pune", "noida", "hyderabad", "mumbai", "delhi", "bangalore", "bengaluru",
    "gurgaon", "gurugram",
]

# Production-action words used for production_signal_count
PRODUCTION_KEYWORDS: list[str] = [
    "deployed", "deploy", "productionized", "shipped", "ship",
    "scaled", "scale", "served", "serving", "production", "prod",
    "users", "latency", "throughput", "real-time", "realtime", "live",
    "live traffic", "a/b test", "a/b tested", "rollout", "launched",
    "in production", "at scale", "millions", "billion", "qps", "rps",
    "uptime", "reliability", "serving infrastructure",
]

# --------------------------------------------------------------------------- #
# Education tier mapping
# --------------------------------------------------------------------------- #

EDU_TIER_SCORES: dict[str, int] = {
    "tier_1": 4,
    "tier_2": 3,
    "tier_3": 2,
    "tier_4": 1,
}

STEM_FIELDS: set[str] = {
    "computer science", "cs", "cse", "information technology", "it",
    "electronics", "ece", "electrical", "eee", "mathematics", "statistics",
    "physics", "data science", "machine learning", "artificial intelligence",
    "ai", "ml", "information systems", "software engineering", "software",
}

# --------------------------------------------------------------------------- #
# Company-size ordinal encoding
# --------------------------------------------------------------------------- #

COMPANY_SIZE_ORDINAL: dict[str, int] = {
    "1-10": 1,
    "11-50": 2,
    "51-200": 3,
    "201-500": 4,
    "501-1000": 5,
    "1001-5000": 6,
    "5001-10000": 7,
    "10001+": 8,
}

# --------------------------------------------------------------------------- #
# Date / duration helpers
# --------------------------------------------------------------------------- #

def days_since(date_str: str, ref: date = REFERENCE_DATE) -> int:
    """Return days between date_str (ISO) and REFERENCE_DATE. Negative = future."""
    try:
        d = date.fromisoformat(date_str)
        return (ref - d).days
    except (ValueError, TypeError):
        return 0


def parse_year(date_str: str | None) -> int | None:
    """Extract year from an ISO date string or bare year string."""
    if not date_str:
        return None
    try:
        return int(str(date_str)[:4])
    except (ValueError, TypeError):
        return None


def notice_penalty(notice_days: int) -> float:
    """
    0 if notice_period_days <= 30.
    Linear decay: 0.5 at 90d, 0.0 at 180d+.
    """
    if notice_days <= 30:
        return 0.0
    if notice_days >= 180:
        return 1.0  # max penalty
    # linear between 30→90: 0→0.5; 90→180: 0.5→1.0
    return min(1.0, (notice_days - 30) / 150.0)


def yoe_fit_score(yoe: float) -> float:
    """
    Gaussian-like score peaked at [6, 8] years.
    - Below 4: steep penalty
    - 4–6: ramp up
    - 6–8: peak (1.0)
    - 8–12: soft decay
    - Above 12: harder decay
    """
    if yoe < 2:
        return 0.0
    if yoe <= 6:
        return max(0.0, (yoe - 2) / 4.0)
    if yoe <= 8:
        return 1.0
    if yoe <= 12:
        return max(0.5, 1.0 - (yoe - 8) / 8.0)
    return max(0.2, 0.5 - (yoe - 12) / 20.0)


# --------------------------------------------------------------------------- #
# Text helpers
# --------------------------------------------------------------------------- #

def career_text(candidate: dict) -> str:
    """Concatenate all career description text, lowercased."""
    parts = [candidate["profile"].get("summary", ""),
             candidate["profile"].get("headline", "")]
    for job in candidate.get("career_history", []):
        parts.append(job.get("description", ""))
    return " ".join(parts).lower()


def tokenize(text: str) -> list[str]:
    """Lowercase alphanumeric tokenization — shared by BM25 build (offline + runtime)."""
    return re.findall(r"[a-z0-9]+", text.lower())


def candidate_to_chunks(c: dict) -> list[str]:
    """
    Text chunks fed to the dense embedder, one vector per chunk.

    SINGLE SOURCE OF TRUTH — used by both offline.precompute_embeddings (training)
    and src.runtime_index (rank time). The XGBoost model is trained on semantic
    features derived from these exact chunks, so offline and runtime MUST chunk
    identically or the feature distribution shifts and the model degrades.
    """
    chunks: list[str] = []
    for job in c.get("career_history", []):
        if job.get("description"):
            chunks.append(f"{job['title']} at {job['company']}: {job['description']}")
    if c["profile"].get("summary"):
        chunks.append(c["profile"]["summary"])
    if c["profile"].get("headline"):
        chunks.append(c["profile"]["headline"])
    return chunks or [c["profile"].get("current_title", "")]


def candidate_to_bm25_text(c: dict) -> str:
    """Concatenated text indexed by BM25. Shared by offline + runtime BM25 build."""
    parts = [
        c["profile"].get("headline", ""),
        c["profile"].get("summary", ""),
        c["profile"].get("current_title", ""),
    ]
    for job in c.get("career_history", []):
        parts.append(job.get("title", ""))
        parts.append(job.get("description", ""))
    for s in c.get("skills", []):
        parts.append(s.get("name", ""))
    return " ".join(parts)


def is_it_services_company(company_name: str) -> bool:
    name_lower = company_name.lower()
    return any(svc in name_lower for svc in IT_SERVICES)


def company_size_ordinal(size_str: str | None) -> int:
    return COMPANY_SIZE_ORDINAL.get(size_str or "", 0)


# --------------------------------------------------------------------------- #
# JSONL streaming loader (memory-efficient for 100K candidates)
# --------------------------------------------------------------------------- #

def stream_candidates(path: str | Path) -> Iterator[dict]:
    """Yield one candidate dict per line without loading the full file."""
    with open(path, encoding="utf-8") as f:
        for line in f:
            line = line.strip()
            if line:
                yield json.loads(line)


def load_candidates_json(path: str | Path) -> list[dict]:
    """Load sample_candidates.json (list format, not JSONL)."""
    with open(path, encoding="utf-8") as f:
        return json.load(f)