settled-reach/whatsinagame/static/tooling/db/ticket

#!/usr/bin/env python3
"""
Ticket CLI — ergonomic interface to the project ticketing database.

Usage:
  ticket list [--status S] [--priority P] [--epic N] [--sprint N] [--assigned A] [--team T]
  ticket show <id>
  ticket done <id> [<id> ...]
  ticket status <id> <new_status>
  ticket assign <id> <agent>
  ticket unassign <id>
  ticket team <id> <teams>
  ticket sprint [--active]
  ticket sprint assign <id> <sprint_id>
  ticket deps <id>
  ticket search <keyword>
  ticket create <type> <title> [--parent N] [--priority P] [--decision D] [--team T]
  ticket epics [--status S]
  ticket children <id>
  ticket count [--status S]

All output is JSON on stdout.
"""

import json
import os
import sqlite3
import sys
from pathlib import Path

SCRIPT_DIR = Path(__file__).resolve().parent
CONFIG_PATH = SCRIPT_DIR / "config.json"


def load_config():
    with open(CONFIG_PATH, "r") as f:
        return json.load(f)


def resolve_db_path():
    """Resolve database path from environment or config."""
    env_path = os.environ.get("PROJECT_DB")
    if env_path:
        return Path(env_path).resolve()
    cfg = load_config()
    db_name = cfg.get("db_name", "project.db")
    db_location = cfg.get("db_location", "parent")
    if db_location == "parent":
        return (SCRIPT_DIR / ".." / ".." / ".." / db_name).resolve()
    elif db_location == "local":
        return (SCRIPT_DIR / ".." / ".." / db_name).resolve()
    else:
        return Path(db_location).resolve() / db_name


def get_connection():
    db_path = resolve_db_path()
    conn = sqlite3.connect(str(db_path))
    conn.execute("PRAGMA journal_mode=WAL;")
    conn.execute("PRAGMA foreign_keys=ON;")
    conn.row_factory = sqlite3.Row
    return conn


def query(conn, sql, params=()):
    cursor = conn.execute(sql, params)
    columns = [desc[0] for desc in cursor.description] if cursor.description else []
    return [dict(zip(columns, row)) for row in cursor.fetchall()]


def execute(conn, sql, params=()):
    cursor = conn.execute(sql, params)
    conn.commit()
    return cursor.rowcount


def out(data):
    print(json.dumps(data, indent=2))


def parse_flags(args, known_flags):
    """Parse --flag value pairs from args, return (flags_dict, positional_args)."""
    flags = {}
    positional = []
    i = 0
    while i < len(args):
        if args[i].startswith("--") and args[i][2:] in known_flags:
            key = args[i][2:]
            if i + 1 < len(args):
                flags[key] = args[i + 1]
                i += 2
            else:
                positional.append(args[i])
                i += 1
        else:
            positional.append(args[i])
            i += 1
    return flags, positional


# ---------------------------------------------------------------------------
# Commands
# ---------------------------------------------------------------------------

def cmd_list(conn, args):
    flags, _ = parse_flags(args, ["status", "priority", "epic", "sprint", "assigned", "team"])
    conditions = []
    params = []
    if "status" in flags:
        conditions.append("t.status = ?")
        params.append(flags["status"])
    if "priority" in flags:
        conditions.append("t.priority = ?")
        params.append(flags["priority"])
    if "epic" in flags:
        conditions.append("t.parent_id = ?")
        params.append(int(flags["epic"]))
    if "sprint" in flags:
        conditions.append("t.sprint_id = ?")
        params.append(int(flags["sprint"]))
    if "assigned" in flags:
        conditions.append("t.assigned_to = ?")
        params.append(flags["assigned"])
    if "team" in flags:
        # Match exact team name within comma-separated list
        conditions.append("(',' || t.team || ',' LIKE '%,' || ? || ',%')")
        params.append(flags["team"])
    where = " AND ".join(conditions) if conditions else "1=1"
    sql = f"""SELECT t.id, t.type, t.title, t.status, t.priority, t.assigned_to,
                     t.team, t.parent_id, t.sprint_id
              FROM tickets t WHERE {where}
              ORDER BY
                CASE t.priority WHEN 'critical' THEN 0 WHEN 'high' THEN 1
                WHEN 'medium' THEN 2 ELSE 3 END, t.id"""
    rows = query(conn, sql, tuple(params))
    out({"ok": True, "count": len(rows), "rows": rows})


