430 lines
13 KiB
Python
430 lines
13 KiB
Python
"""Worker cmdlet: Display workers table in ResultTable format."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
import sys
|
|
from dataclasses import dataclass
|
|
from datetime import datetime, timezone
|
|
from typing import Any, Dict, Sequence, List
|
|
|
|
from cmdlet import register
|
|
from cmdlet._shared import Cmdlet, CmdletArg
|
|
from SYS import pipeline as ctx
|
|
from SYS.logger import log
|
|
from SYS.config import get_local_storage_path
|
|
|
|
DEFAULT_LIMIT = 100
|
|
WORKER_STATUS_FILTERS = {"running",
|
|
"completed",
|
|
"error",
|
|
"cancelled"}
|
|
HELP_FLAGS = {"-?",
|
|
"/?",
|
|
"--help",
|
|
"-h",
|
|
"help",
|
|
"--cmdlet"}
|
|
|
|
CMDLET = Cmdlet(
|
|
name=".worker",
|
|
summary="Display workers table in result table format.",
|
|
usage=".worker [status] [-limit N] [@N]",
|
|
arg=[
|
|
CmdletArg(
|
|
"status",
|
|
description="Filter by status: running, completed, error (default: all)",
|
|
requires_db=True,
|
|
),
|
|
CmdletArg(
|
|
"limit",
|
|
type="integer",
|
|
description="Limit results (default: 100)",
|
|
requires_db=True
|
|
),
|
|
CmdletArg(
|
|
"@N",
|
|
description="Select worker by index (1-based) and display full logs",
|
|
requires_db=True,
|
|
),
|
|
CmdletArg(
|
|
"-id",
|
|
description="Show full logs for a specific worker",
|
|
requires_db=True
|
|
),
|
|
CmdletArg(
|
|
"-clear",
|
|
type="flag",
|
|
description="Remove completed workers from the database",
|
|
requires_db=True,
|
|
),
|
|
],
|
|
detail=[
|
|
"- Shows all background worker tasks and their output",
|
|
"- Can filter by status: running, completed, error",
|
|
"- Search result stdout is captured from each worker",
|
|
"- Use @N to select a specific worker by index and display its full logs",
|
|
"Examples:",
|
|
".worker # Show all workers",
|
|
".worker running # Show running workers only",
|
|
".worker completed -limit 50 # Show 50 most recent completed workers",
|
|
".worker @3 # Show full logs for the 3rd worker",
|
|
".worker running @2 # Show full logs for the 2nd running worker",
|
|
],
|
|
)
|
|
|
|
|
|
def _has_help_flag(args_list: Sequence[str]) -> bool:
|
|
return any(str(arg).lower() in HELP_FLAGS for arg in args_list)
|
|
|
|
|
|
@dataclass
|
|
class WorkerCommandOptions:
|
|
status: str | None = None
|
|
limit: int = DEFAULT_LIMIT
|
|
worker_id: str | None = None
|
|
clear: bool = False
|
|
|
|
|
|
@register([".worker", "worker", "workers"])
|
|
def _run(result: Any, args: Sequence[str], config: Dict[str, Any]) -> int:
|
|
"""Display workers table or show detailed logs for a specific worker."""
|
|
args_list = [str(arg) for arg in (args or [])]
|
|
selection_indices = ctx.get_last_selection()
|
|
selection_requested = bool(selection_indices) and isinstance(result,
|
|
list
|
|
) and len(result) > 0
|
|
|
|
if _has_help_flag(args_list):
|
|
ctx.emit(CMDLET.__dict__)
|
|
return 0
|
|
|
|
options = _parse_worker_args(args_list)
|
|
|
|
library_root = get_local_storage_path(config or {})
|
|
if not library_root:
|
|
log("No library root configured", file=sys.stderr)
|
|
return 1
|
|
|
|
try:
|
|
from API.folder import API_folder_store
|
|
|
|
with API_folder_store(library_root) as db:
|
|
if options.clear:
|
|
count = db.clear_finished_workers()
|
|
log(f"Cleared {count} finished workers.")
|
|
return 0
|
|
|
|
if options.worker_id:
|
|
worker = db.get_worker(options.worker_id)
|
|
if worker:
|
|
events: List[Dict[str, Any]] = []
|
|
try:
|
|
wid = worker.get("worker_id")
|
|
if wid and hasattr(db, "get_worker_events"):
|
|
events = db.get_worker_events(wid)
|
|
except Exception:
|
|
pass
|
|
_emit_worker_detail(worker, events)
|
|
return 0
|
|
log(f"Worker not found: {options.worker_id}", file=sys.stderr)
|
|
return 1
|
|
|
|
if selection_requested:
|
|
return _render_worker_selection(db, result)
|
|
|
|
return _render_worker_list(db, options.status, options.limit)
|
|
except Exception as exc:
|
|
log(f"Workers query failed: {exc}", file=sys.stderr)
|
|
import traceback
|
|
|
|
traceback.print_exc(file=sys.stderr)
|
|
return 1
|
|
|
|
|
|
def _parse_worker_args(args_list: Sequence[str]) -> WorkerCommandOptions:
|
|
options = WorkerCommandOptions()
|
|
i = 0
|
|
while i < len(args_list):
|
|
arg = args_list[i]
|
|
low = arg.lower()
|
|
if low in {"-limit",
|
|
"--limit"} and i + 1 < len(args_list):
|
|
options.limit = _normalize_limit(args_list[i + 1])
|
|
i += 2
|
|
elif low in {"-id",
|
|
"--id"} and i + 1 < len(args_list):
|
|
options.worker_id = args_list[i + 1]
|
|
i += 2
|
|
elif low in {"-clear",
|
|
"--clear"}:
|
|
options.clear = True
|
|
i += 1
|
|
elif low in {"-status",
|
|
"--status"} and i + 1 < len(args_list):
|
|
options.status = args_list[i + 1].lower()
|
|
i += 2
|
|
elif low in WORKER_STATUS_FILTERS:
|
|
options.status = low
|
|
i += 1
|
|
elif not arg.startswith("-"):
|
|
options.status = low
|
|
i += 1
|
|
else:
|
|
i += 1
|
|
return options
|
|
|
|
|
|
def _normalize_limit(value: Any) -> int:
|
|
try:
|
|
return max(1, int(value))
|
|
except (TypeError, ValueError):
|
|
return DEFAULT_LIMIT
|
|
|
|
|
|
def _render_worker_list(db, status_filter: str | None, limit: int) -> int:
|
|
workers = db.get_all_workers(limit=limit)
|
|
if status_filter:
|
|
workers = [
|
|
w for w in workers if str(w.get("status", "")).lower() == status_filter
|
|
]
|
|
|
|
if not workers:
|
|
log("No workers found", file=sys.stderr)
|
|
return 0
|
|
|
|
for worker in workers:
|
|
started = worker.get("started_at", "")
|
|
ended = worker.get("completed_at", worker.get("last_updated", ""))
|
|
|
|
date_str = _extract_date(started)
|
|
start_time = _format_event_timestamp(started)
|
|
end_time = _format_event_timestamp(ended)
|
|
worker_id = str(worker.get("worker_id") or worker.get("id") or "unknown")
|
|
status = str(worker.get("status") or "unknown")
|
|
result_state = str(worker.get("result") or "")
|
|
status_label = status
|
|
if result_state and result_state.lower() not in {"", status.lower()}:
|
|
status_label = f"{status_label} ({result_state})"
|
|
pipe_display = _summarize_pipe(worker.get("pipe"))
|
|
error_message = _normalize_text(worker.get("error_message"))
|
|
description = _normalize_text(worker.get("description"))
|
|
|
|
columns = [
|
|
("ID", worker_id[:8]),
|
|
("Status", status_label),
|
|
("Pipe", pipe_display),
|
|
("Date", date_str),
|
|
("Start", start_time),
|
|
("End", end_time),
|
|
]
|
|
if error_message:
|
|
columns.append(("Error", error_message[:140]))
|
|
if description and description != error_message:
|
|
columns.append(("Details", description[:200]))
|
|
|
|
item = {
|
|
"columns": columns,
|
|
"__worker_metadata": worker,
|
|
"_selection_args": ["-id", worker.get("worker_id")],
|
|
}
|
|
ctx.emit(item)
|
|
log(
|
|
f"Worker {worker_id[:8]} status={status_label} pipe={pipe_display} "
|
|
f"error={error_message or 'none'}",
|
|
file=sys.stderr,
|
|
)
|
|
return 0
|
|
|
|
|
|
def _render_worker_selection(db, selected_items: Any) -> int:
|
|
if not isinstance(selected_items, list):
|
|
log("Selection payload missing", file=sys.stderr)
|
|
return 1
|
|
|
|
emitted = False
|
|
for item in selected_items:
|
|
worker = _resolve_worker_record(db, item)
|
|
if not worker:
|
|
continue
|
|
events: List[Dict[str, Any]] = []
|
|
try:
|
|
events = (
|
|
db.get_worker_events(worker.get("worker_id"))
|
|
if hasattr(db,
|
|
"get_worker_events") else []
|
|
)
|
|
except Exception:
|
|
events = []
|
|
_emit_worker_detail(worker, events)
|
|
emitted = True
|
|
if not emitted:
|
|
log("Selected rows no longer exist", file=sys.stderr)
|
|
return 1
|
|
return 0
|
|
|
|
|
|
def _resolve_worker_record(db, payload: Any) -> Dict[str, Any] | None:
|
|
if not isinstance(payload, dict):
|
|
return None
|
|
worker_data = payload.get("__worker_metadata")
|
|
worker_id = None
|
|
if isinstance(worker_data, dict):
|
|
worker_id = worker_data.get("worker_id")
|
|
else:
|
|
worker_id = payload.get("worker_id")
|
|
worker_data = None
|
|
if worker_id:
|
|
fresh = db.get_worker(worker_id)
|
|
if fresh:
|
|
return fresh
|
|
return worker_data if isinstance(worker_data, dict) else None
|
|
|
|
|
|
def _emit_worker_detail(worker: Dict[str, Any], events: List[Dict[str, Any]]) -> None:
|
|
stdout_content = worker.get("stdout", "") or ""
|
|
|
|
lines = stdout_content.splitlines()
|
|
|
|
for line in lines:
|
|
line = line.strip()
|
|
if not line:
|
|
continue
|
|
|
|
timestamp = ""
|
|
level = "INFO"
|
|
message = line
|
|
|
|
try:
|
|
parts = line.split(" - ", 3)
|
|
if len(parts) >= 4:
|
|
ts_str, _, lvl, msg = parts
|
|
timestamp = _format_event_timestamp(ts_str)
|
|
level = lvl
|
|
message = msg
|
|
elif len(parts) == 3:
|
|
ts_str, lvl, msg = parts
|
|
timestamp = _format_event_timestamp(ts_str)
|
|
level = lvl
|
|
message = msg
|
|
except Exception:
|
|
pass
|
|
|
|
item = {
|
|
"columns": [
|
|
("Time",
|
|
timestamp),
|
|
("Level",
|
|
level),
|
|
("Message",
|
|
message),
|
|
]
|
|
}
|
|
ctx.emit(item)
|
|
|
|
# Events are already always derived from stdout for now.
|
|
|
|
|
|
def _summarize_pipe(pipe_value: Any, limit: int = 200) -> str:
|
|
text = _normalize_text(pipe_value)
|
|
if not text:
|
|
return "(none)"
|
|
|
|
stage_count = text.count("|") + 1 if text else 0
|
|
display = text
|
|
if len(display) > limit:
|
|
trimmed = display[:max(limit - 3, 0)].rstrip()
|
|
if not trimmed:
|
|
trimmed = display[:limit]
|
|
display = f"{trimmed}..."
|
|
if stage_count > 1:
|
|
suffix = f" ({stage_count} stages)"
|
|
if not display.endswith("..."):
|
|
display = f"{display}{suffix}"
|
|
else:
|
|
display = f"{display}{suffix}"
|
|
return display
|
|
|
|
|
|
def _format_event_timestamp(raw_timestamp: Any) -> str:
|
|
dt = _parse_to_local(raw_timestamp)
|
|
if dt:
|
|
return dt.strftime("%H:%M:%S")
|
|
|
|
if not raw_timestamp:
|
|
return "--:--:--"
|
|
text = str(raw_timestamp)
|
|
if "T" in text:
|
|
time_part = text.split("T", 1)[1]
|
|
elif " " in text:
|
|
time_part = text.split(" ", 1)[1]
|
|
else:
|
|
time_part = text
|
|
return time_part[:8] if len(time_part) >= 8 else time_part
|
|
|
|
|
|
def _parse_to_local(timestamp_str: Any) -> datetime | None:
|
|
if not timestamp_str:
|
|
return None
|
|
text = str(timestamp_str).strip()
|
|
if not text:
|
|
return None
|
|
|
|
try:
|
|
if "T" in text:
|
|
return datetime.fromisoformat(text)
|
|
if " " in text:
|
|
dt = datetime.strptime(text, "%Y-%m-%d %H:%M:%S")
|
|
dt = dt.replace(tzinfo=timezone.utc)
|
|
return dt.astimezone()
|
|
except Exception:
|
|
pass
|
|
return None
|
|
|
|
|
|
def _extract_date(raw_timestamp: Any) -> str:
|
|
dt = _parse_to_local(raw_timestamp)
|
|
if dt:
|
|
return dt.strftime("%m-%d-%y")
|
|
|
|
if not raw_timestamp:
|
|
return ""
|
|
text = str(raw_timestamp)
|
|
date_part = ""
|
|
if "T" in text:
|
|
date_part = text.split("T", 1)[0]
|
|
elif " " in text:
|
|
date_part = text.split(" ", 1)[0]
|
|
else:
|
|
date_part = text
|
|
|
|
try:
|
|
parts = date_part.split("-")
|
|
if len(parts) == 3:
|
|
year, month, day = parts
|
|
return f"{month}-{day}-{year[2:]}"
|
|
except Exception:
|
|
pass
|
|
return date_part
|
|
|
|
|
|
def _normalize_text(value: Any) -> str:
|
|
text = str(value or "").strip()
|
|
if not text:
|
|
return ""
|
|
# collapse whitespace to keep table columns aligned
|
|
normalized = re.sub(r"\s+", " ", text)
|
|
return normalized
|
|
|
|
|
|
def _truncate_text(value: str, limit: int) -> str:
|
|
if limit <= 0:
|
|
return ""
|
|
if len(value) <= limit:
|
|
return value
|
|
cutoff = max(limit - 3, 0)
|
|
trimmed = value[:cutoff].rstrip()
|
|
if not trimmed:
|
|
return value[:limit]
|
|
return f"{trimmed}..."
|