332 lines
11 KiB
Python
Executable File
332 lines
11 KiB
Python
Executable File
#!/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()
|