AiDOPWarehouse/doc/_recover_from_transcript.py

#!/usr/bin/env python3
"""Replay Write/StrReplace tool calls from agent transcript onto repo."""
from __future__ import annotations

import json
import re
import sys
from pathlib import Path

REPO = Path(__file__).resolve().parents[1]
TRANSCRIPT = Path(
    r"C:\Users\skygu\.cursor\projects\d-Projects-Ai-DOP-SourceCode-ZZYDOP"
    r"\agent-transcripts\09f6a7e3-4c30-4108-b1d7-f3d16f6412f6"
    r"\09f6a7e3-4c30-4108-b1d7-f3d16f6412f6.jsonl"
)

PREFIXES = ("Web", "server", "doc")


def to_rel(path: str) -> str | None:
    p = path.replace("/", "\\")
    m = re.search(r"ZZYDOP[\\/](.+)$", p, re.I)
    if not m:
        return None
    rel = m.group(1)
    if rel.startswith(PREFIXES):
        return rel
    return None


def iter_ops():
    text = TRANSCRIPT.read_text(encoding="utf-8", errors="replace")
    for line in text.splitlines():
        try:
            obj = json.loads(line)
        except json.JSONDecodeError:
            continue
        msg = obj.get("message", {})
        content = msg.get("content", [])
        if not isinstance(content, list):
            continue
        for part in content:
            if not isinstance(part, dict) or part.get("type") != "tool_use":
                continue
            name = part.get("name")
            inp = part.get("input", {})
            if not isinstance(inp, dict):
                continue
            rel = to_rel(inp.get("path", ""))
            if not rel:
                continue
            if name == "Write":
                yield ("write", rel, inp.get("contents", ""))
            elif name == "StrReplace":
                yield (
                    "replace",
                    rel,
                    inp.get("old_string", ""),
                    inp.get("new_string", ""),
                    bool(inp.get("replace_all")),
                )


def main() -> int:
    writes = 0
    replaces_ok = 0
    replaces_fail: list[str] = []

    for op in iter_ops():
        kind = op[0]
        rel = op[1]
        target = REPO / rel
        if kind == "write":
            contents = op[2]
            target.parent.mkdir(parents=True, exist_ok=True)
            target.write_text(contents, encoding="utf-8", newline="\n")
            writes += 1
            print(f"WRITE {rel}")
            continue

        old, new, replace_all = op[2], op[3], op[4]
        if not target.is_file():
            replaces_fail.append(f"{rel}: file missing")
            continue
        try:
            text = target.read_text(encoding="utf-8")
        except UnicodeDecodeError:
            replaces_fail.append(f"{rel}: not utf-8 text")
            continue
        if old not in text:
            replaces_fail.append(f"{rel}: old_string not found")
            continue
        if replace_all:
            text = text.replace(old, new)
        else:
            text = text.replace(old, new, 1)
        target.write_text(text, encoding="utf-8", newline="\n")
        replaces_ok += 1
        print(f"REPLACE {rel}")

    print(f"\nDone: writes={writes}, replaces_ok={replaces_ok}, replaces_fail={len(replaces_fail)}")
    for item in replaces_fail[:40]:
        print(f"  FAIL {item}")
    if len(replaces_fail) > 40:
        print(f"  ... and {len(replaces_fail) - 40} more")
    return 0 if not replaces_fail else 1


if __name__ == "__main__":
    sys.exit(main())