Jtbd

{
  "name": "jtbd",
  "description": "Terminal-first JTBD engine for founders and product people. Interview fast, kill jargon, capture real switching forces (",
  "author": {
    "name": "Gleb Kalinin"
  },
  "repository": "https://github.com/glebis/claude-skills",
  "license": "MIT"
}

__pycache__/
*.pyc
.pytest_cache/

"""Voice transcript ingest — map transcript chunks to JTBD schema fields.

Reads a transcript file (plain text or markdown), splits it into chunks
(by speaker turn or paragraph), and maps chunks to JTBD schema fields
with confidence scores. Also extracts Switch forces and verbatim quotes.

See references/switch_forces.md "One-minute pass (transcript mode)" for
the phrase patterns to match.

Usage:
    python ingest_transcript.py <path_to_transcript>

Output: JSON proposal to stdout.
"""

import json
import re
import sys
from typing import Optional


# --- Phrase patterns for JTBD schema field detection ---

SITUATION_PHRASES = [
    "when ", "every time", "last week", "during ", "last month",
    "yesterday", "this morning", "the other day", "last time",
    "whenever ", "at work", "in the meeting", "on monday",
]

MOTIVATION_PHRASES = [
    "i want", "i need", "i wish", "looking for", "trying to",
    "i'd like", "i would like", "we want", "we need", "hoping to",
    "my goal", "the goal",
]

OUTCOME_PHRASES = [
    "so that", "in order to", "that way", "goal is",
    "so i can", "so we can", "which means", "which would",
    "the result", "end up with", "outcome",
]

PAIN_PHRASES = [
    "frustrated", "frustrating", "annoying", "annoyed",
    "waste", "wasted", "wasting", "problem", "struggle",
    "struggling", "painful", "pain", "broken", "terrible",
    "awful", "hate", "hated", "worst", "nightmare",
    "can't stand", "drives me crazy", "unbearable",
]

WORKAROUND_PHRASES = [
    "currently using", "right now i", "right now we",
    "we use", "i use", "spreadsheet", "manually",
    "workaround", "hack", "duct tape", "cobbled together",
    "excel", "google sheets", "slack", "email",
    "copy and paste", "by hand",
]

# --- Switch force phrase patterns (from switch_forces.md) ---

PUSH_PHRASES = [
    "tired of", "frustrated", "can't believe", "every time", "wasted",
    "sick of", "fed up", "annoyed", "broken", "failing",
]

PULL_PHRASES = [
    "i wish", "imagine if", "what if i could", "finally",
    "dream of", "would be amazing", "picture this",
    "wouldn't it be great",
]

HABIT_PHRASES = [
    "we currently", "right now i", "i've been using", "for years",
    "we always", "our process", "the way we do it",
    "we've always done", "muscle memory",
]

ANXIETY_PHRASES = [
    "worried", "afraid", "what if", "last time we tried",
    "the team won't", "risky", "scared", "concerned",
    "might break", "could go wrong", "not sure if",
]

# --- Strong signal phrases for verbatim quote extraction ---

QUOTE_SIGNAL_PHRASES = (
    PAIN_PHRASES[:6] + PUSH_PHRASES[:5] + PULL_PHRASES[:4]
    + ["i want", "i need", "i wish", "the problem is",
       "what kills me", "the worst part", "if only"]
)


def split_into_chunks(text: str) -> list[dict]:
    """Split transcript into chunks by speaker turns or paragraphs.

    Returns a list of dicts: {"index": int, "text": str, "speaker": str|None}
    """
    lines = text.strip().split("\n")
    chunks: list[dict] = []

    # Detect format: speaker-turn (Q:/A:, Speaker:, Name:) or paragraph
    speaker_pattern = re.compile(
        r"^(?:([A-Za-z][A-Za-z0-9 ]*?)\s*:|([QA])\s*:)\s*(.*)", re.IGNORECASE
    )

    # Try speaker-turn parsing first
    current_speaker: Optional[str] = None
    current_lines: list[str] = []
    found_speakers = False

    for line in lines:
        m = speaker_pattern.match(line)
        if m:
            found_speakers = True
            # Save previous chunk
            if current_lines:
                chunks.append({
                    "index": len(chunks),
                    "text": " ".join(current_lines).strip(),
                    "speaker": current_speaker,
                })
            current_speaker = m.group(1) or m.group(2)
            current_lines = [m.group(3).strip()] if m.group(3).strip() else []
        else:
            if line.strip():
                current_lines.append(line.strip())

    # Flush last speaker chunk
    if found_speakers and current_lines:
        chunks.append({
            "index": len(chunks),
            "text": " ".join(current_lines).strip(),
            "speaker": current_speaker,
        })

    if found_speakers and chunks:
        return chunks

    # Fallback: paragraph-based splitting (double newline or blank line)
    paragraphs = re.split(r"\n\s*\n", text.strip())
    return [
        {"index": i, "text": p.strip(), "speaker": None}
        for i, p in enumerate(paragraphs)
        if p.strip()
    ]


def _score_chunk(text: str, phrases: list[str]) -> float:
    """Score how well a chunk matches a set of phrases. Returns 0.0-1.0."""
    text_lower = text.lower()
    matches = sum(1 for p in phrases if p in text_lower)
    if matches == 0:
        return 0.0
    # Scale: 1 match = 0.5, 2 = 0.7, 3 = 0.9, 4+ = 1.0
    return min(0.3 + matches * 0.2, 1.0)


def map_field(chunks: list[dict], phrases: list[str]) -> dict:
    """Find the best chunk for a given field. Returns mapping dict."""
    best_score = 0.0
    best_chunk = None
    best_index = -1

    for chunk in chunks:
        score = _score_chunk(chunk["text"], phrases)
        if score > best_score:
            best_score = score
            best_chunk = chunk["text"]
            best_index = chunk["index"]

    if best_chunk is None or best_score == 0.0:
        return {"text": "", "confidence": 0.0, "source_chunk": -1}

    return {
        "text": best_chunk,
        "confidence": round(best_score, 2),
        "source_chunk": best_index,
    }


def map_switch_force(chunks: list[dict], phrases: list[str]) -> dict:
    """Find the best chunk for a switch force. Returns mapping dict."""
    best_score = 0.0
    best_chunk = None

    for chunk in chunks:
        score = _score_chunk(chunk["text"], phrases)
        if score > best_score:
            best_score = score
            best_chunk = chunk["text"]

    if best_chunk is None or best_score == 0.0:
        return {"text": "", "confidence": 0.0}

    return {"text": best_chunk, "confidence": round(best_score, 2)}


def extract_quotes(chunks: list[dict]) -> list[str]:
    """Extract verbatim sentences that contain strong signal phrases."""
    quotes: list[str] = []
    seen: set[str] = set()

    for chunk in chunks:
        # Split chunk into sentences
        sentences = re.split(r"(?<=[.!?])\s+", chunk["text"])
        for sentence in sentences:
            s_lower = sentence.lower()
            for phrase in QUOTE_SIGNAL_PHRASES:
                if phrase in s_lower:
                    normalized = sentence.strip()
                    if normalized and normalized not in seen:
                        quotes.append(normalized)
                        seen.add(normalized)
                    break  # one match per sentence is enough

    return quotes


def identify_gaps(fields: dict) -> list[str]:
    """Return field names with confidence < 0.3 that need follow-up."""
    return [
        f"{name} (confidence: {info['confidence']})"
        for name, info in fields.items()
        if info["confidence"] < 0.3
    ]


