src/skill_ner.py

"""
skill_ner.py
-----------
Named Entity Recognition for skills using pattern matching and rule-based extraction.

Features:
  - Advanced skill taxonomy with domains
  - Contextual skill detection (avoid false positives)
  - Skill confidence scores
  - Skill evolution tracking

Author: SmartHire AI
"""

import logging
import re
from typing import Dict, List, Set, Tuple

logger = logging.getLogger(__name__)

# Skill domains with context
SKILL_DOMAINS = {
    "backend": {
        "skills": ["python", "java", "go", "rust", "node.js", "flask", "django", "fastapi", "spring"],
        "frameworks": ["fastapi", "flask", "django", "spring boot"],
        "keywords": ["api", "rest", "backend", "server"],
    },
    "frontend": {
        "skills": ["javascript", "typescript", "react", "vue", "angular", "css", "html"],
        "frameworks": ["react", "vue", "angular", "svelte"],
        "keywords": ["ui", "ux", "frontend", "client", "web app"],
    },
    "data_science": {
        "skills": ["python", "r", "sql", "pandas", "scikit-learn", "tensorflow", "pytorch"],
        "frameworks": ["tensorflow", "pytorch", "keras", "scikit-learn"],
        "keywords": ["data", "model", "analysis", "statistics", "ml"],
    },
    "devops": {
        "skills": ["docker", "kubernetes", "ci/cd", "terraform", "aws", "gcp", "azure"],
        "frameworks": ["docker", "kubernetes", "terraform"],
        "keywords": ["deployment", "infrastructure", "cloud", "monitoring"],
    },
}


def detect_skill_context(text: str, skill: str, window: int = 50) -> str:
    """
    Extract context window around skill mention.
    
    Returns:
        Text snippet containing the skill with surrounding context.
    """
    pattern = rf'\b{re.escape(skill)}\b'
    matches = list(re.finditer(pattern, text, re.IGNORECASE))
    
    if not matches:
        return ""
    
    match = matches[0]
    start = max(0, match.start() - window)
    end = min(len(text), match.end() + window)
    return text[start:end]


def get_skill_confidence(skill: str, context: str, domain: str = None) -> float:
    """
    Compute confidence score for a skill detection (0-1).
    
    Factors:
      - Frequency in context
      - Domain relevance
      - Surrounding keywords
    """
    confidence = 0.5  # Base
    
    # Frequency boost
    count = len(re.findall(rf'\b{re.escape(skill)}\b', context, re.IGNORECASE))
    confidence += min(0.3, count * 0.1)
    
    # Domain boost
    if domain and domain in SKILL_DOMAINS:
        if skill.lower() in [s.lower() for s in SKILL_DOMAINS[domain].get("skills", [])]:
            confidence += 0.1
        
        domain_keywords = SKILL_DOMAINS[domain].get("keywords", [])
        if any(kw.lower() in context.lower() for kw in domain_keywords):
            confidence += 0.1
    
    return min(1.0, confidence)


def extract_skills_with_confidence(
    text: str,
    skill_list: List[str],
    domain: str = None,
) -> List[Tuple[str, float]]:
    """
    Extract skills with confidence scores.
    
    Returns:
        List of (skill, confidence) tuples sorted by confidence.
    """
    results = []
    seen = set()
    
    for skill in skill_list:
        if skill.lower() in seen:
            continue
        seen.add(skill.lower())
        
        if re.search(rf'\b{re.escape(skill)}\b', text, re.IGNORECASE):
            context = detect_skill_context(text, skill)
            confidence = get_skill_confidence(skill, context, domain)
            results.append((skill, round(confidence, 2)))
    
    results.sort(key=lambda x: x[1], reverse=True)
    return results


def extract_domain_skills(text: str) -> Dict[str, List[Tuple[str, float]]]:
    """
    Extract skills organized by domain.
    
    Returns:
        {
            "backend": [("python", 0.95), ("flask", 0.88), ...],
            "frontend": [...],
            ...
        }
    """
    domain_skills = {}
    
    for domain, domain_config in SKILL_DOMAINS.items():
        skills = extract_skills_with_confidence(
            text,
            domain_config["skills"],
            domain=domain
        )
        if skills:
            domain_skills[domain] = skills
    
    return domain_skills


def track_skill_evolution(
    historical_extractions: List[Dict],
) -> Dict:
    """
    Analyze skill acquisition over time (if timestamps available).
    
    Args:
        historical_extractions: List of {"timestamp", "skills"} dicts.
    
    Returns:
        {
            "new_skills": [...],
            "mastered_skills": [...],
            "declining_skills": [...],
            "trends": {...}
        }
    """
    if len(historical_extractions) < 2:
        return {"status": "insufficient_data"}
    
    earliest = set(s[0] for s in historical_extractions[0].get("skills", []))
    latest = set(s[0] for s in historical_extractions[-1].get("skills", []))
    
    new_skills = latest - earliest
    lost_skills = earliest - latest
    stable_skills = latest & earliest
    
    return {
        "new_skills": sorted(list(new_skills)),
        "mastered_skills": sorted(list(stable_skills)),
        "declining_skills": sorted(list(lost_skills)),
        "total_skill_growth": len(new_skills),
        "skill_retention": len(stable_skills) / len(earliest) if earliest else 0.0,
    }


def get_skill_recommendations(
    current_skills: List[str],
    target_domain: str,
) -> Dict:
    """
    Recommend skills to acquire based on target domain.
    """
    if target_domain not in SKILL_DOMAINS:
        return {"error": f"Unknown domain: {target_domain}"}
    
    target_skills = set(SKILL_DOMAINS[target_domain]["skills"])
    current_set = set(s.lower() for s in current_skills)
    
    missing = [s for s in target_skills if s.lower() not in current_set]
    complementary = []
    
    # Find complementary skills from other domains
    for other_domain, config in SKILL_DOMAINS.items():
        if other_domain != target_domain:
            overlap = set(config["skills"]) & current_set
            if overlap:
                complementary.extend(list(set(config["skills"]) - current_set)[:2])
    
    return {
        "target_domain": target_domain,
        "skill_gaps": missing[:10],
        "complementary_skills": list(set(complementary))[:10],
        "priority_level": "critical" if len(missing) > 5 else "moderate" if len(missing) > 2 else "low",
    }