cli.py

import argparse
import csv
import io
import os
import sys
from typing import Optional

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import db

db.init_db()

def fmt(val, decimals: int = 2) -> str:
    return f"{val: .{decimals}f}" if val is not None else "-"

def trunc(text: str, width: int) -> str:
    return text if len(text) <= width else text[: width - 1] + "..."

def divider(width: int = 72) -> None:
    print("-" * width)

def cmd_stats(args: argparse.Namespace) -> None:
    windows = [args.window] if args.window else ["7d", "30d", "120d", "all"]
    print()
    print("ProperPrompt")
    divider()
    print(f"   {'Window':<8}  {'N':>4}  {'Overall':>7}  {'Clarity':>7}  {'Brevity':>7}  {'Specificity':>8}  {'Resp':>5}  {'Followup':>8}  Weakness")
    divider()
    for w in windows:
        s = db.compute_stats(w)
        n = s["n_prompts"] or 0
        if n == 0:
            print(f"   {w:<8}  no scored prompts")
            continue
        followup_pct = f"{s['followup_rate'] * 100:.0f}%" if s["followup_rate"] is not None else "-"
        print(
            f"   {w:<8}"
            f"   {n:>4}"
            f"   {fmt(s['avg_overall'], 1):>7}"
            f"   {fmt(s['avg_clarity'], 1):>7}"
            f"   {fmt(s['avg_brevity'], 1):>7}"
            f"   {fmt(s['avg_specificity'], 1):>8}"
            f"   {fmt(s['avg_response_rating'], 1):>5}"
            f"   {followup_pct:>8}"
            f"   {s['top_weakness'] or '-'}"
        )
        print()

def cmd_history(args: argparse.Namespace) -> None:
    prompts = db.get_prompt_history(limit = args.limit, session_id = args.session, min_score = args.min_score, scored_only = args.scored_only)
    if not prompts:
        print("\nNo prompts found.\n")
        return
    print(f"\nPrompt History ({len(prompts)} shown)")
    for p in prompts:
        divider()
        followup_str = ""
        if p["needed_followup"] == 1:
            followup_str = f"  followup: {p['followup_reason']}"
        elif p["needed_followup"] == 0:
            followup_str = "   no followup"
        
        rating_str = f"   response: {p['response_rating']}/5" if p["response_rating"] else ""
        print(f"#{p['id']}  {p["logged_at"][:16]}  {p['session_id']}{followup_str}{rating_str}")
        print(f"  {trunc(p['prompt_text'], 90)}")

        if p["scored_at"]:
            print(
                f"  scores - overall: {fmt(p['score_overall'], 1)}"
                f"  clarity: {fmt(p['score_clarity'], 1)}"
                f"  brevity: {fmt(p['score_brevity'], 1)}"
                f"  specificity: {fmt(p['score_specificity'], 1)}"
            )
            if p["score_rationale"]:
                print(f"   rationale: {trunc(p['score_rationale'], 85)}")
        else:
            print("   (not scored)")

    divider()
    print()

def cmd_sessions(args: argparse.Namespace) -> None:
    sessions = db.list_sessions(limit = args.limit)
    if not sessions:
        print("\nNo sessions found.\n")
        return
    
    print(f"\nSessions  ({len(sessions)} shown)")
    divider()
    print(f"  {'ID':<32}  {'Started':<16}  {'Ended':<16}  Notes")
    divider()

    for s in sessions:
        ended = s["ended_at"][:16] if s["ended_at"] else "open"
        print(
            f"   {s['session_id']:<32}"
            f"   {s['started_at'][:16]}"
            f"   {ended:<16}"
            f"   {trunc(s['notes'] or '', 40)}"
        )
    
    print()

def fmt_stars(rating: int) -> str:
    return "*" * rating + "-" * (5 - rating)

