Spaces:
Running
Running
| #!/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()) | |