|
#!/usr/bin/env python3 |
|
"""Enrich a Copilot chat markdown dump by interleaving the "thinking" blocks and |
|
tool-call details that only exist in the JSON dump. |
|
|
|
The JSON dump stores each conversation turn under ``prompts[]``. Every prompt has |
|
an ordered ``logs[]`` timeline whose entries are either: |
|
|
|
* ``kind == "request"`` -> assistant text in ``response.message[]`` |
|
* ``kind == "toolCall"`` -> ``tool`` name, ``args`` (JSON string), tool output |
|
in ``response[]`` and a ``thinking.text`` block. |
|
|
|
This walks that timeline in order and rebuilds the readable markdown, injecting a |
|
blockquote for every thinking area and every tool call/output right where it |
|
happened. |
|
|
|
Usage: |
|
python3 enrich_chat_dump.py DUMP.json |
|
python3 enrich_chat_dump.py DUMP.json --query "julia function" --stdout |
|
""" |
|
|
|
from __future__ import annotations |
|
|
|
import argparse |
|
import json |
|
import sys |
|
from pathlib import Path |
|
|
|
|
|
def blockquote(text: str) -> str: |
|
"""Prefix every line of ``text`` with a markdown blockquote marker.""" |
|
out = [] |
|
for line in text.rstrip("\n").split("\n"): |
|
out.append("> " + line if line else ">") |
|
return "\n".join(out) |
|
|
|
|
|
def fenced_in_quote(body: str, lang: str = "") -> str: |
|
"""Render ``body`` as a fenced code block nested inside a blockquote.""" |
|
lines = ["```" + lang] + body.rstrip("\n").split("\n") + ["```"] |
|
return "\n".join("> " + line if line else ">" for line in lines) |
|
|
|
|
|
def format_args(args) -> str: |
|
"""Pretty-print a tool call's arguments (usually a JSON string).""" |
|
if isinstance(args, str): |
|
try: |
|
return json.dumps(json.loads(args), indent=2, ensure_ascii=False) |
|
except (ValueError, TypeError): |
|
return args |
|
return json.dumps(args, indent=2, ensure_ascii=False) |
|
|
|
|
|
def maybe_truncate(text: str, limit: int) -> str: |
|
if limit and len(text) > limit: |
|
return text[:limit] + f"\n… [truncated {len(text) - limit} chars]" |
|
return text |
|
|
|
|
|
def message_text(log: dict) -> str: |
|
"""Assistant text for a ``request`` log entry.""" |
|
resp = log.get("response") |
|
if isinstance(resp, dict): |
|
msg = resp.get("message") |
|
if isinstance(msg, list): |
|
return "".join(m for m in msg if isinstance(m, str)).strip() |
|
if isinstance(msg, str): |
|
return msg.strip() |
|
return "" |
|
|
|
|
|
def tool_output(log: dict) -> str: |
|
"""Tool output for a ``toolCall`` log entry.""" |
|
resp = log.get("response") |
|
if isinstance(resp, list): |
|
return "\n".join(str(x) for x in resp).strip() |
|
if isinstance(resp, str): |
|
return resp.strip() |
|
return "" |
|
|
|
|
|
def emit_log(log: dict, lines: list[str], limit: int) -> None: |
|
"""Append the markdown for a single timeline entry to ``lines``.""" |
|
thinking = log.get("thinking") |
|
if isinstance(thinking, dict): |
|
ttext = (thinking.get("text") or "").strip() |
|
if ttext: |
|
lines.append("> **Thinking**") |
|
lines.append(">") |
|
lines.append(blockquote(maybe_truncate(ttext, limit))) |
|
lines.append("") |
|
|
|
kind = log.get("kind") |
|
if kind == "request": |
|
msg = message_text(log) |
|
if msg: |
|
lines.append(msg) |
|
lines.append("") |
|
elif kind == "toolCall": |
|
tool = log.get("tool", "unknown") |
|
lines.append(f"> **Tool call: `{tool}`**") |
|
lines.append(">") |
|
args = format_args(log.get("args", "")) |
|
if args.strip(): |
|
lines.append("> Arguments:") |
|
lines.append(">") |
|
lines.append(fenced_in_quote(maybe_truncate(args, limit), "json")) |
|
lines.append(">") |
|
output = tool_output(log) |
|
if output: |
|
lines.append("> Output:") |
|
lines.append(">") |
|
lines.append(fenced_in_quote(maybe_truncate(output, limit))) |
|
lines.append("") |
|
|
|
|
|
def find_start(prompts: list[dict], query: str | None) -> int | None: |
|
"""Index of the first prompt to start from.""" |
|
if query: |
|
q = query.lower() |
|
for i, p in enumerate(prompts): |
|
if q in (p.get("prompt") or "").lower(): |
|
return i |
|
return None |
|
# Default: skip the internal "title" helper prompt. |
|
for i, p in enumerate(prompts): |
|
if (p.get("prompt") or "").strip().lower() != "title": |
|
return i |
|
return 0 if prompts else None |
|
|
|
|
|
def _is_log_entry(obj) -> bool: |
|
"""A timeline log entry (request or toolCall).""" |
|
return isinstance(obj, dict) and ("kind" in obj or "tool" in obj) |
|
|
|
|
|
def _is_prompt_object(obj) -> bool: |
|
"""A prompt turn: has its own ``logs`` timeline.""" |
|
return isinstance(obj, dict) and isinstance(obj.get("logs"), list) |
|
|
|
|
|
def normalize_prompts(data) -> list[dict]: |
|
"""Coerce the various dump shapes into a list of prompt objects (each with ``logs``). |
|
|
|
Accepts: |
|
* ``{"prompts": [ ... ]}`` -> the full multi-turn dump |
|
* ``{"prompt": ..., "logs": [...]}`` -> a single prompt turn |
|
* ``{"kind": ...}`` / ``{"tool": ...}`` -> a single log entry |
|
* ``[ prompt, prompt, ... ]`` -> an array of prompt turns |
|
* ``[ log, log, ... ]`` -> a bare array of log entries |
|
* a single log entry (dict) -> one synthetic turn |
|
""" |
|
if isinstance(data, dict): |
|
prompts = data.get("prompts") |
|
if isinstance(prompts, list): |
|
return prompts |
|
if _is_prompt_object(data): |
|
return [data] |
|
if _is_log_entry(data): |
|
return [{"prompt": "", "logs": [data]}] |
|
return [] |
|
if isinstance(data, list): |
|
if not data: |
|
return [] |
|
if all(_is_prompt_object(x) for x in data): |
|
return list(data) |
|
if any(_is_log_entry(x) for x in data): |
|
return [{"prompt": "", "logs": [x for x in data if isinstance(x, dict)]}] |
|
# Fall back to treating list items as prompt objects. |
|
return [x for x in data if isinstance(x, dict)] |
|
return [] |
|
|
|
|
|
def build_markdown(prompts: list[dict], limit: int) -> str: |
|
lines: list[str] = [] |
|
multiple = len(prompts) > 1 |
|
for p in prompts: |
|
prompt_text = (p.get("prompt") or "").strip() |
|
# Only drop the internal "title" helper turn when there is other content. |
|
if prompt_text.lower() == "title" and multiple: |
|
continue |
|
if prompt_text: |
|
lines.append("User:") |
|
lines.append("") |
|
lines.append(prompt_text) |
|
lines.append("") |
|
lines.append("GitHub Copilot:") |
|
lines.append("") |
|
for log in p.get("logs", []): |
|
emit_log(log, lines, limit) |
|
lines.append("---") |
|
lines.append("") |
|
return "\n".join(lines).rstrip() + "\n" |
|
|
|
|
|
def main(argv=None) -> int: |
|
ap = argparse.ArgumentParser( |
|
description="Interleave thinking + tool calls from a Copilot JSON dump into enriched markdown." |
|
) |
|
ap.add_argument("json_path", type=Path, help="Path to the *.json dump.") |
|
ap.add_argument( |
|
"--query", |
|
default=None, |
|
help="Substring to locate the first user prompt to start from (case-insensitive). " |
|
"Defaults to the first non-'title' prompt.", |
|
) |
|
ap.add_argument( |
|
"--out", |
|
type=Path, |
|
default=None, |
|
help="Output markdown path. Defaults to <json_stem>.enriched.md next to the input.", |
|
) |
|
ap.add_argument("--stdout", action="store_true", help="Write to stdout instead of a file.") |
|
ap.add_argument( |
|
"--only-first", |
|
action="store_true", |
|
help="Only emit the matched prompt, not the following turns.", |
|
) |
|
ap.add_argument( |
|
"--max-chars", |
|
type=int, |
|
default=0, |
|
help="Truncate long thinking/args/output blocks to this many chars (0 = no limit).", |
|
) |
|
args = ap.parse_args(argv) |
|
|
|
try: |
|
data = json.loads(args.json_path.read_text(encoding="utf-8")) |
|
except (OSError, ValueError) as e: |
|
print(f"Failed to read JSON: {e}", file=sys.stderr) |
|
return 1 |
|
|
|
prompts = normalize_prompts(data) |
|
if not prompts: |
|
print("No prompts or logs found in dump (unrecognized shape).", file=sys.stderr) |
|
return 1 |
|
|
|
start = find_start(prompts, args.query) |
|
if start is None: |
|
print(f"No prompt matched query: {args.query!r}", file=sys.stderr) |
|
return 1 |
|
|
|
selected = prompts[start : start + 1] if args.only_first else prompts[start:] |
|
output = build_markdown(selected, args.max_chars) |
|
|
|
if args.stdout: |
|
sys.stdout.write(output) |
|
return 0 |
|
|
|
out_path = args.out or args.json_path.with_name(args.json_path.stem + ".enriched.md") |
|
out_path.write_text(output, encoding="utf-8") |
|
print(f"Wrote {out_path} ({len(selected)} prompt(s), {output.count(chr(10))} lines).") |
|
return 0 |
|
|
|
|
|
if __name__ == "__main__": |
|
raise SystemExit(main()) |