apps/whatsapp_control_api/app/services/gateway_aggregator.py
text
from __future__ import annotations
import json
import re
import subprocess
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Any
from ..core.config import Settings
from .store import Store
ACTIVE_TASK_STATUSES = {"queued", "running", "waiting_approval"}
PROCESS_ACTIVE_MARKERS = (" hermes ", "/hermes", "\\hermes", " codex ", "/codex", "\\codex")
PROCESS_EXCLUDE_MARKERS = (
"whatsapp_control_api",
"hermes_telegram_agent.bot.main",
"hermes_telegram_agent.api.main",
"uvicorn",
"pytest",
)
def collect_gateway_tasks(store: Store, settings: Settings, *, limit: int = 50) -> list[dict[str, Any]]:
items: list[dict[str, Any]] = []
items.extend(_collect_registry_tasks(settings))
items.extend(_collect_whatsapp_tasks(store, settings, limit=max(limit * 2, 100)))
items.extend(_collect_file_tasks(settings))
items.extend(_collect_runtime_process_tasks())
latest_by_id: dict[str, dict[str, Any]] = {}
for item in items:
existing = latest_by_id.get(item["id"])
if existing is None or _coerce_datetime(item.get("updated_at") or item.get("created_at")) > _coerce_datetime(
existing.get("updated_at") or existing.get("created_at")
):
latest_by_id[item["id"]] = item
ordered = sorted(
latest_by_id.values(),
key=lambda item: _coerce_datetime(item.get("updated_at") or item.get("created_at")),
reverse=True,
)
return ordered[:limit]
def get_gateway_task_by_id(task_id: str, store: Store, settings: Settings) -> dict[str, Any] | None:
return next((task for task in collect_gateway_tasks(store, settings, limit=300) if task["id"] == task_id), None)
def read_gateway_task_log_text(task: dict[str, Any]) -> str:
log_path = task.get("task_log_path")
if isinstance(log_path, str) and log_path:
path = Path(log_path)
if path.exists() and path.is_file():
return path.read_text(encoding="utf-8", errors="replace")[-20000:]
raw_log = task.get("raw_log")
if isinstance(raw_log, str):
return raw_log[-20000:]
logs = task.get("logs")
if isinstance(logs, list):
lines: list[str] = []
for entry in logs:
if not isinstance(entry, dict):
continue
created_at = str(entry.get("created_at") or "-")
level = str(entry.get("level") or "INFO")
message = str(entry.get("message") or "")
lines.append(f"[{created_at}] {level}: {message}")
return "\n".join(lines)[-20000:]
return ""
def _collect_registry_tasks(settings: Settings) -> list[dict[str, Any]]:
registry_root = settings.gateway_registry_tasks_dir
if not registry_root.exists():
return []
items: list[dict[str, Any]] = []
for task_file in registry_root.glob("*/task.json"):
task = _read_json(task_file)
if not task:
continue
task.setdefault("logs", [])
task.setdefault("task_log_path", "")
items.append(task)
return items
def _collect_whatsapp_tasks(store: Store, settings: Settings, *, limit: int) -> list[dict[str, Any]]:
items: list[dict[str, Any]] = []
for task in store.get_recent_tasks(limit=limit):
raw_id = str(task.get("id") or "").strip()
if not raw_id:
continue
channel = str(task.get("channel") or "whatsapp").strip().lower() or "whatsapp"
source_name = "WhatsApp"
agent_name = "AI Assistants Control"
if channel == "web":
source_name = "Web Dashboard"
agent_name = "Hermes Gateway Dashboard"
logs = store.get_task_logs(raw_id, limit=12)
items.append(
{
"id": _compose_task_id("wa", raw_id),
"raw_id": raw_id,
"channel": channel,
"source_name": source_name,
"agent_name": agent_name,
"project_name": task.get("project_name") or None,
"status": _normalize_status(task.get("status")),
"current_step": str(task.get("current_step") or "-"),
"progress_percent": int(task.get("progress_percent") or 0),
"prompt": str(task.get("prompt") or ""),
"result_summary": str(task.get("result_summary") or ""),
"error_summary": str(task.get("error_summary") or ""),
"created_at": task.get("created_at"),
"updated_at": task.get("updated_at"),
"started_at": task.get("started_at"),
"finished_at": task.get("finished_at"),
"logs": logs,
"task_log_path": str((settings.task_logs_dir / raw_id / "agent.log").resolve()),
}
)
return items
def _collect_file_tasks(settings: Settings) -> list[dict[str, Any]]:
items: list[dict[str, Any]] = []
projects_root = settings.hermes_projects_root
if not projects_root.exists():
return items
for project_root in projects_root.iterdir():
if not project_root.is_dir() or project_root.name.startswith("."):
continue
project_name = project_root.name
default_channel = _default_channel_for_project(project_name)
tasks_root = project_root / "data" / "tasks"
if not tasks_root.exists():
continue
for task_file in tasks_root.glob("*/task.json"):
task = _read_json(task_file)
if not task:
continue
raw_id = str(task.get("id") or task_file.parent.name).strip()
if not raw_id:
continue
task_source = str(task.get("source") or "").strip().lower()
channel, source_name = _classify_task_source(project_name, task_source, default_channel)
log_path = _resolve_log_path(task, task_file.parent / "agent.log")
logs = _tail_log_entries(log_path)
summary = _summarize_file_task(task, project_name, logs)
items.append(
{
"id": _compose_task_id(_slug(project_name), raw_id),
"raw_id": raw_id,
"channel": channel,
"source_name": source_name,
"agent_name": _agent_name_for_project(project_name),
"project_name": task.get("project_name") or project_name,
"status": _normalize_status(task.get("status")),
"current_step": str(task.get("current_step") or "-"),
"progress_percent": int(task.get("progress") or task.get("progress_percent") or 0),
"prompt": str(task.get("prompt") or ""),
"result_summary": summary if _normalize_status(task.get("status")) == "completed" else "",
"error_summary": str(task.get("error_message") or ""),
"created_at": task.get("created_at"),
"updated_at": task.get("updated_at"),
"started_at": task.get("started_at"),
"finished_at": task.get("finished_at"),
"logs": logs,
"task_log_path": str(log_path.resolve()) if log_path else "",
}
)
return items
def _collect_runtime_process_tasks() -> list[dict[str, Any]]:
try:
result = subprocess.run(
["ps", "-eo", "pid=,etimes=,args="],
capture_output=True,
text=True,
check=False,
)
except OSError:
return []
if result.returncode != 0:
return []
now = datetime.now(timezone.utc)
items: list[dict[str, Any]] = []
for line in result.stdout.splitlines():
parsed = _parse_process_line(line)
if not parsed:
continue
pid, elapsed_seconds, command = parsed
normalized = f" {command.lower()} "
if not any(marker in normalized for marker in PROCESS_ACTIVE_MARKERS):
continue
if any(marker in normalized for marker in PROCESS_EXCLUDE_MARKERS):
continue
created_at = (now - timedelta(seconds=elapsed_seconds)).isoformat()
source_name = "Hermes Runtime"
agent_name = "Hermes CLI Worker"
if "telegram" in normalized:
source_name = "Telegram"
agent_name = "Hermes Telegram Worker"
elif "whatsapp" in normalized:
source_name = "WhatsApp"
agent_name = "WhatsApp Runtime Worker"
items.append(
{
"id": _compose_task_id("proc", str(pid)),
"raw_id": str(pid),
"channel": "workspace",
"source_name": source_name,
"agent_name": agent_name,
"project_name": None,
"status": "running",
"current_step": "Hermes process sedang berjalan di VPS.",
"progress_percent": 15,
"prompt": command[:400],
"result_summary": "",
"error_summary": "",
"created_at": created_at,
"updated_at": now.isoformat(),
"started_at": created_at,
"finished_at": None,
"logs": [
{
"level": "INFO",
"message": f"PID {pid} aktif selama {elapsed_seconds} detik. Command: {command[:500]}",
"created_at": now.isoformat(),
}
],
"task_log_path": "",
"raw_log": command[:2000],
}
)
return items
def _classify_task_source(project_name: str, task_source: str, default_channel: str) -> tuple[str, str]:
normalized_project = project_name.lower()
if "telegram" in normalized_project:
return "telegram", "Telegram"
if "whatsapp" in normalized_project:
return "whatsapp", "WhatsApp"
if default_channel == "telegram":
if task_source == "voice":
return "telegram", "Telegram Voice"
return "telegram", "Telegram"
if default_channel == "bot":
return "bot", project_name
return "workspace", project_name
def _default_channel_for_project(project_name: str) -> str:
normalized = project_name.lower()
if normalized == "hermes-agent":
return "telegram"
if "telegram" in normalized:
return "telegram"
if "whatsapp" in normalized:
return "whatsapp"
if "bot" in normalized:
return "bot"
return "workspace"
def _agent_name_for_project(project_name: str) -> str:
if project_name == "Hermes-Agent":
return "Hermes Telegram Agent"
if "Telegram" in project_name:
return f"{project_name} Agent"
if "Bot" in project_name:
return project_name
return f"{project_name} Agent"
def _resolve_log_path(task: dict[str, Any], fallback: Path) -> Path | None:
raw_path = task.get("log_path")
if isinstance(raw_path, str) and raw_path.strip():
path = Path(raw_path)
if path.exists():
return path
return fallback if fallback.exists() else None
def _summarize_file_task(task: dict[str, Any], project_name: str, logs: list[dict[str, str]]) -> str:
error_message = str(task.get("error_message") or "").strip()
if error_message:
return error_message
current_step = str(task.get("current_step") or "").strip()
if current_step and current_step.lower() not in {"queued", "running", "completed", "failed", "cancelled"}:
return current_step
for entry in reversed(logs):
message = str(entry.get("message") or "").strip()
if message:
return message[:600]
status = _normalize_status(task.get("status"))
if status == "completed":
return f"Task dari {project_name} selesai."
if status == "failed":
return f"Task dari {project_name} gagal."
return f"Task dari {project_name} sedang diproses."
def _tail_log_entries(path: Path | None, limit: int = 10) -> list[dict[str, str]]:
if path is None or not path.exists() or not path.is_file():
return []
lines = path.read_text(encoding="utf-8", errors="replace").splitlines()[-limit:]
entries: list[dict[str, str]] = []
now_iso = datetime.now(timezone.utc).isoformat()
for line in lines:
stripped = line.strip()
if stripped:
entries.append({"level": "INFO", "message": stripped[:1000], "created_at": now_iso})
return entries
def _compose_task_id(source_key: str, raw_id: str) -> str:
return f"{source_key}__{raw_id}"
def _slug(value: str) -> str:
slug = re.sub(r"[^a-z0-9]+", "-", value.lower()).strip("-")
return slug or "task"
def _normalize_status(value: Any) -> str:
status = str(value or "queued").strip().lower()
if status == "cancelled":
return "failed"
if status in {"queued", "running", "waiting_approval", "completed", "failed"}:
return status
return "queued"
def _parse_process_line(line: str) -> tuple[int, int, str] | None:
match = re.match(r"^\s*(\d+)\s+(\d+)\s+(.+?)\s*$", line)
if not match:
return None
try:
pid = int(match.group(1))
elapsed = int(match.group(2))
except ValueError:
return None
command = match.group(3).strip()
if not command:
return None
return pid, elapsed, command
def _read_json(path: Path) -> dict[str, Any] | None:
try:
return json.loads(path.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError):
return None
def _coerce_datetime(value: Any) -> datetime:
if isinstance(value, datetime):
return value.astimezone(timezone.utc)
if isinstance(value, str) and value.strip():
raw = value.strip()
if raw.endswith("Z"):
raw = raw[:-1] + "+00:00"
try:
parsed = datetime.fromisoformat(raw)
except ValueError:
return datetime.fromtimestamp(0, tz=timezone.utc)
if parsed.tzinfo is None:
return parsed.replace(tzinfo=timezone.utc)
return parsed.astimezone(timezone.utc)
return datetime.fromtimestamp(0, tz=timezone.utc)