def cmd_show(conn, ids, brief=False):
    tickets = []
    for ticket_id in ids:
        rows = query(conn, """SELECT t.*, p.title as parent_title
                              FROM tickets t LEFT JOIN tickets p ON t.parent_id = p.id
                              WHERE t.id = ?""", (ticket_id,))
        if not rows:
            tickets.append({"id": ticket_id, "error": f"Ticket #{ticket_id} not found"})
            continue
        ticket = rows[0]
        # Get children
        children = query(conn, "SELECT id, title, status, priority FROM tickets WHERE parent_id = ? ORDER BY id", (ticket_id,))
        # Get dependencies (what blocks this)
        blockers = query(conn, """SELECT t.id, t.title, t.status FROM ticket_deps d
                                  JOIN tickets t ON d.blocker_id = t.id
                                  WHERE d.blocked_id = ?""", (ticket_id,))
        # Get dependents (what this blocks)
        blocks = query(conn, """SELECT t.id, t.title, t.status FROM ticket_deps d
                                JOIN tickets t ON d.blocked_id = t.id
                                WHERE d.blocker_id = ?""", (ticket_id,))
        ticket["children"] = children
        ticket["blocked_by"] = blockers
        ticket["blocks"] = blocks
        tickets.append(ticket)
    if brief:
        _print_brief(tickets)
    elif len(tickets) == 1:
        out({"ok": True, "ticket": tickets[0]})
    else:
        out({"ok": True, "count": len(tickets), "tickets": tickets})


def _print_brief(tickets):
    for i, t in enumerate(tickets):
        if "error" in t:
            print(f"#{t['id']}: NOT FOUND")
            continue
        # Header line
        print(f"#{t['id']}: {t['title']}")
        # Metadata line
        parts = [f"{t['type']}", f"P:{t['priority']}", f"S:{t['status']}"]
        if t.get("assigned_to"):
            parts.append(f"@{t['assigned_to']}")
        if t.get("team"):
            parts.append(f"Team:{t['team']}")
        if t.get("parent_id"):
            parts.append(f"Epic:#{t['parent_id']} ({t.get('parent_title', '?')})")
        if t.get("sprint_id"):
            parts.append(f"Sprint:{t['sprint_id']}")
        if t.get("decision_ref"):
            parts.append(f"Ref:{t['decision_ref']}")
        print(f"  {' | '.join(parts)}")
        # Description
        desc = t.get("description") or ""
        if desc:
            # Truncate long descriptions
            if len(desc) > 200:
                desc = desc[:197] + "..."
            print(f"  {desc}")
        # Dependencies
        if t.get("blocked_by"):
            blockers = ", ".join(f"#{b['id']} ({b['status']})" for b in t["blocked_by"])
            print(f"  Blocked by: {blockers}")
        if t.get("blocks"):
            blocks = ", ".join(f"#{b['id']}" for b in t["blocks"])
            print(f"  Blocks: {blocks}")
        if i < len(tickets) - 1:
            print()


def cmd_done(conn, ids):
    updated = 0
    for tid in ids:
        updated += execute(conn, "UPDATE tickets SET status='done', updated_at=datetime('now') WHERE id=?", (int(tid),))
    out({"ok": True, "updated": updated, "ids": [int(i) for i in ids]})


def cmd_status(conn, ticket_id, new_status):
    valid = ('backlog', 'ready', 'in_progress', 'review', 'done', 'cancelled')
    if new_status not in valid:
        out({"ok": False, "error": f"Invalid status '{new_status}'. Valid: {', '.join(valid)}"})
        return
    updated = execute(conn, "UPDATE tickets SET status=?, updated_at=datetime('now') WHERE id=?", (new_status, int(ticket_id)))
    out({"ok": True, "updated": updated, "id": int(ticket_id), "status": new_status})


def cmd_assign(conn, ticket_id, agent):
    updated = execute(conn, "UPDATE tickets SET assigned_to=?, updated_at=datetime('now') WHERE id=?", (agent, int(ticket_id)))
    out({"ok": True, "updated": updated, "id": int(ticket_id), "assigned_to": agent})