def ingest(text: str) -> dict:
    """Main entry point: ingest transcript text and return JSON proposal."""
    chunks = split_into_chunks(text)

    if not chunks:
        return {
            "fields": {
                "situation": {"text": "", "confidence": 0.0, "source_chunk": -1},
                "motivation": {"text": "", "confidence": 0.0, "source_chunk": -1},
                "outcome": {"text": "", "confidence": 0.0, "source_chunk": -1},
                "what_hurts": {"text": "", "confidence": 0.0, "source_chunk": -1},
                "current_workaround": {"text": "", "confidence": 0.0, "source_chunk": -1},
            },
            "switch_forces": {
                "push": {"text": "", "confidence": 0.0},
                "pull": {"text": "", "confidence": 0.0},
                "habit": {"text": "", "confidence": 0.0},
                "anxiety": {"text": "", "confidence": 0.0},
            },
            "quotes": [],
            "gaps": [
                "situation (confidence: 0.0)",
                "motivation (confidence: 0.0)",
                "outcome (confidence: 0.0)",
                "what_hurts (confidence: 0.0)",
                "current_workaround (confidence: 0.0)",
            ],
        }

    fields = {
        "situation": map_field(chunks, SITUATION_PHRASES),
        "motivation": map_field(chunks, MOTIVATION_PHRASES),
        "outcome": map_field(chunks, OUTCOME_PHRASES),
        "what_hurts": map_field(chunks, PAIN_PHRASES),
        "current_workaround": map_field(chunks, WORKAROUND_PHRASES),
    }

    switch_forces = {
        "push": map_switch_force(chunks, PUSH_PHRASES),
        "pull": map_switch_force(chunks, PULL_PHRASES),
        "habit": map_switch_force(chunks, HABIT_PHRASES),
        "anxiety": map_switch_force(chunks, ANXIETY_PHRASES),
    }

    quotes = extract_quotes(chunks)
    gaps = identify_gaps(fields)

    return {
        "fields": fields,
        "switch_forces": switch_forces,
        "quotes": quotes,
        "gaps": gaps,
    }


def main() -> None:
    if len(sys.argv) < 2:
        print("Usage: python ingest_transcript.py <path_to_transcript>",
              file=sys.stderr)
        sys.exit(1)

    path = sys.argv[1]
    try:
        with open(path, "r", encoding="utf-8") as f:
            text = f.read()
    except FileNotFoundError:
        print(f"Error: file not found: {path}", file=sys.stderr)
        sys.exit(1)

    result = ingest(text)
    print(json.dumps(result, indent=2, ensure_ascii=False))


if __name__ == "__main__":
    main()

"""Review mining — cluster reviews on three axes: pain, outcome, workaround.

Accepts CSV or JSON input.  Emits a Markdown brief (review-brief.md) and
prints a JSON summary to stdout.

See references/review_taxonomy.md for axis vocabularies and cluster labels.
"""

import argparse
import csv
import json
import os
import re
import sys
from collections import Counter, defaultdict
from datetime import datetime
from pathlib import Path

# ── Keyword maps ────────────────────────────────────────────────────────

PAIN_KEYWORDS = {
    "time_cost": [
        "takes forever", "slow", "waste", "wasted", "wasting", "too long",
        "hours", "all day", "wait", "waiting", "delays", "delayed",
    ],
    "quality_cost": [
        "wrong", "error", "errors", "unreliable", "inaccurate", "buggy",
        "broken", "glitch", "incorrect", "mistake", "mistakes", "fails",
    ],
    "social_cost": [
        "embarrassing", "my boss", "my manager", "look bad", "looks bad",
        "unprofessional", "in front of", "colleagues",
    ],
    "cognitive_cost": [
        "confusing", "can't figure", "hard to understand", "complicated",
        "overwhelming", "unintuitive", "not intuitive", "steep learning",
        "learning curve", "too complex",
    ],
    "money_cost": [
        "expensive", "not worth", "pricey", "overpriced", "costs too much",
        "too much money", "waste of money", "rip off", "ripoff",
    ],
    "trust_cost": [
        "lost data", "lost my data", "can't count on", "can't trust",
        "broke", "crashed", "crash", "unreliable", "data loss",
        "disappeared", "went down",
    ],
}

OUTCOME_KEYWORDS = {
    "speed": [
        "faster", "quicker", "quick", "instant", "rapid", "save time",
        "saves time", "time-saving", "efficient", "efficiency",
    ],
    "accuracy": [
        "accurate", "correct", "reliable", "precise", "precision",
        "dependable", "consistent", "no errors", "error-free",
    ],
    "control": [
        "control", "customize", "customizable", "configure", "configurable",
        "flexible", "flexibility", "options", "settings", "adjust",
    ],
    "simplicity": [
        "simple", "easy", "easier", "straightforward", "intuitive",
        "user-friendly", "user friendly", "no-brainer", "clean",
        "minimal", "just works",
    ],
    "trust": [
        "predictable", "no surprises", "dependable", "stable", "solid",
        "reliable", "always works", "never fails", "count on",
    ],
    "status": [
        "professional", "impressive", "looks great", "polished",
        "beautiful", "sleek", "modern", "cutting-edge", "premium",
    ],
}

WORKAROUND_KEYWORDS = {
    "manual": [
        "spreadsheet", "manually", "by hand", "pen and paper",
        "excel", "google sheets", "paper", "handwritten", "notepad",
        "sticky notes", "whiteboard",
    ],
    "abandoned": [
        "gave up", "stopped", "quit", "abandoned", "don't bother",
        "stopped trying", "no longer", "moved on",
    ],
    "hybrid": [
        "combine", "alongside", "in addition to", "supplement",
        "workaround", "work around", "plus", "together with",
    ],
}

# Generic phrases that appear in any business category (hygiene factors).
GENERIC_PRAISE = [
    "great customer service", "easy to use", "fast delivery",
    "friendly staff", "good value", "highly recommend",
    "love this product", "works great", "very helpful",
    "excellent service", "best ever", "amazing product",
    "five stars", "5 stars", "would recommend",
]


# ── Helpers ──────────────────────────────────────────────────────────────

def _normalize(text):
    """Lowercase and collapse whitespace."""
    return re.sub(r"\s+", " ", text.lower().strip())


def _match_axis(text, keyword_map):
    """Return the best-matching axis label or None.

    If multiple labels match, pick the one with the most keyword hits.
    """
    text_norm = _normalize(text)
    scores = {}
    for label, keywords in keyword_map.items():
        hits = sum(1 for kw in keywords if kw in text_norm)
        if hits > 0:
            scores[label] = hits
    if not scores:
        return None
    return max(scores, key=scores.get)


def _detect_competitor(text):
    """Very basic competitor mention detection.

    Returns competitor name if found, else None.  Looks for patterns like
    "switched to X", "using X instead", "moved to X".
    """
    text_norm = _normalize(text)
    patterns = [
        r"switch(?:ed)? to (\w+)",
        r"moved? to (\w+)",
        r"using (\w+) instead",
        r"went (?:back )?to (\w+)",
        r"prefer (\w+)",
    ]
    for pat in patterns:
        m = re.search(pat, text_norm)
        if m:
            candidate = m.group(1)
            # Filter out generic words that aren't tool names.
            if candidate not in {"it", "the", "a", "an", "this", "that",
                                 "them", "something", "nothing", "another"}:
                return candidate
    return None


def _is_generic(text):
    """Return True if the review text matches a generic praise pattern."""
    text_norm = _normalize(text)
    return any(phrase in text_norm for phrase in GENERIC_PRAISE)


# ── Core pipeline ────────────────────────────────────────────────────────

