sysops
457 lines
19 KiB
Python
457 lines
19 KiB
Python
"""Interactive agent terminal — parse and execute whitelisted commands."""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
from typing import Any, Optional
|
|
|
|
import httpx
|
|
|
|
from app.db import fetch_all, fetch_one
|
|
from app.services import agent_integration, agent_souls, herman, webbuilder_agent
|
|
from app.services.agent_events_log import log_agent_event
|
|
from app.services.agent_names import normalize_agent_key
|
|
|
|
TOOLS_URL = os.getenv("TOOLS_API_URL", "http://tools-api:8700").rstrip("/")
|
|
BROWSER_URL = os.getenv("BROWSER_AGENT_URL", "http://browser-agent:7790").rstrip("/")
|
|
|
|
Line = dict[str, Any]
|
|
|
|
GLOBAL_HELP = [
|
|
"help — dit overzicht",
|
|
"status — huidige status & taak",
|
|
"history — laatste events (in terminal)",
|
|
"handoff <agent> <type> [notitie]",
|
|
"say <bericht> — log een notitie",
|
|
"ask <vraag> — AI-antwoord in rol van deze agent",
|
|
]
|
|
|
|
AGENT_HELP: dict[str, list[str]] = {
|
|
"herman": [
|
|
"briefing — dagrapport genereren",
|
|
"<tekst> — vraag / opdracht aan Herman",
|
|
],
|
|
"browser": [
|
|
"monitor add <url> — site toevoegen aan monitor",
|
|
"monitor list — actieve sites",
|
|
"browse <url> — pagina openen",
|
|
"crawl — alle sites crawlen + parse-overzicht",
|
|
"parse — laatste parse-resultaten",
|
|
"parse <id> — parse van specifieke site",
|
|
"intel — hype & trend analyse (samenvatting)",
|
|
"intel <zoekterm> — filter analyse",
|
|
],
|
|
"research": ["run — research pipeline starten", "briefs — recente briefs"],
|
|
"retail": ["rss — retail RSS verversen", "scores — halal opportunity scores"],
|
|
"sysops": [
|
|
"backup — config backup (approval)",
|
|
"scan — maintenance scan",
|
|
"topology — infra overzicht",
|
|
"status — sysops status",
|
|
],
|
|
"marketing": ["reco — aanbevelingen genereren"],
|
|
"packaging": ["design <brief> — packaging opdracht (via Herman)"],
|
|
"webbuilder": [
|
|
"build <opdracht> — website bouwen (agy op Hermes)",
|
|
"projects — bestaande website-projecten",
|
|
"status [project] — build-status",
|
|
"preview [project] — preview-URL",
|
|
],
|
|
"email": ["sync — inbox sync trigger"],
|
|
"finance": ["margins — marge-overzicht (placeholder)"],
|
|
}
|
|
|
|
|
|
def _line(msg: str, *, detail: str = "", line_type: str = "output") -> Line:
|
|
return {"type": line_type, "message": msg, "detail": detail}
|
|
|
|
|
|
def _help_lines(key: str) -> list[Line]:
|
|
lines = [_line("Beschikbare commando's:", line_type="output")]
|
|
for h in GLOBAL_HELP:
|
|
lines.append(_line(" " + h, line_type="output"))
|
|
extra = AGENT_HELP.get(key, [])
|
|
if extra:
|
|
lines.append(_line(f"— {key} —", line_type="output"))
|
|
for h in extra:
|
|
lines.append(_line(" " + h, line_type="output"))
|
|
return lines
|
|
|
|
|
|
def _status_lines(key: str, soul: dict[str, Any]) -> list[Line]:
|
|
recent = soul.get("recent_events") or []
|
|
last = recent[0] if recent else {}
|
|
lines = [
|
|
_line(f"{soul.get('display_name') or key} · {soul.get('role_title') or 'agent'}"),
|
|
_line(f"Status: {last.get('status') or soul.get('current_status') or 'standby'}"),
|
|
_line(f"Taak: {last.get('title') or soul.get('current_task') or '—'}"),
|
|
_line(f"Events totaal: {soul.get('event_count') or 0}"),
|
|
]
|
|
resp = (soul.get("responsibilities") or "").strip()
|
|
if resp:
|
|
lines.append(_line(resp[:220], detail=resp if len(resp) > 220 else ""))
|
|
return lines
|
|
|
|
|
|
async def _tools_post(path: str, payload: dict | None = None, timeout: float = 180.0) -> dict[str, Any]:
|
|
async with httpx.AsyncClient(timeout=timeout) as client:
|
|
resp = await client.post(f"{TOOLS_URL}{path}", json=payload or {})
|
|
try:
|
|
data = resp.json()
|
|
except Exception:
|
|
data = {"ok": False, "detail": resp.text[:500]}
|
|
if resp.status_code >= 400:
|
|
data.setdefault("ok", False)
|
|
data.setdefault("detail", resp.text[:500])
|
|
return data
|
|
|
|
|
|
async def _tools_get(path: str, timeout: float = 60.0) -> dict[str, Any]:
|
|
async with httpx.AsyncClient(timeout=timeout) as client:
|
|
resp = await client.get(f"{TOOLS_URL}{path}")
|
|
try:
|
|
return resp.json()
|
|
except Exception:
|
|
return {"ok": False, "detail": resp.text[:500]}
|
|
|
|
|
|
async def _browser_post(path: str, payload: dict, timeout: float = 120.0) -> dict[str, Any]:
|
|
async with httpx.AsyncClient(timeout=timeout) as client:
|
|
resp = await client.post(f"{BROWSER_URL}{path}", json=payload)
|
|
try:
|
|
return resp.json()
|
|
except Exception:
|
|
return {"ok": False, "detail": resp.text[:500]}
|
|
|
|
|
|
def _handoff(src: str, dst_raw: str, htype: str, note: str) -> list[Line]:
|
|
dst = normalize_agent_key(dst_raw)
|
|
if not dst:
|
|
return [_line("Onbekende agent: " + dst_raw, line_type="error")]
|
|
payload = {"note": note} if note else {}
|
|
try:
|
|
agent_integration.create_handoff(src, dst, handoff_type=htype, payload=payload)
|
|
return [_line(f"Handoff → {dst} ({htype})" + (f": {note}" if note else ""))]
|
|
except Exception as exc:
|
|
return [_line("Handoff mislukt: " + str(exc), line_type="error")]
|
|
|
|
|
|
async def _agent_ask(key: str, soul: dict[str, Any], question: str) -> list[Line]:
|
|
name = soul.get("display_name") or key
|
|
role = soul.get("role_title") or ""
|
|
prompt = (
|
|
f"Je bent {name} ({key}), {role}. "
|
|
f"Beantwoord kort en actiegericht in het Nederlands.\n\nOpdracht: {question}"
|
|
)
|
|
result = await herman.chat(prompt, channel=f"terminal:{key}")
|
|
reply = (result.get("reply") or "").strip()
|
|
delegated = result.get("delegated_agents") or []
|
|
lines = [_line(reply or "(geen antwoord)", detail=reply)]
|
|
if delegated:
|
|
lines.append(_line("Gedelegeerd: " + ", ".join(delegated), line_type="handoff_out"))
|
|
log_agent_event(
|
|
key,
|
|
"terminal_out",
|
|
reply[:255] if reply else "Antwoord",
|
|
reply[:4000],
|
|
channel="terminal",
|
|
metadata={"delegated": delegated, "question": question[:500]},
|
|
)
|
|
return lines
|
|
|
|
|
|
def _format_crawl_results(result: dict[str, Any]) -> list[Line]:
|
|
if result.get("message") and not result.get("results"):
|
|
return [_line(result["message"], line_type="output")]
|
|
lines: list[Line] = [
|
|
_line(
|
|
f"Crawl klaar — {result.get('sites', 0)} site(s), "
|
|
f"{result.get('changed', 0)} wijziging(en), {result.get('errors', 0)} fout(en)"
|
|
)
|
|
]
|
|
for r in result.get("results") or []:
|
|
if r.get("status") == "ERROR":
|
|
lines.append(_line(f"✗ {r.get('name') or r.get('url')}: {r.get('error')}", line_type="error"))
|
|
continue
|
|
flag = "△" if r.get("changed") else "✓"
|
|
lines.append(
|
|
_line(
|
|
f"{flag} {r.get('title') or r.get('name')} — {r.get('word_count', 0)} woorden, "
|
|
f"{r.get('links_count', 0)} links",
|
|
detail=(r.get("excerpt") or "")[:200],
|
|
)
|
|
)
|
|
headings = r.get("headings") or []
|
|
if headings:
|
|
lines.append(_line(" H: " + " · ".join(headings[:4])[:180], line_type="output"))
|
|
for link in (r.get("links_sample") or [])[:3]:
|
|
lbl = link.get("label") or link.get("href") or ""
|
|
lines.append(_line(" → " + lbl[:70], line_type="output"))
|
|
return lines
|
|
|
|
|
|
def _format_parse_items(items: list[dict[str, Any]]) -> list[Line]:
|
|
if not items:
|
|
return [_line("Geen parse-resultaten — voer eerst crawl uit.")]
|
|
lines: list[Line] = [_line(f"Parse-overzicht ({len(items)} pagina's):")]
|
|
for p in items:
|
|
lines.append(
|
|
_line(
|
|
f"#{p.get('site_id') or '—'} {p.get('title') or p.get('url')} — "
|
|
f"{p.get('word_count', 0)} woorden",
|
|
detail=(p.get("excerpt") or "")[:200],
|
|
)
|
|
)
|
|
if p.get("meta_description"):
|
|
lines.append(_line(" desc: " + str(p["meta_description"])[:120], line_type="output"))
|
|
for h in (p.get("headings") or [])[:3]:
|
|
lines.append(_line(" · " + h[:90], line_type="output"))
|
|
return lines
|
|
|
|
|
|
async def _run_agent_command(key: str, soul: dict[str, Any], cmd: str, args: list[str], raw: str) -> list[Line]:
|
|
if key == "herman":
|
|
if cmd == "briefing":
|
|
try:
|
|
content = await herman.generate_briefing()
|
|
preview = (content or "")[:400]
|
|
log_agent_event("herman", "terminal_out", "Briefing gegenereerd", content[:4000], channel="terminal")
|
|
return [_line("CEO briefing gegenereerd.", detail=preview)]
|
|
except Exception as exc:
|
|
return [_line("Briefing mislukt: " + str(exc), line_type="error")]
|
|
result = await herman.chat(raw, channel="terminal")
|
|
reply = (result.get("reply") or "").strip()
|
|
delegated = result.get("delegated_agents") or []
|
|
lines = [_line(reply or "(geen antwoord)", detail=reply)]
|
|
if delegated:
|
|
lines.append(_line("→ " + ", ".join(delegated), line_type="handoff_out"))
|
|
log_agent_event("herman", "terminal_out", reply[:255] if reply else "Antwoord", reply[:4000], channel="terminal")
|
|
return lines
|
|
|
|
if key == "browser":
|
|
if cmd == "monitor" and args and args[0].lower() == "add" and len(args) >= 2:
|
|
url = args[1]
|
|
from app.services.monitor import add_site
|
|
|
|
site = add_site(url, name=url[:80])
|
|
log_agent_event(
|
|
"browser",
|
|
"monitor_site_added",
|
|
f"Terminal: site toegevoegd {url[:120]}",
|
|
"",
|
|
channel="terminal",
|
|
metadata={"url": url, "site_id": site.get("id")},
|
|
)
|
|
return [_line(f"Monitor: {url} toegevoegd (id {site.get('id')})")]
|
|
if cmd == "monitor" and args and args[0].lower() == "list":
|
|
rows = fetch_all(
|
|
"SELECT id, url, is_active FROM monitored_sites ORDER BY id DESC LIMIT 15"
|
|
)
|
|
if not rows:
|
|
return [_line("Geen monitor-sites.")]
|
|
return [_line(f"#{r['id']} {'✓' if r.get('is_active') else '○'} {r['url']}") for r in rows]
|
|
if cmd == "browse" and args:
|
|
url = args[0]
|
|
data = await _browser_post("/browse", {"url": url, "purpose": "terminal"})
|
|
log_agent_event("browser", "browse", f"Terminal browse: {url[:120]}", json.dumps(data)[:500], channel="terminal")
|
|
title = data.get("title") or data.get("url") or url
|
|
return [_line(f"Browse OK: {title}")]
|
|
if cmd == "crawl":
|
|
from app.services.monitor import trigger_crawl
|
|
|
|
site_id = int(args[0]) if args and str(args[0]).isdigit() else None
|
|
result = trigger_crawl(site_id)
|
|
log_agent_event(
|
|
"browser",
|
|
"monitor_crawl",
|
|
f"Terminal crawl ({result.get('sites', 0)} sites)",
|
|
json.dumps([{k: r.get(k) for k in ("url", "title", "word_count", "status")} for r in result.get("results", [])])[:800],
|
|
channel="terminal",
|
|
)
|
|
return _format_crawl_results(result)
|
|
if cmd == "parse":
|
|
from app.services.monitor import list_parse_results
|
|
|
|
sid = int(args[0]) if args and str(args[0]).isdigit() else None
|
|
return _format_parse_items(list_parse_results(site_id=sid, limit=10))
|
|
if cmd == "intel":
|
|
from app.services.monitor import build_parse_intelligence
|
|
|
|
q = " ".join(args) if args else None
|
|
data = build_parse_intelligence(query=q, limit=20)
|
|
lines = [
|
|
_line(
|
|
f"Analyse: {data['summary']['pages']} pagina's · "
|
|
f"{data['summary']['total_words']} woorden · "
|
|
f"{data['summary']['themes_detected']} trend-termen"
|
|
),
|
|
_line("Open Browser → tab Parse Analyse voor volledig overzicht", line_type="output"),
|
|
]
|
|
for t in (data.get("hype_terms") or [])[:12]:
|
|
flag = "★ " if t.get("cross_site") else ""
|
|
lines.append(
|
|
_line(f"{flag}{t['term']} ({t['score']}) — {', '.join(t.get('sites') or [])[:3]}", line_type="output")
|
|
)
|
|
return lines
|
|
|
|
if key == "research":
|
|
if cmd == "run":
|
|
data = await _tools_post("/research/run")
|
|
ok = data.get("ok", True)
|
|
log_agent_event("research", "terminal_action", "Research run (terminal)", json.dumps(data)[:500], channel="terminal")
|
|
return [_line("Research pipeline: " + ("OK" if ok else "fout"), detail=json.dumps(data)[:300])]
|
|
if cmd == "briefs":
|
|
data = await _tools_get("/research/briefs")
|
|
items = data if isinstance(data, list) else data.get("items") or []
|
|
if not items:
|
|
return [_line("Geen briefs.")]
|
|
return [_line(str(b.get("title") or b.get("id") or b))[:120] for b in items[:8]]
|
|
|
|
if key == "retail":
|
|
if cmd == "rss":
|
|
data = await _tools_post("/retail/rss/refresh")
|
|
log_agent_event("retail", "terminal_action", "RSS refresh (terminal)", json.dumps(data)[:300], channel="terminal")
|
|
return [_line("RSS verversd.", detail=json.dumps(data)[:200])]
|
|
if cmd == "scores":
|
|
data = await _tools_post("/retail/compute-opportunities")
|
|
return [_line("Opportunity scores bijgewerkt.", detail=json.dumps(data)[:200])]
|
|
|
|
if key == "sysops":
|
|
if cmd == "backup":
|
|
data = await _tools_post("/ops/backup/run")
|
|
log_agent_event("sysops", "terminal_action", "Backup (terminal)", json.dumps(data)[:400], channel="terminal")
|
|
return [_line("Backup uitgevoerd.", detail=json.dumps(data)[:300])]
|
|
if cmd == "scan":
|
|
data = await _tools_post("/ops/maintenance/scan")
|
|
return [_line("Maintenance scan klaar.", detail=json.dumps(data)[:300])]
|
|
if cmd == "topology":
|
|
data = await _tools_get("/ops/topology")
|
|
return [_line("Topology geladen.", detail=json.dumps(data)[:400])]
|
|
if cmd == "status":
|
|
data = await _tools_get("/ops/status")
|
|
return [_line(json.dumps(data)[:350], detail=json.dumps(data)[:800])]
|
|
|
|
if key == "marketing" and cmd == "reco":
|
|
data = await _tools_post("/recommendations/generate")
|
|
return [_line("Aanbevelingen gegenereerd.", detail=json.dumps(data)[:300])]
|
|
|
|
if key == "packaging" and cmd == "design":
|
|
brief = " ".join(args) or raw
|
|
result = await herman.chat(f"Packaging design: {brief}", channel="terminal:packaging")
|
|
reply = (result.get("reply") or "").strip()
|
|
return [_line(reply, detail=reply)]
|
|
|
|
if key == "webbuilder":
|
|
if cmd == "build":
|
|
brief = rest or raw
|
|
if not brief:
|
|
return [_line("Gebruik: build <opdracht>", line_type="error")]
|
|
try:
|
|
outcome = await webbuilder_agent.generate_from_message(brief, channel="terminal", wait=False)
|
|
proj = outcome.get("project") or "?"
|
|
preview = outcome.get("preview_url") or "—"
|
|
return [
|
|
_line(f"Build gestart: {proj}"),
|
|
_line(f"Preview: {preview}", line_type="output"),
|
|
_line("Volg voortgang live in dit terminal-venster.", line_type="output"),
|
|
]
|
|
except Exception as exc:
|
|
return [_line("Build mislukt: " + str(exc), line_type="error")]
|
|
if cmd == "projects":
|
|
data = await webbuilder_agent.list_projects()
|
|
items = data.get("items") or []
|
|
if not items:
|
|
return [_line("Geen projecten op Hermes/NAS.")]
|
|
return [
|
|
_line(
|
|
f"{p.get('slug')} — {p.get('files', 0)} bestand(en)"
|
|
+ (" · index.html ✓" if p.get("has_index") else "")
|
|
)
|
|
for p in items
|
|
]
|
|
if cmd == "status":
|
|
slug = args[0] if args else ""
|
|
if not slug:
|
|
return [_line("Gebruik: status <project>", line_type="error")]
|
|
st = await webbuilder_agent.get_build_status(slug)
|
|
lines = [_line(f"{st.get('project')} · {st.get('status')} — {st.get('message') or ''}")]
|
|
tail = (st.get("log_tail") or "")[:300]
|
|
if tail:
|
|
lines.append(_line(tail, line_type="output"))
|
|
return lines
|
|
if cmd == "preview":
|
|
slug = args[0] if args else ""
|
|
if slug:
|
|
st = await webbuilder_agent.get_build_status(slug)
|
|
url = st.get("preview_url") or webbuilder_agent.HERMES_PREVIEW_BASE
|
|
else:
|
|
url = webbuilder_agent.HERMES_PREVIEW_BASE
|
|
return [_line(f"Preview: {url}")]
|
|
|
|
return await _agent_ask(key, soul, raw)
|
|
|
|
|
|
async def execute_terminal_command(
|
|
agent_key: str,
|
|
command: str,
|
|
*,
|
|
issued_by: str = "ceo",
|
|
) -> dict[str, Any]:
|
|
key = normalize_agent_key(agent_key)
|
|
soul = agent_souls.get_soul(key)
|
|
if not soul:
|
|
raise ValueError("Agent niet gevonden")
|
|
|
|
raw = (command or "").strip()
|
|
if not raw:
|
|
raise ValueError("Leeg commando")
|
|
|
|
log_agent_event(
|
|
key,
|
|
"terminal_in",
|
|
raw[:255],
|
|
"",
|
|
channel="terminal",
|
|
metadata={"issued_by": issued_by, "command": raw[:500]},
|
|
)
|
|
|
|
parts = raw.split()
|
|
cmd = parts[0].lower()
|
|
args = parts[1:]
|
|
rest = " ".join(args)
|
|
|
|
if cmd in ("help", "?", "h"):
|
|
lines = _help_lines(key)
|
|
elif cmd == "status":
|
|
lines = _status_lines(key, soul)
|
|
elif cmd == "handoff":
|
|
if len(args) < 2:
|
|
lines = [_line("Gebruik: handoff <agent> <type> [notitie]", line_type="error")]
|
|
else:
|
|
note = " ".join(args[2:]) if len(args) > 2 else ""
|
|
lines = _handoff(key, args[0], args[1], note)
|
|
elif cmd == "say":
|
|
if not rest:
|
|
lines = [_line("Gebruik: say <bericht>", line_type="error")]
|
|
else:
|
|
log_agent_event(key, "terminal_note", rest[:255], rest, channel="terminal")
|
|
lines = [_line("Genoteerd.")]
|
|
elif cmd == "ask":
|
|
if not rest:
|
|
lines = [_line("Gebruik: ask <vraag>", line_type="error")]
|
|
else:
|
|
lines = await _agent_ask(key, soul, rest)
|
|
elif cmd == "history":
|
|
events = agent_souls.list_agent_events(key, limit=8)
|
|
if not events:
|
|
lines = [_line("Geen history.")]
|
|
else:
|
|
lines = []
|
|
for ev in reversed(events):
|
|
t = (ev.get("created_at") or "")[11:19] or "--:--"
|
|
lines.append(_line(f"{t} {ev.get('title') or ev.get('event_type')}"))
|
|
else:
|
|
lines = await _run_agent_command(key, soul, cmd, args, raw)
|
|
|
|
return {"ok": True, "agent_key": key, "command": raw, "lines": lines}
|