| """The command: run git log here, grade it, print the files most likely overwritten. Nothing leaves this machine.""" |
| import argparse |
| import json |
| import subprocess |
| import sys |
| from datetime import datetime |
|
|
| from . import __version__ |
| from .grade import grade, score_line |
|
|
| PREORDER = "https://overflow.games/memento/?ref=claudeplugin" |
| FIX = ("memento Team gives each file one owner at a time across machines: " |
| "a session claims a path before it edits, and the next one waits.") |
|
|
|
|
| def git_log(repo, since): |
| cmd = ["git", "-C", repo, "log", "--all", "--source", "--name-only", |
| "--since=" + since, "--format=%H|%an|%aI|%S"] |
| try: |
| r = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE) |
| except OSError as e: |
| raise SystemExit("agent-overwrite-grader: git is not on the PATH (%s)" % e) |
| if r.returncode: |
| raise SystemExit("agent-overwrite-grader: git log failed in %s: %s" |
| % (repo, r.stderr.decode("utf-8", "replace").strip())) |
| return r.stdout.decode("utf-8", "replace") |
|
|
|
|
| def when(ms): |
| t = datetime.fromtimestamp(ms / 1000) |
| return "%s %d %s" % (t.strftime("%b"), t.day, t.strftime("%H:%M")) |
|
|
|
|
| def who(c): |
| return c["hash"][:7] + " " + c["author"] + (" on " + c["ref"] if c["ref"] else "") |
|
|
|
|
| def text(report, top): |
| out = [score_line(report), "%d commits by %d sessions read, on this machine only." % (report["commits"], report["sessions"]), ""] |
| for f in report["files"][:top]: |
| p = f["pairs"][0] |
| line = " %d %s" % (f["count"], f["file"]) |
| pair = " %s, %s, then %d min later %s" % (who(p["a"]), when(p["a"]["when"]), p["minutes"], who(p["b"])) |
| if f["count"] > 1: |
| pair += " (%d such pairs)" % f["count"] |
| out += [line, pair] |
| n = len(report["files"]) |
| if n > top: |
| out.append("And %d more %s." % (n - top, "file" if n - top == 1 else "files")) |
| elif n == 0: |
| out.append("Nothing collided in this history. The more agents run at once, the sooner it does.") |
| out += ["", FIX, "Pre-order memento Team, 10 USD per agent a month: " + PREORDER] |
| return "\n".join(out) |
|
|
|
|
| def main(argv=None): |
| ap = argparse.ArgumentParser( |
| prog="agent-overwrite-grader", |
| description="List every file two sessions (two authors, or two branches) changed less than an hour apart: " |
| "the edits most likely to overwrite each other. It reads the local git log; nothing is sent anywhere.") |
| ap.add_argument("repo", nargs="?", default=".", help="the repository to read (default: here)") |
| ap.add_argument("--since", default="1 month ago", help="how far back to read (default: 1 month ago)") |
| ap.add_argument("--log", metavar="FILE", help="grade a saved log instead ('-' for stdin), in the format of: " |
| "git log --all --source --name-only --format='%%H|%%an|%%aI|%%S'") |
| ap.add_argument("--top", type=int, default=12, help="files to show (default: 12)") |
| ap.add_argument("--json", action="store_true", help="print the report as JSON") |
| ap.add_argument("--version", action="version", version="%(prog)s " + __version__) |
| a = ap.parse_args(argv) |
| try: |
| if a.log == "-": |
| log = sys.stdin.read() |
| elif a.log: |
| with open(a.log, encoding="utf-8", errors="replace") as f: |
| log = f.read() |
| else: |
| log = git_log(a.repo, a.since) |
| except SystemExit as e: |
| print(e, file=sys.stderr) |
| return 2 |
| report = grade(log) |
| if not report["commits"] and a.log: |
| print("agent-overwrite-grader: no commit in that text; save the output of the git log command in --help.", |
| file=sys.stderr) |
| return 1 |
| if a.json: |
| report["preorder"] = PREORDER |
| print(json.dumps(report, indent=2)) |
| else: |
| print(text(report, a.top)) |
| return 0 |
|
|
