Bot & Automation

AI Assistants

/root/hermes-projects/AI Assistants

apps/whatsapp_control_api/app/services/settings_control.py text
from __future__ import annotations

import os
import shlex
from dataclasses import dataclass
from pathlib import Path

from .env_manager import (
    is_sensitive_env_key,
    is_valid_env_key,
    parse_env_file,
    redact_sensitive_assignments,
    reload_runtime_settings,
    resolve_project_env_path,
    resolve_runtime_env_path,
    update_env_file,
)

APP_RUNTIME_KEYS = {
    "CHAT_ENABLED",
    "CHAT_PROVIDER",
    "CHAT_BASE_URL",
    "CHAT_MODEL",
    "CHAT_API_KEY",
    "CHAT_TEMPERATURE",
    "CHAT_MAX_TOKENS",
    "WEB_SEARCH_ENABLED",
    "WEB_SEARCH_PROVIDER",
    "WEB_SEARCH_API_KEY",
    "WEB_SEARCH_BASE_URL",
    "WEB_SEARCH_MAX_RESULTS",
    "WEB_SEARCH_TIMEOUT_SECONDS",
    "DASHBOARD_ADMIN_TOKEN",
    "WHATSAPP_ALLOWED_NUMBERS",
}


@dataclass(slots=True)
class ParsedControlCommand:
    scope: str
    action: str
    target_name: str | None
    assignments: dict[str, str]


def looks_like_control_command(text: str) -> bool:
    try:
        return parse_control_command(text) is not None
    except ValueError:
        return False


def parse_control_command(text: str) -> ParsedControlCommand | None:
    cleaned = text.strip()
    if not cleaned:
        return None

    try:
        tokens = shlex.split(cleaned)
    except ValueError:
        return None

    if not tokens:
        return None

    first = tokens[0].lower()
    if first in {"settings", "setting", "config"}:
        if len(tokens) == 1 or tokens[1].lower() in {"show", "list", "view"}:
            return ParsedControlCommand(scope="app", action="show", target_name=None, assignments={})
        if tokens[1].lower() == "set":
            assignments = _parse_assignments(tokens[2:])
            if assignments:
                return ParsedControlCommand(scope="app", action="set", target_name=None, assignments=assignments)
        return None

    if first in {"set", "update", "change"}:
        assignments = _parse_assignments(tokens[1:])
        if assignments:
            return ParsedControlCommand(scope="app", action="set", target_name=None, assignments=assignments)
        return None

    if first == "show" and len(tokens) == 1:
        return ParsedControlCommand(scope="app", action="show", target_name=None, assignments={})

    if len(tokens) >= 4 and first == "project" and tokens[1].lower() == "env":
        project_name = tokens[2]
        verb = tokens[3].lower()
        if verb in {"show", "list", "view"}:
            return ParsedControlCommand(scope="project", action="show", target_name=project_name, assignments={})
        if verb == "set":
            assignments = _parse_assignments(tokens[4:])
            if assignments:
                return ParsedControlCommand(
                    scope="project",
                    action="set",
                    target_name=project_name,
                    assignments=assignments,
                )
        return None

    return None


def _parse_assignments(tokens: list[str]) -> dict[str, str]:
    assignments: dict[str, str] = {}
    for token in tokens:
        if "=" not in token:
            continue
        key, value = token.split("=", 1)
        key = key.strip()
        value = value.strip()
        if key:
            assignments[key] = value
    return assignments


def render_control_help() -> str:
    return "\n".join(
        [
            "Settings control aktif.",
            "",
            "App settings:",
            "settings show",
            "settings set CHAT_PROVIDER=openrouter CHAT_MODEL=... ",
            "settings set CHAT_PROVIDER=openai CHAT_BASE_URL=https://api.openai.com/v1 CHAT_MODEL=... ",
            "settings set WEB_SEARCH_ENABLED=true WEB_SEARCH_PROVIDER=bing",
            "settings set DASHBOARD_ADMIN_TOKEN=token-rahasia-baru",
            "",
            "Project env:",
            'project env "Nama Project" show',
            'project env "Nama Project" set OPENAI_API_KEY=... APP_BASE_URL=https://... ',
            "",
            "Catatan:",
            "- Key sensitif seperti *_KEY, *_TOKEN, *_SECRET akan minta approval.",
            "- Nomor Anda harus masuk allowlist WhatsApp.",
        ]
    )


