comfyui

1#!/usr/bin/env python32"""3fetch_logs.py — Retrieve workflow execution diagnostics from a ComfyUI server.4 5When a workflow errors, the server's /history (local) or /jobs (cloud) entry6contains the full Python traceback. This script makes it easy to fetch by7prompt_id, with sensible formatting.8 9Usage:10    python3 fetch_logs.py <prompt_id>11    python3 fetch_logs.py <prompt_id> --host https://cloud.comfy.org12    python3 fetch_logs.py --tail-queue            # show currently queued/running jobs13"""14 15from __future__ import annotations16 17import argparse18import sys19from pathlib import Path20 21sys.path.insert(0, str(Path(__file__).resolve().parent))22from _common import (  # noqa: E40223    DEFAULT_LOCAL_HOST, ENV_API_KEY, emit_json, http_get, is_cloud_host,24    resolve_api_key, resolve_url,25)26 27 28def fetch_history_entry(host: str, headers: dict, prompt_id: str, *, is_cloud: bool) -> dict:29    if is_cloud:30        # Try /jobs/{id} first31        url = resolve_url(host, f"/jobs/{prompt_id}", is_cloud=True)32        r = http_get(url, headers=headers, retries=2, timeout=30)33        if r.status == 200:34            try:35                return {"ok": True, "entry": r.json(), "source": "/api/jobs"}36            except Exception:37                pass38        # Fallback to history_v239        url = resolve_url(host, f"/history/{prompt_id}", is_cloud=True)40        r = http_get(url, headers=headers, retries=2, timeout=30)41        try:42            data = r.json()43        except Exception:44            data = None45        if r.status == 200 and data:46            return {"ok": True, "entry": data, "source": "/api/history_v2"}47        return {"ok": False, "http_status": r.status, "body": r.text()[:500]}48 49    url = resolve_url(host, f"/history/{prompt_id}", is_cloud=False)50    r = http_get(url, headers=headers, retries=2, timeout=30)51    if r.status != 200:52        return {"ok": False, "http_status": r.status, "body": r.text()[:500]}53    try:54        data = r.json()55    except Exception:56        return {"ok": False, "reason": "non-JSON response"}57    if not isinstance(data, dict) or prompt_id not in data:58        return {"ok": False, "reason": "prompt_id not found in history",59                "history_keys": list(data.keys())[:5] if isinstance(data, dict) else []}60    return {"ok": True, "entry": data[prompt_id], "source": "/history"}61 62 63def fetch_queue(host: str, headers: dict) -> dict:64    url = resolve_url(host, "/queue")65    r = http_get(url, headers=headers, retries=2, timeout=15)66    try:67        data = r.json()68    except Exception:69        data = {"raw": r.text()[:500]}70    return {"http_status": r.status, "data": data}71 72 73def extract_diagnostics(entry: dict) -> dict:74    """Pull out the parts a human cares about: status, errors, traceback, timing."""75    diag: dict = {}76    status = entry.get("status") or {}77    diag["status_str"] = status.get("status_str")78    diag["completed"] = status.get("completed")79 80    messages = status.get("messages") or []81    diag["execution_log"] = []82    for msg in messages:83        if isinstance(msg, list) and len(msg) >= 2:84            mtype, mdata = msg[0], msg[1]85            diag["execution_log"].append({"type": mtype, "data": mdata})86        else:87            diag["execution_log"].append(msg)88 89    # Look for execution_error inside messages90    errors = []91    for msg in messages:92        if isinstance(msg, list) and len(msg) >= 2 and msg[0] == "execution_error":93            errors.append(msg[1])94    if errors:95        diag["errors"] = errors96 97    # Cloud's /jobs response shape: top-level outputs / status / etc.98    if "outputs" in entry:99        out = entry["outputs"] or {}100        if isinstance(out, dict):101            diag["output_node_ids"] = list(out.keys())102            # Count file refs across all output buckets (images / video / etc.)103            total = 0104            for node_output in out.values():105                if not isinstance(node_output, dict):106                    continue107                for v in node_output.values():108                    if isinstance(v, list):109                        total += len(v)110            diag["output_count"] = total111        else:112            diag["output_node_ids"] = []113            diag["output_count"] = 0114    return diag115 116 117def main(argv: list[str] | None = None) -> int:118    p = argparse.ArgumentParser(description="Fetch workflow execution diagnostics")119    p.add_argument("prompt_id", nargs="?", help="prompt_id to look up")120    p.add_argument("--host", default=DEFAULT_LOCAL_HOST)121    p.add_argument("--api-key", help=f"or set ${ENV_API_KEY}")122    p.add_argument("--raw", action="store_true",123                   help="Print the full history entry instead of the digest")124    p.add_argument("--tail-queue", action="store_true",125                   help="Show currently running/pending jobs instead")126    args = p.parse_args(argv)127 128    api_key = resolve_api_key(args.api_key)129    headers = {"X-API-Key": api_key} if api_key else {}130    is_cloud = is_cloud_host(args.host)131 132    if args.tail_queue:133        emit_json(fetch_queue(args.host, headers))134        return 0135 136    if not args.prompt_id:137        print("Error: prompt_id is required (or use --tail-queue)", file=sys.stderr)138        return 1139 140    res = fetch_history_entry(args.host, headers, args.prompt_id, is_cloud=is_cloud)141    if not res.get("ok"):142        emit_json(res)143        return 1144 145    if args.raw:146        emit_json(res)147        return 0148 149    diag = extract_diagnostics(res["entry"])150    diag["source"] = res.get("source")151    diag["prompt_id"] = args.prompt_id152    emit_json(diag)153    return 0 if diag.get("status_str") not in {"error",} else 1154 155 156if __name__ == "__main__":157    sys.exit(main())158