apps/whatsapp_control_api/app/services/hermes_control.py
text
from __future__ import annotations
import re
import shlex
from dataclasses import dataclass
from pathlib import Path
from ..core.config import get_settings
SAFE_PROVIDER_RE = re.compile(r"^[a-z0-9_-]{2,40}$")
SAFE_MODEL_RE = re.compile(r"^[A-Za-z0-9._:/+-]{2,120}$")
PROVIDER_DEFAULT_BASE_URLS = {
"openrouter": "https://openrouter.ai/api/v1",
"openai": "https://api.openai.com/v1",
"openai_compatible": "https://api.openai.com/v1",
"gemini": "",
}
PROVIDER_ENV_KEYS = {
"openrouter": "OPENROUTER_API_KEY",
"openai": "OPENAI_API_KEY",
"openai_compatible": "OPENAI_API_KEY",
"gemini": "GOOGLE_API_KEY",
}
@dataclass(slots=True)
class HermesModelConfig:
model: str
provider: str
default: str
base_url: str
api_mode: str
@dataclass(slots=True)
class ParsedHermesCommand:
command: str
action: str
provider: str | None = None
model: str | None = None
base_url: str | None = None
def parse_hermes_command(text: str) -> ParsedHermesCommand | None:
try:
tokens = shlex.split(text.strip())
except ValueError:
return None
if not tokens:
return None
first = tokens[0].lower()
if first not in {"/model", "model"}:
return None
if len(tokens) == 1:
return ParsedHermesCommand(command="model", action="show")
action = tokens[1].lower()
if action in {"show", "list", "view", "current", "status"}:
return ParsedHermesCommand(command="model", action="show")
if action in {"help", "-h", "--help"}:
return ParsedHermesCommand(command="model", action="help")
if action == "set":
if len(tokens) < 4:
return ParsedHermesCommand(command="model", action="help")
provider = tokens[2].strip().lower()
model = tokens[3].strip()
base_url = _extract_base_url(tokens[4:])
return ParsedHermesCommand(command="model", action="set", provider=provider, model=model, base_url=base_url)
if len(tokens) >= 3:
provider = tokens[1].strip().lower()
model = tokens[2].strip()
base_url = _extract_base_url(tokens[3:])
return ParsedHermesCommand(command="model", action="set", provider=provider, model=model, base_url=base_url)
return ParsedHermesCommand(command="model", action="help")
def build_hermes_command_message(command: ParsedHermesCommand) -> str:
if command.command != "model":
return render_model_help()
if command.action == "show":
return render_model_status()
if command.action == "help":
return render_model_help()
if command.action == "set":
assert command.provider is not None
assert command.model is not None
return set_hermes_model(command.provider, command.model, command.base_url)
return render_model_help()
def render_model_status() -> str:
config = read_hermes_model_config()
key_status = _provider_key_status(config.provider)
lines = [
"Hermes model aktif",
f"Provider: {config.provider or '-'}",
f"Model: {config.model or '-'}",
f"Default: {config.default or '-'}",
f"Base URL: {config.base_url or '-'}",
f"API mode: {config.api_mode or '-'}",
f"Credential: {key_status}",
"",
"Ganti model:",
"/model set openrouter deepseek-v4-flash",
"/model set openrouter gpt-4o-mini",
"/model set gemini gemini-2.5-flash",
]
return "\n".join(lines)
def render_model_help() -> str:
return "\n".join(
[
"Command /model untuk Hermes Agent",
"",
"Lihat model aktif:",
"/model",
"/model show",
"",
"Ganti model Hermes:",
"/model set <provider> <model>",
"",
"Contoh:",
"/model set openrouter deepseek-v4-flash",
"/model set openrouter gpt-4o-mini",
"/model set gemini gemini-2.5-flash",
"",
"Catatan: provider harus punya API key di /root/.hermes/.env.",
]
)
def set_hermes_model(provider: str, model: str, base_url: str | None = None) -> str:
provider = provider.strip().lower()
model = model.strip()
if not SAFE_PROVIDER_RE.fullmatch(provider):
return "Provider tidak valid. Gunakan format huruf kecil seperti openrouter, openai, atau gemini."
if not SAFE_MODEL_RE.fullmatch(model):
return "Model tidak valid. Gunakan ID model tanpa spasi, contoh: deepseek-v4-flash."
selected_base_url = base_url if base_url is not None else PROVIDER_DEFAULT_BASE_URLS.get(provider, "")
if selected_base_url and not selected_base_url.startswith(("https://", "http://")):
return "Base URL tidak valid. Gunakan URL lengkap, contoh: https://openrouter.ai/api/v1"
config_path = _hermes_config_path()
if not config_path.exists():
return f"Config Hermes tidak ditemukan: {config_path}"
before = read_hermes_model_config()
updates = {
"model": model,
"provider": provider,
"default": model,
"api_mode": "chat_completions",
}
if selected_base_url:
updates["base_url"] = selected_base_url
elif provider == "gemini":
updates["base_url"] = ""
_write_model_config(config_path, updates)
after = read_hermes_model_config()
key_status = _provider_key_status(provider)
warning = ""
if key_status.endswith("belum ada"):
warning = f"\n\nPeringatan: credential {PROVIDER_ENV_KEYS.get(provider, 'API key')} belum ada. Hermes task bisa gagal sampai key diisi."
return "\n".join(
[
"Model Hermes berhasil diubah",
f"Sebelumnya: {before.provider or '-'} / {before.model or '-'}",
f"Sekarang: {after.provider or '-'} / {after.model or '-'}",
f"Base URL: {after.base_url or '-'}",
f"Credential: {key_status}",
"",
"Task Hermes berikutnya akan memakai model ini.",
]
) + warning
def read_hermes_model_config() -> HermesModelConfig:
config_path = _hermes_config_path()
if not config_path.exists():
return HermesModelConfig(model="", provider="", default="", base_url="", api_mode="")
values = {"model": "", "provider": "", "default": "", "base_url": "", "api_mode": ""}
in_model_block = False
for raw_line in config_path.read_text(encoding="utf-8", errors="replace").splitlines():
if raw_line.strip() == "model:":
in_model_block = True
continue
if in_model_block and raw_line and not raw_line.startswith((" ", "\t")):
break
if not in_model_block:
continue
match = re.match(r"^\s{2}([A-Za-z0-9_-]+):\s*(.*)$", raw_line)
if match and match.group(1) in values:
values[match.group(1)] = match.group(2).strip().strip("'\"")
return HermesModelConfig(**values)
def _write_model_config(config_path: Path, updates: dict[str, str]) -> None:
lines = config_path.read_text(encoding="utf-8", errors="replace").splitlines()
output: list[str] = []
in_model_block = False
seen_model_block = False
written_keys: set[str] = set()
for line in lines:
if line.strip() == "model:" and not line.startswith((" ", "\t")):
in_model_block = True
seen_model_block = True
output.append(line)
continue
if in_model_block and line and not line.startswith((" ", "\t")):
for key, value in updates.items():
if key not in written_keys:
output.append(f" {key}: {value}")
written_keys.add(key)
in_model_block = False
if in_model_block:
match = re.match(r"^\s{2}([A-Za-z0-9_-]+):", line)
if match and match.group(1) in updates:
key = match.group(1)
output.append(f" {key}: {updates[key]}")
written_keys.add(key)
continue
output.append(line)
if in_model_block:
for key, value in updates.items():
if key not in written_keys:
output.append(f" {key}: {value}")
written_keys.add(key)
if not seen_model_block:
output = ["model:"] + [f" {key}: {value}" for key, value in updates.items()] + [""] + output
config_path.write_text("\n".join(output).rstrip() + "\n", encoding="utf-8")
def _provider_key_status(provider: str) -> str:
key_name = PROVIDER_ENV_KEYS.get(provider)
if not key_name:
return "provider custom, credential tidak diverifikasi"
env_map = _read_env_file(_hermes_env_path())
return f"{key_name} sudah ada" if env_map.get(key_name) else f"{key_name} belum ada"
def _read_env_file(path: Path) -> dict[str, str]:
values: dict[str, str] = {}
if not path.exists():
return values
for raw_line in path.read_text(encoding="utf-8", errors="replace").splitlines():
stripped = raw_line.strip()
if not stripped or stripped.startswith("#") or "=" not in stripped:
continue
key, value = stripped.split("=", 1)
values[key.strip()] = value.strip().strip("'\"")
return values
def _extract_base_url(tokens: list[str]) -> str | None:
for index, token in enumerate(tokens):
if token.startswith("--base-url="):
return token.split("=", 1)[1].strip()
if token == "--base-url" and index + 1 < len(tokens):
return tokens[index + 1].strip()
return None
def _hermes_config_path() -> Path:
return get_settings().hermes_workspace_root / ".hermes" / "config.yaml"
def _hermes_env_path() -> Path:
return get_settings().hermes_workspace_root / ".hermes" / ".env"