zhi
9e22c97ae8
Milestone: open/freeze/undergoing/completed/closed (was open/pending/deferred/progressing/closed) Task: open/pending/undergoing/completed/closed (was open/pending/progressing/closed) - Add MilestoneStatusEnum to schemas with typed validation - Add started_at field to Milestone model - Update all router/CLI references from progressing->undergoing - Add completed status handling in task transition logic
329 lines
11 KiB
Python
Executable File
329 lines
11 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""HarborForge CLI - 简易命令行工具"""
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import sys
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
|
|
BASE_URL = os.environ.get("HARBORFORGE_URL", "http://localhost:8000")
|
|
TOKEN = os.environ.get("HARBORFORGE_TOKEN", "")
|
|
|
|
|
|
STATUS_ICON = {
|
|
"open": "🟢",
|
|
"pending": "🟡",
|
|
"freeze": "🧊",
|
|
"undergoing": "🔵",
|
|
"completed": "✅",
|
|
"closed": "⚫",
|
|
}
|
|
TYPE_ICON = {
|
|
"resolution": "⚖️",
|
|
"task": "📋",
|
|
"story": "📖",
|
|
"test": "🧪",
|
|
"issue": "📌",
|
|
"maintenance": "🛠️",
|
|
"research": "🔬",
|
|
"review": "🧐",
|
|
}
|
|
|
|
|
|
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 is not None 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
|
|
raw = resp.read()
|
|
return json.loads(raw) if raw else None
|
|
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_tasks(args):
|
|
params = []
|
|
if args.project:
|
|
params.append(f"project_id={args.project}")
|
|
if args.type:
|
|
params.append(f"task_type={args.type}")
|
|
if args.status:
|
|
params.append(f"task_status={args.status}")
|
|
qs = f"?{'&'.join(params)}" if params else ""
|
|
result = _request("GET", f"/tasks{qs}")
|
|
items = result.get("items", result if isinstance(result, list) else [])
|
|
for task in items:
|
|
status_icon = STATUS_ICON.get(task["status"], "❓")
|
|
type_icon = TYPE_ICON.get(task.get("task_type"), "📌")
|
|
print(f" {status_icon} {type_icon} #{task['id']} [{task['priority']}] {task['title']}")
|
|
|
|
|
|
def cmd_task_create(args):
|
|
data = {
|
|
"title": args.title,
|
|
"project_id": args.project,
|
|
"milestone_id": args.milestone,
|
|
"reporter_id": args.reporter,
|
|
"task_type": args.type,
|
|
"priority": args.priority or "medium",
|
|
}
|
|
if args.description:
|
|
data["description"] = args.description
|
|
if args.assignee:
|
|
data["assignee_id"] = args.assignee
|
|
if args.subtype:
|
|
data["task_subtype"] = args.subtype
|
|
|
|
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", "/tasks", data)
|
|
print(f"Created task #{result['id']}: {result['title']}")
|
|
|
|
|
|
def cmd_projects(args):
|
|
projects = _request("GET", "/projects")
|
|
for project in projects:
|
|
print(f" #{project['id']} {project['name']} - {project.get('description', '')}")
|
|
|
|
|
|
def cmd_users(args):
|
|
users = _request("GET", "/users")
|
|
for user in users:
|
|
role = "👑" if user["is_admin"] else "👤"
|
|
print(f" {role} #{user['id']} {user['username']} ({user.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={urllib.parse.quote(args.query)}"]
|
|
if args.project:
|
|
params.append(f"project_id={args.project}")
|
|
result = _request("GET", f"/search/tasks?{'&'.join(params)}")
|
|
items = result.get("items", result if isinstance(result, list) else [])
|
|
if not items:
|
|
print(" No results found.")
|
|
return
|
|
for task in items:
|
|
status_icon = STATUS_ICON.get(task["status"], "❓")
|
|
type_icon = TYPE_ICON.get(task.get("task_type"), "📌")
|
|
print(f" {status_icon} {type_icon} #{task['id']} [{task['priority']}] {task['title']}")
|
|
|
|
|
|
def cmd_transition(args):
|
|
result = _request("POST", f"/tasks/{args.task_id}/transition?new_status={args.status}")
|
|
print(f"Task #{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_tasks']}")
|
|
print("By status:")
|
|
for status_name, count in stats["by_status"].items():
|
|
if count > 0:
|
|
print(f" {status_name}: {count}")
|
|
print("By type:")
|
|
for task_type, count in stats["by_type"].items():
|
|
if count > 0:
|
|
print(f" {task_type}: {count}")
|
|
|
|
|
|
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 milestone in milestones:
|
|
status_icon = STATUS_ICON.get(milestone["status"], "⚪")
|
|
due = f" (due: {milestone['due_date'][:10]})" if milestone.get("due_date") else ""
|
|
print(f" {status_icon} #{milestone['id']} {milestone['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_tasks']})")
|
|
|
|
|
|
def cmd_notifications(args):
|
|
params = []
|
|
if args.unread:
|
|
params.append("unread_only=true")
|
|
qs = f"?{'&'.join(params)}" if params else ""
|
|
notifications = _request("GET", f"/notifications{qs}")
|
|
if not notifications:
|
|
print(" No notifications.")
|
|
return
|
|
for notification in notifications:
|
|
icon = "🔴" if not notification["is_read"] else "⚪"
|
|
print(f" {icon} [{notification['type']}] {notification.get('message') or notification['title']}")
|
|
|
|
|
|
def cmd_overdue(args):
|
|
print("Overdue tasks are not supported by the current milestone-based task schema.")
|
|
|
|
|
|
def cmd_log_time(args):
|
|
from datetime import datetime
|
|
|
|
data = {
|
|
"task_id": args.task_id,
|
|
"user_id": args.user_id,
|
|
"hours": args.hours,
|
|
"logged_date": datetime.utcnow().isoformat(),
|
|
}
|
|
if args.desc:
|
|
data["description"] = args.desc
|
|
result = _request("POST", "/worklogs", data)
|
|
print(f"Logged {result['hours']}h on task #{result['task_id']} (log #{result['id']})")
|
|
|
|
|
|
def cmd_worklogs(args):
|
|
logs = _request("GET", f"/tasks/{args.task_id}/worklogs")
|
|
for log in logs:
|
|
desc = f" - {log['description']}" if log.get("description") else ""
|
|
print(f" [{log['id']}] {log['hours']}h by user#{log['user_id']} on {log['logged_date']}{desc}")
|
|
summary = _request("GET", f"/tasks/{args.task_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")
|
|
|
|
p_login = sub.add_parser("login", help="Login and get token")
|
|
p_login.add_argument("username")
|
|
p_login.add_argument("password")
|
|
|
|
p_tasks = sub.add_parser("tasks", aliases=["issues"], help="List tasks")
|
|
p_tasks.add_argument("--project", "-p", type=int)
|
|
p_tasks.add_argument("--type", "-t", choices=["task", "story", "test", "resolution", "issue", "maintenance", "research", "review"])
|
|
p_tasks.add_argument("--status", "-s", choices=["open", "pending", "undergoing", "completed", "closed"])
|
|
|
|
p_create = sub.add_parser("create-task", aliases=["create-issue"], help="Create a task")
|
|
p_create.add_argument("title")
|
|
p_create.add_argument("--project", "-p", type=int, required=True)
|
|
p_create.add_argument("--milestone", "-m", 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", "issue", "maintenance", "research", "review"])
|
|
p_create.add_argument("--subtype")
|
|
p_create.add_argument("--priority", choices=["low", "medium", "high", "critical"])
|
|
p_create.add_argument("--description", "-d")
|
|
p_create.add_argument("--assignee", "-a", type=int)
|
|
p_create.add_argument("--summary")
|
|
p_create.add_argument("--positions")
|
|
p_create.add_argument("--pending")
|
|
|
|
sub.add_parser("projects", help="List projects")
|
|
sub.add_parser("users", help="List users")
|
|
sub.add_parser("version", help="Show version")
|
|
sub.add_parser("health", help="Health check")
|
|
|
|
p_search = sub.add_parser("search", help="Search tasks")
|
|
p_search.add_argument("query")
|
|
p_search.add_argument("--project", "-p", type=int)
|
|
|
|
p_trans = sub.add_parser("transition", help="Transition task status")
|
|
p_trans.add_argument("task_id", type=int)
|
|
p_trans.add_argument("status", choices=["open", "pending", "undergoing", "completed", "closed"])
|
|
|
|
p_stats = sub.add_parser("stats", help="Dashboard stats")
|
|
p_stats.add_argument("--project", "-p", type=int)
|
|
|
|
p_ms = sub.add_parser("milestones", help="List milestones")
|
|
p_ms.add_argument("--project", "-p", type=int)
|
|
|
|
p_msp = sub.add_parser("milestone-progress", help="Show milestone progress")
|
|
p_msp.add_argument("milestone_id", type=int)
|
|
|
|
p_notif = sub.add_parser("notifications", help="List notifications for current token user")
|
|
p_notif.add_argument("--unread", action="store_true")
|
|
|
|
p_overdue = sub.add_parser("overdue", help="Explain overdue-task support status")
|
|
p_overdue.add_argument("--project", "-p", type=int)
|
|
|
|
p_logtime = sub.add_parser("log-time", help="Log time on a task")
|
|
p_logtime.add_argument("task_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 a task")
|
|
p_worklogs.add_argument("task_id", type=int)
|
|
|
|
args = parser.parse_args()
|
|
if not args.command:
|
|
parser.print_help()
|
|
sys.exit(1)
|
|
|
|
cmds = {
|
|
"login": cmd_login,
|
|
"tasks": cmd_tasks,
|
|
"issues": cmd_tasks,
|
|
"create-task": cmd_task_create,
|
|
"create-issue": cmd_task_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__":
|
|
main()
|