Spaces:
Running
Running
File size: 3,451 Bytes
62f3e9a 02321f4 62f3e9a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 | #!/usr/bin/env python
"""The tracker from a shell: what the agent worker (scripts/agent_worker.sh)
and a person at a terminal need, over the same store the MCP tools use.
.venv/bin/python scripts/tracker_cli.py next --owner demoprep.agent
.venv/bin/python scripts/tracker_cli.py list --owner jack
.venv/bin/python scripts/tracker_cli.py get T-42
.venv/bin/python scripts/tracker_cli.py update T-42 --by mike.boone --status in_progress
.venv/bin/python scripts/tracker_cli.py update T-42 --by demoprep.agent --status review \\
--pr-url https://github.com/boone10/DemoPrep/pull/57 --note "PR opened"
.venv/bin/python scripts/tracker_cli.py note T-42 --by demoprep.agent --note "no change made: ..."
Output is JSON on stdout. Exit 1 with the refusal on stderr when the store
refuses (unknown person, bad status, no such task).
"""
from __future__ import annotations
import argparse
import json
import sys
from pathlib import Path
REPO_ROOT = Path(__file__).resolve().parents[1]
if str(REPO_ROOT) not in sys.path:
sys.path.insert(0, str(REPO_ROOT))
from dotenv import load_dotenv # noqa: E402
load_dotenv(REPO_ROOT / ".env")
load_dotenv() # or the nearest .env up the tree, when run from a worktree
from demoprep_app.tracker import store as S # noqa: E402
from demoprep_app.tracker.store import TrackerError # noqa: E402
def main(argv: list[str] | None = None) -> int:
ap = argparse.ArgumentParser(description="DemoPrep team tracker")
sub = ap.add_subparsers(dest="cmd", required=True)
p = sub.add_parser("next", help="the top open task for an owner, or nothing")
p.add_argument("--owner", required=True)
p = sub.add_parser("list")
p.add_argument("--owner", default="")
p.add_argument("--status", default="")
p.add_argument("--include-done", action="store_true")
p = sub.add_parser("get")
p.add_argument("task")
p = sub.add_parser("update")
p.add_argument("task")
p.add_argument("--by", required=True)
for f in ("title", "body", "status", "priority", "label", "owner", "due-date", "pr-url",
"run-id", "liveboard-url", "note"):
p.add_argument(f"--{f}", default="")
p = sub.add_parser("note")
p.add_argument("task")
p.add_argument("--by", required=True)
p.add_argument("--note", required=True)
a = ap.parse_args(argv)
try:
if a.cmd == "next":
tasks = [t for t in S.list_tasks(owner=a.owner, limit=200) if t["status"] in ("open", "in_progress")]
out = tasks[0] if tasks else None
elif a.cmd == "list":
out = S.list_tasks(owner=a.owner, status=a.status, include_done=a.include_done, limit=200)
elif a.cmd == "get":
out = S.get_task(a.task)
elif a.cmd == "update":
fields = {k.replace("-", "_"): getattr(a, k.replace("-", "_"))
for k in ("title", "body", "status", "priority", "label", "owner", "due-date",
"pr-url", "run-id", "liveboard-url")}
changes = {k: v for k, v in fields.items() if v}
out = S.update_task(a.task, changes, by=a.by, note=a.note)
else:
out = {"note_id": S.add_note(a.task, a.note, by=a.by)}
except TrackerError as e:
print(str(e), file=sys.stderr)
return 1
print(json.dumps(out, indent=2, default=str))
return 0
if __name__ == "__main__":
sys.exit(main())
|