def parse_input(filepath):
    """Parse CSV or JSON file.  Returns list of dicts with at least 'text'."""
    path = Path(filepath)
    suffix = path.suffix.lower()

    if suffix == ".json":
        with open(path, encoding="utf-8") as f:
            data = json.load(f)
        if not isinstance(data, list):
            raise ValueError("JSON input must be an array of objects.")
        for item in data:
            if "text" not in item:
                raise ValueError("Each JSON object must have a 'text' field.")
        return data

    # Default: treat as CSV.
    rows = []
    with open(path, newline="", encoding="utf-8") as f:
        reader = csv.DictReader(f)
        if "text" not in (reader.fieldnames or []):
            raise ValueError("CSV must have a 'text' column.")
        for row in reader:
            rows.append(row)
    return rows


def classify_review(review_text):
    """Classify a single review on all three axes.

    Returns dict: {pain, outcome, workaround, competitor, generic}.
    """
    pain = _match_axis(review_text, PAIN_KEYWORDS)
    outcome = _match_axis(review_text, OUTCOME_KEYWORDS)

    # Workaround: check competitor first, then keyword map.
    competitor = _detect_competitor(review_text)
    if competitor:
        workaround = "competitor"
    else:
        workaround = _match_axis(review_text, WORKAROUND_KEYWORDS) or "unknown"

    return {
        "pain": pain,
        "outcome": outcome,
        "workaround": workaround,
        "competitor": competitor,
        "generic": _is_generic(review_text),
    }


def cluster_reviews(reviews):
    """Cluster a list of review dicts.

    Returns (clusters, classified_reviews) where clusters is a dict keyed
    by (pain, outcome) tuples with metadata.
    """
    classified = []
    cluster_map = defaultdict(list)

    for review in reviews:
        text = review.get("text", "")
        cls = classify_review(text)
        cls["text"] = text
        cls["rating"] = review.get("rating")
        cls["date"] = review.get("date")
        cls["source"] = review.get("source")
        cls["author"] = review.get("author")
        classified.append(cls)

        key = (cls["pain"], cls["outcome"])
        cluster_map[key].append(cls)

    # Build cluster summaries, sorted by volume descending.
    clusters = []
    for (pain, outcome), members in sorted(
        cluster_map.items(), key=lambda kv: -len(kv[1])
    ):
        # Determine dominant workaround.
        wa_counts = Counter(m["workaround"] for m in members)
        dominant_wa = wa_counts.most_common(1)[0][0]

        # Competitor name if applicable.
        competitors = [m["competitor"] for m in members if m["competitor"]]
        comp_name = Counter(competitors).most_common(1)[0][0] if competitors else None

        # Confidence based on convergence threshold.
        count = len(members)
        if count >= 3:
            sources = set(m["source"] for m in members if m.get("source"))
            confidence = "high" if len(sources) > 1 else "medium"
        else:
            confidence = "low"

        # Check if cluster is mostly generic.
        generic_count = sum(1 for m in members if m["generic"])
        is_generic = generic_count > len(members) / 2

        # Build label.
        pain_label = pain or "mixed_pain"
        outcome_label = outcome or "unclear_outcome"
        wa_label = dominant_wa
        if comp_name and wa_label == "competitor":
            wa_label = f"competitor ({comp_name})"

        label = f"{pain_label} — users want {outcome_label} but currently {wa_label}"

        # Representative quotes (up to 3).
        quotes = [m["text"] for m in members[:3]]

        clusters.append({
            "pain": pain,
            "outcome": outcome,
            "workaround": dominant_wa,
            "competitor": comp_name,
            "label": label,
            "count": count,
            "confidence": confidence,
            "generic": is_generic,
            "quotes": quotes,
        })

    return clusters, classified


def render_brief(clusters, total_count, source_name, output_dir):
    """Render review-brief.md from clusters into output_dir."""
    template_path = Path(__file__).resolve().parent.parent / "templates" / "review-brief.md"

    date_str = datetime.now().strftime("%Y-%m-%d")

    # Filter out generic clusters for the top-3 ranking.
    featured = [c for c in clusters if not c["generic"]]
    # If everything is generic, show them anyway.
    if not featured:
        featured = clusters
    top3 = featured[:3]

    lines = []
    lines.append(f"# Review Brief — {source_name}")
    lines.append("")
    lines.append(f"**Reviews parsed:** {total_count}")
    lines.append(f"**Source:** {source_name}")
    lines.append(f"**Date:** {date_str}")
    lines.append("")
    lines.append("## Top 3 clusters (ranked by volume)")
    lines.append("")

    for rank, cluster in enumerate(top3, 1):
        pct = round(cluster["count"] / total_count * 100) if total_count else 0
        lines.append(f"### {rank}. {cluster['label']} — {cluster['count']} reviews ({pct}%)")
        lines.append("")
        lines.append("**Representative quotes:**")
        for q in cluster["quotes"]:
            lines.append(f'- "{q}"')
        lines.append("")
        lines.append("**Hypothesized job:**")
        lines.append(f"When [situation], I want to [{cluster['outcome'] or '...'}], so I can [outcome].")
        lines.append("")
        lines.append(f"**Confidence:** {cluster['confidence']}")
        lines.append("")

    # Underserved forces.
    lines.append("## Underserved forces (interview priorities)")
    lines.append("")
    lines.append("Reviews are systematically weak on these two Switch forces. Probe them in the interview.")
    lines.append("")
    lines.append('- **Habit:** reviewers rarely describe the muscle memory / workflow inertia that keeps them with the old. Ask: "What have you been using, even if it\'s duct tape and a spreadsheet?"')
    lines.append('- **Anxiety:** only unhappy switchers leave reviews — happy stayers are invisible. Ask at least one real user what worries them about switching.')
    lines.append("")

    # Conflicts.
    lines.append("## Conflicts / tensions")
    lines.append("")
    conflict_pairs = _find_conflicts(clusters)
    if conflict_pairs:
        for conflict in conflict_pairs:
            lines.append(f"- {conflict}")
    else:
        lines.append("- No obvious conflicts detected in this review set.")
    lines.append("")

    # Interview prep.
    lines.append("## Interview prep")
    lines.append("")
    if total_count >= 30:
        lines.append("Given these clusters, skip these questions in the interview (reviews already answered them):")
        lines.append("")
        if top3:
            lines.append(f"- [ ] Pain around {top3[0]['pain'] or 'top cluster'} is well-documented")
        if len(top3) > 1:
            lines.append(f"- [ ] Desired outcome of {top3[1]['outcome'] or 'second cluster'} is clear")
    else:
        lines.append("Volume < 30 — treat clusters as hypotheses. Run the full interview.")
    lines.append("")
    lines.append("Emphasize these instead:")
    lines.append("")
    lines.append("- [ ] Walk through one specific switching moment")
    lines.append("- [ ] Probe all four Switch forces (especially habit + anxiety)")
    lines.append("- [ ] Push for measurable outcomes")
    lines.append("")
    lines.append("---")
    lines.append("")
    lines.append("**Next step:** run `/jtbd` in Interview mode, using this brief as pre-seed.")
    lines.append("")

    out_path = Path(output_dir) / "review-brief.md"
    out_path.parent.mkdir(parents=True, exist_ok=True)
    out_path.write_text("\n".join(lines), encoding="utf-8")
    return str(out_path)


