zhi/HarborForge.Backend
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: replace issues backend with milestone tasks

Browse Source
This commit is contained in:
zhi
2026-03-16 13:22:14 +00:00
parent dc5d06489d
commit 214a9b109d
20 changed files with 836 additions and 1066 deletions
Download Patch File Download Diff File Expand all files Collapse all files

233
cli.py
View File

@@ -5,27 +5,47 @@ import argparse
import json
import os
import sys
import urllib.request
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": "🟡",
"progressing": "🔵",
"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 else None
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
return json.loads(resp.read())
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)
@@ -45,36 +65,39 @@ def cmd_login(args):
sys.exit(1)
def cmd_issues(args):
def cmd_tasks(args):
params = []
if args.project:
params.append(f"project_id={args.project}")
if args.type:
params.append(f"issue_type={args.type}")
params.append(f"task_type={args.type}")
if args.status:
params.append(f"issue_status={args.status}")
params.append(f"task_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']}")
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_issue_create(args):
def cmd_task_create(args):
data = {
"title": args.title,
"project_id": args.project,
"milestone_id": args.milestone,
"reporter_id": args.reporter,
"issue_type": args.type,
"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
# Resolution specific
if args.type == "resolution":
if args.summary:
data["resolution_summary"] = args.summary
@@ -83,21 +106,21 @@ def cmd_issue_create(args):
if args.pending:
data["pending_matters"] = args.pending
result = _request("POST", "/issues", data)
print(f"Created issue #{result['id']}: {result['title']}")
result = _request("POST", "/tasks", data)
print(f"Created task #{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', '')}")
for project in projects:
print(f" #{project['id']} {project['name']} - {project.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', '')})")
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):
@@ -110,41 +133,38 @@ def cmd_health(args):
print(f"Status: {result['status']}")
def cmd_search(args):
params = [f"q={args.query}"]
params = [f"q={urllib.parse.quote(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:
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 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']}")
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"/issues/{args.issue_id}/transition?new_status={args.status}")
print(f"Issue #{result['id']} transitioned to: {result['status']}")
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']}")
print(f"Total: {stats['total_tasks']}")
print("By status:")
for s, c in stats["by_status"].items():
if c > 0:
print(f" {s}: {c}")
for status_name, count in stats["by_status"].items():
if count > 0:
print(f" {status_name}: {count}")
print("By type:")
for t, c in stats["by_type"].items():
if c > 0:
print(f" {t}: {c}")
for task_type, count in stats["by_type"].items():
if count > 0:
print(f" {task_type}: {count}")
def cmd_milestones(args):
@@ -153,10 +173,10 @@ def cmd_milestones(args):
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}")
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):
@@ -165,140 +185,114 @@ def cmd_milestone_progress(args):
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']})")
print(f" [{bar}] {result['progress_pct']}% ({result['completed']}/{result['total_tasks']})")
def cmd_notifications(args):
params = [f"user_id={args.user}"]
params = []
if args.unread:
params.append("unread_only=true")
qs = "&".join(params)
notifs = _request("GET", f"/notifications?{qs}")
if not notifs:
qs = f"?{'&'.join(params)}" if params else ""
notifications = _request("GET", f"/notifications{qs}")
if not notifications:
print(" No notifications.")
return
for n in notifs:
icon = "🔴" if not n["is_read"] else "⚪"
print(f" {icon} [{n['type']}] {n['title']}")
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):
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})")
print("Overdue tasks are not supported by the current milestone-based task schema.")
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(),
"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
r = api('POST', '/worklogs', json=data)
print(f'Logged {r["hours"]}h on issue #{r["issue_id"]} (log #{r["id"]})')
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 = 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)')
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")
# 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")
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", "progressing", "closed"])
# issue create
p_create = sub.add_parser("create-issue", help="Create an issue")
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"])
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)
# 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 = sub.add_parser("search", help="Search tasks")
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"])
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", "progressing", "closed"])
# 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 = sub.add_parser("notifications", help="List notifications for current token user")
p_notif.add_argument("--unread", action="store_true")
# overdue
p_overdue = sub.add_parser("overdue", help="List overdue issues")
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 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_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 an issue')
p_worklogs.add_argument('issue_id', type=int)
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:
@@ -307,8 +301,10 @@ def main():
cmds = {
"login": cmd_login,
"issues": cmd_issues,
"create-issue": cmd_issue_create,
"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,
@@ -327,5 +323,4 @@ def main():
if __name__ == "__main__":
import urllib.parse
main()