Talent Acquisition

#!/usr/bin/env python3
"""
Candidate Pipeline Tracker - Track and analyze recruiting funnel metrics.

Reads a CSV of candidate pipeline data and computes stage-by-stage conversion
rates, time-in-stage metrics, source effectiveness, and bottleneck identification.

Usage:
    python candidate_pipeline_tracker.py --file pipeline.csv
    python candidate_pipeline_tracker.py --file pipeline.csv --json

Input CSV columns:
    candidate_id    - Unique candidate identifier
    role            - Job title or requisition name
    source          - Sourcing channel (e.g., LinkedIn, Referral, Job Board)
    stage           - Current pipeline stage (Applied, Screened, Interviewed, Offered, Accepted, Rejected, Withdrawn)
    stage_date      - Date the candidate entered this stage (YYYY-MM-DD)
    applied_date    - Date of initial application (YYYY-MM-DD)

Output: Funnel metrics, conversion rates, source effectiveness, and bottleneck analysis.
"""

import argparse
import csv
import json
import os
import sys
from collections import defaultdict
from datetime import datetime


STAGE_ORDER = ["Applied", "Screened", "Interviewed", "Offered", "Accepted"]
TERMINAL_STAGES = ["Rejected", "Withdrawn"]

BENCHMARKS = {
    "Applied_to_Screened": (0.40, 0.50),
    "Screened_to_Interviewed": (0.30, 0.40),
    "Interviewed_to_Offered": (0.15, 0.25),
    "Offered_to_Accepted": (0.80, 0.90),
}


def read_csv(path: str) -> list:
    """Read CSV file and return list of dicts."""
    if not os.path.isfile(path):
        print(f"Error: File not found: {path}", file=sys.stderr)
        sys.exit(1)
    with open(path, "r", encoding="utf-8") as f:
        reader = csv.DictReader(f)
        rows = list(reader)
    required = {"candidate_id", "stage", "applied_date"}
    if rows:
        missing = required - set(rows[0].keys())
        if missing:
            print(f"Error: Missing required columns: {', '.join(missing)}", file=sys.stderr)
            sys.exit(1)
    return rows


def parse_date(date_str: str) -> datetime:
    """Parse date string to datetime."""
    if not date_str:
        return None
    for fmt in ("%Y-%m-%d", "%m/%d/%Y", "%d/%m/%Y"):
        try:
            return datetime.strptime(date_str.strip(), fmt)
        except ValueError:
            continue
    return None


def get_candidate_max_stage(rows: list) -> dict:
    """Get the furthest stage each candidate reached."""
    candidates = defaultdict(lambda: {"max_stage_idx": -1, "stage": "Unknown", "source": "", "role": "", "applied_date": None, "stage_date": None})

    for row in rows:
        cid = row["candidate_id"]
        stage = row.get("stage", "").strip()
        source = row.get("source", "").strip()
        role = row.get("role", "").strip()
        applied_date = parse_date(row.get("applied_date", ""))
        stage_date = parse_date(row.get("stage_date", ""))

        if stage in STAGE_ORDER:
            idx = STAGE_ORDER.index(stage)
        elif stage in TERMINAL_STAGES:
            idx = -1  # terminal, but track separately
        else:
            idx = -1

        if source:
            candidates[cid]["source"] = source
        if role:
            candidates[cid]["role"] = role
        if applied_date:
            candidates[cid]["applied_date"] = applied_date

        if stage in TERMINAL_STAGES:
            candidates[cid]["terminal"] = stage
            if stage_date:
                candidates[cid]["terminal_date"] = stage_date

        if idx > candidates[cid]["max_stage_idx"]:
            candidates[cid]["max_stage_idx"] = idx
            candidates[cid]["stage"] = stage
            if stage_date:
                candidates[cid]["stage_date"] = stage_date

    return candidates


def compute_funnel(candidates: dict) -> dict:
    """Compute funnel metrics."""
    stage_counts = defaultdict(int)
    for cid, data in candidates.items():
        idx = data["max_stage_idx"]
        # Count each candidate at their max stage AND all previous stages
        for i in range(idx + 1):
            stage_counts[STAGE_ORDER[i]] += 1

    funnel = []
    for i, stage in enumerate(STAGE_ORDER):
        count = stage_counts.get(stage, 0)
        entry = {"stage": stage, "count": count}

        if i > 0:
            prev_count = stage_counts.get(STAGE_ORDER[i - 1], 0)
            if prev_count > 0:
                rate = count / prev_count
                entry["conversion_rate"] = round(rate, 3)
                key = f"{STAGE_ORDER[i-1]}_to_{stage}"
                bench = BENCHMARKS.get(key)
                if bench:
                    entry["benchmark_range"] = f"{bench[0]*100:.0f}-{bench[1]*100:.0f}%"
                    if rate < bench[0]:
                        entry["status"] = "BELOW_BENCHMARK"
                    elif rate > bench[1]:
                        entry["status"] = "ABOVE_BENCHMARK"
                    else:
                        entry["status"] = "ON_TARGET"

        funnel.append(entry)

    return funnel


def compute_source_effectiveness(candidates: dict) -> list:
    """Compute source channel effectiveness."""
    source_data = defaultdict(lambda: {"applied": 0, "screened": 0, "interviewed": 0, "offered": 0, "accepted": 0})

    for cid, data in candidates.items():
        source = data.get("source", "Unknown") or "Unknown"
        idx = data["max_stage_idx"]
        for i in range(idx + 1):
            stage_key = STAGE_ORDER[i].lower()
            source_data[source][stage_key] += 1

    results = []
    for source, counts in sorted(source_data.items()):
        applied = counts["applied"]
        if applied == 0:
            continue
        entry = {
            "source": source,
            "applied": applied,
            "screened": counts["screened"],
            "accepted": counts["accepted"],
            "overall_conversion": round(counts["accepted"] / applied, 3) if applied > 0 else 0,
            "screen_rate": round(counts["screened"] / applied, 3) if applied > 0 else 0,
        }
        results.append(entry)

    results.sort(key=lambda x: x["overall_conversion"], reverse=True)
    return results