def _find_conflicts(clusters):
    """Detect clusters that contradict each other."""
    conflicts = []
    # Conflict: same pain, opposite desired outcomes.
    by_pain = defaultdict(list)
    for c in clusters:
        if c["pain"]:
            by_pain[c["pain"]].append(c)
    for pain, group in by_pain.items():
        outcomes = set(c["outcome"] for c in group if c["outcome"])
        if "simplicity" in outcomes and "control" in outcomes:
            conflicts.append(
                f"Tension in {pain}: some users want simplicity, others want control."
            )
        if "speed" in outcomes and "accuracy" in outcomes:
            conflicts.append(
                f"Tension in {pain}: some users want speed, others want accuracy."
            )
    return conflicts


def build_summary(clusters, total_count):
    """Build a JSON-serializable summary dict."""
    return {
        "total_reviews": total_count,
        "cluster_count": len(clusters),
        "clusters": [
            {
                "label": c["label"],
                "count": c["count"],
                "confidence": c["confidence"],
                "generic": c["generic"],
            }
            for c in clusters
        ],
    }


# ── CLI ──────────────────────────────────────────────────────────────────

def main(argv=None):
    parser = argparse.ArgumentParser(
        description="Mine product reviews and cluster on pain/outcome/workaround axes."
    )
    parser.add_argument("input", help="Path to CSV or JSON file of reviews.")
    parser.add_argument(
        "-o", "--output-dir", default=".",
        help="Directory to write review-brief.md (default: current dir).",
    )
    parser.add_argument(
        "--source-name", default=None,
        help="Human-readable name for the review source (default: filename).",
    )
    args = parser.parse_args(argv)

    reviews = parse_input(args.input)
    clusters, classified = cluster_reviews(reviews)
    total = len(reviews)
    source = args.source_name or Path(args.input).stem

    brief_path = render_brief(clusters, total, source, args.output_dir)
    summary = build_summary(clusters, total)
    summary["brief_path"] = brief_path

    print(json.dumps(summary, indent=2))
    return summary


if __name__ == "__main__":
    main()

"""Outcome-Driven Innovation (ODI) opportunity scoring."""


def score(importance, satisfaction):
    return importance + max(0, importance - satisfaction)


def tier(opportunity_score):
    if opportunity_score >= 12:
        return "prioritize"
    if opportunity_score <= 8:
        return "skip"
    return "marginal"

"""Granularity Gate validator — pattern-based heuristic scorer.

Scores a draft JTBD JSON on five dimensions (0-2 each).
Any dimension at 0 blocks save until rewrite.
See references/granularity_fixes.md for rewrite prompts.
"""

import json
import re
import sys


GENERIC_ACTORS = {
    "user", "users", "people", "person", "someone", "everyone",
    "customer", "customers", "anybody", "they", "them",
}

ALWAYS_WORDS = {
    "always", "in general", "whenever", "all the time", "every time",
    "usually", "often", "sometimes",
}

VAGUE_WORKAROUNDS = {
    "nothing", "none", "they don't", "various tools", "various",
    "whatever they have", "different things", "a few things",
}

VAGUE_OUTCOMES = {
    "better", "improved", "more efficient", "good", "great",
    "faster", "easier", "smoother", "nicer", "enhanced",
}


def score_actor(text):
    if not text:
        return 0
    tokens = text.lower().strip().split()
    non_filler = [t for t in tokens if t not in {"the", "a", "an", "some", "all", "any"}]
    if not non_filler:
        return 0
    if len(non_filler) <= 2 and non_filler[0] in GENERIC_ACTORS:
        return 0
    if any(t in GENERIC_ACTORS for t in non_filler) and len(non_filler) <= 3:
        return 1
    if len(non_filler) >= 4:
        return 2
    return 1


def score_context(text):
    if not text:
        return 0
    lower = text.lower().strip()
    if any(w in lower for w in ALWAYS_WORDS):
        return 0
    if re.search(r"when\s+\w+", lower) or re.search(r"after\s+\w+", lower):
        if len(lower.split()) >= 8:
            return 2
        return 1
    if len(lower.split()) < 5:
        return 0
    return 1


def score_workaround(text):
    if not text:
        return 0
    lower = text.lower().strip()
    for phrase in VAGUE_WORKAROUNDS:
        if lower == phrase or lower.startswith(phrase):
            return 0
    if any(char in text for char in ["→", "->", "then", "paste", "copy", "manually"]):
        return 2
    if len(text.split()) >= 6:
        return 1
    return 0


def score_outcome(text):
    if not text:
        return 0
    lower = text.lower().strip()
    for phrase in VAGUE_OUTCOMES:
        if lower == phrase or (lower.startswith(phrase) and len(lower.split()) <= 3):
            return 0
    if re.search(r"\d+", text):
        return 2
    if any(w in lower for w in ["half", "double", "reduce", "cut", "increase", "decrease", "from", "under"]):
        return 1
    if len(text.split()) >= 6:
        return 1
    return 0


def score_evidence(quotes):
    if not quotes:
        return 0
    if isinstance(quotes, list) and len(quotes) == 0:
        return 0
    if isinstance(quotes, list):
        has_attribution = any("—" in q or " - " in q or "said" in q.lower() for q in quotes)
        has_verbatim = any(q.startswith('"') or q.startswith("'") or q.startswith("“") for q in quotes)
        if has_attribution and has_verbatim:
            return 2
        return 1
    return 0


def validate(data):
    situation = data.get("jtbd", {}).get("situation", "")
    outcome = data.get("jtbd", {}).get("outcome", "")
    workaround = data.get("problem", {}).get("what_hurts", "")
    quotes = data.get("evidence", {}).get("quotes", [])
    actor = situation.split(",")[0] if situation else ""

    scores = {
        "actor_specificity": score_actor(actor),
        "context_trigger": score_context(situation),
        "current_workaround": score_workaround(workaround),
        "measurable_outcome": score_outcome(outcome),
        "evidence_quote": score_evidence(quotes),
    }

    total = sum(scores.values())
    blocking = [k for k, v in scores.items() if v == 0]

    return {
        "scores": scores,
        "total": total,
        "max": 10,
        "blocking": blocking,
        "passes": len(blocking) == 0,
    }


if __name__ == "__main__":
    if len(sys.argv) < 2:
        print("Usage: python validate_granularity.py <path-to-jtbd.json>")
        sys.exit(1)
    with open(sys.argv[1]) as f:
        data = json.load(f)
    result = validate(data)
    print(json.dumps(result, indent=2))
    sys.exit(0 if result["passes"] else 1)

"""ODI outcome statement validator.

Validates that outcome statements follow the strict ODI format:
[Direction] the [metric] it takes to [action] when [situation]
"""

import re
import sys
import json


DIRECTIONS = {"minimize", "increase", "maintain", "reduce"}
METRICS = {
    "time", "likelihood", "amount", "number", "frequency",
    "cost", "effort", "risk", "errors", "steps",
}


def validate_statement(statement):
    lower = statement.lower().strip()
    issues = []

    has_direction = any(lower.startswith(d) for d in DIRECTIONS)
    if not has_direction:
        issues.append(f"Must start with a direction ({', '.join(sorted(DIRECTIONS))})")

    has_metric = any(m in lower for m in METRICS)
    if not has_metric:
        issues.append(f"Must include a metric ({', '.join(sorted(METRICS))})")

    has_action = "to " in lower and len(lower.split("to ", 1)) > 1
    if not has_action:
        issues.append("Must include 'to [action]'")

    has_context = "when " in lower or "while " in lower or "during " in lower
    if not has_context:
        issues.append("Should include 'when [situation]' for context (optional but recommended)")

    solution_words = ["button", "tool", "app", "dashboard", "plugin", "widget", "modal", "sidebar"]
    describes_solution = any(f" {w} " in f" {lower} " for w in solution_words)
    if describes_solution:
        issues.append("Describes a solution, not an outcome. Rewrite as what the user achieves.")

    return {
        "statement": statement,
        "valid": len([i for i in issues if "optional" not in i]) == 0,
        "issues": issues,
    }