def export_md(session: dict, prompts: list[dict]) -> str:
    lines = [
        f"# Session: {session['session_id']}",
        f"",
        f"Started: {session['started_at']}",
        f"Ended: {session['ended_at'] or 'open'}",
        f"Notes: {session['notes'] or '-'}",
        f"",
        f"---",
        f""
    ]
    for p in prompts:
        lines.append(f"## Prompt#{p['id']}  seq {p['sequence_num']}  {p['logged_at'][:16]}")
        lines.append(f"")
        lines.append(f"> {p['prompt_text']}")
        lines.append(f"")

        if p["scored_at"]:
            lines += [
                f"| Dimension | Score |",
                f"| --------- | ----- |",
                f"| Overall   | {fmt(p["score_overall"])} |",
                f"| Brevity   | {fmt(p["score_brevity"])} |",
                f"| Clarity   | {fmt(p["score_clarity"])} |",
                f"| Specificity   | {fmt(p["score_specificity"])} |"
            ]
            if p["score_rationale"]:
                lines += [f"", f"Rationale: {p['score_rationale']}"]
        else:
            lines.append("Not scored.")

        if p['response_rating']:
            note = f" - {p['rating_note']}" if p['rating_note'] else ""
            lines += [f"", f"Response rating: {fmt_stars(p['response_rating'])}{note}"]
        if p["needed_followup"] == 1:
            lines += [f"", f"Followup needed: yes ({p['followup_reason']})"]
        lines += [f"", f"---", f""]
    return "\n".join(lines)

def export_csv(prompts: list[dict]) -> str:
    buf = io.StringIO()
    fields = ["id", "session_id", "sequence_num", "logged_at", "prompt_text", "score_overall", "score_clarity", "score_brevity", "score_specificity", "score_rationale", "scored_at", "needed_followup", "followup_reason", "response_rating", "rating_note"]
    writer = csv.DictWriter(buf, fieldnames = fields, extrasaction = "ignore")
    writer.writeheader()
    for p in prompts:
        writer.writerow(p)
    return buf.getvalue()


def cmd_export(args: argparse.Namespace) -> None:
    session = db.get_session(args.session_id)
    if not session:
        print(f"\nSession '{args.session_id}' not found.\n")
        sys.exit(1)
    
    prompts = db.get_prompt_history(limit = 1000, session_id = args.session_id)
    prompts.reverse()

    content = export_md(session, prompts) if args.format == "md" else export_csv(prompts)
    ext = ".md" if args.format == "md" else ".csv"
    out_path = args.out or os.path.join(os.path.dirname(os.path.abspath(__file__)), f"{args.session_id}{ext}")
    with open(out_path, "w", encoding = "utf-8") as f:
        f.write(content)
    print(f"\nExported to {out_path}.\n")

def cmd_close_all(args: argparse.Namespace) -> None:
    sessions = db.list_sessions(limit = 1000)
    open_sessions = [s for s in sessions if not s["ended_at"]]

    if not open_sessions:
        print("\nNo open sessions.\n")
        return
    
    print(f"\n{len(open_sessions)} open session(s) will be closed:")
    for s in open_sessions:
        print(f"   {s['session_id']}  {s['started_at'][:16]}  {s['notes'] or ''}")

    if not args.yes:
        confirm = input("\nClose all? [y/N] ").strip().lower()
        if confirm != "y":
            print("Aborted.\n")
            return
    for s in open_sessions:
        db.end_session(s['session_id'])
        last = db.get_last_prompt(s['session_id'])
        if last and last["needed_followup"] is None:
            db.update_followup(last["id"], False, "session_ended", "keyword")

    print(f"Closed {len(open_sessions)} session(s).\n")

def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(prog = "python cli.py", description = "ProperPrompt Terminal Interface")
    sub = parser.add_subparsers(dest = "command", required = True)
    p_stats = sub.add_parser("stats", help = "Show aggregate quality stats")
    p_stats.add_argument("--window", choices = ["7d", "30d", "120d", "all"], help = "Single window to show (default: all four)")
    p_hist = sub.add_parser("history", help = "browse logged prompts")
    p_hist.add_argument("--limit", type = int, default = 20)
    p_hist.add_argument("--session", metavar = "session_id")
    p_hist.add_argument("--scored-only", action = "store_true")
    p_hist.add_argument("--min-score", type = float, metavar = "X")

    p_sess = sub.add_parser("sessions", help = "list sessions")
    p_sess.add_argument("--limit", type = int, default = 20)

    p_exp = sub.add_parser("export", help = "export a session to markdown or CSV")
    p_exp.add_argument("session_id")
    p_exp.add_argument("--format", choices = ["md", "csv"], default = "md")
    p_exp.add_argument("--out", metavar = "file")

    p_close = sub.add_parser("close-all", help = "Close all open sessions")
    p_close.add_argument("-y", "--yes", action = "store_true", help = "Skip confirmation prompt")

    return parser

def main() -> None:
    parser = build_parser()
    args = parser.parse_args()
    {"stats": cmd_stats, "history": cmd_history, "sessions": cmd_sessions, "export": cmd_export, "close-all": cmd_close_all}[args.command](args)

if __name__ == "__main__":
    main()