def cmd_unassign(conn, ticket_id):
    updated = execute(conn, "UPDATE tickets SET assigned_to=NULL, updated_at=datetime('now') WHERE id=?", (int(ticket_id),))
    out({"ok": True, "updated": updated, "id": int(ticket_id), "assigned_to": None})


def cmd_sprint(conn, args):
    flags, positional = parse_flags(args, ["active"])
    if positional and positional[0] == "assign" and len(positional) >= 3:
        ticket_id, sprint_id = int(positional[1]), int(positional[2])
        updated = execute(conn, "UPDATE tickets SET sprint_id=?, updated_at=datetime('now') WHERE id=?", (sprint_id, ticket_id))
        out({"ok": True, "updated": updated, "id": ticket_id, "sprint_id": sprint_id})
        return
    conditions = []
    params = []
    if "active" in flags:
        conditions.append("s.status = 'active'")
    where = " AND ".join(conditions) if conditions else "1=1"
    sprints = query(conn, f"""SELECT s.*, COUNT(t.id) as ticket_count,
                              SUM(CASE WHEN t.status='done' THEN 1 ELSE 0 END) as done_count
                              FROM sprints s LEFT JOIN tickets t ON t.sprint_id = s.id
                              WHERE {where} GROUP BY s.id ORDER BY s.id DESC""", tuple(params))
    out({"ok": True, "count": len(sprints), "sprints": sprints})


def cmd_deps(conn, ticket_id):
    blockers = query(conn, """SELECT t.id, t.title, t.status, t.priority FROM ticket_deps d
                              JOIN tickets t ON d.blocker_id = t.id
                              WHERE d.blocked_id = ? ORDER BY t.id""", (int(ticket_id),))
    blocks = query(conn, """SELECT t.id, t.title, t.status, t.priority FROM ticket_deps d
                            JOIN tickets t ON d.blocked_id = t.id
                            WHERE d.blocker_id = ? ORDER BY t.id""", (int(ticket_id),))
    out({"ok": True, "id": int(ticket_id), "blocked_by": blockers, "blocks": blocks})


def cmd_search(conn, keyword):
    rows = query(conn, """SELECT id, type, title, status, priority, assigned_to, team
                          FROM tickets WHERE title LIKE ? OR description LIKE ?
                          ORDER BY id""", (f"%{keyword}%", f"%{keyword}%"))
    out({"ok": True, "count": len(rows), "rows": rows})


def cmd_create(conn, args):
    flags, positional = parse_flags(args, ["parent", "priority", "decision", "team"])
    if len(positional) < 2:
        out({"ok": False, "error": "Usage: ticket create <type> <title> [--parent N] [--priority P] [--decision D] [--team T]"})
        return
    ticket_type = positional[0]
    title = " ".join(positional[1:])
    parent_id = int(flags["parent"]) if "parent" in flags else None
    priority = flags.get("priority", "medium")
    decision_ref = flags.get("decision")
    team = flags.get("team")
    conn.execute(
        "INSERT INTO tickets (type, title, parent_id, priority, decision_ref, team) VALUES (?, ?, ?, ?, ?, ?)",
        (ticket_type, title, parent_id, priority, decision_ref, team))
    conn.commit()
    last_id = query(conn, "SELECT last_insert_rowid() as id")[0]["id"]
    out({"ok": True, "id": last_id, "title": title})


def cmd_epics(conn, args):
    flags, _ = parse_flags(args, ["status"])
    conditions = ["t.type = 'epic'"]
    params = []
    if "status" in flags:
        conditions.append("t.status = ?")
        params.append(flags["status"])
    where = " AND ".join(conditions)
    rows = query(conn, f"""SELECT t.id, t.title, t.status, t.priority, t.assigned_to, t.team,
                           COUNT(c.id) as child_count,
                           SUM(CASE WHEN c.status='done' THEN 1 ELSE 0 END) as done_count
                           FROM tickets t LEFT JOIN tickets c ON c.parent_id = t.id
                           WHERE {where} GROUP BY t.id
                           ORDER BY CASE t.priority WHEN 'critical' THEN 0 WHEN 'high' THEN 1
                           WHEN 'medium' THEN 2 ELSE 3 END, t.id""", tuple(params))
    out({"ok": True, "count": len(rows), "rows": rows})


def cmd_children(conn, ticket_id):
    rows = query(conn, """SELECT id, type, title, status, priority, assigned_to, team
                          FROM tickets WHERE parent_id = ? ORDER BY id""", (int(ticket_id),))
    out({"ok": True, "count": len(rows), "parent_id": int(ticket_id), "rows": rows})


