import argparse
import asyncio
import json
import os
import sys
from typing import Any, Callable, Dict, Iterable, List, Optional, Sequence, TextIO, Tuple
from .api import GitCodeAPI
ClientFactory = Callable[[argparse.Namespace], GitCodeAPI]
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(
prog="gitcode",
description="GitCode command-line client inspired by GitHub's gh CLI.",
)
parser.add_argument("--json", action="store_true", help="Print raw JSON responses.")
parser.add_argument("--token", help="GitCode personal access token. Defaults to GITCODE_TOKEN.")
parser.add_argument(
"--api-base-url",
default="https://api.gitcode.com/api/v5",
help="GitCode API base URL.",
)
subparsers = parser.add_subparsers(dest="command")
subparsers.required = True
auth = subparsers.add_parser("auth", help="Authentication helpers.")
auth_subparsers = auth.add_subparsers(dest="auth_command")
auth_subparsers.required = True
auth_subparsers.add_parser("status", help="Show token status.")
repo = subparsers.add_parser("repo", help="Repository commands.")
repo_subparsers = repo.add_subparsers(dest="repo_command")
repo_subparsers.required = True
repo_list = repo_subparsers.add_parser("list", help="List your repositories.")
repo_list.add_argument("--visibility", choices=["all", "public", "private"])
repo_list.add_argument("--sort", choices=["created", "updated", "pushed", "full_name"])
repo_view = repo_subparsers.add_parser("view", help="View a repository.")
repo_view.add_argument("repo", help="Repository in OWNER/NAME format.")
repo_create = repo_subparsers.add_parser("create", help="Create a repository.")
repo_create.add_argument("name", help="Repository name.")
repo_create.add_argument("--description", default="", help="Repository description.")
repo_create.add_argument("--private", action="store_true", help="Create a private repository.")
issue = subparsers.add_parser("issue", help="Issue commands.")
issue_subparsers = issue.add_subparsers(dest="issue_command")
issue_subparsers.required = True
issue_list = issue_subparsers.add_parser("list", help="List issues for a repository.")
issue_list.add_argument("repo", help="Repository in OWNER/NAME format.")
issue_list.add_argument("--state", choices=["open", "closed", "all"], default="open")
issue_view = issue_subparsers.add_parser("view", help="View an issue.")
issue_view.add_argument("repo", help="Repository in OWNER/NAME format.")
issue_view.add_argument("number", type=int, help="Issue number.")
issue_create = issue_subparsers.add_parser("create", help="Create an issue.")
issue_create.add_argument("repo", help="Repository in OWNER/NAME format.")
issue_create.add_argument("--title", required=True, help="Issue title.")
issue_create.add_argument("--body", default="", help="Issue body.")
issue_comment = issue_subparsers.add_parser("comment", help="Comment on an issue.")
issue_comment.add_argument("repo", help="Repository in OWNER/NAME format.")
issue_comment.add_argument("number", type=int, help="Issue number.")
issue_comment.add_argument("--body", required=True, help="Comment body.")
pr = subparsers.add_parser("pr", help="Pull request commands.")
pr_subparsers = pr.add_subparsers(dest="pr_command")
pr_subparsers.required = True
pr_list = pr_subparsers.add_parser("list", help="List pull requests for a repository.")
pr_list.add_argument("repo", help="Repository in OWNER/NAME format.")
pr_list.add_argument("--state", choices=["open", "closed", "all"], default="open")
pr_view = pr_subparsers.add_parser("view", help="View a pull request.")
pr_view.add_argument("repo", help="Repository in OWNER/NAME format.")
pr_view.add_argument("number", type=int, help="Pull request number.")
pr_create = pr_subparsers.add_parser("create", help="Create a pull request.")
pr_create.add_argument("repo", help="Repository in OWNER/NAME format.")
pr_create.add_argument("--title", required=True, help="Pull request title.")
pr_create.add_argument("--head", required=True, help="Source branch.")
pr_create.add_argument("--base", required=True, help="Target branch.")
pr_create.add_argument("--body", default="", help="Pull request body.")
pr_comment = pr_subparsers.add_parser("comment", help="Comment on a pull request.")
pr_comment.add_argument("repo", help="Repository in OWNER/NAME format.")
pr_comment.add_argument("number", type=int, help="Pull request number.")
pr_comment.add_argument("--body", required=True, help="Comment body.")
pr_comment.add_argument("--path", help="Optional file path for a review comment.")
pr_comment.add_argument("--position", type=int, help="Optional line position for a review comment.")
pr_merge = pr_subparsers.add_parser("merge", help="Merge a pull request.")
pr_merge.add_argument("repo", help="Repository in OWNER/NAME format.")
pr_merge.add_argument("number", type=int, help="Pull request number.")
pr_merge.add_argument("--method", choices=["merge", "squash", "rebase"], help="Merge strategy.")
pr_merge.add_argument("--title", help="Merge commit title.")
pr_merge.add_argument("--body", help="Merge commit message.")
search = subparsers.add_parser("search", help="Search GitCode resources.")
search_subparsers = search.add_subparsers(dest="search_command")
search_subparsers.required = True
for subject, help_text in (
("repos", "Search repositories."),
("issues", "Search issues and pull requests."),
("users", "Search users."),
):
search_parser = search_subparsers.add_parser(subject, help=help_text)
search_parser.add_argument("query", nargs="+", help="Search query.")
search_parser.add_argument("--sort", help="Sort field supported by the API.")
search_parser.add_argument("--order", choices=["asc", "desc"], help="Sort order.")
search_parser.add_argument("--limit", type=int, dest="per_page", help="Results per page.")
search_parser.add_argument("--page", type=int, help="Page number.")
api = subparsers.add_parser("api", help="Call a GitCode API endpoint directly.")
api.add_argument("method", help="HTTP method.")
api.add_argument("path", help="API path, such as /user/repos.")
api.add_argument(
"--field",
action="append",
default=[],
metavar="KEY=VALUE",
help="Request field. Reuse for multiple values.",
)
return parser
def parse_repo_slug(value: str) -> Tuple[str, str]:
if "/" not in value:
raise ValueError("Repository must be in OWNER/NAME format.")
owner, repo = value.split("/", 1)
if not owner or not repo:
raise ValueError("Repository must be in OWNER/NAME format.")
return owner, repo
def default_client_factory(args: argparse.Namespace) -> GitCodeAPI:
token = args.token or os.environ.get("GITCODE_TOKEN")
return GitCodeAPI(token=token, api_base_url=args.api_base_url)
def infer_field_value(raw_value: str) -> Any:
try:
return json.loads(raw_value)
except json.JSONDecodeError:
return raw_value
def parse_fields(items: Iterable[str]) -> Dict[str, Any]:
parsed: Dict[str, Any] = {}
for item in items:
if "=" not in item:
raise ValueError(f"Invalid field '{item}'. Expected KEY=VALUE.")
key, raw_value = item.split("=", 1)
if not key:
raise ValueError(f"Invalid field '{item}'. Expected KEY=VALUE.")
parsed[key] = infer_field_value(raw_value)
return parsed
def print_json(data: Any, stdout: TextIO) -> None:
stdout.write(json.dumps(data, ensure_ascii=False, indent=2) + "\n")
def print_lines(lines: Iterable[str], stdout: TextIO) -> None:
for line in lines:
stdout.write(line + "\n")
def format_repo_summary(repo: Dict[str, Any]) -> List[str]:
visibility = "private" if repo.get("private") else "public"
description = repo.get("description") or ""
stars = repo.get("stargazers_count", 0)
forks = repo.get("forks_count", 0)
default_branch = repo.get("default_branch", "main")
return [
repo.get("full_name", repo.get("name", "unknown")),
f" visibility: {visibility} | stars: {stars} | forks: {forks} | default branch: {default_branch}",
f" {description}" if description else "",
]
def format_issue_summary(issue: Dict[str, Any]) -> str:
number = issue.get("number", "?")
state = issue.get("state", "unknown")
title = issue.get("title", "")
return f"#{number} [{state}] {title}"
def format_pr_summary(pr: Dict[str, Any]) -> str:
number = pr.get("number", "?")
state = pr.get("state", "unknown")
title = pr.get("title", "")
return f"#{number} [{state}] {title}"
def format_search_item(item: Dict[str, Any], kind: str) -> str:
if kind == "repos":
return item.get("full_name", item.get("name", "unknown"))
if kind == "users":
return item.get("login", item.get("name", "unknown"))
return format_issue_summary(item)
def emit_output(data: Any, args: argparse.Namespace, stdout: TextIO, formatter: Optional[Callable[[Any], List[str]]] = None) -> None:
if args.json:
print_json(data, stdout)
return
if formatter is None:
print_json(data, stdout)
return
print_lines(formatter(data), stdout)
async def execute_command(
args: argparse.Namespace,
client_factory: ClientFactory,
stdout: TextIO,
) -> int:
if args.command == "auth":
token = args.token or os.environ.get("GITCODE_TOKEN")
if not token:
raise ValueError("GitCode token not found. Set GITCODE_TOKEN or pass --token.")
stdout.write("Authenticated with GitCode token\n")
stdout.write(f"API base URL: {args.api_base_url}\n")
return 0
client = client_factory(args)
try:
if args.command == "repo":
return await handle_repo_command(args, client, stdout)
if args.command == "issue":
return await handle_issue_command(args, client, stdout)
if args.command == "pr":
return await handle_pr_command(args, client, stdout)
if args.command == "search":
return await handle_search_command(args, client, stdout)
if args.command == "api":
return await handle_api_command(args, client, stdout)
raise ValueError(f"Unsupported command: {args.command}")
finally:
await client.close()
async def handle_repo_command(args: argparse.Namespace, client: GitCodeAPI, stdout: TextIO) -> int:
if args.repo_command == "list":
repos = await client.repos.list_user_repos(visibility=args.visibility, sort=args.sort)
emit_output(repos, args, stdout, formatter=lambda items: format_repo_list(items))
return 0
if args.repo_command == "view":
owner, repo = parse_repo_slug(args.repo)
repo_data = await client.repos.get_repo(owner=owner, repo=repo)
emit_output(repo_data, args, stdout, formatter=format_repo_detail)
return 0
if args.repo_command == "create":
repo_data = await client.repos.create_repo(
name=args.name,
description=args.description,
private=args.private,
)
emit_output(repo_data, args, stdout, formatter=format_repo_detail)
return 0
raise ValueError(f"Unsupported repo command: {args.repo_command}")
async def handle_issue_command(args: argparse.Namespace, client: GitCodeAPI, stdout: TextIO) -> int:
owner, repo = parse_repo_slug(args.repo)
if args.issue_command == "list":
issues = await client.issues.list_issues(owner=owner, repo=repo, state=args.state)
emit_output(issues, args, stdout, formatter=lambda items: [format_issue_summary(item) for item in items])
return 0
if args.issue_command == "view":
issue = await client.issues.get_issue(owner=owner, repo=repo, issue_number=args.number)
emit_output(issue, args, stdout, formatter=format_issue_detail)
return 0
if args.issue_command == "create":
issue = await client.issues.create_issue(owner=owner, repo=repo, title=args.title, body=args.body)
emit_output(issue, args, stdout, formatter=format_issue_detail)
return 0
if args.issue_command == "comment":
comment = await client.issues.add_comment(owner=owner, repo=repo, issue_number=args.number, body=args.body)
emit_output(comment, args, stdout, formatter=format_comment_detail)
return 0
raise ValueError(f"Unsupported issue command: {args.issue_command}")
async def handle_pr_command(args: argparse.Namespace, client: GitCodeAPI, stdout: TextIO) -> int:
owner, repo = parse_repo_slug(args.repo)
if args.pr_command == "list":
prs = await client.pulls.list_pull_requests(owner=owner, repo=repo, state=args.state)
emit_output(prs, args, stdout, formatter=lambda items: [format_pr_summary(item) for item in items])
return 0
if args.pr_command == "view":
pr = await client.pulls.get_pull_request(owner=owner, repo=repo, pr_number=args.number)
emit_output(pr, args, stdout, formatter=format_pr_detail)
return 0
if args.pr_command == "create":
pr = await client.pulls.create_pull_request(
owner=owner,
repo=repo,
title=args.title,
head=args.head,
base=args.base,
body=args.body,
)
emit_output(pr, args, stdout, formatter=format_pr_detail)
return 0
if args.pr_command == "comment":
comment = await client.pulls.create_pull_request_comment(
owner=owner,
repo=repo,
pr_number=args.number,
body=args.body,
path=args.path,
position=args.position,
)
emit_output(comment, args, stdout, formatter=format_comment_detail)
return 0
if args.pr_command == "merge":
result = await client.pulls.merge_pull_request(
owner=owner,
repo=repo,
pr_number=args.number,
commit_title=args.title,
commit_message=args.body,
merge_method=args.method,
)
emit_output(result, args, stdout, formatter=format_merge_result)
return 0
raise ValueError(f"Unsupported pr command: {args.pr_command}")
async def handle_search_command(args: argparse.Namespace, client: GitCodeAPI, stdout: TextIO) -> int:
query = " ".join(args.query)
if args.search_command == "repos":
results = await client.search.search_repositories(
query=query,
sort=args.sort,
order=args.order,
per_page=args.per_page,
page=args.page,
)
elif args.search_command == "issues":
results = await client.search.search_issues(
query=query,
sort=args.sort,
order=args.order,
per_page=args.per_page,
page=args.page,
)
elif args.search_command == "users":
results = await client.search.search_users(
query=query,
sort=args.sort,
order=args.order,
per_page=args.per_page,
page=args.page,
)
else:
raise ValueError(f"Unsupported search command: {args.search_command}")
emit_output(results, args, stdout, formatter=lambda payload: format_search_results(payload, args.search_command))
return 0
async def handle_api_command(args: argparse.Namespace, client: GitCodeAPI, stdout: TextIO) -> int:
fields = parse_fields(args.field)
method = args.method.upper()
path = args.path if args.path.startswith("/") else f"/{args.path}"
request_kwargs: Dict[str, Any]
if method == "GET":
request_kwargs = {"params": fields}
else:
request_kwargs = {"data": fields}
result = await client.request(method, path, **request_kwargs)
emit_output(result, args, stdout)
return 0
def format_repo_list(repos: List[Dict[str, Any]]) -> List[str]:
lines: List[str] = []
for repo in repos:
lines.extend(line for line in format_repo_summary(repo) if line)
return lines or ["No repositories found."]
def format_repo_detail(repo: Dict[str, Any]) -> List[str]:
lines = format_repo_summary(repo)
homepage = repo.get("homepage")
if homepage:
lines.append(f" homepage: {homepage}")
clone_url = repo.get("clone_url")
if clone_url:
lines.append(f" clone: {clone_url}")
return [line for line in lines if line]
def format_issue_detail(issue: Dict[str, Any]) -> List[str]:
body = issue.get("body") or ""
return [
format_issue_summary(issue),
f" author: {issue.get('user', {}).get('login', 'unknown')}",
f" updated: {issue.get('updated_at', 'unknown')}",
f" url: {issue.get('html_url', issue.get('url', ''))}",
f" {body}" if body else "",
]
def format_pr_detail(pr: Dict[str, Any]) -> List[str]:
body = pr.get("body") or ""
head_ref = pr.get("head", {}).get("ref", "unknown")
base_ref = pr.get("base", {}).get("ref", "unknown")
return [
format_pr_summary(pr),
f" branches: {head_ref} -> {base_ref}",
f" author: {pr.get('user', {}).get('login', 'unknown')}",
f" updated: {pr.get('updated_at', 'unknown')}",
f" url: {pr.get('html_url', pr.get('url', ''))}",
f" {body}" if body else "",
]
def format_comment_detail(comment: Dict[str, Any]) -> List[str]:
return [
f"Comment #{comment.get('id', 'unknown')}",
f" author: {comment.get('user', {}).get('login', 'unknown')}",
f" {comment.get('body', '')}",
]
def format_merge_result(result: Dict[str, Any]) -> List[str]:
message = result.get("message", "Merge completed.")
sha = result.get("sha")
lines = [message]
if sha:
lines.append(f" sha: {sha}")
return lines
def format_search_results(payload: Dict[str, Any], kind: str) -> List[str]:
items = payload.get("items", [])
total_count = payload.get("total_count")
lines = [f"Found {total_count} result(s)."] if total_count is not None else []
lines.extend(format_search_item(item, kind) for item in items)
return lines or ["No results found."]
def main(
argv: Optional[Sequence[str]] = None,
client_factory: Optional[ClientFactory] = None,
stdout: Optional[TextIO] = None,
stderr: Optional[TextIO] = None,
) -> int:
parser = build_parser()
args = parser.parse_args(argv)
client_factory = client_factory or default_client_factory
stdout_stream: TextIO = stdout if stdout is not None else sys.stdout
stderr_stream: TextIO = stderr if stderr is not None else sys.stderr
try:
return asyncio.run(execute_command(args, client_factory, stdout_stream))
except KeyboardInterrupt:
stderr_stream.write("Interrupted.\n")
return 130
except Exception as exc:
stderr_stream.write(f"Error: {exc}\n")
return 1
if __name__ == "__main__":
raise SystemExit(main())