def validate_all(outcomes):
    results = [validate_statement(o["statement"]) for o in outcomes]
    return {
        "results": results,
        "all_valid": all(r["valid"] for r in results),
        "count": len(results),
        "valid_count": sum(1 for r in results if r["valid"]),
    }


if __name__ == "__main__":
    if len(sys.argv) < 2:
        print("Usage: python validate_outcome.py <path-to-jtbd.json>")
        sys.exit(1)
    with open(sys.argv[1]) as f:
        data = json.load(f)
    outcomes = data.get("odi", {}).get("outcomes", [])
    if not outcomes:
        print("No ODI outcomes found in JSON.")
        sys.exit(0)
    result = validate_all(outcomes)
    print(json.dumps(result, indent=2))
    sys.exit(0 if result["all_valid"] else 1)

{
  "_note": "This file demonstrates the Granularity Gate in action. The 'bad' version passes a naive check but fails all 5 gate dimensions. The 'fixed' version is what the gate forces out through its rewrite prompts. Compare side-by-side.",

  "bad_version": {
    "name": "productivity-booster",
    "hook": "A delightful productivity tool that helps users work better.",
    "jtbd": {
      "situation": "When users need to be more productive",
      "motivation": "I want to use a tool that helps me",
      "outcome": "So I can get more done"
    },
    "problem": {
      "what_hurts": "Current tools don't meet user needs"
    },
    "needs": {
      "functional": ["Be fast", "Be easy to use"],
      "emotional": ["Feel empowered"]
    },
    "switch_forces": {
      "push": "Current tools are not great",
      "pull": "Our tool is better",
      "habit": "Users are used to what they have",
      "anxiety": "Change is hard"
    },
    "outputs": ["Productivity improvements"],
    "evidence": {
      "source": "interview",
      "quotes": [],
      "weaknesses": []
    },
    "_gate_failures": {
      "actor_specificity": "0 — 'users' is generic",
      "context_trigger": "0 — 'when users need to be more productive' is always",
      "current_workaround": "0 — 'current tools don't meet needs' is content-free",
      "measurable_outcome": "0 — 'get more done' has no direction or number",
      "evidence_quote": "0 — empty quotes array",
      "jargon_violations": "'delightful', 'empowered' — both in jargon_blacklist.md"
    }
  },

  "fixed_version": {
    "name": "sprint-retro-compiler",
    "hook": "For engineering leads at 10-30 person teams: turns Friday retro notes + sprint board state into a Monday-ready exec summary in under 5 minutes.",
    "jtbd": {
      "situation": "When an engineering lead at a Series-A/B SaaS finishes Friday sprint retro and faces a Monday exec update with no time to write it up over the weekend",
      "motivation": "I want to convert retro bullets and sprint board state into a structured exec summary with velocity, risks, and asks",
      "outcome": "So I can cut summary prep from 45 minutes to under 5 and keep the weekend clean"
    },
    "problem": {
      "what_hurts": "Rewriting the same retro material into exec-speak every Monday while trying to reconstruct what 'shipped' means without re-reading 40 PRs",
      "cost_today": "45 minutes every Monday morning under pressure; occasionally misremember what shipped and get caught out in exec review"
    },
    "needs": {
      "functional": [
        "Pull sprint board state (closed tickets, open blockers, velocity numbers)",
        "Ingest retro notes from Miro or a Google Doc",
        "Populate a structured exec-summary template with velocity, risks, and specific asks"
      ],
      "emotional": [
        "Walk into Monday's exec review without scrambling",
        "Trust the summary accurately reflects the sprint"
      ]
    },
    "switch_forces": {
      "push": "'I spend every Monday morning rewriting Friday retro into CEO-speak — by the time I'm done my coffee is cold and the exec call is in 10 minutes.' — Eng lead at 22-person Series-B SaaS",
      "pull": "'If the summary drafted itself from what the team already produced, I'd get my Monday back.'",
      "habit": "Already uses Linear for sprints and Miro for retros. Team expects exec updates in the same Google Doc format every week.",
      "anxiety": "'If it auto-generates and gets velocity wrong, I'll look like I don't know my own team. Last month my deputy caught a 30% error in a manual summary — I can't risk that happening in front of the CEO.'"
    },
    "outputs": [
      "Exec-summary Google Doc populated with velocity (closed story points), risks (open blockers), and asks (unblocked dependencies)",
      "Source-link column showing where each claim came from (Linear ticket IDs, Miro sticky IDs, PR links)",
      "Gap report flagging sections where retro or board data was too thin"
    ],
    "evidence": {
      "source": "interview",
      "quotes": [
        "\"I spend every Monday morning rewriting Friday retro into CEO-speak — by the time I'm done my coffee is cold and the exec call is in 10 minutes.\" — Eng lead, 22-person Series-B SaaS",
        "\"Last month my deputy caught a 30% error in a manual summary — I can't risk that happening in front of the CEO.\" — same eng lead"
      ],
      "weaknesses": []
    }
  }
}

{
  "name": "readiness-doc-helper",
  "hook": "Cuts Series-B PMs' launch-readiness doc prep from 90 min to under 15 by pulling PR/Linear/Slack signal directly into the template.",
  "jtbd": {
    "situation": "When a PM at a Series-B SaaS is preparing launch-readiness docs the Sunday before a Monday go/no-go review",
    "motivation": "I want to assemble the doc from sources of truth (PRs merged, Linear tickets closed, Slack decisions) without re-interviewing the team",
    "outcome": "So I can cut prep from 90 minutes to under 15 and show up to the meeting with a defensible readiness view"
  },
  "problem": {
    "what_hurts": "Rebuilding the readiness context from scratch every launch because the tracker is always 24-48 hours stale",
    "cost_today": "Every Sunday for 2-3 hours pre-launch, plus the stress of walking into Monday under-prepared"
  },
  "needs": {
    "functional": [
      "Pull PR/ticket/Slack signal for a named feature or milestone",
      "Populate a structured readiness template with source links",
      "Flag gaps where signal is missing"
    ],
    "emotional": [
      "Walk into Monday confident the doc is defensible",
      "Reclaim the Sunday evening"
    ]
  },
  "switch_forces": {
    "push": "Every Sunday rebuilding the doc by hand while team is offline — 'I spent my entire Sunday chasing Slack threads'",
    "pull": "'What if the doc wrote itself from the signal my team already produces?'",
    "habit": "Already using Notion for the readiness template; team expects the same format",
    "anxiety": "'If it auto-generates and misses something, I still own it on Monday — last launch we missed a flag and got burned'"
  },
  "outputs": [
    "Populated readiness doc (Markdown, compatible with existing Notion template)",
    "Signal-to-section map showing where each claim came from",
    "Gap list — sections where signal was too thin to auto-fill"
  ],
  "evidence": {
    "source": "interview",
    "quotes": [
      "\"I spend my entire Sunday rebuilding the readiness doc because nobody updated the tracker\" — PM at Series-B SaaS",
      "\"Last launch I missed a rollback flag that was in a Slack thread from 3am — we shipped broken.\" — same PM, 2 weeks later"
    ],
    "weaknesses": []
  }
}

"""Tests for ingest_transcript.py — transcript parsing, field mapping,
switch force detection, quote extraction, and confidence scoring."""

