import argparse import json import sys from .errors import TrackerError # type: ignore[reportMissingImports] from .models import Envelope, RetryPolicy # type: ignore[reportMissingImports] from .service import Tracker # type: ignore[reportMissingImports] class JsonArgumentParser(argparse.ArgumentParser): def error(self, message): raise TrackerError("invalid_input", message) def _parse_attempts(value): try: return int(value) except (TypeError, ValueError) as error: raise TrackerError("invalid_input", f"invalid retry attempt count: {value}") from error def _take_global_options(argv): remaining = [] provider = repo = None attempts = 3 index = 0 while index < len(argv): item = argv[index] if item == "--provider" and index + 1 < len(argv): provider = argv[index + 1] index += 2 elif item.startswith("--provider="): provider = item.split("=", 1)[1] index += 1 elif item == "--repo" and index + 1 < len(argv): repo = argv[index + 1] index += 2 elif item.startswith("--repo="): repo = item.split("=", 1)[1] index += 1 elif item == "--retry-attempts" and index + 1 < len(argv): attempts = _parse_attempts(argv[index + 1]) index += 2 elif item.startswith("--retry-attempts="): attempts = _parse_attempts(item.split("=", 1)[1]) index += 1 else: remaining.append(item) index += 1 return remaining, provider, repo, attempts def _add_resource_commands(subparsers, kind): resource = subparsers.add_parser(kind) commands = resource.add_subparsers(dest="action", required=True) create = commands.add_parser("create") create.add_argument("--title", required=True) create.add_argument("--body", default="") create.add_argument("--label", action="append", default=[]) create.add_argument("--assignee", action="append", default=[]) if kind == "pr": create.add_argument("--head") create.add_argument("--base") get = commands.add_parser("get") get.add_argument("number", type=int) get.add_argument("--no-comments", action="store_true") if kind == "pr": get.add_argument("--diff", action="store_true") listing = commands.add_parser("list") listing.add_argument("--state", default="open") listing.add_argument("--label", action="append", default=[]) listing.add_argument("--limit", type=int, default=100) if kind == "pr": listing.add_argument("--external-only", action="store_true") comment = commands.add_parser("comment") comment.add_argument("number", type=int) comment.add_argument("--body", required=True) edit = commands.add_parser("edit") edit.add_argument("number", type=int) edit.add_argument("--title") edit.add_argument("--body") assign = commands.add_parser("assign") assign.add_argument("number", type=int) assign.add_argument("--user", required=True) close = commands.add_parser("close") close.add_argument("number", type=int) close.add_argument("--explanation") if kind == "pr": diff = commands.add_parser("diff") diff.add_argument("number", type=int) return resource def build_parser(): parser = JsonArgumentParser(prog="tracker") commands = parser.add_subparsers(dest="resource", required=True) _add_resource_commands(commands, "issue") _add_resource_commands(commands, "pr") labels = commands.add_parser("label") label_commands = labels.add_subparsers(dest="action", required=True) ensure = label_commands.add_parser("ensure") ensure.add_argument("name") ensure.add_argument("--color", default="ededed") ensure.add_argument("--description") add = label_commands.add_parser("add") add.add_argument("kind", choices=("issue", "pr")) add.add_argument("number", type=int) add.add_argument("name") remove = label_commands.add_parser("remove") remove.add_argument("kind", choices=("issue", "pr")) remove.add_argument("number", type=int) remove.add_argument("name") map_parser = commands.add_parser("map") map_commands = map_parser.add_subparsers(dest="action", required=True) map_create = map_commands.add_parser("create") map_create.add_argument("--title", required=True) map_create.add_argument("--body", default="") map_create.add_argument("--label", action="append", default=[]) child = commands.add_parser("child") child_commands = child.add_subparsers(dest="action", required=True) child_create = child_commands.add_parser("create") child_create.add_argument("map_number", type=int) child_create.add_argument("--title", required=True) child_create.add_argument("--type", dest="wayfinder_type", choices=("research", "prototype", "grilling", "task"), default="task") child_create.add_argument("--body", default="") child_create.add_argument("--label", action="append", default=[]) dependency = commands.add_parser("dependency") dependency_commands = dependency.add_subparsers(dest="action", required=True) dependency_add = dependency_commands.add_parser("add") dependency_add.add_argument("child", type=int) dependency_add.add_argument("blocker", type=int) frontier = commands.add_parser("frontier") frontier.add_argument("map_number", type=int) claim = commands.add_parser("claim") claim.add_argument("kind", choices=("issue", "pr")) claim.add_argument("number", type=int) claim.add_argument("--user") resolve = commands.add_parser("resolve") resolve.add_argument("kind", choices=("issue", "pr")) resolve.add_argument("number", type=int) resolve.add_argument("--answer", required=True) resolve.add_argument("--map", dest="map_number", type=int) reference = commands.add_parser("resolve-reference") reference.add_argument("number", type=int) return parser def _dispatch(tracker, args): resource = args.resource if resource in ("issue", "pr"): if args.action == "create": method = tracker.create_issue if resource == "issue" else tracker.create_pr kwargs = {"body": args.body, "labels": args.label, "assignees": args.assignee} if resource == "pr": kwargs.update(head=args.head, base=args.base) return method(args.title, **kwargs) if args.action == "get": if resource == "issue": return tracker.get_issue(args.number, comments=not args.no_comments) return tracker.get_pr(args.number, comments=not args.no_comments, diff=args.diff) if args.action == "list": if resource == "issue": return tracker.list_issues(state=args.state, labels=args.label, limit=args.limit) return tracker.list_prs(state=args.state, limit=args.limit, external_only=args.external_only) if args.action == "comment": return tracker.comment(resource, args.number, args.body) if args.action == "edit": return (tracker.edit_issue if resource == "issue" else tracker.edit_pr)(args.number, title=args.title, body=args.body) if args.action == "assign": return tracker.assign(resource, args.number, args.user) if args.action == "close": return tracker.close(resource, args.number, explanation=args.explanation) if args.action == "diff": return tracker.diff(args.number) if resource == "label": if args.action == "ensure": return tracker.ensure_label(args.name, color=args.color, description=args.description) if args.action == "add": return tracker.add_label(args.kind, args.number, args.name) return tracker.remove_label(args.kind, args.number, args.name) if resource == "map": return tracker.create_map(args.title, body=args.body, labels=args.label) if resource == "child": return tracker.create_child(args.map_number, args.title, wayfinder_type=args.wayfinder_type, body=args.body, labels=args.label) if resource == "dependency": return tracker.add_dependency(args.child, args.blocker) if resource == "frontier": return tracker.frontier(args.map_number) if resource == "claim": return tracker.claim(args.kind, args.number, user=args.user) if resource == "resolve": return tracker.resolve(args.kind, args.number, args.answer, map_number=args.map_number) return tracker.resolve_reference(args.number) def main(argv=None): argv = list(sys.argv[1:] if argv is None else argv) try: command_argv, provider, repo, attempts = _take_global_options(argv) args = build_parser().parse_args(command_argv) tracker = Tracker(provider=provider, repo=repo, retry=RetryPolicy(attempts=attempts)) envelope = _dispatch(tracker, args) except TrackerError as error: operation = error.operation if operation is None: operation = "cli" error.operation = operation envelope = Envelope(False, getattr(error, "provider", None), operation, error=error.to_dict()).to_dict() except (ValueError, TypeError, OSError) as error: failure = TrackerError("invalid_input", str(error), operation="cli") envelope = Envelope(False, None, "cli", error=failure.to_dict()).to_dict() print(json.dumps(envelope, sort_keys=True)) return 0 if envelope.get("ok") else 1 if __name__ == "__main__": raise SystemExit(main())