def cmd_team(conn, ticket_id, teams):
    updated = execute(conn, "UPDATE tickets SET team=?, updated_at=datetime('now') WHERE id=?", (teams, int(ticket_id)))
    out({"ok": True, "updated": updated, "id": int(ticket_id), "team": teams})


def cmd_count(conn, args):
    flags, _ = parse_flags(args, ["status"])
    if "status" in flags:
        rows = query(conn, "SELECT COUNT(*) as count FROM tickets WHERE status = ?", (flags["status"],))
    else:
        rows = query(conn, "SELECT status, COUNT(*) as count FROM tickets GROUP BY status ORDER BY count DESC")
    out({"ok": True, "rows": rows})


# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------

HELP = """ticket — project ticket CLI

Usage:
  ticket list [--status S] [--priority P] [--epic N] [--sprint N] [--assigned A] [--team T]
  ticket show [--brief] <id> [<id>...] Full ticket detail (--brief for summary)
  ticket done <id> [<id> ...]         Mark tickets as done
  ticket status <id> <new_status>     Change ticket status
  ticket assign <id> <agent>          Assign ticket to agent/branch
  ticket unassign <id>                Remove assignment
  ticket team <id> <teams>            Set team(s) (comma-separated, e.g. server,client)
  ticket sprint [--active]            List sprints
  ticket sprint assign <id> <sprint>  Assign ticket to sprint
  ticket deps <id>                    Show ticket dependencies
  ticket search <keyword>             Search tickets by title/description
  ticket create <type> <title> [--parent N] [--priority P] [--decision D] [--team T]
  ticket epics [--status S]           List epics with child counts
  ticket children <id>                List children of a ticket
  ticket count [--status S]           Count tickets by status"""


def main():
    if len(sys.argv) < 2 or sys.argv[1] in ("--help", "-h", "help"):
        print(HELP)
        sys.exit(0)

    conn = get_connection()
    cmd = sys.argv[1]
    args = sys.argv[2:]

    try:
        if cmd == "list":
            cmd_list(conn, args)
        elif cmd == "show":
            brief = "--brief" in args
            id_args = [a for a in args if a != "--brief"]
            if not id_args:
                out({"ok": False, "error": "Usage: ticket show [--brief] <id> [<id> ...]"})
            else:
                cmd_show(conn, [int(a) for a in id_args], brief=brief)
        elif cmd == "done":
            if not args:
                out({"ok": False, "error": "Usage: ticket done <id> [<id> ...]"})
            else:
                cmd_done(conn, args)
        elif cmd == "status":
            if len(args) < 2:
                out({"ok": False, "error": "Usage: ticket status <id> <new_status>"})
            else:
                cmd_status(conn, args[0], args[1])
        elif cmd == "assign":
            if len(args) < 2:
                out({"ok": False, "error": "Usage: ticket assign <id> <agent>"})
            else:
                cmd_assign(conn, args[0], args[1])
        elif cmd == "unassign":
            if not args:
                out({"ok": False, "error": "Usage: ticket unassign <id>"})
            else:
                cmd_unassign(conn, args[0])
        elif cmd == "team":
            if len(args) < 2:
                out({"ok": False, "error": "Usage: ticket team <id> <teams>"})
            else:
                cmd_team(conn, args[0], args[1])
        elif cmd == "sprint":
            cmd_sprint(conn, args)
        elif cmd == "deps":
            if not args:
                out({"ok": False, "error": "Usage: ticket deps <id>"})
            else:
                cmd_deps(conn, args[0])
        elif cmd == "search":
            if not args:
                out({"ok": False, "error": "Usage: ticket search <keyword>"})
            else:
                cmd_search(conn, " ".join(args))
        elif cmd == "create":
            cmd_create(conn, args)
        elif cmd == "epics":
            cmd_epics(conn, args)
        elif cmd == "children":
            if not args:
                out({"ok": False, "error": "Usage: ticket children <id>"})
            else:
                cmd_children(conn, args[0])
        elif cmd == "count":
            cmd_count(conn, args)
        else:
            out({"ok": False, "error": f"Unknown command: {cmd}. Use --help for usage."})
    finally:
        conn.close()


if __name__ == "__main__":
    main()