def render_env_summary(env_path: Path) -> str:
    env_map = parse_env_file(env_path)
    if not env_map:
        return "(empty)"

    lines = []
    for key in sorted(env_map):
        value = env_map[key]
        if is_sensitive_env_key(key):
            value = "[redacted]"
        lines.append(f"{key}={value}")
    return "\n".join(lines[:25])


def summarize_assignments(assignments: dict[str, str]) -> str:
    parts = []
    for key in sorted(assignments):
        value = assignments[key]
        display = "[redacted]" if is_sensitive_env_key(key) else value
        parts.append(f"{key}={display}")
    return ", ".join(parts)


def split_sensitive_assignments(assignments: dict[str, str]) -> tuple[dict[str, str], dict[str, str]]:
    sensitive: dict[str, str] = {}
    non_sensitive: dict[str, str] = {}
    for key, value in assignments.items():
        if is_sensitive_env_key(key):
            sensitive[key] = value
        else:
            non_sensitive[key] = value
    return non_sensitive, sensitive


def validate_assignments(assignments: dict[str, str], *, scope: str) -> None:
    for key in assignments:
        if not is_valid_env_key(key):
            raise ValueError(f"Invalid env key: {key}")
        if scope == "app" and key not in APP_RUNTIME_KEYS:
            raise ValueError(f"App setting {key} is not allowed via WhatsApp.")


def apply_app_settings(assignments: dict[str, str]) -> None:
    validate_assignments(assignments, scope="app")
    env_path = resolve_runtime_env_path()
    update_env_file(env_path, assignments)
    for key, value in assignments.items():
        os.environ[key] = value
    reload_runtime_settings()


def apply_project_settings(project_name: str, assignments: dict[str, str]) -> Path:
    validate_assignments(assignments, scope="project")
    env_path = resolve_project_env_path(project_name)
    if not env_path.parent.exists():
        raise ValueError(f"Project folder not found: {project_name}")
    update_env_file(env_path, assignments)
    return env_path


def build_control_message(command: ParsedControlCommand) -> str:
    if command.scope == "app" and command.action == "show":
        env_path = resolve_runtime_env_path()
        return "Current app env:\n" + render_env_summary(env_path)

    if command.scope == "project" and command.action == "show":
        if not command.target_name:
            return "Project name is missing."
        env_path = resolve_project_env_path(command.target_name)
        if not env_path.exists():
            return f"Project env not found for {command.target_name}."
        return f"Project env for {command.target_name}:\n" + render_env_summary(env_path)

    if command.scope == "app" and command.action == "set":
        non_sensitive, sensitive = split_sensitive_assignments(command.assignments)
        if sensitive:
            return (
                "Setting change butuh approval karena ada key sensitif.\n"
                f"Request: {summarize_assignments(command.assignments)}"
            )
        apply_app_settings(non_sensitive)
        return "App settings updated:\n" + summarize_assignments(non_sensitive)

    if command.scope == "project" and command.action == "set":
        if not command.target_name:
            return "Project name is missing."
        non_sensitive, sensitive = split_sensitive_assignments(command.assignments)
        if sensitive:
            return (
                "Project env change butuh approval karena ada key sensitif.\n"
                f"Project: {command.target_name}\n"
                f"Request: {summarize_assignments(command.assignments)}"
            )
        path = apply_project_settings(command.target_name, non_sensitive)
        return f"Project env updated at {path}:\n" + summarize_assignments(non_sensitive)

    return render_control_help()


def redact_control_text(text: str) -> str:
    return redact_sensitive_assignments(text)


def load_env_preview_for_path(path: Path) -> str:
    return render_env_summary(path)