import sys
import os

# Add scripts dir to path
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "scripts"))

from ingest_transcript import (
    split_into_chunks,
    _score_chunk,
    map_field,
    map_switch_force,
    extract_quotes,
    identify_gaps,
    ingest,
    SITUATION_PHRASES,
    MOTIVATION_PHRASES,
    OUTCOME_PHRASES,
    PAIN_PHRASES,
    WORKAROUND_PHRASES,
    PUSH_PHRASES,
    PULL_PHRASES,
    HABIT_PHRASES,
    ANXIETY_PHRASES,
)


# ── Sample transcripts ──────────────────────────────────────────────

QA_TRANSCRIPT = """Q: Tell me about the problem you're facing.
A: Every time I need to generate a report, I have to manually copy data from three different spreadsheets. It's incredibly frustrating and I waste at least two hours every week on it.

Q: What are you trying to achieve?
A: I want a single dashboard so that I can see all metrics in one place. Right now I use Excel and it's painful.

Q: What worries you about switching?
A: I'm worried that the team won't adopt a new tool. Last time we tried something new, it was a disaster.

Q: What would the ideal solution look like?
A: I wish I could just click a button and have everything updated. Imagine if I never had to touch a spreadsheet again. Finally, I could focus on actual analysis."""

SPEAKER_TRANSCRIPT = """Interviewer: What brought you here today?
Sarah: When our team grew past 10 people, the old process broke. I need a way to track everyone's tasks without manually checking in.

Interviewer: How do you handle it now?
Sarah: We currently use a shared Google Sheet. I've been using it for years but it's terrible. Drives me crazy.

Interviewer: What's your biggest fear about changing?
Sarah: I'm afraid the team won't learn the new system. What if we lose data during migration?"""

PARAGRAPH_TRANSCRIPT = """The problem started last week when we had three client calls back to back. I was trying to find the notes from previous meetings but everything was scattered across email, Slack, and random documents.

I want a single place where all meeting notes live. I need to search across all of them quickly. The goal is to never lose context between calls.

The worst part is that I waste at least 30 minutes before every call just hunting for old notes. It's frustrating and makes me look unprepared. I can't stand it.

Right now I use a combination of Apple Notes and Google Docs. I copy and paste between them manually. It's a terrible workaround but I've been using it for years."""


# ── Chunk splitting tests ────────────────────────────────────────────

class TestSplitIntoChunks:

    def test_qa_format(self):
        chunks = split_into_chunks(QA_TRANSCRIPT)
        assert len(chunks) >= 4
        # Should detect Q and A as speakers
        speakers = {c["speaker"] for c in chunks}
        assert "Q" in speakers or "A" in speakers

    def test_speaker_format(self):
        chunks = split_into_chunks(SPEAKER_TRANSCRIPT)
        assert len(chunks) >= 4
        speakers = {c["speaker"] for c in chunks}
        assert "Interviewer" in speakers or "Sarah" in speakers

    def test_paragraph_format(self):
        chunks = split_into_chunks(PARAGRAPH_TRANSCRIPT)
        assert len(chunks) == 4
        # No speakers in paragraph mode
        assert all(c["speaker"] is None for c in chunks)

    def test_chunk_indices(self):
        chunks = split_into_chunks(QA_TRANSCRIPT)
        indices = [c["index"] for c in chunks]
        assert indices == list(range(len(chunks)))

    def test_empty_text(self):
        chunks = split_into_chunks("")
        assert chunks == []

    def test_single_paragraph(self):
        chunks = split_into_chunks("Just one paragraph with no breaks.")
        assert len(chunks) == 1
        assert chunks[0]["text"] == "Just one paragraph with no breaks."


# ── Scoring tests ────────────────────────────────────────────────────

class TestScoreChunk:

    def test_no_match(self):
        score = _score_chunk("The weather is nice today.", PAIN_PHRASES)
        assert score == 0.0

    def test_single_match(self):
        score = _score_chunk("This is really frustrating.", PAIN_PHRASES)
        assert 0.4 <= score <= 0.6

    def test_multiple_matches(self):
        score = _score_chunk(
            "I'm frustrated and it's a waste of time. The problem is painful.",
            PAIN_PHRASES,
        )
        assert score >= 0.7

    def test_case_insensitive(self):
        score = _score_chunk("I WANT this to work. I NEED it now.", MOTIVATION_PHRASES)
        assert score > 0.0

    def test_max_cap(self):
        # Even with many matches, score caps at 1.0
        text = "frustrated annoying waste problem struggle painful broken terrible"
        score = _score_chunk(text, PAIN_PHRASES)
        assert score <= 1.0


# ── Field mapping tests ─────────────────────────────────────────────

class TestFieldMapping:

    def test_situation_detected(self):
        chunks = split_into_chunks(QA_TRANSCRIPT)
        result = map_field(chunks, SITUATION_PHRASES)
        assert result["confidence"] > 0.0
        assert result["source_chunk"] >= 0
        assert "every time" in result["text"].lower() or "right now" in result["text"].lower()

    def test_motivation_detected(self):
        chunks = split_into_chunks(QA_TRANSCRIPT)
        result = map_field(chunks, MOTIVATION_PHRASES)
        assert result["confidence"] > 0.0
        assert "want" in result["text"].lower() or "need" in result["text"].lower()

    def test_outcome_detected(self):
        chunks = split_into_chunks(QA_TRANSCRIPT)
        result = map_field(chunks, OUTCOME_PHRASES)
        assert result["confidence"] > 0.0
        assert "so that" in result["text"].lower() or "so i can" in result["text"].lower()

    def test_pain_detected(self):
        chunks = split_into_chunks(QA_TRANSCRIPT)
        result = map_field(chunks, PAIN_PHRASES)
        assert result["confidence"] > 0.0
        assert "frustrat" in result["text"].lower() or "waste" in result["text"].lower()

    def test_workaround_detected(self):
        chunks = split_into_chunks(QA_TRANSCRIPT)
        result = map_field(chunks, WORKAROUND_PHRASES)
        assert result["confidence"] > 0.0

    def test_no_match_returns_empty(self):
        chunks = [{"index": 0, "text": "Hello world.", "speaker": None}]
        result = map_field(chunks, PAIN_PHRASES)
        assert result["confidence"] == 0.0
        assert result["text"] == ""
        assert result["source_chunk"] == -1


# ── Switch force detection tests ─────────────────────────────────────

class TestSwitchForces:

    def test_push_detected(self):
        chunks = split_into_chunks(QA_TRANSCRIPT)
        result = map_switch_force(chunks, PUSH_PHRASES)
        assert result["confidence"] > 0.0
        assert "frustrated" in result["text"].lower() or "every time" in result["text"].lower()

    def test_pull_detected(self):
        chunks = split_into_chunks(QA_TRANSCRIPT)
        result = map_switch_force(chunks, PULL_PHRASES)
        assert result["confidence"] > 0.0
        assert "wish" in result["text"].lower() or "imagine" in result["text"].lower()

    def test_habit_detected(self):
        chunks = split_into_chunks(SPEAKER_TRANSCRIPT)
        result = map_switch_force(chunks, HABIT_PHRASES)
        assert result["confidence"] > 0.0

    def test_anxiety_detected(self):
        chunks = split_into_chunks(QA_TRANSCRIPT)
        result = map_switch_force(chunks, ANXIETY_PHRASES)
        assert result["confidence"] > 0.0
        assert "worried" in result["text"].lower() or "last time we tried" in result["text"].lower()

    def test_no_force_match(self):
        chunks = [{"index": 0, "text": "The sky is blue.", "speaker": None}]
        result = map_switch_force(chunks, PUSH_PHRASES)
        assert result["confidence"] == 0.0
        assert result["text"] == ""


