extract_thread.py

#!/usr/bin/env python3
"""
Enrichment: Slack Thread Decision Extractor

For each unique thread_ts detected in the fetched Slack messages, fetches the
full thread via conversations.replies and classifies it for decision/blocker signal.
Noise threads are silently dropped. Only signal threads reach the final summary.

Input:
  --slack    path to slack.json (output of fetch_slack.py)
  --token    Slack User OAuth token (xoxp-...)
  --user-id  Slack user ID

Output (stdout): JSON
{
  "enrichment": "extract_thread",
  "since": "...",
  "until": "...",
  "results": [
    {
      "channel":            "engineering",
      "date":               "YYYY-MM-DD",
      "thread_type":        "decision|blocker|discussion",
      "summary":            "Team aligned on using Postgres for job queue instead of Redis.",
      "action_item_for_me": "Spike Postgres job queue implementation"  // or null
    },
    ...
  ]
}
"""

import json
import argparse
import sys
import subprocess

try:
    import requests
except ImportError:
    print("[extract_thread] Error: 'requests' not installed. Run: pip install requests", file=sys.stderr)
    sys.exit(1)

SLACK_API = "https://slack.com/api"


def slack_get(token, method, params=None):
    headers = {"Authorization": f"Bearer {token}"}
    try:
        resp = requests.get(f"{SLACK_API}/{method}", headers=headers, params=params or {}, timeout=15)
    except requests.RequestException as e:
        print(f"[extract_thread] Request failed ({method}): {e}", file=sys.stderr)
        return None
    if resp.status_code != 200:
        print(f"[extract_thread] HTTP {resp.status_code} from {method}", file=sys.stderr)
        return None
    data = resp.json()
    if not data.get("ok"):
        err = data.get("error", "unknown")
        print(f"[extract_thread] Slack API error ({method}): {err}", file=sys.stderr)
        if err == "missing_scope":
            print("[extract_thread] Token needs 'channels:history' or 'groups:history' user scope.", file=sys.stderr)
        return None
    return data


def fetch_thread(token, channel_id, thread_ts):
    """Fetch all messages in a thread. Returns a list of message dicts."""
    data = slack_get(token, "conversations.replies", {
        "channel": channel_id,
        "ts": thread_ts,
        "limit": 50,
    })
    if not data:
        return []
    return data.get("messages", [])


def call_claude(prompt):
    """Run claude -p and return the response text, or None on failure."""
    result = subprocess.run(
        ["claude", "-p", prompt],
        capture_output=True, text=True, timeout=90
    )
    if result.returncode != 0:
        print(f"[extract_thread] claude error: {result.stderr.strip()[:120]}", file=sys.stderr)
        return None
    return result.stdout.strip()


def extract_json(text):
    """Extract a JSON object from Claude's response (handles ```json fences)."""
    if not text:
        return None
    s = text.strip()
    if "```" in s:
        parts = s.split("```")
        for part in parts:
            candidate = part.lstrip("json").strip()
            try:
                return json.loads(candidate)
            except json.JSONDecodeError:
                continue
    try:
        return json.loads(s)
    except json.JSONDecodeError:
        start = s.find("{")
        end = s.rfind("}") + 1
        if start >= 0 and end > start:
            try:
                return json.loads(s[start:end])
            except json.JSONDecodeError:
                pass
    return None


def classify_thread(channel, date, messages):
    """Ask Claude to classify a thread and extract signal."""
    msg_lines = []
    for i, m in enumerate(messages):
        text = (m.get("text") or "").replace("\n", " ")[:300]
        if text:
            msg_lines.append(f"[{i+1}] {text}")
    if not msg_lines:
        return None

    messages_block = "\n".join(msg_lines)

    prompt = f"""Classify the following Slack thread from #{channel} on {date}.

Thread messages (chronological):
{messages_block}

Return ONLY a valid JSON object — no explanation, no markdown:
{{
  "channel": "{channel}",
  "date": "{date}",
  "thread_type": "decision|blocker|discussion|noise",
  "summary": "one sentence describing what happened, or null if noise",
  "action_item_for_me": "specific next action assigned or mentioned, or null"
}}

Use "noise" for casual chat, greetings, social conversation, or off-topic threads.
Use "decision" when the thread concludes with an agreed-upon direction.
Use "blocker" when something is blocking progress.
Use "discussion" for substantive technical or work-related exchanges."""

    raw = call_claude(prompt)
    result = extract_json(raw)
    if result and isinstance(result, dict) and "thread_type" in result:
        return result
    return None


def main():
    parser = argparse.ArgumentParser(description="Slack thread decision extractor enrichment")
    parser.add_argument("--slack",   required=True, help="Path to slack.json from fetch_slack.py")
    parser.add_argument("--token",   required=True, help="Slack User OAuth token (xoxp-...)")
    parser.add_argument("--user-id", required=True, help="Slack user ID")
    args = parser.parse_args()

    try:
        with open(args.slack) as f:
            slack_data = json.load(f)
    except Exception as e:
        print(f"[extract_thread] Could not load {args.slack}: {e}", file=sys.stderr)
        sys.exit(1)

    if slack_data.get("service") != "slack":
        print("[extract_thread] Input is not slack service data", file=sys.stderr)
        sys.exit(1)

    # Collect unique threads (thread_ts -> {channel_id, channel_name, date})
    threads = {}
    for msg in slack_data.get("messages", []):
        thread_ts = msg.get("thread_ts")
        if not thread_ts:
            continue
        if thread_ts in threads:
            continue
        channel_id = msg.get("channel_id", "")
        if not channel_id:
            # Can't fetch thread without channel_id
            continue
        threads[thread_ts] = {
            "channel_id": channel_id,
            "channel":    msg.get("channel", "unknown"),
            "date":       msg.get("date", "?"),
        }

    results = []
    noise_count = 0

    for thread_ts, info in threads.items():
        messages = fetch_thread(args.token, info["channel_id"], thread_ts)
        if not messages:
            continue

        result = classify_thread(info["channel"], info["date"], messages)
        if not result:
            continue

        if result.get("thread_type") == "noise":
            noise_count += 1
            continue  # Drop noise threads from output

        results.append(result)

    output = {
        "enrichment": "extract_thread",
        "since": slack_data.get("since", ""),
        "until": slack_data.get("until", ""),
        "results": results,
    }
    print(json.dumps(output, indent=2))
    print(
        f"[extract_thread] {len(threads)} thread(s) processed, "
        f"{len(results)} signal, {noise_count} noise dropped",
        file=sys.stderr
    )


if __name__ == "__main__":
    main()