impactflow_discovery/scripts/reprocess_csv.py

#!/usr/bin/env python3
"""Re-run AI extraction over a saved discovery CSV, offline.

Reads a per-conversation CSV (``data/questions/{id}.csv``) or a row-slice of
the master log (``data/questions_master.csv`` filtered by ``--conversation-id``)
and runs the same DiscoveryExtractor the web app uses. By default it prints the
resulting profile JSON to stdout so it is safe to run for inspection; pass
``--write-db`` to also persist a new DiscoveryProfile.

Examples:
    python scripts/reprocess_csv.py data/questions/<id>.csv
    python scripts/reprocess_csv.py data/questions_master.csv \\
        --conversation-id <id> --write-db

Requires ANTHROPIC_API_KEY (and optionally ANTHROPIC_MODEL) in the environment.
"""
import argparse
import asyncio
import csv
import json
import os
import sys
import uuid
from datetime import datetime, timezone

# Allow running as a plain script (python scripts/reprocess_csv.py ...).
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

from app.services.answer_store import DISCOVERY_PROMPTS  # noqa: E402
from app.services.extractor import (  # noqa: E402
    DiscoveryExtractionError,
    DiscoveryExtractor,
)

PROMPT_KEYS = [p["key"] for p in DISCOVERY_PROMPTS]


def read_answers(path: str, conversation_id: str | None) -> tuple[str, dict]:
    """Return (conversation_id, {prompt_key: answer}) from a CSV file.

    If ``conversation_id`` is given, only rows for that conversation are used
    (needed for the master log). Later rows win, so the most recent saved
    answers take precedence.
    """
    answers: dict[str, str] = {}
    found_id = conversation_id
    with open(path, newline="", encoding="utf-8") as fh:
        for row in csv.DictReader(fh):
            row_id = row.get("conversation_id")
            if conversation_id and row_id != conversation_id:
                continue
            found_id = found_id or row_id
            key = row.get("prompt_key")
            if key in PROMPT_KEYS:
                answers[key] = row.get("answer", "") or ""
    if not answers:
        raise SystemExit(
            f"No matching answer rows found in {path}"
            + (f" for conversation {conversation_id}" if conversation_id else "")
        )
    return found_id or "", answers


async def run_extraction(answers: dict) -> dict:
    api_key = os.getenv("ANTHROPIC_API_KEY")
    model = os.getenv("ANTHROPIC_MODEL", "claude-sonnet-4-6")
    responses = {k: answers.get(k, "") for k in PROMPT_KEYS}
    extractor = DiscoveryExtractor(api_key=api_key, model=model)
    return await extractor.extract(responses)


async def write_db(conversation_id: str, data: dict) -> str:
    """Persist a new DiscoveryProfile for an existing conversation."""
    from app.database import AsyncSessionLocal
    from app.models import DiscoveryConversation, DiscoveryProfile

    def _as_int(value):
        try:
            return int(value)
        except (TypeError, ValueError):
            return None

    async with AsyncSessionLocal() as db:
        conversation = await db.get(DiscoveryConversation, conversation_id)
        if conversation is None:
            raise SystemExit(
                f"Conversation {conversation_id} not found in the database; "
                "cannot --write-db."
            )
        profile = DiscoveryProfile(
            id=str(uuid.uuid4()),
            user_id=conversation.user_id,
            conversation_id=conversation.id,
            generated_at=datetime.now(timezone.utc),
            triad=data.get("triad"),
            probable_type=_as_int(data.get("probable_type")),
            wing=_as_int(data.get("wing")),
            instinctual_variant=data.get("instinctual_variant"),
            instinctual_stack=data.get("instinctual_stack"),
            love_summary=data.get("love_summary"),
            strength_summary=data.get("strength_summary"),
            mission_summary=data.get("mission_summary"),
            vocation_summary=data.get("vocation_summary"),
            overlap_narrative=data.get("overlap_narrative"),
            short_term_goals=data.get("short_term_goals"),
            long_term_goals=data.get("long_term_goals"),
            confidence_json=json.dumps(data.get("confidence", {})),
            locked=False,
        )
        db.add(profile)
        await db.commit()
        return profile.id


async def main_async(args: argparse.Namespace) -> int:
    conversation_id, answers = read_answers(args.csv_path, args.conversation_id)
    try:
        data = await run_extraction(answers)
    except DiscoveryExtractionError as exc:
        print(f"Extraction failed: {exc}", file=sys.stderr)
        return 1

    print(json.dumps(data, indent=2, ensure_ascii=False))

    if args.write_db:
        if not conversation_id:
            print(
                "Cannot --write-db: no conversation_id in the CSV.",
                file=sys.stderr,
            )
            return 1
        profile_id = await write_db(conversation_id, data)
        print(
            f"\nWrote profile {profile_id} for conversation {conversation_id}.",
            file=sys.stderr,
        )
    return 0


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "csv_path", help="Path to a per-conversation or master CSV file."
    )
    parser.add_argument(
        "--conversation-id",
        help="Only use rows for this conversation (required for the master log "
        "when it holds more than one conversation).",
    )
    parser.add_argument(
        "--write-db",
        action="store_true",
        help="Persist a new DiscoveryProfile to the database (default: print "
        "only).",
    )
    args = parser.parse_args()
    return asyncio.run(main_async(args))


if __name__ == "__main__":
    raise SystemExit(main())