def compute_time_metrics(candidates: dict) -> dict:
    """Compute time-based metrics."""
    times_to_current = []
    times_to_accept = []

    for cid, data in candidates.items():
        applied = data.get("applied_date")
        stage_date = data.get("stage_date")
        if applied and stage_date:
            days = (stage_date - applied).days
            if days >= 0:
                times_to_current.append(days)
                if data["stage"] == "Accepted":
                    times_to_accept.append(days)

    def stats(values):
        if not values:
            return {"count": 0, "avg": 0, "median": 0, "min": 0, "max": 0}
        s = sorted(values)
        n = len(s)
        return {
            "count": n,
            "avg": round(sum(s) / n, 1),
            "median": s[n // 2],
            "min": s[0],
            "max": s[-1],
        }

    return {
        "time_in_pipeline_days": stats(times_to_current),
        "time_to_accept_days": stats(times_to_accept),
    }


def compute_role_breakdown(candidates: dict) -> list:
    """Compute metrics by role."""
    role_data = defaultdict(lambda: defaultdict(int))

    for cid, data in candidates.items():
        role = data.get("role", "Unknown") or "Unknown"
        idx = data["max_stage_idx"]
        for i in range(idx + 1):
            role_data[role][STAGE_ORDER[i]] += 1

    results = []
    for role, counts in sorted(role_data.items()):
        applied = counts.get("Applied", 0)
        accepted = counts.get("Accepted", 0)
        offered = counts.get("Offered", 0)
        results.append({
            "role": role,
            "applied": applied,
            "accepted": accepted,
            "overall_conversion": round(accepted / applied, 3) if applied > 0 else 0,
            "offer_accept_rate": round(accepted / offered, 3) if offered > 0 else 0,
        })

    return results


def identify_bottlenecks(funnel: list) -> list:
    """Identify pipeline bottlenecks."""
    bottlenecks = []
    for entry in funnel:
        if entry.get("status") == "BELOW_BENCHMARK":
            bottlenecks.append({
                "stage_transition": f"{STAGE_ORDER[STAGE_ORDER.index(entry['stage'])-1]} -> {entry['stage']}",
                "actual_rate": f"{entry['conversion_rate']*100:.1f}%",
                "benchmark": entry["benchmark_range"],
                "severity": "HIGH" if entry["conversion_rate"] < 0.5 * float(entry["benchmark_range"].split("-")[0].replace("%", "")) / 100 else "MEDIUM",
            })
    return bottlenecks


def format_human(funnel: list, sources: list, time_metrics: dict, roles: list, bottlenecks: list, total: int) -> str:
    """Format results for human-readable output."""
    lines = []
    lines.append("=" * 65)
    lines.append("CANDIDATE PIPELINE REPORT")
    lines.append("=" * 65)
    lines.append(f"  Total candidates: {total}")
    lines.append("")

    lines.append("-" * 65)
    lines.append("FUNNEL ANALYSIS")
    lines.append("-" * 65)
    lines.append(f"  {'Stage':<20} {'Count':>8} {'Conv Rate':>12} {'Benchmark':>14} {'Status':>16}")
    lines.append(f"  {'-'*20} {'-'*8} {'-'*12} {'-'*14} {'-'*16}")
    for entry in funnel:
        conv = f"{entry['conversion_rate']*100:.1f}%" if "conversion_rate" in entry else "--"
        bench = entry.get("benchmark_range", "--")
        status = entry.get("status", "--")
        lines.append(f"  {entry['stage']:<20} {entry['count']:>8} {conv:>12} {bench:>14} {status:>16}")

    lines.append("")
    lines.append("-" * 65)
    lines.append("SOURCE EFFECTIVENESS")
    lines.append("-" * 65)
    lines.append(f"  {'Source':<25} {'Applied':>8} {'Screened':>9} {'Accepted':>9} {'Conv %':>8}")
    lines.append(f"  {'-'*25} {'-'*8} {'-'*9} {'-'*9} {'-'*8}")
    for s in sources:
        lines.append(f"  {s['source']:<25} {s['applied']:>8} {s['screened']:>9} {s['accepted']:>9} {s['overall_conversion']*100:>7.1f}%")

    tm = time_metrics
    lines.append("")
    lines.append("-" * 65)
    lines.append("TIME METRICS")
    lines.append("-" * 65)
    pip = tm["time_in_pipeline_days"]
    lines.append(f"  Time in pipeline:  avg {pip['avg']} days | median {pip['median']} days | range {pip['min']}-{pip['max']} days")
    acc = tm["time_to_accept_days"]
    if acc["count"] > 0:
        lines.append(f"  Time to accept:    avg {acc['avg']} days | median {acc['median']} days | range {acc['min']}-{acc['max']} days")

    if bottlenecks:
        lines.append("")
        lines.append("-" * 65)
        lines.append("BOTTLENECKS IDENTIFIED")
        lines.append("-" * 65)
        for b in bottlenecks:
            lines.append(f"  [{b['severity']}] {b['stage_transition']}: {b['actual_rate']} (benchmark: {b['benchmark']})")

    if roles:
        lines.append("")
        lines.append("-" * 65)
        lines.append("BY ROLE")
        lines.append("-" * 65)
        lines.append(f"  {'Role':<30} {'Applied':>8} {'Accepted':>9} {'Conv %':>8}")
        lines.append(f"  {'-'*30} {'-'*8} {'-'*9} {'-'*8}")
        for r in roles:
            lines.append(f"  {r['role']:<30} {r['applied']:>8} {r['accepted']:>9} {r['overall_conversion']*100:>7.1f}%")

    lines.append("")
    return "\n".join(lines)


def main():
    parser = argparse.ArgumentParser(
        description="Track and analyze candidate pipeline funnel metrics."
    )
    parser.add_argument("--file", required=True, help="Path to pipeline CSV file")
    parser.add_argument("--json", action="store_true", help="Output results as JSON")
    args = parser.parse_args()

    rows = read_csv(args.file)
    if not rows:
        print("Error: No data found in CSV file.", file=sys.stderr)
        sys.exit(1)

    candidates = get_candidate_max_stage(rows)
    funnel = compute_funnel(candidates)
    sources = compute_source_effectiveness(candidates)
    time_metrics = compute_time_metrics(candidates)
    roles = compute_role_breakdown(candidates)
    bottlenecks = identify_bottlenecks(funnel)

    if args.json:
        output = {
            "total_candidates": len(candidates),
            "funnel": funnel,
            "source_effectiveness": sources,
            "time_metrics": time_metrics,
            "role_breakdown": roles,
            "bottlenecks": bottlenecks,
        }
        print(json.dumps(output, indent=2, default=str))
    else:
        print(format_human(funnel, sources, time_metrics, roles, bottlenecks, len(candidates)))


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Interview Scorecard Generator - Generate structured interview scorecards.

Creates competency-based interview scorecards tailored to role and level,
with behavioral interview questions, rating scales, and evaluation criteria.

Usage:
    python interview_scorecard.py --role "Senior Engineer" --level IC3
    python interview_scorecard.py --role "Product Manager" --level IC2 --stage technical --json
    python interview_scorecard.py --role "Sales Manager" --level M1 --competencies leadership,negotiation,analytics

Input: Role name, level, optional stage and custom competencies.

Output: Structured interview scorecard in markdown or JSON format.
"""

import argparse
import json
import sys
from datetime import date


# --- Competency library ---

COMPETENCY_LIBRARY = {
    "technical_depth": {
        "name": "Technical Depth",
        "description": "Demonstrates deep expertise in relevant technical domain",
        "questions": [
            "Describe the most technically challenging project you have worked on. What made it difficult and how did you approach it?",
            "Walk me through a technical decision you made that had significant trade-offs. How did you evaluate the options?",
            "Tell me about a time you had to learn a new technology or framework quickly to deliver a project. What was your approach?",
        ],
        "signals_strong": ["Explains complex concepts clearly", "Demonstrates depth beyond surface level", "Shows awareness of trade-offs and edge cases"],
        "signals_weak": ["Cannot explain past work in detail", "Lacks awareness of alternatives", "Struggles with follow-up questions"],
    },
    "problem_solving": {
        "name": "Problem Solving",
        "description": "Approaches complex problems systematically and creatively",
        "questions": [
            "Describe a complex problem you solved that did not have an obvious solution. Walk me through your approach step by step.",
            "Tell me about a time you identified a problem before anyone else noticed. What did you do?",
            "Give an example of when you had to make a decision with incomplete information. How did you handle the uncertainty?",
        ],
        "signals_strong": ["Breaks problems into components", "Considers multiple approaches", "Uses data to validate assumptions"],
        "signals_weak": ["Jumps to solutions without analysis", "Cannot articulate reasoning", "Relies on single approach"],
    },
    "communication": {
        "name": "Communication",
        "description": "Communicates clearly and adapts style to audience",
        "questions": [
            "Tell me about a time you had to explain a complex idea to a non-technical audience. How did you ensure understanding?",
            "Describe a situation where miscommunication caused a problem. How did you resolve it and what did you learn?",
            "Give an example of how you gave constructive feedback to a peer or direct report.",
        ],
        "signals_strong": ["Adapts communication to audience", "Listens actively", "Structures information clearly"],
        "signals_weak": ["Uses excessive jargon", "Interrupts or talks over others", "Cannot simplify complex ideas"],
    },
    "leadership": {
        "name": "Leadership",
        "description": "Leads teams effectively and drives outcomes through others",
        "questions": [
            "Tell me about a time you led a team through a difficult situation. What was the outcome?",
            "Describe how you have developed someone on your team. What was your approach and what was the result?",
            "Give an example of when you had to make an unpopular decision. How did you handle the pushback?",
        ],
        "signals_strong": ["Empowers team members", "Takes accountability for outcomes", "Develops others proactively"],
        "signals_weak": ["Micromanages", "Avoids difficult decisions", "Takes credit for team work"],
    },
    "collaboration": {
        "name": "Collaboration",
        "description": "Works effectively across teams and functions",
        "questions": [
            "Give an example of a successful cross-functional project you contributed to. What was your role?",
            "Tell me about a time you disagreed with a colleague on an approach. How did you resolve it?",
            "Describe a situation where you had to build alignment across multiple stakeholders with competing priorities.",
        ],
        "signals_strong": ["Seeks input from others", "Builds consensus", "Gives credit to team"],
        "signals_weak": ["Works in isolation", "Cannot compromise", "Creates friction with other teams"],
    },
    "execution": {
        "name": "Execution & Delivery",
        "description": "Delivers results consistently and manages competing priorities",
        "questions": [
            "Tell me about a project where the requirements changed significantly mid-stream. How did you adapt?",
            "Describe a time you had to deliver under a tight deadline. What trade-offs did you make?",
            "Give an example of how you prioritized competing demands when you could not do everything.",
        ],
        "signals_strong": ["Delivers on commitments", "Manages scope effectively", "Communicates blockers early"],
        "signals_weak": ["Misses deadlines without communication", "Cannot prioritize", "Over-commits"],
    },
    "strategic_thinking": {
        "name": "Strategic Thinking",
        "description": "Thinks beyond immediate scope and connects work to business outcomes",
        "questions": [
            "Tell me about a time you identified a strategic opportunity that others had missed. What did you do?",
            "Describe how you have connected your team's work to broader business objectives.",
            "Give an example of a long-term bet you made that paid off. How did you build conviction?",
        ],
        "signals_strong": ["Connects work to business impact", "Thinks in systems", "Anticipates future needs"],
        "signals_weak": ["Focuses only on immediate tasks", "Cannot articulate business context", "Reactive rather than proactive"],
    },
    "customer_focus": {
        "name": "Customer Focus",
        "description": "Puts customer needs at the center of decisions",
        "questions": [
            "Tell me about a time you went above and beyond for a customer or end user.",
            "Describe a decision you made based on customer feedback that changed your team's direction.",
            "Give an example of how you balanced customer requests with technical or business constraints.",
        ],
        "signals_strong": ["References customer impact in decisions", "Seeks direct customer feedback", "Advocates for user experience"],
        "signals_weak": ["Never mentions customers", "Prioritizes internal convenience over user needs", "Cannot describe user impact"],
    },
    "negotiation": {
        "name": "Negotiation",
        "description": "Negotiates effectively to achieve mutually beneficial outcomes",
        "questions": [
            "Describe a negotiation where you achieved a better outcome than expected. What was your strategy?",
            "Tell me about a time you had to negotiate with limited leverage. How did you approach it?",
            "Give an example of a negotiation that did not go well. What would you do differently?",
        ],
        "signals_strong": ["Prepares thoroughly", "Finds creative solutions", "Maintains relationships"],
        "signals_weak": ["Confrontational approach", "Caves too easily", "Cannot articulate value proposition"],
    },
    "analytics": {
        "name": "Analytical Skills",
        "description": "Uses data and analysis to drive decisions",
        "questions": [
            "Tell me about a time you used data to change a decision or strategy.",
            "Describe a situation where the data was ambiguous. How did you reach a conclusion?",
            "Give an example of a metric or dashboard you created that drove a business outcome.",
        ],
        "signals_strong": ["Asks for data before deciding", "Distinguishes correlation from causation", "Builds frameworks for analysis"],
        "signals_weak": ["Relies on gut feeling", "Cherry-picks data", "Cannot interpret basic metrics"],
    },
}

# --- Default competency sets by role type ---

ROLE_COMPETENCIES = {
    "engineer": ["technical_depth", "problem_solving", "communication", "collaboration", "execution"],
    "manager": ["leadership", "communication", "strategic_thinking", "execution", "collaboration"],
    "product": ["strategic_thinking", "communication", "customer_focus", "analytics", "collaboration"],
    "sales": ["negotiation", "communication", "customer_focus", "execution", "analytics"],
    "design": ["customer_focus", "communication", "collaboration", "problem_solving", "execution"],
    "data": ["technical_depth", "analytics", "problem_solving", "communication", "execution"],
    "default": ["problem_solving", "communication", "collaboration", "execution", "strategic_thinking"],
}

# --- Competency weights by level ---

LEVEL_WEIGHTS = {
    "IC1": {"technical_depth": 40, "problem_solving": 25, "communication": 20, "collaboration": 15},
    "IC2": {"technical_depth": 35, "problem_solving": 25, "communication": 20, "collaboration": 20},
    "IC3": {"technical_depth": 30, "problem_solving": 25, "communication": 20, "collaboration": 15, "leadership": 10},
    "IC4": {"technical_depth": 25, "problem_solving": 20, "communication": 20, "strategic_thinking": 20, "leadership": 15},
    "IC5": {"technical_depth": 20, "strategic_thinking": 25, "leadership": 20, "communication": 20, "problem_solving": 15},
    "M1": {"leadership": 30, "communication": 25, "execution": 20, "collaboration": 15, "strategic_thinking": 10},
    "M2": {"leadership": 25, "strategic_thinking": 25, "communication": 20, "execution": 15, "collaboration": 15},
    "VP": {"strategic_thinking": 30, "leadership": 30, "communication": 20, "execution": 20},
}

STAGES = ["phone_screen", "technical", "behavioral", "final", "hiring_manager"]

RATING_SCALE = {
    1: "Does Not Meet - No evidence of competency; significant concerns",
    2: "Partially Meets - Limited evidence; below expectations for level",
    3: "Meets Expectations - Solid evidence; appropriate for level",
    4: "Exceeds Expectations - Strong evidence; above level expectations",
    5: "Exceptional - Outstanding evidence; role-model level performance",
}


def detect_role_type(role: str) -> str:
    """Detect role type from role name."""
    role_lower = role.lower()
    for key in ROLE_COMPETENCIES:
        if key in role_lower:
            return key
    if any(w in role_lower for w in ["software", "backend", "frontend", "fullstack", "devops", "sre", "infrastructure", "platform"]):
        return "engineer"
    if any(w in role_lower for w in ["director", "vp", "head of", "lead"]):
        return "manager"
    if any(w in role_lower for w in ["product", "pm"]):
        return "product"
    if any(w in role_lower for w in ["account", "sales", "bd", "business development"]):
        return "sales"
    if any(w in role_lower for w in ["ux", "ui", "design"]):
        return "design"
    if any(w in role_lower for w in ["data", "analytics", "ml", "machine learning"]):
        return "data"
    return "default"


def get_competencies(role: str, level: str, custom: list = None) -> list:
    """Get competencies for the role and level."""
    if custom:
        comp_keys = []
        for c in custom:
            c_clean = c.strip().lower().replace(" ", "_")
            if c_clean in COMPETENCY_LIBRARY:
                comp_keys.append(c_clean)
            else:
                # Create a basic custom competency
                COMPETENCY_LIBRARY[c_clean] = {
                    "name": c.strip().title(),
                    "description": f"Demonstrates proficiency in {c.strip().lower()}",
                    "questions": [f"Tell me about your experience with {c.strip().lower()}."],
                    "signals_strong": ["Provides specific examples", "Shows depth of knowledge"],
                    "signals_weak": ["Cannot provide examples", "Surface-level understanding"],
                }
                comp_keys.append(c_clean)
        return comp_keys

    role_type = detect_role_type(role)
    return ROLE_COMPETENCIES.get(role_type, ROLE_COMPETENCIES["default"])


def get_weights(level: str, competencies: list) -> dict:
    """Get competency weights for the level."""
    level_upper = level.upper()
    base_weights = LEVEL_WEIGHTS.get(level_upper, {})

    weights = {}
    total_assigned = 0
    for comp in competencies:
        if comp in base_weights:
            weights[comp] = base_weights[comp]
            total_assigned += base_weights[comp]

    remaining = 100 - total_assigned
    unweighted = [c for c in competencies if c not in weights]
    if unweighted:
        per_comp = remaining // len(unweighted)
        for c in unweighted:
            weights[c] = per_comp

    # Normalize to 100%
    total = sum(weights.values())
    if total > 0 and total != 100:
        factor = 100 / total
        weights = {k: round(v * factor) for k, v in weights.items()}

    return weights


def build_scorecard(role: str, level: str, stage: str, competencies: list) -> dict:
    """Build the complete scorecard."""
    weights = get_weights(level, competencies)

    comp_details = []
    for comp_key in competencies:
        comp = COMPETENCY_LIBRARY[comp_key]
        comp_details.append({
            "key": comp_key,
            "name": comp["name"],
            "description": comp["description"],
            "weight": weights.get(comp_key, 20),
            "questions": comp["questions"],
            "signals_strong": comp["signals_strong"],
            "signals_weak": comp["signals_weak"],
        })

    return {
        "role": role,
        "level": level,
        "stage": stage,
        "date": date.today().isoformat(),
        "competencies": comp_details,
        "rating_scale": RATING_SCALE,
        "instructions": {
            "before_interview": [
                "Review the candidate's resume and any prior interview notes",
                "Prepare 1-2 follow-up questions per competency",
                "Clear 15 minutes after the interview for note-taking",
            ],
            "during_interview": [
                "Ask behavioral questions using STAR format (Situation, Task, Action, Result)",
                "Take brief notes on specific examples and evidence",
                "Rate each competency independently before discussing with other interviewers",
            ],
            "after_interview": [
                "Complete all ratings within 24 hours while memory is fresh",
                "Provide specific behavioral evidence for each rating",
                "Submit scorecard before the debrief to prevent anchoring bias",
            ],
        },
    }


def format_markdown(scorecard: dict) -> str:
    """Format scorecard as markdown."""
    lines = []
    lines.append(f"# Interview Scorecard: {scorecard['role']} ({scorecard['level']})")
    lines.append("")
    lines.append(f"**Stage:** {scorecard['stage'].replace('_', ' ').title()}")
    lines.append(f"**Date:** {scorecard['date']}")
    lines.append(f"**Interviewer:** ____________________")
    lines.append(f"**Candidate:** ____________________")
    lines.append("")

    lines.append("---")
    lines.append("")
    lines.append("## Rating Scale")
    lines.append("")
    for rating, desc in scorecard["rating_scale"].items():
        lines.append(f"- **{rating}** - {desc}")
    lines.append("")

    lines.append("---")
    lines.append("")
    lines.append("## Competency Ratings")
    lines.append("")
    lines.append("| Competency | Weight | Rating (1-5) | Evidence |")
    lines.append("|------------|--------|:------------:|----------|")
    for comp in scorecard["competencies"]:
        lines.append(f"| {comp['name']} | {comp['weight']}% | ____ | |")
    lines.append("")
    lines.append("**Weighted Score:** ______ / 5.0")
    lines.append("")

    lines.append("---")
    lines.append("")
    for comp in scorecard["competencies"]:
        lines.append(f"## {comp['name']} ({comp['weight']}%)")
        lines.append(f"*{comp['description']}*")
        lines.append("")
        lines.append("### Suggested Questions")
        for q in comp["questions"]:
            lines.append(f"- {q}")
        lines.append("")
        lines.append("### Strong Signals")
        for s in comp["signals_strong"]:
            lines.append(f"- {s}")
        lines.append("")
        lines.append("### Weak Signals")
        for s in comp["signals_weak"]:
            lines.append(f"- {s}")
        lines.append("")
        lines.append(f"**Rating:** ____ / 5")
        lines.append(f"**Evidence:** ")
        lines.append("")
        lines.append("---")
        lines.append("")

    lines.append("## Overall Assessment")
    lines.append("")
    lines.append("**Recommendation:**")
    lines.append("- [ ] Strong Hire")
    lines.append("- [ ] Hire")
    lines.append("- [ ] No Hire")
    lines.append("- [ ] Strong No Hire")
    lines.append("")
    lines.append("**Key Strengths:**")
    lines.append("1. ")
    lines.append("2. ")
    lines.append("")
    lines.append("**Key Concerns:**")
    lines.append("1. ")
    lines.append("2. ")
    lines.append("")

    lines.append("## Interview Instructions")
    lines.append("")
    for phase, instructions in scorecard["instructions"].items():
        lines.append(f"### {phase.replace('_', ' ').title()}")
        for inst in instructions:
            lines.append(f"- {inst}")
        lines.append("")

    return "\n".join(lines)


def format_human(scorecard: dict) -> str:
    """Format scorecard for terminal output."""
    lines = []
    lines.append("=" * 65)
    lines.append(f"INTERVIEW SCORECARD: {scorecard['role']} ({scorecard['level']})")
    lines.append("=" * 65)
    lines.append(f"  Stage: {scorecard['stage'].replace('_', ' ').title()}")
    lines.append(f"  Date:  {scorecard['date']}")
    lines.append("")

    lines.append("-" * 65)
    lines.append("COMPETENCIES")
    lines.append("-" * 65)
    lines.append(f"  {'Competency':<30} {'Weight':>8}")
    lines.append(f"  {'-'*30} {'-'*8}")
    for comp in scorecard["competencies"]:
        lines.append(f"  {comp['name']:<30} {comp['weight']:>7}%")

    lines.append("")
    lines.append("-" * 65)
    lines.append("QUESTIONS BY COMPETENCY")
    lines.append("-" * 65)
    for comp in scorecard["competencies"]:
        lines.append(f"\n  [{comp['name']}]")
        for i, q in enumerate(comp["questions"], 1):
            lines.append(f"    {i}. {q}")

    lines.append("")
    lines.append("-" * 65)
    lines.append("RATING SCALE")
    lines.append("-" * 65)
    for rating, desc in scorecard["rating_scale"].items():
        lines.append(f"  {rating} = {desc}")

    lines.append("")
    lines.append("Use --json for machine-readable output or pipe to a file for the full markdown scorecard.")
    lines.append("")
    return "\n".join(lines)


def main():
    parser = argparse.ArgumentParser(
        description="Generate structured interview scorecards for a given role and level."
    )
    parser.add_argument("--role", required=True, help="Role title (e.g., 'Senior Engineer', 'Product Manager')")
    parser.add_argument("--level", required=True, help="Level code (IC1-IC5, M1-M2, VP)")
    parser.add_argument("--stage", default="behavioral", choices=STAGES, help="Interview stage (default: behavioral)")
    parser.add_argument("--competencies", default=None, help="Comma-separated list of competencies to override defaults")
    parser.add_argument("--json", action="store_true", help="Output results as JSON")
    parser.add_argument("--markdown", action="store_true", help="Output as full markdown scorecard")
    args = parser.parse_args()

    custom_comps = args.competencies.split(",") if args.competencies else None
    competencies = get_competencies(args.role, args.level, custom_comps)
    scorecard = build_scorecard(args.role, args.level, args.stage, competencies)

    if args.json:
        print(json.dumps(scorecard, indent=2))
    elif args.markdown:
        print(format_markdown(scorecard))
    else:
        print(format_human(scorecard))


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Job Posting Analyzer - Analyze job descriptions for bias, readability, and quality.

Reads a job description file (plain text or markdown) and scores it across
multiple dimensions: inclusive language, readability, structure completeness,
requirement inflation, and gendered language.

Usage:
    python job_posting_analyzer.py --file job_description.md
    python job_posting_analyzer.py --file job_description.md --json

Input: Plain text or markdown file containing a job description.

Output: Quality score with category breakdowns and actionable recommendations.
"""

import argparse
import json
import os
import re
import sys
from collections import Counter


# --- Bias and language detection word lists ---

MASCULINE_CODED = [
    "aggressive", "ambitious", "analytical", "assertive", "autonomous",
    "boast", "champion", "competitive", "confident", "courage",
    "decisive", "determined", "dominant", "driven", "fearless",
    "force", "headstrong", "hero", "hostile", "hustle",
    "independent", "individual", "lead", "logic", "ninja",
    "objective", "opinion", "outspoken", "persist", "principle",
    "rockstar", "self-reliant", "strong", "superior", "tackle",
    "thrust", "warrior",
]

FEMININE_CODED = [
    "agree", "affectionate", "caring", "collaborate", "commit",
    "communal", "compassion", "connect", "considerate", "cooperate",
    "depend", "emotional", "empathy", "feel", "gentle",
    "honest", "interpersonal", "kind", "loyal", "modesty",
    "nurture", "pleasant", "polite", "quiet", "responsible",
    "share", "submit", "support", "sympathetic", "tender",
    "together", "trust", "understand", "warm", "yield",
]

EXCLUSIONARY_PHRASES = [
    "native english speaker",
    "young and dynamic",
    "digital native",
    "culture fit",
    "man enough",
    "manpower",
    "chairman",
    "he/his",
    "she/her",
    "able-bodied",
    "normal",
    "walk to",
    "stand for extended",
    "clean-shaven",
    "must be available 24/7",
]

JARGON_WORDS = [
    "synergy", "leverage", "paradigm", "disrupt", "bleeding edge",
    "move the needle", "circle back", "deep dive", "low-hanging fruit",
    "thought leader", "game changer", "best-in-class", "world-class",
    "fast-paced environment", "wear many hats", "self-starter",
    "hit the ground running", "rock star", "guru", "wizard",
]

REQUIRED_SECTIONS = [
    "responsibilities",
    "requirements",
    "qualifications",
    "compensation",
    "salary",
    "benefits",
    "about",
    "the role",
    "what you",
    "who you",
]


def read_file(path: str) -> str:
    """Read file contents."""
    if not os.path.isfile(path):
        print(f"Error: File not found: {path}", file=sys.stderr)
        sys.exit(1)
    with open(path, "r", encoding="utf-8") as f:
        return f.read()


def count_words(text: str) -> int:
    """Count words in text."""
    return len(text.split())


def count_sentences(text: str) -> int:
    """Count sentences in text."""
    sentences = re.split(r"[.!?]+", text)
    return max(1, len([s for s in sentences if s.strip()]))


def flesch_reading_ease(text: str) -> float:
    """
    Approximate Flesch Reading Ease score.
    Higher = easier to read. Target: 60-70 for job postings.
    """
    words = text.split()
    word_count = len(words)
    if word_count == 0:
        return 0.0
    sentence_count = count_sentences(text)
    # Approximate syllables: count vowel groups per word
    syllable_count = 0
    for word in words:
        word_lower = word.lower().strip(".,!?;:'\"()-")
        vowels = re.findall(r"[aeiouy]+", word_lower)
        syllable_count += max(1, len(vowels))
    score = 206.835 - 1.015 * (word_count / sentence_count) - 84.6 * (syllable_count / word_count)
    return round(max(0, min(100, score)), 1)


def find_word_matches(text: str, word_list: list) -> list:
    """Find words from word_list that appear in text."""
    text_lower = text.lower()
    found = []
    for word in word_list:
        pattern = r"\b" + re.escape(word) + r"\w*\b"
        if re.search(pattern, text_lower):
            found.append(word)
    return found


def find_phrase_matches(text: str, phrase_list: list) -> list:
    """Find phrases from phrase_list that appear in text."""
    text_lower = text.lower()
    return [p for p in phrase_list if p in text_lower]


def count_requirements(text: str) -> dict:
    """Count and categorize requirements."""
    lines = text.split("\n")
    bullet_lines = [l.strip() for l in lines if re.match(r"^\s*[-*]\s+", l)]

    years_mentions = re.findall(r"(\d+)\+?\s*(?:years?|yrs?)", text.lower())
    max_years = max([int(y) for y in years_mentions], default=0)

    must_have = []
    nice_to_have = []
    in_must = False
    in_nice = False

    for line in lines:
        lower = line.lower().strip()
        if "must have" in lower or "required" in lower or "requirements" in lower:
            in_must = True
            in_nice = False
        elif "nice to have" in lower or "preferred" in lower or "bonus" in lower:
            in_nice = True
            in_must = False
        elif re.match(r"^#{1,3}\s+", line):
            in_must = False
            in_nice = False
        elif re.match(r"^\s*[-*]\s+", line):
            if in_must:
                must_have.append(line.strip())
            elif in_nice:
                nice_to_have.append(line.strip())

    return {
        "total_bullet_points": len(bullet_lines),
        "must_have_count": len(must_have),
        "nice_to_have_count": len(nice_to_have),
        "max_years_experience": max_years,
    }


def check_sections(text: str) -> dict:
    """Check which recommended sections are present."""
    text_lower = text.lower()
    found = []
    missing = []
    for section in REQUIRED_SECTIONS:
        if section in text_lower:
            found.append(section)
        else:
            missing.append(section)
    return {"found": found, "missing": missing}


def calculate_scores(text: str) -> dict:
    """Calculate all analysis scores."""
    word_count = count_words(text)
    readability = flesch_reading_ease(text)

    masculine = find_word_matches(text, MASCULINE_CODED)
    feminine = find_word_matches(text, FEMININE_CODED)
    exclusionary = find_phrase_matches(text, EXCLUSIONARY_PHRASES)
    jargon = find_word_matches(text, JARGON_WORDS)
    jargon += find_phrase_matches(text, JARGON_WORDS)
    jargon = list(set(jargon))

    requirements = count_requirements(text)
    sections = check_sections(text)

    # --- Scoring (0-100 per category) ---

    # Gender neutrality: penalize imbalance and exclusionary terms
    masc_count = len(masculine)
    fem_count = len(feminine)
    gender_balance = abs(masc_count - fem_count)
    gender_score = max(0, 100 - gender_balance * 10 - len(exclusionary) * 20)

    # Readability: target 60-70
    if 55 <= readability <= 75:
        readability_score = 100
    elif 45 <= readability < 55 or 75 < readability <= 85:
        readability_score = 80
    elif 30 <= readability < 45 or 85 < readability <= 95:
        readability_score = 60
    else:
        readability_score = 40

    # Length: target 300-800 words
    if 300 <= word_count <= 800:
        length_score = 100
    elif 200 <= word_count < 300 or 800 < word_count <= 1000:
        length_score = 80
    elif 100 <= word_count < 200 or 1000 < word_count <= 1500:
        length_score = 60
    else:
        length_score = 40

    # Requirements: penalize excessive must-haves and high year requirements
    must_haves = requirements["must_have_count"]
    if must_haves <= 7:
        req_score = 100
    elif must_haves <= 10:
        req_score = 80
    elif must_haves <= 15:
        req_score = 60
    else:
        req_score = 40
    if requirements["max_years_experience"] > 10:
        req_score = max(0, req_score - 20)

    # Structure: based on sections found
    section_ratio = len(sections["found"]) / len(REQUIRED_SECTIONS)
    structure_score = round(section_ratio * 100)

    # Jargon: penalize buzzwords
    jargon_score = max(0, 100 - len(jargon) * 15)

    # Overall weighted score
    overall = round(
        gender_score * 0.25
        + readability_score * 0.20
        + length_score * 0.10
        + req_score * 0.15
        + structure_score * 0.15
        + jargon_score * 0.15
    )

    return {
        "overall_score": overall,
        "word_count": word_count,
        "flesch_reading_ease": readability,
        "categories": {
            "gender_neutrality": {
                "score": gender_score,
                "masculine_coded_words": masculine,
                "feminine_coded_words": feminine,
                "exclusionary_phrases": exclusionary,
            },
            "readability": {
                "score": readability_score,
                "flesch_reading_ease": readability,
            },
            "length": {
                "score": length_score,
                "word_count": word_count,
                "target_range": "300-800 words",
            },
            "requirements": {
                "score": req_score,
                "must_have_count": requirements["must_have_count"],
                "nice_to_have_count": requirements["nice_to_have_count"],
                "max_years_experience": requirements["max_years_experience"],
                "total_bullet_points": requirements["total_bullet_points"],
            },
            "structure": {
                "score": structure_score,
                "sections_found": sections["found"],
                "sections_missing": sections["missing"],
            },
            "jargon": {
                "score": jargon_score,
                "jargon_found": jargon,
            },
        },
    }


def build_recommendations(results: dict) -> list:
    """Generate actionable recommendations from scores."""
    recs = []
    cats = results["categories"]

    if cats["gender_neutrality"]["score"] < 80:
        if cats["gender_neutrality"]["masculine_coded_words"]:
            recs.append(
                f"Replace masculine-coded words ({', '.join(cats['gender_neutrality']['masculine_coded_words'][:5])}) "
                "with neutral alternatives to broaden the candidate pool."
            )
        if cats["gender_neutrality"]["exclusionary_phrases"]:
            recs.append(
                f"Remove exclusionary phrases: {', '.join(cats['gender_neutrality']['exclusionary_phrases'])}."
            )

    if cats["readability"]["score"] < 80:
        recs.append(
            f"Improve readability (current Flesch score: {cats['readability']['flesch_reading_ease']}). "
            "Use shorter sentences and simpler vocabulary. Target a Flesch score of 60-70."
        )

    if cats["length"]["score"] < 80:
        wc = cats["length"]["word_count"]
        if wc < 300:
            recs.append(f"Expand the posting ({wc} words). Add more detail about responsibilities, team, and benefits. Target 300-800 words.")
        else:
            recs.append(f"Shorten the posting ({wc} words). Consolidate overlapping requirements and remove filler. Target 300-800 words.")

    if cats["requirements"]["score"] < 80:
        if cats["requirements"]["must_have_count"] > 7:
            recs.append(
                f"Reduce must-have requirements from {cats['requirements']['must_have_count']} to 5-7. "
                "Move non-essential items to nice-to-have. Excessive requirements deter qualified candidates, especially from underrepresented groups."
            )
        if cats["requirements"]["max_years_experience"] > 8:
            recs.append(
                f"Reconsider the {cats['requirements']['max_years_experience']}+ year experience requirement. "
                "Research shows years of experience is a weak predictor of performance beyond 5 years."
            )

    if cats["structure"]["score"] < 80:
        recs.append(
            f"Add missing sections: {', '.join(cats['structure']['sections_missing'][:5])}. "
            "Job postings with compensation info receive 30% more applications."
        )

    if cats["jargon"]["score"] < 80:
        recs.append(
            f"Remove jargon and buzzwords: {', '.join(cats['jargon']['jargon_found'][:5])}. "
            "Use concrete language that describes actual work."
        )

    if not recs:
        recs.append("This job posting scores well across all dimensions. No critical improvements needed.")

    return recs


def format_human(results: dict, recommendations: list) -> str:
    """Format results for human-readable output."""
    lines = []
    lines.append("=" * 60)
    lines.append("JOB POSTING ANALYSIS REPORT")
    lines.append("=" * 60)
    lines.append("")

    overall = results["overall_score"]
    grade = "A" if overall >= 85 else "B" if overall >= 70 else "C" if overall >= 55 else "D" if overall >= 40 else "F"
    lines.append(f"  Overall Score: {overall}/100 (Grade: {grade})")
    lines.append(f"  Word Count: {results['word_count']}")
    lines.append(f"  Readability: {results['flesch_reading_ease']} (Flesch Reading Ease)")
    lines.append("")

    lines.append("-" * 60)
    lines.append("CATEGORY SCORES")
    lines.append("-" * 60)

    cats = results["categories"]
    for name, data in cats.items():
        label = name.replace("_", " ").title()
        score = data["score"]
        bar = "#" * (score // 5) + "." * (20 - score // 5)
        lines.append(f"  {label:.<30} {score:>3}/100  [{bar}]")

    lines.append("")
    lines.append("-" * 60)
    lines.append("DETAILS")
    lines.append("-" * 60)

    gender = cats["gender_neutrality"]
    if gender["masculine_coded_words"]:
        lines.append(f"  Masculine-coded: {', '.join(gender['masculine_coded_words'])}")
    if gender["feminine_coded_words"]:
        lines.append(f"  Feminine-coded:  {', '.join(gender['feminine_coded_words'])}")
    if gender["exclusionary_phrases"]:
        lines.append(f"  Exclusionary:    {', '.join(gender['exclusionary_phrases'])}")

    reqs = cats["requirements"]
    lines.append(f"  Must-haves: {reqs['must_have_count']}  |  Nice-to-haves: {reqs['nice_to_have_count']}  |  Max years: {reqs['max_years_experience']}")

    if cats["jargon"]["jargon_found"]:
        lines.append(f"  Jargon found: {', '.join(cats['jargon']['jargon_found'])}")

    lines.append("")
    lines.append("-" * 60)
    lines.append("RECOMMENDATIONS")
    lines.append("-" * 60)
    for i, rec in enumerate(recommendations, 1):
        lines.append(f"  {i}. {rec}")

    lines.append("")
    return "\n".join(lines)


def main():
    parser = argparse.ArgumentParser(
        description="Analyze job descriptions for bias, readability, and quality."
    )
    parser.add_argument("--file", required=True, help="Path to job description file (text or markdown)")
    parser.add_argument("--json", action="store_true", help="Output results as JSON")
    args = parser.parse_args()

    text = read_file(args.file)
    results = calculate_scores(text)
    recommendations = build_recommendations(results)

    if args.json:
        output = {**results, "recommendations": recommendations}
        print(json.dumps(output, indent=2))
    else:
        print(format_human(results, recommendations))


if __name__ == "__main__":
    main()