# ── Quote extraction tests ───────────────────────────────────────────

class TestQuoteExtraction:

    def test_extracts_pain_quotes(self):
        chunks = split_into_chunks(QA_TRANSCRIPT)
        quotes = extract_quotes(chunks)
        assert len(quotes) > 0
        # Should find at least one quote with frustration language
        has_pain = any("frustrat" in q.lower() or "waste" in q.lower() for q in quotes)
        assert has_pain

    def test_no_duplicates(self):
        chunks = split_into_chunks(QA_TRANSCRIPT)
        quotes = extract_quotes(chunks)
        assert len(quotes) == len(set(quotes))

    def test_empty_transcript_no_quotes(self):
        quotes = extract_quotes([])
        assert quotes == []

    def test_preserves_verbatim(self):
        text = "I wish I could just click a button."
        chunks = [{"index": 0, "text": text, "speaker": None}]
        quotes = extract_quotes(chunks)
        assert any(text in q for q in quotes)


# ── Gap identification tests ─────────────────────────────────────────

class TestGapIdentification:

    def test_low_confidence_flagged(self):
        fields = {
            "situation": {"text": "context", "confidence": 0.8, "source_chunk": 0},
            "motivation": {"text": "", "confidence": 0.0, "source_chunk": -1},
        }
        gaps = identify_gaps(fields)
        assert len(gaps) == 1
        assert "motivation" in gaps[0]

    def test_all_high_confidence_no_gaps(self):
        fields = {
            "situation": {"text": "x", "confidence": 0.5, "source_chunk": 0},
            "motivation": {"text": "y", "confidence": 0.4, "source_chunk": 1},
        }
        gaps = identify_gaps(fields)
        assert gaps == []

    def test_threshold_at_03(self):
        fields = {
            "f1": {"text": "x", "confidence": 0.3, "source_chunk": 0},
            "f2": {"text": "y", "confidence": 0.29, "source_chunk": 1},
        }
        gaps = identify_gaps(fields)
        assert len(gaps) == 1
        assert "f2" in gaps[0]


# ── Full ingest integration tests ────────────────────────────────────

class TestIngest:

    def test_qa_transcript_full(self):
        result = ingest(QA_TRANSCRIPT)
        assert "fields" in result
        assert "switch_forces" in result
        assert "quotes" in result
        assert "gaps" in result

        # All 5 fields present
        assert set(result["fields"].keys()) == {
            "situation", "motivation", "outcome", "what_hurts", "current_workaround"
        }

        # All 4 switch forces present
        assert set(result["switch_forces"].keys()) == {
            "push", "pull", "habit", "anxiety"
        }

        # This transcript has strong signals, so most fields should map
        high_conf = sum(
            1 for f in result["fields"].values() if f["confidence"] >= 0.3
        )
        assert high_conf >= 3

    def test_speaker_transcript_full(self):
        result = ingest(SPEAKER_TRANSCRIPT)
        assert result["fields"]["what_hurts"]["confidence"] > 0.0
        assert result["switch_forces"]["habit"]["confidence"] > 0.0
        assert result["switch_forces"]["anxiety"]["confidence"] > 0.0

    def test_paragraph_transcript_full(self):
        result = ingest(PARAGRAPH_TRANSCRIPT)
        assert result["fields"]["motivation"]["confidence"] > 0.0
        assert result["fields"]["current_workaround"]["confidence"] > 0.0
        assert len(result["quotes"]) > 0

    def test_empty_transcript(self):
        result = ingest("")
        assert all(f["confidence"] == 0.0 for f in result["fields"].values())
        assert all(f["confidence"] == 0.0 for f in result["switch_forces"].values())
        assert result["quotes"] == []
        assert len(result["gaps"]) == 5

    def test_confidence_scores_bounded(self):
        result = ingest(QA_TRANSCRIPT)
        for f in result["fields"].values():
            assert 0.0 <= f["confidence"] <= 1.0
        for f in result["switch_forces"].values():
            assert 0.0 <= f["confidence"] <= 1.0

    def test_output_structure_matches_spec(self):
        result = ingest(QA_TRANSCRIPT)
        # Fields have text, confidence, source_chunk
        for f in result["fields"].values():
            assert "text" in f
            assert "confidence" in f
            assert "source_chunk" in f
        # Switch forces have text, confidence
        for f in result["switch_forces"].values():
            assert "text" in f
            assert "confidence" in f
        # Quotes is a list of strings
        assert isinstance(result["quotes"], list)
        # Gaps is a list of strings
        assert isinstance(result["gaps"], list)

"""Tests for mine_reviews.py — review clustering on pain/outcome/workaround axes."""
import csv
import json
import os
import sys
import tempfile
import unittest
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "scripts"))

import mine_reviews  # noqa: E402


class ParseInputTests(unittest.TestCase):
    """CSV and JSON parsing."""

    def test_csv_basic(self):
        with tempfile.NamedTemporaryFile(
            mode="w", suffix=".csv", delete=False, newline=""
        ) as f:
            writer = csv.DictWriter(f, fieldnames=["text", "rating", "source"])
            writer.writeheader()
            writer.writerow({"text": "Too slow", "rating": "2", "source": "G2"})
            writer.writerow({"text": "Great tool", "rating": "5", "source": "G2"})
            path = f.name
        try:
            rows = mine_reviews.parse_input(path)
            self.assertEqual(len(rows), 2)
            self.assertEqual(rows[0]["text"], "Too slow")
            self.assertEqual(rows[1]["rating"], "5")
        finally:
            os.unlink(path)

    def test_json_basic(self):
        data = [
            {"text": "Takes forever to load", "rating": 1},
            {"text": "Love the simplicity", "rating": 5},
        ]
        with tempfile.NamedTemporaryFile(
            mode="w", suffix=".json", delete=False
        ) as f:
            json.dump(data, f)
            path = f.name
        try:
            rows = mine_reviews.parse_input(path)
            self.assertEqual(len(rows), 2)
            self.assertEqual(rows[0]["text"], "Takes forever to load")
        finally:
            os.unlink(path)

    def test_csv_missing_text_column_raises(self):
        with tempfile.NamedTemporaryFile(
            mode="w", suffix=".csv", delete=False, newline=""
        ) as f:
            writer = csv.DictWriter(f, fieldnames=["review", "rating"])
            writer.writeheader()
            writer.writerow({"review": "hello", "rating": "3"})
            path = f.name
        try:
            with self.assertRaises(ValueError):
                mine_reviews.parse_input(path)
        finally:
            os.unlink(path)

    def test_json_missing_text_field_raises(self):
        data = [{"body": "no text field"}]
        with tempfile.NamedTemporaryFile(
            mode="w", suffix=".json", delete=False
        ) as f:
            json.dump(data, f)
            path = f.name
        try:
            with self.assertRaises(ValueError):
                mine_reviews.parse_input(path)
        finally:
            os.unlink(path)

    def test_json_not_array_raises(self):
        with tempfile.NamedTemporaryFile(
            mode="w", suffix=".json", delete=False
        ) as f:
            json.dump({"text": "single object"}, f)
            path = f.name
        try:
            with self.assertRaises(ValueError):
                mine_reviews.parse_input(path)
        finally:
            os.unlink(path)


