#!/usr/bin/env python3 """ Deploymentcenter CLI Tool (dc.py) Umfassendes Befehlszeilenwerkzeug zur Interaktion mit dem Deploymentcenter (Bugtracker, Roadmap, Watchdog, Supervisor und Projects). Funktioniert out-of-the-box ohne externe Abhängigkeiten (nur Python Standardbibliothek). """ import os import sys import json import urllib.request import urllib.error import urllib.parse import argparse from typing import Optional, Dict, Any, List DEFAULT_BASE_URL = "https://dc.mhdf.de" DEFAULT_PROJECT = "deploymentcenter" FALLBACK_TOKEN = "" # ANSI Farbcodes (unterstützt auf modernen Windows- und POSIX-Terminals) class Colors: HEADER = '\033[95m' BLUE = '\033[94m' CYAN = '\033[96m' GREEN = '\033[92m' YELLOW = '\033[93m' RED = '\033[91m' BOLD = '\033[1m' DIM = '\033[2m' RESET = '\033[0m' @classmethod def disable(cls): cls.HEADER = '' cls.BLUE = '' cls.CYAN = '' cls.GREEN = '' cls.YELLOW = '' cls.RED = '' cls.BOLD = '' cls.DIM = '' cls.RESET = '' if not sys.stdout.isatty() or os.name == 'nt' and 'WT_SESSION' not in os.environ and 'TERM' not in os.environ: # Unter Windows CMD ohne ANSI-Support Farben beibehalten wenn möglich try: import ctypes kernel32 = ctypes.windll.kernel32 kernel32.SetConsoleMode(kernel32.GetStdHandle(-11), 7) except Exception: pass def find_token() -> str: """Ermittelt das Deploymentcenter-Token nach Priorität.""" # 1. Umgebungsvariable env_token = os.environ.get("DC_TOKEN") if env_token and env_token.strip(): return env_token.strip() # 2. .mcp.json im aktuellen Verzeichnis oder Repo-Root candidate_paths = [ os.path.join(os.getcwd(), ".mcp.json"), os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), ".mcp.json"), os.path.expanduser("~/.mcp.json") ] for cp in candidate_paths: if os.path.isfile(cp): try: with open(cp, "r", encoding="utf-8") as f: data = json.load(f) servers = data.get("mcpServers", {}) dc_server = servers.get("deploymentcenter", {}) token = dc_server.get("env", {}).get("DC_TOKEN") if token and token.strip(): return token.strip() except Exception: pass # 3. Fallback return FALLBACK_TOKEN class DcClient: def __init__(self, base_url: str, token: str): self.base_url = base_url.rstrip("/") self.token = token def request(self, endpoint: str, method: str = "GET", data: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: url = f"{self.base_url}{endpoint}" headers = { "Authorization": f"Bearer {self.token}", "Content-Type": "application/json", "User-Agent": "dc-cli/1.0" } body = json.dumps(data).encode("utf-8") if data is not None else None req = urllib.request.Request(url, data=body, headers=headers, method=method) try: with urllib.request.urlopen(req, timeout=15) as resp: raw = resp.read().decode("utf-8") return json.loads(raw) if raw.strip() else {} except urllib.error.HTTPError as e: raw_err = e.read().decode("utf-8") try: err_json = json.loads(raw_err) return {"_error": True, "status_code": e.code, "error": err_json} except Exception: return {"_error": True, "status_code": e.code, "message": raw_err or str(e)} except Exception as e: return {"_error": True, "status_code": 0, "message": str(e)} def format_status(status: str) -> str: s = (status or "").lower() if s in ["resolved", "closed", "verified"]: return f"{Colors.GREEN}{status}{Colors.RESET}" if s in ["in_progress", "in_verification", "awaiting_verification"]: return f"{Colors.YELLOW}{status}{Colors.RESET}" if s in ["triage_external", "triage"]: return f"{Colors.CYAN}{status}{Colors.RESET}" if s in ["open", "fix_insufficient"]: return f"{Colors.RED}{status}{Colors.RESET}" return status def format_severity(sev: str) -> str: s = (sev or "").lower() if s in ["critical", "high"]: return f"{Colors.RED}{Colors.BOLD}{sev}{Colors.RESET}" if s == "medium": return f"{Colors.YELLOW}{sev}{Colors.RESET}" if s == "low": return f"{Colors.BLUE}{sev}{Colors.RESET}" return f"{Colors.DIM}{sev}{Colors.RESET}" def cmd_list(client: DcClient, args): query_params = { "action": "list", "limit": args.limit } if args.project: query_params["project_slug"] = args.project if args.status and args.status != "all": query_params["status"] = args.status if args.type and args.type != "all": query_params["type"] = args.type if args.search: query_params["q"] = args.search url = "/api/bugtracker/v1/manage?" + urllib.parse.urlencode(query_params) res = client.request(url) if res.get("_error"): print(f"{Colors.RED}Fehler beim Abrufen:{Colors.RESET} {res}") return 1 items = res.get("items", []) total = res.get("total", len(items)) # Optionaler client-seitiger Sort if args.sort == "id_desc": items.sort(key=lambda x: x.get("id", 0), reverse=True) elif args.sort == "id_asc": items.sort(key=lambda x: x.get("id", 0)) elif args.sort == "severity": order = {"critical": 0, "high": 1, "medium": 2, "low": 3, "wishlist": 4, "idea": 5} items.sort(key=lambda x: order.get(x.get("severity", "").lower(), 99)) print(f"{Colors.BOLD}Bugtracker Items ({len(items)} von {total} angezeigt):{Colors.RESET}\n") for it in items: iid = it.get("id") status = format_status(it.get("status", "")) itype = it.get("type", "") sev = format_severity(it.get("severity", "")) proj = it.get("project_slug", "") title = it.get("title", "") epic = it.get("epic") epic_str = f" [{Colors.CYAN}{epic}{Colors.RESET}]" if epic else "" print(f"#{Colors.BOLD}{iid:<4}{Colors.RESET} [{proj}] [{status}] [{itype}] ({sev}){epic_str} {title}") return 0 def cmd_get(client: DcClient, args): res = client.request(f"/api/bugtracker/v1/manage?action=get&id={args.id}") if res.get("_error"): print(f"{Colors.RED}Fehler beim Abrufen von #{args.id}:{Colors.RESET} {res}") return 1 item = res.get("item") if not item: print(f"{Colors.RED}Item #{args.id} nicht gefunden.{Colors.RESET}") return 1 print(f"{Colors.BOLD}{'='*80}{Colors.RESET}") print(f"#{Colors.BOLD}{item.get('id')} {item.get('title')}{Colors.RESET}") print(f"Projekt: {item.get('project_slug')} (gemeldet von: {item.get('reported_from', 'intern')})") print(f"Status: {format_status(item.get('status'))} | Typ: {item.get('type')} | Schweregrad: {format_severity(item.get('severity'))}") if item.get('epic'): print(f"Roadmap: Bereich {Colors.CYAN}{item.get('epic')}{Colors.RESET} | Rang: {item.get('roadmap_rank')}") if item.get('file_path'): print(f"Fundort: {item.get('file_path')}:{item.get('line_no')}") if item.get('build_version'): print(f"Version: {item.get('build_version')} (Commit: {item.get('commit_sha')})") if item.get('target_build'): print(f"Zielversion:{item.get('target_build')}") print(f"Erstellt: {item.get('created_at')} von {item.get('created_by')}") print(f"{Colors.BOLD}{'='*80}{Colors.RESET}\n") desc = item.get("description") if desc: print(f"{Colors.BOLD}Beschreibung:{Colors.RESET}\n{desc.strip()}\n") err_msg = item.get("error_message") if err_msg: print(f"{Colors.BOLD}Fehlermeldung:{Colors.RESET}\n{Colors.RED}{err_msg.strip()}{Colors.RESET}\n") trace = item.get("stack_trace") if trace: print(f"{Colors.BOLD}Stack Trace:{Colors.RESET}\n{Colors.DIM}{trace.strip()}{Colors.RESET}\n") comments = item.get("comments", []) if comments: print(f"{Colors.BOLD}Kommentare & Verlauf ({len(comments)}):{Colors.RESET}") for c in comments: action = f" [{c.get('action_taken')}]" if c.get('action_taken') else "" print(f" - {Colors.DIM}{c.get('created_at')}{Colors.RESET} {Colors.BOLD}{c.get('author')}{Colors.RESET}{action}:") comment_text = c.get('comment', '').strip() for line in comment_text.splitlines(): print(f" {line}") print() return 0 def cmd_comments(client: DcClient, args): res = client.request(f"/api/bugtracker/v1/manage?action=get&id={args.id}") if res.get("_error"): print(f"{Colors.RED}Fehler beim Abrufen von #{args.id}:{Colors.RESET} {res}") return 1 item = res.get("item") if not item: print(f"{Colors.RED}Item #{args.id} nicht gefunden.{Colors.RESET}") return 1 comments = item.get("comments", []) if not comments: print(f"{Colors.YELLOW}Keine Kommentare zu Item #{args.id} vorhanden.{Colors.RESET}") return 0 print(f"{Colors.BOLD}Kommentare & Diskussionsverlauf zu #{item.get('id')} ({item.get('title')}):{Colors.RESET}\n") for c in comments: action = f" [{Colors.CYAN}{c.get('action_taken')}{Colors.RESET}]" if c.get('action_taken') else "" print(f"- {Colors.DIM}{c.get('created_at')}{Colors.RESET} {Colors.BOLD}{c.get('author')}{Colors.RESET}{action}:") for line in (c.get('comment') or '').strip().splitlines(): print(f" {line}") print() return 0 def cmd_comment(client: DcClient, args): payload = { "comment": args.text, "action_taken": args.action or "commented" } res = client.request(f"/api/bugtracker/v1/manage?action=comment&id={args.id}", method="POST", data=payload) if res.get("_error"): print(f"{Colors.RED}Fehler beim Kommentieren:{Colors.RESET} {res}") return 1 print(f"{Colors.GREEN}Kommentar zu #{args.id} erfolgreich hinzugefügt.{Colors.RESET}") return 0 def cmd_resolve(client: DcClient, args): payload = { "resolved_in_build": args.build, "resolution_notes": args.notes or "" } res = client.request(f"/api/bugtracker/v1/manage?action=resolve&id={args.id}", method="POST", data=payload) if res.get("_error"): print(f"{Colors.RED}Fehler beim Abschließen:{Colors.RESET} {res}") return 1 print(f"{Colors.GREEN}Item #{args.id} erfolgreich auf 'resolved' gesetzt (Build: {args.build}).{Colors.RESET}") return 0 def cmd_release(client: DcClient, args): payload = {"note": args.note or ""} res = client.request(f"/api/bugtracker/v1/manage?action=release&id={args.id}", method="POST", data=payload) if res.get("_error"): print(f"{Colors.RED}Fehler beim Freigeben:{Colors.RESET} {res}") return 1 print(f"{Colors.GREEN}Item #{args.id} erfolgreich freigegeben.{Colors.RESET}") return 0 def cmd_next(client: DcClient, args): payload = { "project_slug": args.project, "limit": args.limit } res = client.request("/api/bugtracker/v1/manage?action=next", method="POST", data=payload) if res.get("_error"): print(f"{Colors.RED}Fehler beim Abrufen der nächsten Aufgabe:{Colors.RESET} {res}") return 1 items = res.get("items", []) if not items: print(f"{Colors.YELLOW}Keine anstehenden Aufgaben für Projekt '{args.project}' in der Warteschlange.{Colors.RESET}") return 0 print(f"{Colors.GREEN}Nächste Aufgabe(n) übernommen:{Colors.RESET}\n") for it in items: print(f"#{Colors.BOLD}{it.get('id')}{Colors.RESET} [{it.get('status')}] ({it.get('severity')}): {it.get('title')}") if it.get('description'): print(f" {it.get('description')[:150]}...") return 0 def cmd_report(client: DcClient, args): payload = { "project_slug": args.project, "type": args.type, "title": args.title, "description": args.desc, "severity": args.severity } if args.file: payload["file_path"] = args.file if args.line: payload["line_no"] = args.line if args.ref: payload["client_ref"] = args.ref if args.version: payload["build_version"] = args.version if args.env: payload["environment"] = args.env res = client.request("/api/bugtracker/v1/report", method="POST", data=payload) if res.get("_error"): print(f"{Colors.RED}Fehler beim Erstellen des Eintrags:{Colors.RESET} {res}") return 1 iid = res.get("id") or res.get("item_id") print(f"{Colors.GREEN}Item #{iid} erfolgreich angelegt!{Colors.RESET}") return 0 def cmd_roadmap(client: DcClient, args): query_params = { "action": args.action or "list", "project": args.project, "limit": args.limit } url = "/api/bugtracker/v1/roadmap?" + urllib.parse.urlencode(query_params) res = client.request(url) if res.get("_error"): print(f"{Colors.RED}Fehler beim Abrufen der Roadmap:{Colors.RESET} {res}") return 1 items = res.get("items", []) print(f"{Colors.BOLD}Roadmap für {args.project} ({len(items)} Items):{Colors.RESET}\n") for it in items: iid = it.get("id") epic = it.get("epic", "-") rank = it.get("roadmap_rank", "-") status = format_status(it.get("status", "")) title = it.get("title", "") print(f"#{Colors.BOLD}{iid:<4}{Colors.RESET} [{Colors.CYAN}{epic}{Colors.RESET} #{rank}] [{status}] {title}") return 0 def cmd_projects(client: DcClient, args): res = client.request("/api/bugtracker/v1/projects") if res.get("_error"): print(f"{Colors.RED}Fehler beim Abrufen der Projekte:{Colors.RESET} {res}") return 1 projects = res.get("projects", []) print(f"{Colors.BOLD}Verfügbare Projekte ({len(projects)}):{Colors.RESET}\n") for p in projects: slug = p.get("slug", "") name = p.get("name", "") open_bugs = p.get("open_bugs", 0) monitors = p.get("active_monitors", 0) print(f"- {Colors.BOLD}{slug:<16}{Colors.RESET} {name} (Offene Bugs: {open_bugs}, Monitore: {monitors})") return 0 def main(): parser = argparse.ArgumentParser( description="Deploymentcenter CLI (dc.py) — Direkter Zugriff auf Bugtracker, Roadmap & Services", formatter_class=argparse.RawDescriptionHelpFormatter ) parser.add_argument("--base-url", default=os.environ.get("DC_BASE_URL", DEFAULT_BASE_URL), help="Basis-URL (Vorgabe: https://dc.mhdf.de)") parser.add_argument("--token", default=None, help="Deploymentcenter API-Token (Vorgabe: $DC_TOKEN oder .mcp.json)") parser.add_argument("--no-color", action="store_true", help="Farbige Terminalausgabe deaktivieren") subparsers = parser.add_subparsers(dest="command", help="Befehle") # list p_list = subparsers.add_parser("list", help="Items auflisten") p_list.add_argument("--project", "-p", default=DEFAULT_PROJECT, help=f"Projekt-Slug (Vorgabe: {DEFAULT_PROJECT}, 'all' für alle)") p_list.add_argument("--status", "-s", default="open", help="Status-Filter (open, in_progress, triage_external, resolved, all)") p_list.add_argument("--type", "-t", default="all", help="Typ-Filter (bug, feature_request, idea, all)") p_list.add_argument("--search", "-q", default=None, help="Suchbegriff") p_list.add_argument("--limit", "-n", type=int, default=50, help="Maximale Trefferzahl (Vorgabe: 50)") p_list.add_argument("--sort", choices=["id_desc", "id_asc", "severity"], default="id_desc", help="Sortierung") # get p_get = subparsers.add_parser("get", help="Details zu einem Item anzeigen (inkl. Kommentare)") p_get.add_argument("id", type=int, help="ID des Bugtracker-Items") # comments p_comms = subparsers.add_parser("comments", help="Nur Kommentare & Diskussionsverlauf eines Items anzeigen") p_comms.add_argument("id", type=int, help="ID des Bugtracker-Items") # comment p_comm = subparsers.add_parser("comment", help="Kommentar zu einem Item hinzufügen") p_comm.add_argument("id", type=int, help="ID des Bugtracker-Items") p_comm.add_argument("text", help="Kommentartext") p_comm.add_argument("--action", "-a", choices=["investigated", "fix_proposed", "pr_opened", "needs_human", "blocked", "commented"], default="commented", help="Aktion") # resolve p_res = subparsers.add_parser("resolve", help="Item als gelöst markieren") p_res.add_argument("id", type=int, help="ID des Bugtracker-Items") p_res.add_argument("--build", "-b", required=True, help="Build-Version, in der der Fehler behoben wurde") p_res.add_argument("--notes", help="Zusammenfassung der Änderung / Notizen") # release p_rel = subparsers.add_parser("release", help="Reserviertes Item freigeben") p_rel.add_argument("id", type=int, help="ID des Bugtracker-Items") p_rel.add_argument("--note", help="Grund der Freigabe") # next p_next = subparsers.add_parser("next", help="Nächste Aufgabe aus der Warteschlange übernehmen") p_next.add_argument("--project", "-p", default=DEFAULT_PROJECT, help="Projekt-Slug") p_next.add_argument("--limit", "-n", type=int, default=1, help="Anzahl Items") # report p_rep = subparsers.add_parser("report", help="Neuen Fehler oder Wunsch melden") p_rep.add_argument("--title", required=True, help="Titel") p_rep.add_argument("--desc", required=True, help="Beschreibung") p_rep.add_argument("--project", "-p", default=DEFAULT_PROJECT, help="Projekt-Slug") p_rep.add_argument("--type", choices=["bug", "feature_request", "idea"], default="bug", help="Typ") p_rep.add_argument("--severity", choices=["critical", "high", "medium", "low", "wishlist", "idea"], default="medium", help="Schweregrad") p_rep.add_argument("--file", help="Betroffener Dateipfad") p_rep.add_argument("--line", type=int, help="Zeilennummer") p_rep.add_argument("--ref", help="Eindeutige client_ref zur Duplikaterkennung") p_rep.add_argument("--version", help="Build-Version") p_rep.add_argument("--env", default="development", help="Umgebung") # roadmap p_road = subparsers.add_parser("roadmap", help="Roadmap anzeigen") p_road.add_argument("--project", "-p", default=DEFAULT_PROJECT, help="Projekt-Slug") p_road.add_argument("--action", choices=["list", "next", "render", "stats"], default="list", help="Roadmap-Aktion") p_road.add_argument("--limit", "-n", type=int, default=30, help="Anzahl Einträge") # projects subparsers.add_parser("projects", help="Alle Projekte auflisten") args = parser.parse_args() if args.no_color: Colors.disable() token = args.token or find_token() if not token: print(f"{Colors.RED}FEHLER: Kein Deploymentcenter-Token gefunden.{Colors.RESET}") print("Bitte setze die Umgebungsvariable DC_TOKEN oder gib --token an.") sys.exit(1) client = DcClient(base_url=args.base_url, token=token) if args.command == "list": sys.exit(cmd_list(client, args)) elif args.command == "get": sys.exit(cmd_get(client, args)) elif args.command == "comments": sys.exit(cmd_comments(client, args)) elif args.command == "comment": sys.exit(cmd_comment(client, args)) elif args.command == "resolve": sys.exit(cmd_resolve(client, args)) elif args.command == "release": sys.exit(cmd_release(client, args)) elif args.command == "next": sys.exit(cmd_next(client, args)) elif args.command == "report": sys.exit(cmd_report(client, args)) elif args.command == "roadmap": sys.exit(cmd_roadmap(client, args)) elif args.command == "projects": sys.exit(cmd_projects(client, args)) else: parser.print_help() sys.exit(0) if __name__ == "__main__": main()