lib/domain.py

"""
lib/domain.py — V5 Dynamic Domain Taxonomy

Builds a 3-tier skill taxonomy dynamically from the JD understanding.
No hardcoded "search_spec_score" — the domain engine derives importance
weights from what the JD actually asks for.

Tier 1: Skills explicitly mentioned in "absolutely need" / required section
Tier 2: Skills mentioned in "nice to have" / preferred section
Tier 3: Skills that are adjacent/related but not mentioned in JD

This drives:
  - skill_coverage scoring (Tier 1 weighted more than Tier 2)
  - domain_specialization scoring (how deep in the JD's domain)
  - evidence extraction prioritization (look for Tier 1 evidence first)
"""
from __future__ import annotations
from lib.jd_parser import JDUnderstanding, get_jd


from dataclasses import dataclass, field


@dataclass
class DomainTaxonomy:
    """Dynamic 3-tier taxonomy derived from JD."""
    tier1: dict[str, list[str]] = field(default_factory=dict)
    tier2: dict[str, list[str]] = field(default_factory=dict)
    tier3: dict[str, list[str]] = field(default_factory=dict)
    skill_tier: dict[str, int] = field(default_factory=dict)
    skill_domain: dict[str, str] = field(default_factory=dict)
    domain_importance: dict[str, float] = field(default_factory=dict)