class ClassifyPainTests(unittest.TestCase):
    """Pain axis classification."""

    def test_time_cost(self):
        result = mine_reviews.classify_review("This takes forever to process")
        self.assertEqual(result["pain"], "time_cost")

    def test_time_cost_slow(self):
        result = mine_reviews.classify_review("The app is incredibly slow")
        self.assertEqual(result["pain"], "time_cost")

    def test_quality_cost(self):
        result = mine_reviews.classify_review("Results are always wrong and unreliable")
        self.assertEqual(result["pain"], "quality_cost")

    def test_social_cost(self):
        result = mine_reviews.classify_review("It was embarrassing in front of my boss")
        self.assertEqual(result["pain"], "social_cost")

    def test_cognitive_cost(self):
        result = mine_reviews.classify_review("So confusing, I can't figure out anything")
        self.assertEqual(result["pain"], "cognitive_cost")

    def test_money_cost(self):
        result = mine_reviews.classify_review("Way too expensive, not worth the price")
        self.assertEqual(result["pain"], "money_cost")

    def test_trust_cost(self):
        result = mine_reviews.classify_review("It crashed and I lost my data")
        self.assertEqual(result["pain"], "trust_cost")

    def test_no_pain_detected(self):
        result = mine_reviews.classify_review("It exists.")
        self.assertIsNone(result["pain"])


class ClassifyOutcomeTests(unittest.TestCase):
    """Outcome axis classification."""

    def test_speed(self):
        result = mine_reviews.classify_review("I need something faster")
        self.assertEqual(result["outcome"], "speed")

    def test_accuracy(self):
        result = mine_reviews.classify_review("I need accurate and reliable results")
        self.assertEqual(result["outcome"], "accuracy")

    def test_control(self):
        result = mine_reviews.classify_review("I want more control and customizable options")
        self.assertEqual(result["outcome"], "control")

    def test_simplicity(self):
        result = mine_reviews.classify_review("Should be easier and more intuitive")
        self.assertEqual(result["outcome"], "simplicity")

    def test_trust_outcome(self):
        result = mine_reviews.classify_review("I need something stable and predictable")
        self.assertEqual(result["outcome"], "trust")

    def test_status(self):
        result = mine_reviews.classify_review("I need a more professional looking tool")
        self.assertEqual(result["outcome"], "status")


class ClassifyWorkaroundTests(unittest.TestCase):
    """Workaround axis classification."""

    def test_manual(self):
        result = mine_reviews.classify_review("I do it manually in a spreadsheet")
        self.assertEqual(result["workaround"], "manual")

    def test_abandoned(self):
        result = mine_reviews.classify_review("I just gave up trying")
        self.assertEqual(result["workaround"], "abandoned")

    def test_competitor(self):
        result = mine_reviews.classify_review("I switched to Notion instead")
        self.assertEqual(result["workaround"], "competitor")
        self.assertEqual(result["competitor"], "notion")

    def test_hybrid(self):
        result = mine_reviews.classify_review(
            "I use it alongside another tool to work around the gaps"
        )
        self.assertEqual(result["workaround"], "hybrid")

    def test_unknown_default(self):
        result = mine_reviews.classify_review("The color is blue.")
        self.assertEqual(result["workaround"], "unknown")


class ConvergenceThresholdTests(unittest.TestCase):
    """Clusters with <3 reviews get confidence: low."""

    def _make_reviews(self, texts):
        return [{"text": t} for t in texts]

    def test_cluster_below_threshold_is_low(self):
        reviews = self._make_reviews([
            "Too slow, takes forever",
            "Very slow app",
        ])
        clusters, _ = mine_reviews.cluster_reviews(reviews)
        # Both should land in time_cost cluster with <3 reviews.
        time_clusters = [c for c in clusters if c["pain"] == "time_cost"]
        self.assertTrue(len(time_clusters) > 0)
        self.assertEqual(time_clusters[0]["confidence"], "low")

    def test_cluster_at_threshold_is_not_low(self):
        reviews = self._make_reviews([
            "Too slow, takes forever",
            "Very slow app, wastes my time",
            "So slow, I waste all day waiting",
        ])
        clusters, _ = mine_reviews.cluster_reviews(reviews)
        time_clusters = [c for c in clusters if c["pain"] == "time_cost"]
        self.assertTrue(len(time_clusters) > 0)
        # 3 reviews, no source info -> medium.
        self.assertIn(time_clusters[0]["confidence"], ("medium", "high"))

    def test_cross_source_boosts_to_high(self):
        reviews = [
            {"text": "Too slow", "source": "G2"},
            {"text": "Very slow", "source": "App Store"},
            {"text": "Wasting my time", "source": "Google Maps"},
        ]
        clusters, _ = mine_reviews.cluster_reviews(reviews)
        time_clusters = [c for c in clusters if c["pain"] == "time_cost"]
        self.assertTrue(len(time_clusters) > 0)
        self.assertEqual(time_clusters[0]["confidence"], "high")


class GenericFilterTests(unittest.TestCase):
    """Unique-to-business filter flags generic praise."""

    def test_generic_praise_flagged(self):
        result = mine_reviews.classify_review("Great customer service, highly recommend!")
        self.assertTrue(result["generic"])

    def test_specific_review_not_flagged(self):
        result = mine_reviews.classify_review(
            "The Kanban board crashes when I drag cards between columns"
        )
        self.assertFalse(result["generic"])


class EmptyInputTests(unittest.TestCase):
    """Edge cases: empty and minimal inputs."""

    def test_empty_review_list(self):
        clusters, classified = mine_reviews.cluster_reviews([])
        self.assertEqual(len(clusters), 0)
        self.assertEqual(len(classified), 0)

    def test_single_review(self):
        reviews = [{"text": "App is slow"}]
        clusters, classified = mine_reviews.cluster_reviews(reviews)
        self.assertEqual(len(classified), 1)
        # Single review cluster should be low confidence.
        for c in clusters:
            self.assertEqual(c["confidence"], "low")

    def test_reviews_with_empty_text(self):
        reviews = [{"text": ""}, {"text": ""}]
        clusters, classified = mine_reviews.cluster_reviews(reviews)
        self.assertEqual(len(classified), 2)


class RenderBriefTests(unittest.TestCase):
    """Brief generation writes a valid file."""

    def test_brief_file_created(self):
        reviews = [{"text": f"Too slow number {i}"} for i in range(5)]
        clusters, _ = mine_reviews.cluster_reviews(reviews)
        with tempfile.TemporaryDirectory() as tmpdir:
            path = mine_reviews.render_brief(clusters, 5, "test-source", tmpdir)
            self.assertTrue(os.path.exists(path))
            content = Path(path).read_text()
            self.assertIn("Review Brief", content)
            self.assertIn("test-source", content)
            self.assertIn("Reviews parsed:** 5", content)


class CLITests(unittest.TestCase):
    """End-to-end CLI invocation."""

    def test_main_with_csv(self):
        with tempfile.NamedTemporaryFile(
            mode="w", suffix=".csv", delete=False, newline=""
        ) as f:
            writer = csv.DictWriter(f, fieldnames=["text", "rating"])
            writer.writeheader()
            for i in range(5):
                writer.writerow({"text": f"This is way too slow {i}", "rating": "2"})
            path = f.name
        try:
            with tempfile.TemporaryDirectory() as tmpdir:
                summary = mine_reviews.main([path, "-o", tmpdir])
                self.assertEqual(summary["total_reviews"], 5)
                self.assertIn("clusters", summary)
                self.assertTrue(os.path.exists(summary["brief_path"]))
        finally:
            os.unlink(path)


if __name__ == "__main__":
    unittest.main()