research-agent-plugin/tools.py

"""
Обработчики инструментов research-agent плагина.

Каждый обработчик:
  1. Собирает аргументы командной строки из параметров вызова
  2. Запускает соответствующий Python-скрипт через subprocess
  3. Захватывает stdout/stderr
  4. Возвращает результат как JSON

Скрипты лежат в scripts/ рядом с плагином — плагин самодостаточен.
"""

import json
import os
import subprocess
import sys
from pathlib import Path

# ─── Пути к скриптам ──────────────────────────────────────────────────────────

_PLUGIN_DIR = Path(__file__).parent
_SCRIPTS_DIR = _PLUGIN_DIR / "scripts"

DEEP_RESEARCH_SCRIPT = _SCRIPTS_DIR / "deep_research.py"
PERPLEXITY_SEARCH_SCRIPT = _SCRIPTS_DIR / "perplexity_search.py"
YOUTUBE_ANALYZER_SCRIPT = _SCRIPTS_DIR / "youtube_analyzer.py"

# ─── Общие настройки ──────────────────────────────────────────────────────────

# Таймауты (секунд) — скрипты долгие из-за LLM-синтеза
TIMEOUT_DEEP_RESEARCH = 420   # 7 минут
TIMEOUT_PERPLEXITY = 120      # 2 минуты
TIMEOUT_YOUTUBE = 180         # 3 минуты


def _run_script(script_path: Path, args: list[str], timeout: int) -> dict:
    """
    Запустить Python-скрипт и вернуть результат.

    Args:
        script_path: абсолютный путь к .py файлу
        args: аргументы командной строки (без имени скрипта)
        timeout: таймаут в секундах

    Returns:
        dict с ключами: success, report (текст отчёта), error (если ошибка),
        exit_code, stderr
    """
    if not script_path.exists():
        return {
            "success": False,
            "error": f"Скрипт не найден: {script_path}\n"
                     f"Установите плагин полностью: hermes plugins install research-agent",
            "report": "",
            "exit_code": -1,
            "stderr": "",
        }

    cmd = [sys.executable, str(script_path)] + args

    try:
        result = subprocess.run(
            cmd,
            capture_output=True,
            text=True,
            timeout=timeout,
            cwd=str(script_path.parent),
            env={**os.environ, "PYTHONUNBUFFERED": "1"},
        )

        # Скрипт пишет отчёт в stdout
        report = result.stdout.strip()
        stderr = result.stderr.strip()

        if result.returncode != 0:
            # Частичный отчёт может быть на stdout даже при ошибке
            if not report and stderr:
                report = f"[Ошибка выполнения]\n{stderr[-2000:]}"
            return {
                "success": False,
                "error": stderr[-2000:] if stderr else f"Exit code {result.returncode}",
                "report": report,
                "exit_code": result.returncode,
                "stderr": stderr[-3000:],
            }

        return {
            "success": True,
            "report": report,
            "error": "",
            "exit_code": 0,
            "stderr": stderr[-1000:] if stderr else "",
        }

    except subprocess.TimeoutExpired:
        return {
            "success": False,
            "error": f"Таймаут ({timeout}с). Исследование заняло слишком много времени.",
            "report": "",
            "exit_code": -1,
            "stderr": f"Timeout after {timeout}s",
        }

    except Exception as e:
        return {
            "success": False,
            "error": f"Ошибка запуска: {e}",
            "report": "",
            "exit_code": -1,
            "stderr": str(e),
        }


# ─── Обработчики ──────────────────────────────────────────────────────────────


def handle_deep_research(params: dict, **kwargs) -> str:
    """
    Запустить deep_research.py с параметрами из tool call.

    Ожидаемые параметры:
        query (str): тема исследования
        lang (str, опционально): 'ru' или 'en' (default 'ru')
    """
    query = params.get("query", "")
    if not query:
        return json.dumps({"success": False, "error": "Параметр 'query' обязателен"})

    args = [query]

    lang = params.get("lang", "ru")
    if lang:
        args.extend(["--lang", lang])

    # Опциональные параметры из env или переопределения
    model = params.get("model") or os.environ.get("LLM_MODEL")
    temperature = params.get("temperature") or os.environ.get("LLM_TEMPERATURE")
    if model:
        args.extend(["--model", model])
    if temperature:
        args.extend(["--temperature", str(temperature)])

    result = _run_script(DEEP_RESEARCH_SCRIPT, args, TIMEOUT_DEEP_RESEARCH)
    return json.dumps(result, ensure_ascii=False)


def handle_perplexity_search(params: dict, **kwargs) -> str:
    """
    Запустить perplexity_search.py с параметрами из tool call.

    Ожидаемые параметры:
        query (str): поисковый запрос
        category (str, опционально): 'general', 'news', 'it', 'science'
        results (int, опционально): 1-5 (default 3)
    """
    query = params.get("query", "")
    if not query:
        return json.dumps({"success": False, "error": "Параметр 'query' обязателен"})

    args = [query]

    category = params.get("category", "general")
    if category and category != "general":
        args.extend(["--category", category])

    results = params.get("results", 3)
    args.extend(["--results", str(results)])

    # Опциональные параметры
    model = params.get("model") or os.environ.get("LLM_MODEL")
    temperature = params.get("temperature") or os.environ.get("LLM_TEMPERATURE")
    if model:
        args.extend(["--model", model])
    if temperature:
        args.extend(["--temperature", str(temperature)])

    result = _run_script(PERPLEXITY_SEARCH_SCRIPT, args, TIMEOUT_PERPLEXITY)
    return json.dumps(result, ensure_ascii=False)


def handle_youtube_analyze(params: dict, **kwargs) -> str:
    """
    Запустить youtube_analyzer.py с параметрами из tool call.

    Ожидаемые параметры:
        url (str, опционально): URL видео (взаимоисключается с search)
        search (str, опционально): поисковый запрос (взаимоисключается с url)
        lang (str, опционально): 'ru' или 'en'
        limit (int, опционально): 1-5
        compare (bool, опционально): сравнить несколько видео
    """
    url = params.get("url", "")
    search = params.get("search", "")

    if not url and not search:
        return json.dumps({
            "success": False,
            "error": "Укажите 'url' или 'search' параметр",
        })

    args = []

    if url:
        args.extend(["--url", url])
    else:
        args.extend(["--search", search])

    lang = params.get("lang", "ru")
    if lang:
        args.extend(["--lang", lang])

    limit = params.get("limit")
    if limit and search:
        args.extend(["--limit", str(limit)])

    if params.get("compare") and search:
        args.append("--compare")

    result = _run_script(YOUTUBE_ANALYZER_SCRIPT, args, TIMEOUT_YOUTUBE)
    return json.dumps(result, ensure_ascii=False)