class DomainEngine:
    """
    Builds a dynamic domain taxonomy from JD understanding.

    The engine takes the parsed JD and creates a 3-tier skill taxonomy
    that drives all downstream scoring.
    """

    # Adjacency map: which domains are related to which
    DOMAIN_ADJACENCY = {
        "search": ["ranking", "embeddings", "vector_db", "nlp"],
        "ranking": ["search", "embeddings", "evaluation", "nlp"],
        "embeddings": ["vector_db", "llm", "nlp", "search"],
        "vector_db": ["embeddings", "search", "infrastructure"],
        "llm": ["embeddings", "nlp", "engineering"],
        "evaluation": ["ranking", "search", "llm"],
        "engineering": ["infrastructure", "llm", "embeddings"],
        "infrastructure": ["engineering", "vector_db"],
        "hr_tech": ["ranking", "search"],
        "open_source": [],
        "nlp": ["search", "ranking", "embeddings", "llm"],
        "cv": [],
        "speech": [],
        "robotics": [],
    }

    # Extended skill lists for Tier 3 (related but not in JD)
    EXTENDED_SKILLS = {
        "search": ["lucene", "solr", "query understanding", "spell correction",
                    "facet", "filter", "re-ranking", "query expansion",
                    "relevance feedback", "learning to rank", "click model"],
        "ranking": [" ctr model", "conversion rate", "personalization",
                     "feature engineering for ranking", "position bias",
                     "counterfactual", "offline evaluation", "online evaluation"],
        "embeddings": ["word2vec", "glove", "fasttext", "cohere embedding",
                       "voyage embedding", "instructor", "colbert",
                       "cross-encoder", "bi-encoder", "late interaction"],
        "vector_db": ["hnsw", "ivf", "pq", "product quantization",
                      "ann", "approximate nearest neighbor", "index management",
                      "sharding", "replication", "consistency level"],
        "llm": ["gpt", "bert", "t5", "llama", "mistral", "transformer",
                "attention", "decoder", "encoder", "sequence to sequence",
                "instruction tuning", "rlhf", "dpo", "constitutional ai"],
        "evaluation": ["precision", "recall", "f1", "hit rate", "mrr",
                       "ndcg", "map", "coverage", "diversity", "novelty",
                       "calibration", "fairness", "A/B testing"],
        "engineering": ["ci/cd", "git", "unit test", "integration test",
                        "code review", "agile", "scrum", "technical design",
                        "api design", "rest", "graphql", "microservice",
                        "observability", "monitoring", "logging"],
        "infrastructure": ["aws", "gcp", "azure", "docker", "kubernetes",
                           "terraform", "helm", "kafka", "redis", "postgresql",
                           "mongodb", "cassandra", "elasticsearch cluster",
                           "load balancing", "auto-scaling", "cdn"],
        "hr_tech": ["ats", "applicant tracking", "resume parsing",
                     "candidate scoring", "talent pool", "sourcing",
                     "recruitment funnel", "interview scheduling"],
        "open_source": ["github", "gitlab", "pull request", "code contribution",
                        "library", "package", "pypi", "npm"],
        "nlp": ["tokenization", "lemmatization", "stemming", "pos tagging",
                "named entity recognition", "sentiment analysis", "text classification",
                "language model", "sequence labeling", "summarization",
                "translation", "question answering"],
    }

    def __init__(self, jd: JDUnderstanding | None = None):
        self.jd = jd or get_jd()
        self._taxonomy: DomainTaxonomy | None = None

    def build(self) -> DomainTaxonomy:
        """Build the 3-tier taxonomy from JD."""
        if self._taxonomy is not None:
            return self._taxonomy

        tier1 = dict(self.jd.required_skills)  # deep copy
        tier2 = dict(self.jd.preferred_skills)
        tier3: dict[str, list[str]] = {}

        # Collect all Tier 1+2 skills and domains
        all_skills = set()
        all_domains = set()
        for t in [tier1, tier2]:
            for domain, skills in t.items():
                all_skills.update(skills)
                all_domains.add(domain)

        # Build Tier 3: related skills not in Tier 1 or 2
        for domain in all_domains:
            related_domains = self.DOMAIN_ADJACENCY.get(domain, [])
            for rd in related_domains:
                for skill in self.EXTENDED_SKILLS.get(rd, []):
                    if skill.lower() not in all_skills and rd not in tier1 and rd not in tier2:
                        if rd not in tier3:
                            tier3[rd] = []
                        tier3[rd].append(skill.lower())

        # Also add extended skills for Tier 1/2 domains that weren't fully listed
        for domain in all_domains:
            for skill in self.EXTENDED_SKILLS.get(domain, []):
                if skill.lower() not in all_skills:
                    if domain not in tier3:
                        tier3[domain] = []
                    tier3[domain].append(skill.lower())

        # Build flat lookup maps
        skill_tier: dict[str, int] = {}
        skill_domain: dict[str, str] = {}

        for domain, skills in tier1.items():
            for s in skills:
                skill_tier[s] = 1
                skill_domain[s] = domain
        for domain, skills in tier2.items():
            for s in skills:
                skill_tier[s] = 2
                skill_domain[s] = domain
        for domain, skills in tier3.items():
            for s in skills:
                if s not in skill_tier:  # Don't override higher tiers
                    skill_tier[s] = 3
                    skill_domain[s] = domain

        # Domain importance: weighted by number of Tier 1 skills + bonus for Tier 2
        domain_importance: dict[str, float] = {}
        for domain, skills in tier1.items():
            domain_importance[domain] = len(skills) * 1.0
        for domain, skills in tier2.items():
            domain_importance[domain] = domain_importance.get(domain, 0) + len(skills) * 0.5

        # Normalize
        max_imp = max(domain_importance.values()) if domain_importance else 1.0
        for d in domain_importance:
            domain_importance[d] /= max_imp

        self._taxonomy = DomainTaxonomy(
            tier1=tier1, tier2=tier2, tier3=tier3,
            skill_tier=skill_tier, skill_domain=skill_domain,
            domain_importance=domain_importance,
        )
        return self._taxonomy

    def get_skill_weight(self, skill: str) -> float:
        """Get weight for a skill based on its tier. 1.0 for Tier 1, 0.5 for Tier 2, 0.1 for Tier 3."""
        tax = self.build()
        tier = tax.skill_tier.get(skill.lower(), 3)
        return {1: 1.0, 2: 0.5, 3: 0.1}.get(tier, 0.05)

    def get_all_skills_flat(self) -> list[tuple[str, int, str]]:
        """Return all skills as (skill, tier, domain) tuples."""
        tax = self.build()
        result = []
        for skill, tier in tax.skill_tier.items():
            domain = tax.skill_domain.get(skill, "unknown")
            result.append((skill, tier, domain))
        return sorted(result, key=lambda x: (x[1], x[0]))


# Singleton
def get_taxonomy() -> DomainTaxonomy:
    """Get the domain taxonomy (cached after first build)."""
    if not hasattr(get_taxonomy, "_cache"):
        engine = DomainEngine()
        get_taxonomy._cache = engine.build()
    return get_taxonomy._cache