#!/usr/bin/env python3 """HarborForge CLI - 简易命令行工具""" import argparse import json import os import sys import urllib.request import urllib.error BASE_URL = os.environ.get("HARBORFORGE_URL", "http://localhost:8000") TOKEN = os.environ.get("HARBORFORGE_TOKEN", "") def _request(method, path, data=None): url = f"{BASE_URL}{path}" headers = {"Content-Type": "application/json"} if TOKEN: headers["Authorization"] = f"Bearer {TOKEN}" body = json.dumps(data).encode() if data else None req = urllib.request.Request(url, data=body, headers=headers, method=method) try: with urllib.request.urlopen(req) as resp: if resp.status == 204: return None return json.loads(resp.read()) except urllib.error.HTTPError as e: print(f"Error {e.code}: {e.read().decode()}", file=sys.stderr) sys.exit(1) def cmd_login(args): data = urllib.parse.urlencode({"username": args.username, "password": args.password}).encode() req = urllib.request.Request(f"{BASE_URL}/auth/token", data=data, method="POST") req.add_header("Content-Type", "application/x-www-form-urlencoded") try: with urllib.request.urlopen(req) as resp: result = json.loads(resp.read()) print(f"Token: {result['access_token']}") print(f"\nExport it:\nexport HARBORFORGE_TOKEN={result['access_token']}") except urllib.error.HTTPError as e: print(f"Login failed: {e.read().decode()}", file=sys.stderr) sys.exit(1) def cmd_issues(args): params = [] if args.project: params.append(f"project_id={args.project}") if args.type: params.append(f"issue_type={args.type}") if args.status: params.append(f"issue_status={args.status}") qs = f"?{'&'.join(params)}" if params else "" issues = _request("GET", f"/issues{qs}") for i in issues: status_icon = {"open": "🟢", "in_progress": "🔵", "resolved": "✅", "closed": "⚫", "blocked": "🔴"}.get(i["status"], "❓") type_icon = {"resolution": "⚖️", "task": "📋", "story": "📖", "test": "🧪"}.get(i["issue_type"], "📌") print(f" {status_icon} {type_icon} #{i['id']} [{i['priority']}] {i['title']}") def cmd_issue_create(args): data = { "title": args.title, "project_id": args.project, "reporter_id": args.reporter, "issue_type": args.type, "priority": args.priority or "medium", } if args.description: data["description"] = args.description if args.assignee: data["assignee_id"] = args.assignee # Resolution specific if args.type == "resolution": if args.summary: data["resolution_summary"] = args.summary if args.positions: data["positions"] = args.positions if args.pending: data["pending_matters"] = args.pending result = _request("POST", "/issues", data) print(f"Created issue #{result['id']}: {result['title']}") def cmd_projects(args): projects = _request("GET", "/projects") for p in projects: print(f" #{p['id']} {p['name']} - {p.get('description', '')}") def cmd_users(args): users = _request("GET", "/users") for u in users: role = "👑" if u["is_admin"] else "👤" print(f" {role} #{u['id']} {u['username']} ({u.get('full_name', '')})") def cmd_version(args): result = _request("GET", "/version") print(f"{result['name']} v{result['version']}") def cmd_health(args): result = _request("GET", "/health") print(f"Status: {result['status']}") def cmd_search(args): params = [f"q={args.query}"] if args.project: params.append(f"project_id={args.project}") qs = "&".join(params) issues = _request("GET", f"/search/issues?{qs}") if not issues: print(" No results found.") return for i in issues: status_icon = {"open": "\U0001f7e2", "in_progress": "\U0001f535", "resolved": "\u2705", "closed": "\u26ab", "blocked": "\U0001f534"}.get(i["status"], "\u2753") type_icon = {"resolution": "\u2696\ufe0f", "task": "\U0001f4cb", "story": "\U0001f4d6", "test": "\U0001f9ea"}.get(i["issue_type"], "\U0001f4cc") print(f" {status_icon} {type_icon} #{i['id']} [{i['priority']}] {i['title']}") def cmd_transition(args): result = _request("POST", f"/issues/{args.issue_id}/transition?new_status={args.status}") print(f"Issue #{result['id']} transitioned to: {result['status']}") def cmd_stats(args): params = f"?project_id={args.project}" if args.project else "" stats = _request("GET", f"/dashboard/stats{params}") print(f"Total: {stats['total']}") print("By status:") for s, c in stats["by_status"].items(): if c > 0: print(f" {s}: {c}") print("By type:") for t, c in stats["by_type"].items(): if c > 0: print(f" {t}: {c}") def cmd_milestones(args): params = f"?project_id={args.project}" if args.project else "" milestones = _request("GET", f"/milestones{params}") if not milestones: print(" No milestones found.") return for m in milestones: status_icon = "🟢" if m["status"] == "open" else "⚫" due = f" (due: {m['due_date'][:10]})" if m.get("due_date") else "" print(f" {status_icon} #{m['id']} {m['title']}{due}") def cmd_milestone_progress(args): result = _request("GET", f"/milestones/{args.milestone_id}/progress") bar_len = 20 filled = int(bar_len * result["progress_pct"] / 100) bar = "█" * filled + "░" * (bar_len - filled) print(f" {result['title']}") print(f" [{bar}] {result['progress_pct']}% ({result['completed']}/{result['total_issues']})") def cmd_notifications(args): params = [f"user_id={args.user}"] if args.unread: params.append("unread_only=true") qs = "&".join(params) notifs = _request("GET", f"/notifications?{qs}") if not notifs: print(" No notifications.") return for n in notifs: icon = "🔴" if not n["is_read"] else "⚪" print(f" {icon} [{n['type']}] {n['title']}") def cmd_overdue(args): params = f"?project_id={args.project}" if args.project else "" issues = _request("GET", f"/issues/overdue{params}") if not issues: print(" No overdue issues! 🎉") return for i in issues: due = i.get("due_date", "?")[:10] if i.get("due_date") else "?" print(f" ⏰ #{i['id']} [{i['priority']}] {i['title']} (due: {due})") def cmd_log_time(args): from datetime import datetime data = { 'issue_id': args.issue_id, 'user_id': args.user_id, 'hours': args.hours, 'logged_date': datetime.utcnow().isoformat(), } if args.desc: data['description'] = args.desc r = api('POST', '/worklogs', json=data) print(f'Logged {r["hours"]}h on issue #{r["issue_id"]} (log #{r["id"]})') def cmd_worklogs(args): logs = api('GET', f'/issues/{args.issue_id}/worklogs') for l in logs: desc = f' - {l["description"]}' if l.get('description') else '' print(f' [{l["id"]}] {l["hours"]}h by user#{l["user_id"]} on {l["logged_date"]}{desc}') summary = api('GET', f'/issues/{args.issue_id}/worklogs/summary') print(f' Total: {summary["total_hours"]}h ({summary["log_count"]} logs)') def main(): parser = argparse.ArgumentParser(description="HarborForge CLI") sub = parser.add_subparsers(dest="command") # login p_login = sub.add_parser("login", help="Login and get token") p_login.add_argument("username") p_login.add_argument("password") # issues p_issues = sub.add_parser("issues", help="List issues") p_issues.add_argument("--project", "-p", type=int) p_issues.add_argument("--type", "-t", choices=["task", "story", "test", "resolution"]) p_issues.add_argument("--status", "-s") # issue create p_create = sub.add_parser("create-issue", help="Create an issue") p_create.add_argument("title") p_create.add_argument("--project", "-p", type=int, required=True) p_create.add_argument("--reporter", "-r", type=int, required=True) p_create.add_argument("--type", "-t", default="task", choices=["task", "story", "test", "resolution"]) p_create.add_argument("--priority", choices=["low", "medium", "high", "critical"]) p_create.add_argument("--description", "-d") p_create.add_argument("--assignee", "-a", type=int) # Resolution fields p_create.add_argument("--summary") p_create.add_argument("--positions") p_create.add_argument("--pending") # projects sub.add_parser("projects", help="List projects") # users sub.add_parser("users", help="List users") # version sub.add_parser("version", help="Show version") # health sub.add_parser("health", help="Health check") # search p_search = sub.add_parser("search", help="Search issues") p_search.add_argument("query") p_search.add_argument("--project", "-p", type=int) # transition p_trans = sub.add_parser("transition", help="Transition issue status") p_trans.add_argument("issue_id", type=int) p_trans.add_argument("status", choices=["open", "in_progress", "resolved", "closed", "blocked"]) # stats p_stats = sub.add_parser("stats", help="Dashboard stats") p_stats.add_argument("--project", "-p", type=int) # milestones p_ms = sub.add_parser("milestones", help="List milestones") p_ms.add_argument("--project", "-p", type=int) # milestone progress p_msp = sub.add_parser("milestone-progress", help="Show milestone progress") p_msp.add_argument("milestone_id", type=int) # notifications p_notif = sub.add_parser("notifications", help="List notifications") p_notif.add_argument("--user", "-u", type=int, required=True) p_notif.add_argument("--unread", action="store_true") # overdue p_overdue = sub.add_parser("overdue", help="List overdue issues") p_overdue.add_argument("--project", "-p", type=int) p_logtime = sub.add_parser('log-time', help='Log time on an issue') p_logtime.add_argument('issue_id', type=int) p_logtime.add_argument('user_id', type=int) p_logtime.add_argument('hours', type=float) p_logtime.add_argument('--desc', '-d', type=str) p_worklogs = sub.add_parser('worklogs', help='List work logs for an issue') p_worklogs.add_argument('issue_id', type=int) args = parser.parse_args() if not args.command: parser.print_help() sys.exit(1) cmds = { "login": cmd_login, "issues": cmd_issues, "create-issue": cmd_issue_create, "projects": cmd_projects, "users": cmd_users, "version": cmd_version, "health": cmd_health, "search": cmd_search, "transition": cmd_transition, "stats": cmd_stats, "milestones": cmd_milestones, "milestone-progress": cmd_milestone_progress, "notifications": cmd_notifications, "overdue": cmd_overdue, "log-time": cmd_log_time, "worklogs": cmd_worklogs, } cmds[args.command](args) if __name__ == "__main__": import urllib.parse main()