diff --git a/.gitignore b/.gitignore index cbebc85..b96a96f 100644 --- a/.gitignore +++ b/.gitignore @@ -7,3 +7,4 @@ dist/ # Локальный конфиг может содержать API-ключи config.yaml +commands.yaml diff --git a/README.md b/README.md index aaf3675..d47e67b 100644 --- a/README.md +++ b/README.md @@ -7,6 +7,8 @@ в течение пары минут после ответа видят предыдущие вопросы - Журнал запросов и ответов со стримингом - Долговременная память: «запомни…», «забудь…», а устойчивые факты о вас модель сохраняет сама +- Настраиваемые команды: запуск программ и скриптов, ссылки и папки, HTTP-запросы (Home Assistant), + клавиши и медиа — через модель или мгновенно по точной фразе, опасные — с подтверждением - Озвучка ответов голосом Silero: фразы проговариваются по мере генерации, блоки кода пропускаются - Голосовой ввод по глобальной горячей клавише: faster-whisper на видеокарте, конец фразы по паузе (Silero VAD) - Слово активации («ассистент») через Vosk — без нажатия клавиш @@ -50,6 +52,43 @@ Silero читает только кириллицу: числа переводя (например, `qwen2.5:7b` в Ollama или большинство моделей OpenRouter). С другими моделями ассистент просто отвечает без памяти и один раз предупреждает об этом. +### Команды + +Команды описываются в `commands.yaml` рядом с `config.yaml`: при первом запуске он создаётся +с рабочими примерами (пауза, громкость, поиск, папка «Загрузки», блокировка) и +закомментированными шаблонами для скриптов, Home Assistant и выключения компьютера. +После сохранения файл перечитывается сам; список команд и ошибки видны в настройках. + +```yaml +commands: + - name: room_light + description: Включить или выключить свет в комнате + phrases: ["свет {state}"] # мгновенно, без модели + parameters: + state: {type: string, enum: ["on", "off"]} + confirm: false # true — спросить «да/нет» перед выполнением + action: + type: http + method: POST + url: http://homeassistant.local:8123/api/services/light/turn_{state} + headers: {Authorization: "Bearer ${HA_TOKEN}"} + json: {entity_id: light.room} +``` + +Как команда выполняется: + +- **Модель** выбирает команду по описанию и подставляет параметры («сделай потише на десять + шагов»). Результат (вывод скрипта, ответ сервера) возвращается модели, и она отвечает. +- **Точная фраза** выполняется сразу, без модели и даже без интернета, если запрос совпал + с ней целиком (регистр, «ё» и знаки препинания не важны). +- **Подтверждение** (`confirm: true`): ассистент спрашивает «Выполнить …?» и ждёт «да» или «нет»; + если вопрос был голосовым, микрофон включается сам. + +Безопасность: модель может только выбрать команду из файла и передать параметры, которые +проверяются по описанию. Программы запускаются без командной оболочки; для `cmd`, PowerShell +и `.bat`/`.cmd` значения со спецсимволами отклоняются. `${ПЕРЕМЕННЫЕ}` подставляются только +из шаблона, поэтому секреты не попадают ни в модель, ни в параметры. + ### Голосовой ввод Нажмите `Win+Alt+Space` (настраивается в `voice.hotkey`), дождитесь короткого сигнала и говорите — diff --git a/agr-assistent.spec b/agr-assistent.spec index c75b068..a4dda8f 100644 --- a/agr-assistent.spec +++ b/agr-assistent.spec @@ -7,7 +7,10 @@ from pathlib import Path import nvidia from PyInstaller.utils.hooks import collect_data_files, collect_dynamic_libs, collect_submodules -datas = [("src/agr_assistent/default_config.yaml", "agr_assistent")] +datas = [ + ("src/agr_assistent/default_config.yaml", "agr_assistent"), + ("src/agr_assistent/default_commands.yaml", "agr_assistent"), +] datas += collect_data_files("silero_vad") datas += collect_data_files("faster_whisper") diff --git a/src/agr_assistent/app.py b/src/agr_assistent/app.py index 4296b4d..236f2f2 100644 --- a/src/agr_assistent/app.py +++ b/src/agr_assistent/app.py @@ -11,13 +11,14 @@ import threading from logging.handlers import RotatingFileHandler from pathlib import Path -from PySide6.QtCore import QLibraryInfo, QLocale, QTranslator +from PySide6.QtCore import QLibraryInfo, QLocale, Qt, QTranslator from PySide6.QtWidgets import QApplication, QMessageBox, QSystemTrayIcon from agr_assistent import APP_NAME, __version__, system from agr_assistent.audio.player import AudioPlayer from agr_assistent.audio.recorder import SpeechRecorder from agr_assistent.audio.wakeword import VoskWakeWord +from agr_assistent.commands.catalog import COMMANDS_FILE_NAME, CommandCatalog from agr_assistent.config import AppConfig, ConfigError, data_dir, load_config from agr_assistent.core.assistant import Assistant from agr_assistent.core.memory import MemoryStore @@ -106,6 +107,8 @@ def main(argv: list[str] | None = None) -> int: memory = MemoryStore(data_dir() / "memory.sqlite3") app.aboutToQuit.connect(memory.close) + commands = CommandCatalog(config.path.parent / COMMANDS_FILE_NAME, app) + commands.ensure_file() assistant = Assistant( config.llm, speaker, @@ -113,17 +116,27 @@ def main(argv: list[str] | None = None) -> int: wake_word, memory=memory, memory_auto_save=config.memory.auto_save, + commands=commands, ) app.setWindowIcon(state_icon(assistant.state)) window = ChatWindow(assistant) instance.activated.connect(window.show_and_raise) + for error in commands.errors: + assistant.error_occurred.emit(error) dialog: SettingsDialog | None = None + def forget_dialog() -> None: + nonlocal dialog + dialog = None + def open_settings() -> None: nonlocal dialog - if dialog is None or not dialog.isVisible(): - dialog = SettingsDialog(settings, memory, window) + if dialog is None: + dialog = SettingsDialog(settings, memory, commands, window) + # Окно удаляется при закрытии, чтобы не копить подписки на каталог команд + dialog.setAttribute(Qt.WidgetAttribute.WA_DeleteOnClose) + dialog.destroyed.connect(forget_dialog) dialog.show() dialog.raise_() dialog.activateWindow() diff --git a/src/agr_assistent/commands/__init__.py b/src/agr_assistent/commands/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/agr_assistent/commands/actions.py b/src/agr_assistent/commands/actions.py new file mode 100644 index 0000000..98486fa --- /dev/null +++ b/src/agr_assistent/commands/actions.py @@ -0,0 +1,246 @@ +"""Выполнение действий команд: запуск программ, открытие ссылок, HTTP-запросы, клавиши.""" + +from __future__ import annotations + +import ctypes +import json +import logging +import os +import shutil +import subprocess +import sys +import time +import urllib.error +import urllib.parse +import urllib.request +import webbrowser +from dataclasses import dataclass +from typing import Any + +from agr_assistent.commands.model import PLACEHOLDER, Command, parse_key_combo +from agr_assistent.config import expand_env + +log = logging.getLogger(__name__) + +# Сколько текста вывода отдаём модели +MAX_OUTPUT_CHARS = 4000 +_MAX_HTTP_BYTES = 256_000 + +# Программы запускаются без оболочки, но cmd и PowerShell сами разбирают свои аргументы +# (а .bat/.cmd всегда исполняет cmd). Через значение параметра в них можно было бы +# подсунуть лишнюю команду, поэтому значения с такими символами не пропускаем +_CMD_METACHARACTERS = frozenset('&|<>^%!"()\r\n') +_POWERSHELL_METACHARACTERS = frozenset(";&|<>`$(){}[]@\"'#\r\n") +_INTERPRETERS = { + "cmd.exe": _CMD_METACHARACTERS, + "cmd": _CMD_METACHARACTERS, + "powershell.exe": _POWERSHELL_METACHARACTERS, + "powershell": _POWERSHELL_METACHARACTERS, + "pwsh.exe": _POWERSHELL_METACHARACTERS, + "pwsh": _POWERSHELL_METACHARACTERS, +} + +_KEYEVENTF_EXTENDEDKEY = 0x0001 +_KEYEVENTF_KEYUP = 0x0002 +_EXTENDED_KEYS = { + 0x21, 0x22, 0x23, 0x24, 0x25, 0x26, 0x27, 0x28, 0x2D, 0x2E, # навигация + 0x5B, # win + 0xAD, 0xAE, 0xAF, 0xB0, 0xB1, 0xB2, 0xB3, # громкость и медиа +} # fmt: skip +_KEY_PAUSE_SECONDS = 0.01 + + +@dataclass(frozen=True) +class Outcome: + ok: bool + content: str # для модели + display: str # для журнала + + +def execute(command: Command, arguments: dict[str, Any]) -> Outcome: + values = command.with_defaults(arguments) + action = command.action + handlers = {"run": _run, "open": _open, "http": _http, "keys": _keys} + log.info("Команда %s: %s %s", command.name, command.action_type, values) + return handlers[command.action_type](command, action, values) + + +def substitute(template: str, values: dict[str, Any], *, url_encode: bool = False) -> str: + """Подставляет {параметр}; отсутствующий необязательный параметр даёт пустую строку.""" + + def replace(match: Any) -> str: + value = values.get(match.group(1)) + text = "" if value is None else _as_text(value) + return urllib.parse.quote(text, safe="") if url_encode else text + + return PLACEHOLDER.sub(replace, template) + + +def substitute_json(value: Any, values: dict[str, Any]) -> Any: + """Строка ровно «{параметр}» заменяется значением с сохранением типа (число, true/false); + ${VAR} раскрываются только в шаблоне, не в значениях параметров.""" + if isinstance(value, str): + if match := PLACEHOLDER.fullmatch(value): + return values.get(match.group(1)) + return substitute(expand_env(value), values) + if isinstance(value, dict): + return {key: substitute_json(item, values) for key, item in value.items()} + if isinstance(value, list): + return [substitute_json(item, values) for item in value] + return value + + +def open_target(target: str) -> None: + """Вынесено отдельно, чтобы тесты не открывали браузер.""" + if target.startswith(("http://", "https://")): + webbrowser.open(target) + elif sys.platform == "win32": + os.startfile(target) # type: ignore[attr-defined] + else: + subprocess.Popen(["xdg-open", target]) + + +def send_keys(codes: list[int]) -> None: + user32 = ctypes.windll.user32 # type: ignore[attr-defined] + for code in codes: + user32.keybd_event(code, 0, _extended(code), 0) + time.sleep(_KEY_PAUSE_SECONDS) + for code in reversed(codes): + user32.keybd_event(code, 0, _extended(code) | _KEYEVENTF_KEYUP, 0) + time.sleep(_KEY_PAUSE_SECONDS) + + +def _run(command: Command, action: dict[str, Any], values: dict[str, Any]) -> Outcome: + program = substitute(_expand(action["program"]), values) + args = [substitute(_expand(arg), values) for arg in action["args"]] + resolved = shutil.which(program) or program + if not os.path.exists(resolved): + return _failed(f"Программа не найдена: {program}") + + forbidden = _interpreter_metacharacters(resolved) + if forbidden and any(set(_as_text(value)) & forbidden for value in values.values()): + return _failed( + f"Параметры для {os.path.basename(resolved)} содержат недопустимые символы — " + "команда не выполнена" + ) + + flags = 0 + if sys.platform == "win32" and action["hidden"]: + flags = subprocess.CREATE_NO_WINDOW # type: ignore[attr-defined] + cwd = action.get("cwd") + cwd = substitute(_expand(str(cwd)), values) if cwd else None + + if not action["wait"]: + subprocess.Popen([resolved, *args], cwd=cwd, creationflags=flags, close_fds=True) + return Outcome(True, "Программа запущена", f"Запустил: {command.summary(values)}") + + try: + completed = subprocess.run( + [resolved, *args], + cwd=cwd, + capture_output=True, + timeout=action["timeout_seconds"], + creationflags=flags, + ) + except subprocess.TimeoutExpired: + return _failed(f"Программа не завершилась за {action['timeout_seconds']:g} с") + output = (_decode(completed.stdout) + _decode(completed.stderr)).strip() + output = _truncate(output) or "(вывод пуст)" + ok = completed.returncode == 0 + display = f"{'Выполнил' if ok else 'Ошибка'}: {command.summary(values)}" + if not ok: + display += f" — код {completed.returncode}" + return Outcome(ok, f"Код завершения {completed.returncode}. Вывод:\n{output}", display) + + +def _open(command: Command, action: dict[str, Any], values: dict[str, Any]) -> Outcome: + template = _expand(str(action["target"])) + is_url = template.startswith(("http://", "https://")) + target = substitute(template, values, url_encode=is_url) + open_target(target) + return Outcome(True, f"Открыто: {target}", f"Открыл: {command.summary(values)}") + + +def _http(command: Command, action: dict[str, Any], values: dict[str, Any]) -> Outcome: + url = substitute(expand_env(str(action["url"])), values, url_encode=True) + headers = {key: substitute(expand_env(value), values) for key, value in action["headers"].items()} + data = None + if "json" in action: + data = json.dumps(substitute_json(action["json"], values), ensure_ascii=False).encode("utf-8") + headers.setdefault("Content-Type", "application/json") + request = urllib.request.Request(url, data=data, headers=headers, method=action["method"]) + + try: + with urllib.request.urlopen(request, timeout=action["timeout_seconds"]) as response: + status = response.status + body = response.read(_MAX_HTTP_BYTES) + except urllib.error.HTTPError as exc: + status = exc.code + body = exc.read(_MAX_HTTP_BYTES) + except (urllib.error.URLError, TimeoutError, OSError) as exc: + reason = getattr(exc, "reason", exc) + return _failed(f"Запрос не выполнен: {reason}") + + ok = 200 <= status < 300 + text = _truncate(body.decode("utf-8", errors="replace").strip()) or "(пустой ответ)" + display = f"{'Выполнил' if ok else 'Ошибка'}: {command.summary(values)}" + if not ok: + display += f" — HTTP {status}" + return Outcome(ok, f"HTTP {status}. Ответ:\n{text}", display) + + +def _keys(command: Command, action: dict[str, Any], values: dict[str, Any]) -> Outcome: + if sys.platform != "win32": + return _failed("Нажатие клавиш пока поддерживается только в Windows") + codes = parse_key_combo(substitute(str(action["keys"]), values)) + try: + repeat = int(substitute(str(action.get("repeat", 1)), values) or 1) + except ValueError: + return _failed("repeat должен быть целым числом") + for _ in range(max(1, min(repeat, 100))): + send_keys(codes) + return Outcome(True, "Клавиши нажаты", f"Выполнил: {command.summary(values)}") + + +def _interpreter_metacharacters(program: str) -> frozenset[str]: + name = os.path.basename(program).lower() + if name.endswith((".bat", ".cmd")): + return _CMD_METACHARACTERS + return _INTERPRETERS.get(name, frozenset()) + + +def _expand(template: str) -> str: + """${VAR} и %VAR% раскрываются в шаблоне до подстановки параметров, + чтобы значение от модели не могло сослаться на переменную окружения с секретом.""" + return os.path.expandvars(expand_env(template)) + + +def _failed(message: str) -> Outcome: + return Outcome(False, message, message) + + +def _extended(code: int) -> int: + return _KEYEVENTF_EXTENDEDKEY if code in _EXTENDED_KEYS else 0 + + +def _as_text(value: Any) -> str: + if isinstance(value, bool): + return "true" if value else "false" + return str(value) + + +def _truncate(text: str) -> str: + if len(text) <= MAX_OUTPUT_CHARS: + return text + return text[:MAX_OUTPUT_CHARS] + "\n…(вывод обрезан)" + + +def _decode(data: bytes) -> str: + """Консольные программы Windows пишут в кодировке OEM (cp866), новые — в UTF-8.""" + try: + return data.decode("utf-8") + except UnicodeDecodeError: + pass + if sys.platform == "win32": + return data.decode(f"cp{ctypes.windll.kernel32.GetOEMCP()}", errors="replace") # type: ignore[attr-defined] + return data.decode("utf-8", errors="replace") diff --git a/src/agr_assistent/commands/catalog.py b/src/agr_assistent/commands/catalog.py new file mode 100644 index 0000000..6c6d144 --- /dev/null +++ b/src/agr_assistent/commands/catalog.py @@ -0,0 +1,93 @@ +"""Загруженные команды: файл commands.yaml, перезагрузка при изменении, инструменты для модели.""" + +from __future__ import annotations + +import logging +from importlib import resources +from pathlib import Path +from typing import Any + +from PySide6.QtCore import QFileSystemWatcher, QObject, QTimer, Signal + +from agr_assistent.commands import actions +from agr_assistent.commands.model import Command, load_commands +from agr_assistent.llm.tools import Tool, ToolResult + +log = logging.getLogger(__name__) + +COMMANDS_FILE_NAME = "commands.yaml" + +# Редакторы сохраняют файл в несколько приёмов — перечитываем после паузы +_RELOAD_DELAY_MS = 300 + + +def default_commands_text() -> str: + return ( + resources.files("agr_assistent").joinpath("default_commands.yaml").read_text(encoding="utf-8") + ) + + +def command_tool(command: Command) -> Tool: + def handler(arguments: dict[str, Any]) -> ToolResult: + outcome = actions.execute(command, arguments) + return ToolResult(outcome.ok, outcome.content, outcome.display) + + return Tool( + name=command.name, + description=command.description, + parameters=command.schema(), + handler=handler, + confirm=command.confirm, + describe=command.summary, + ) + + +class CommandCatalog(QObject): + changed = Signal() + + def __init__(self, path: Path, parent: QObject | None = None) -> None: + super().__init__(parent) + self._path = path + self._commands: list[Command] = [] + self._errors: list[str] = [] + + self._reload_timer = QTimer(self) + self._reload_timer.setSingleShot(True) + self._reload_timer.setInterval(_RELOAD_DELAY_MS) + self._reload_timer.timeout.connect(self.reload) + self._watcher = QFileSystemWatcher(self) + self._watcher.fileChanged.connect(lambda _path: self._reload_timer.start()) + self.reload() + + @property + def path(self) -> Path: + return self._path + + @property + def commands(self) -> list[Command]: + return self._commands + + @property + def errors(self) -> list[str]: + return self._errors + + def ensure_file(self) -> None: + """Создаёт commands.yaml с примерами, если его ещё нет.""" + if not self._path.exists(): + self._path.parent.mkdir(parents=True, exist_ok=True) + self._path.write_text(default_commands_text(), encoding="utf-8") + log.info("Создан файл команд: %s", self._path) + self.reload() + + def tools(self) -> list[Tool]: + return [command_tool(command) for command in self._commands] + + def reload(self) -> None: + self._commands, self._errors = load_commands(self._path) + # После сохранения «заменой файла» наблюдение слетает — ставим заново + if self._path.exists() and str(self._path) not in self._watcher.files(): + self._watcher.addPath(str(self._path)) + log.info("Команды: %d загружено, %d ошибок", len(self._commands), len(self._errors)) + for error in self._errors: + log.warning(error) + self.changed.emit() diff --git a/src/agr_assistent/commands/matching.py b/src/agr_assistent/commands/matching.py new file mode 100644 index 0000000..0ea5c5b --- /dev/null +++ b/src/agr_assistent/commands/matching.py @@ -0,0 +1,59 @@ +"""Точные фразы команд без LLM и ответы «да/нет» на подтверждение.""" + +from __future__ import annotations + +import re +from typing import Any + +from agr_assistent.commands.model import PLACEHOLDER, Command +from agr_assistent.llm.tools import validate_arguments + +_NON_WORD = re.compile(r"[^\w\s{}]+") +_SPACES = re.compile(r"\s+") + +_YES = {"да", "ага", "конечно", "подтверждаю", "выполняй", "давай", "да давай", "да выполняй", "угу"} +_NO = {"нет", "отмена", "отмени", "не надо", "не нужно", "стоп", "не выполняй", "нет не надо"} + + +def normalize(text: str) -> str: + """Нижний регистр, ё → е, без знаков препинания (фигурные скобки сохраняются для шаблонов).""" + text = text.lower().replace("ё", "е") + return _SPACES.sub(" ", _NON_WORD.sub(" ", text)).strip() + + +def match_phrase(commands: list[Command], text: str) -> tuple[Command, dict[str, Any]] | None: + """Первая команда, фраза которой совпала с запросом целиком, и её аргументы.""" + request = normalize(text.replace("{", " ").replace("}", " ")) + if not request: + return None + for command in commands: + for phrase in command.phrases: + match = _phrase_pattern(phrase).fullmatch(request) + if match is None: + continue + arguments, errors = validate_arguments(command.schema(), match.groupdict()) + if not errors: + return command, arguments + return None + + +def confirmation_decision(text: str) -> bool | None: + """True — «да», False — «нет», None — это не ответ на вопрос.""" + request = normalize(text) + if request in _YES: + return True + if request in _NO: + return False + return None + + +def _phrase_pattern(phrase: str) -> re.Pattern[str]: + parts = [] + position = 0 + normalized = normalize(phrase) + for placeholder in PLACEHOLDER.finditer(normalized): + parts.append(re.escape(normalized[position : placeholder.start()])) + parts.append(f"(?P<{placeholder.group(1)}>.+?)") + position = placeholder.end() + parts.append(re.escape(normalized[position:])) + return re.compile("".join(parts)) diff --git a/src/agr_assistent/commands/model.py b/src/agr_assistent/commands/model.py new file mode 100644 index 0000000..2ed5002 --- /dev/null +++ b/src/agr_assistent/commands/model.py @@ -0,0 +1,285 @@ +"""Описание пользовательских команд из commands.yaml и их проверка.""" + +from __future__ import annotations + +import re +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +import yaml + +from agr_assistent.hotkey import key_code + +NAME_PATTERN = re.compile(r"^[A-Za-z0-9_-]{1,64}$") +# Только имена параметров: никаких выражений вроде {x.attr} или {x[0]}. +# ${VAR} — переменная окружения, а не параметр +PLACEHOLDER = re.compile(r"(? dict[str, Any]: + schema: dict[str, Any] = {"type": self.type} + if self.description: + schema["description"] = self.description + if self.enum is not None: + schema["enum"] = list(self.enum) + if self.minimum is not None: + schema["minimum"] = self.minimum + if self.maximum is not None: + schema["maximum"] = self.maximum + return schema + + +@dataclass(frozen=True) +class Command: + name: str + description: str + parameters: tuple[Parameter, ...] + action: dict[str, Any] + phrases: tuple[str, ...] + confirm: bool + reply: str + + @property + def action_type(self) -> str: + return str(self.action["type"]) + + def schema(self) -> dict[str, Any]: + return { + "type": "object", + "properties": {parameter.name: parameter.schema() for parameter in self.parameters}, + "required": [p.name for p in self.parameters if not p.optional], + } + + def with_defaults(self, arguments: dict[str, Any]) -> dict[str, Any]: + values = {p.name: p.default for p in self.parameters if p.default is not None} + values.update(arguments) + return values + + def summary(self, arguments: dict[str, Any]) -> str: + values = self.with_defaults(arguments) + details = ", ".join(f"{name}: {value}" for name, value in values.items()) + return f"{self.description} ({details})" if details else self.description + + +def load_commands(path: Path) -> tuple[list[Command], list[str]]: + """Возвращает корректные команды и описания ошибок; одна ошибка не ломает остальные.""" + if not path.exists(): + return [], [] + try: + data = yaml.safe_load(path.read_text(encoding="utf-8")) or {} + except yaml.YAMLError as exc: + return [], [f"{path.name}: не удалось разобрать YAML — {exc}"] + items = data.get("commands") if isinstance(data, dict) else None + if items is None: + return [], [] + if not isinstance(items, list): + return [], [f"{path.name}: commands должен быть списком"] + + commands: list[Command] = [] + errors: list[str] = [] + seen: set[str] = set() + for index, item in enumerate(items, start=1): + label = item.get("name") if isinstance(item, dict) and item.get("name") else f"№{index}" + try: + command = parse_command(item) + except CommandError as exc: + errors.append(f"{path.name}, команда {label}: {exc}") + continue + if command.name in seen: + errors.append(f"{path.name}, команда {label}: имя уже используется") + continue + seen.add(command.name) + commands.append(command) + return commands, errors + + +def parse_command(item: Any) -> Command: + if not isinstance(item, dict): + raise CommandError("ожидалось описание команды (словарь)") + name = str(item.get("name") or "") + if not NAME_PATTERN.fullmatch(name): + raise CommandError("name — латиница, цифры, _ и -, до 64 символов") + if name in RESERVED_NAMES: + raise CommandError("это имя занято встроенным инструментом памяти") + description = str(item.get("description") or "").strip() + if not description: + raise CommandError("нужно описание description — по нему модель выбирает команду") + + parameters = tuple(_parse_parameters(item.get("parameters"))) + names = {parameter.name for parameter in parameters} + action = _parse_action(item.get("action")) + unknown = _placeholders(action) - names + if unknown: + raise CommandError(f"в action используются неизвестные параметры: {', '.join(sorted(unknown))}") + + phrases = item.get("phrases") or [] + if isinstance(phrases, str): + phrases = [phrases] + phrases = tuple(str(phrase).strip() for phrase in phrases if str(phrase).strip()) + required = {p.name for p in parameters if not p.optional and p.default is None} + for phrase in phrases: + used = set(PLACEHOLDER.findall(phrase)) + if used - names: + raise CommandError(f"во фразе «{phrase}» неизвестные параметры: {', '.join(sorted(used - names))}") + if required - used: + raise CommandError( + f"во фразе «{phrase}» не хватает обязательных параметров: {', '.join(sorted(required - used))}" + ) + + return Command( + name=name, + description=description, + parameters=parameters, + action=action, + phrases=phrases, + confirm=bool(item.get("confirm", False)), + reply=str(item.get("reply") or "").strip(), + ) + + +def parse_key_combo(spec: str) -> list[int]: + """«ctrl+shift+esc» или «volume_up» -> виртуальные коды в порядке нажатия.""" + parts = [part.strip().lower() for part in str(spec).split("+")] + if not all(parts): + raise CommandError(f"некорректное сочетание клавиш «{spec}»") + *modifiers, key = parts + codes = [] + for modifier in modifiers: + if modifier not in MODIFIER_KEYS: + raise CommandError(f"неизвестный модификатор «{modifier}»") + codes.append(MODIFIER_KEYS[modifier]) + code = MEDIA_KEYS.get(key) or key_code(key) + if code is None: + raise CommandError(f"неизвестная клавиша «{key}»") + codes.append(code) + return codes + + +def _parse_parameters(data: Any) -> list[Parameter]: + if data is None: + return [] + if not isinstance(data, dict): + raise CommandError("parameters — словарь «имя: описание»") + parameters = [] + for name, spec in data.items(): + name = str(name) + if not re.fullmatch(r"\w+", name): + raise CommandError(f"некорректное имя параметра «{name}»") + if not isinstance(spec, dict): + spec = {"type": "string", "description": str(spec or "")} + kind = str(spec.get("type", "string")) + if kind not in PARAMETER_TYPES: + raise CommandError(f"параметр {name}: тип должен быть одним из {', '.join(sorted(PARAMETER_TYPES))}") + enum = spec.get("enum") + if enum is not None and not isinstance(enum, list): + raise CommandError(f"параметр {name}: enum должен быть списком") + default = spec.get("default") + parameters.append( + Parameter( + name=name, + type=kind, + description=str(spec.get("description") or ""), + enum=tuple(enum) if enum is not None else None, + minimum=spec.get("minimum"), + maximum=spec.get("maximum"), + default=default, + optional=bool(spec.get("optional", default is not None)), + ) + ) + return parameters + + +def _parse_action(data: Any) -> dict[str, Any]: + if not isinstance(data, dict): + raise CommandError("нужно действие action с полем type") + kind = data.get("type") + action = dict(data) + if kind == "run": + if not str(data.get("program") or "").strip(): + raise CommandError("для run нужна program") + args = data.get("args") or [] + if not isinstance(args, list): + raise CommandError("args должен быть списком") + action["args"] = [str(arg) for arg in args] + action["wait"] = bool(data.get("wait", False)) + action["hidden"] = bool(data.get("hidden", action["wait"])) + action["timeout_seconds"] = _positive(data.get("timeout_seconds", _DEFAULT_TIMEOUT_SECONDS)) + elif kind == "open": + if not str(data.get("target") or "").strip(): + raise CommandError("для open нужен target — ссылка, файл или папка") + elif kind == "http": + if not str(data.get("url") or "").startswith(("http://", "https://")): + raise CommandError("для http нужен url, начинающийся с http:// или https://") + method = str(data.get("method", "GET")).upper() + if method not in HTTP_METHODS: + raise CommandError(f"метод должен быть одним из {', '.join(sorted(HTTP_METHODS))}") + headers = data.get("headers") or {} + if not isinstance(headers, dict): + raise CommandError("headers должен быть словарём") + action["method"] = method + action["headers"] = {str(key): str(value) for key, value in headers.items()} + action["timeout_seconds"] = _positive(data.get("timeout_seconds", _DEFAULT_TIMEOUT_SECONDS)) + elif kind == "keys": + if not str(data.get("keys") or "").strip(): + raise CommandError("для keys нужно поле keys, например volume_up или ctrl+shift+esc") + parse_key_combo(data["keys"]) + repeat = data.get("repeat", 1) + if not (isinstance(repeat, int) or PLACEHOLDER.fullmatch(str(repeat))): + raise CommandError("repeat — целое число или параметр вида {steps}") + else: + raise CommandError("type действия должен быть run, open, http или keys") + return action + + +def _placeholders(value: Any) -> set[str]: + if isinstance(value, str): + return set(PLACEHOLDER.findall(value)) + if isinstance(value, dict): + return set().union(*(_placeholders(item) for item in value.values())) if value else set() + if isinstance(value, list): + return set().union(*(_placeholders(item) for item in value)) if value else set() + return set() + + +def _positive(value: Any) -> float: + try: + number = float(value) + except (TypeError, ValueError): + number = 0 + if number <= 0: + raise CommandError("timeout_seconds должен быть положительным числом") + return number diff --git a/src/agr_assistent/core/assistant.py b/src/agr_assistent/core/assistant.py index 1607450..eed845c 100644 --- a/src/agr_assistent/core/assistant.py +++ b/src/agr_assistent/core/assistant.py @@ -2,14 +2,19 @@ from __future__ import annotations +import json import logging import threading +import time +from dataclasses import dataclass from datetime import datetime from enum import Enum from typing import Any -from PySide6.QtCore import QObject, Signal, Slot +from PySide6.QtCore import QObject, QTimer, Signal, Slot +from agr_assistent.commands.catalog import CommandCatalog +from agr_assistent.commands.matching import confirmation_decision, match_phrase from agr_assistent.config import LLMConfig from agr_assistent.core.context import FollowUpContext, build_messages from agr_assistent.core.memory import MemoryStore, memory_prompt, memory_tools @@ -24,12 +29,33 @@ from agr_assistent.llm.client import ( ToolCalls, ToolsNotSupportedError, ) -from agr_assistent.llm.tools import ToolRegistry +from agr_assistent.llm.tools import Tool, ToolRegistry, ToolResult log = logging.getLogger(__name__) # Сколько раз подряд модель может вызвать инструменты в одном ответе MAX_TOOL_ROUNDS = 5 +# Сколько ждать ответа «да/нет» на вопрос-подтверждение +CONFIRMATION_TIMEOUT_SECONDS = 60 + +COMMANDS_PROMPT = ( + "Тебе доступны команды для управления компьютером — это инструменты, кроме инструментов " + "памяти. Если просьба пользователя соответствует команде, вызови её сразу, без лишних " + "уточнений. Сообщай результат коротко и только тот, что вернула команда; не придумывай " + "его. Если команда вернула, что нужно подтверждение, задай пользователю короткий вопрос " + "«да или нет» и больше ничего не делай." +) +_CONFIRMATION_RESULT = ( + "Команда НЕ выполнена: требуется подтверждение пользователя. Одним коротким вопросом спроси, " + "выполнить ли «{summary}». Пользователь ответит «да» или «нет» — это обработается само." +) + + +@dataclass(frozen=True) +class PendingConfirmation: + tool: Tool + arguments: dict[str, Any] + created_at: float class AssistantState(Enum): @@ -58,6 +84,7 @@ class Assistant(QObject): # Мост из фонового потока в главный; int — номер генерации _worker_chunk = Signal(int, str) _worker_tool = Signal(int, str, bool) + _worker_confirm = Signal(int, str, str) # имя инструмента, аргументы JSON _worker_notice = Signal(int, str) _worker_failed = Signal(int, str) _worker_done = Signal(int) @@ -71,6 +98,7 @@ class Assistant(QObject): *, memory: MemoryStore | None = None, memory_auto_save: bool = True, + commands: CommandCatalog | None = None, parent: QObject | None = None, ) -> None: super().__init__(parent) @@ -80,7 +108,13 @@ class Assistant(QObject): self._wake_word = wake_word self._memory = memory self._memory_auto_save = memory_auto_save - self._tools = ToolRegistry(memory_tools(memory) if memory is not None else ()) + self._commands = commands + self._tools = ToolRegistry() + self._rebuild_tools() + self._pending: PendingConfirmation | None = None + self._request_by_voice = False + # После вопроса-подтверждения на голосовой запрос сразу слушаем ответ + self._listen_after_reply = False # Модели, которые отказались работать с инструментами: больше не предлагаем им инструменты self._models_without_tools: set[tuple[str, str]] = set() self._client: LLMClient | None = None @@ -94,6 +128,7 @@ class Assistant(QObject): self._worker_chunk.connect(self._on_worker_chunk) self._worker_tool.connect(self._on_worker_tool) + self._worker_confirm.connect(self._on_worker_confirm) self._worker_notice.connect(self._on_worker_notice) self._worker_failed.connect(self._on_worker_failed) self._worker_done.connect(self._on_worker_done) @@ -106,6 +141,8 @@ class Assistant(QObject): voice.finished.connect(self._update_state) voice.recognized.connect(self._on_voice_recognized) voice.error_occurred.connect(self.error_occurred) + if commands is not None: + commands.changed.connect(self._on_commands_changed) if wake_word is not None: wake_word.detected.connect(self._on_wake_word) wake_word.enabled_changed.connect(self.wake_word_enabled_changed) @@ -201,12 +238,40 @@ class Assistant(QObject): self._speaker.set_enabled(enabled) self.speech_enabled_changed.emit(enabled) - def send(self, text: str) -> None: - """Новый запрос прерывает озвучку предыдущего ответа.""" + def send(self, text: str, *, by_voice: bool = False) -> None: + """Новый запрос прерывает озвучку предыдущего ответа. + + Порядок: ответ «да/нет» на ожидающее подтверждение, точная фраза команды, модель. + """ text = text.strip() if not text or self._generating: return self._speaker.stop() + self._request_by_voice = by_voice + + pending, self._pending = self._pending, None + if pending and time.monotonic() - pending.created_at <= CONFIRMATION_TIMEOUT_SECONDS: + decision = confirmation_decision(text) + if decision is not None: + self.request_added.emit(text, True) + if decision: + self._start_direct(text, pending.tool, pending.arguments) + else: + self._start_direct(text, reply="Хорошо, не выполняю.") + return + + if self._commands is not None and (match := match_phrase(self._commands.commands, text)): + command, arguments = match + tool = self._tools.get(command.name) + if tool is not None: + self.request_added.emit(text, False) + if command.confirm: + self._ask_confirmation(tool, arguments) + self._start_direct(text, reply=_confirmation_question(tool.summary(arguments))) + else: + self._start_direct(text, tool, arguments, reply=command.reply or None) + return + context = self._context.recent() self.request_added.emit(text, bool(context)) @@ -223,6 +288,8 @@ class Assistant(QObject): sections = [] if self._memory is not None: sections.append(memory_prompt(self._memory.facts(), self._memory_auto_save)) + if self._commands is not None and self._commands.commands: + sections.append(COMMANDS_PROMPT) messages = build_messages( self._config.system_prompt, context, text, datetime.now(), sections ) @@ -233,13 +300,14 @@ class Assistant(QObject): self.reply_started.emit() threading.Thread( target=self._run_reply, - args=(self._generation, client, messages, use_tools, model_key), + args=(self._generation, client, messages, self._tools, use_tools, model_key), name=f"llm-reply-{self._generation}", daemon=True, ).start() def cancel(self) -> None: """Прерывает запись голоса, генерацию и озвучку.""" + self._listen_after_reply = False if self._voice is not None: self._voice.cancel() if self._generating: @@ -251,8 +319,65 @@ class Assistant(QObject): """Очищает журнал и забывает контекст уточнений.""" self.cancel() self._context.clear() + self._pending = None self.journal_cleared.emit() + def _start_direct( + self, + request: str, + tool: Tool | None = None, + arguments: dict[str, Any] | None = None, + *, + reply: str | None = None, + ) -> None: + """Ответ без модели: выполнить команду (если есть) и сказать короткую фразу.""" + self._generation += 1 + self._generating = True + self._request = request + self._reply_parts = [] + self._speaker.begin() + self._update_state() + self.reply_started.emit() + threading.Thread( + target=self._run_direct, + args=(self._generation, tool, arguments or {}, reply), + name=f"command-{self._generation}", + daemon=True, + ).start() + + def _run_direct( + self, generation: int, tool: Tool | None, arguments: dict[str, Any], reply: str | None + ) -> None: + """Выполняется в фоновом потоке.""" + try: + if tool is not None: + result = ToolRegistry.run(tool, arguments) + if generation != self._generation: + return + self._worker_tool.emit(generation, result.display, result.ok) + if not result.ok: + reply = "Не получилось." + elif reply is None: + reply = "Готово." + if reply: + self._worker_chunk.emit(generation, reply) + except Exception as exc: + log.exception("Сбой выполнения команды") + self._worker_failed.emit(generation, f"Непредвиденная ошибка: {exc}") + else: + self._worker_done.emit(generation) + + def _ask_confirmation(self, tool: Tool, arguments: dict[str, Any]) -> None: + self._pending = PendingConfirmation(tool, arguments, time.monotonic()) + self._listen_after_reply = self._request_by_voice + + def _rebuild_tools(self) -> None: + tools = memory_tools(self._memory) if self._memory is not None else [] + if self._commands is not None: + tools += self._commands.tools() + # Запросы, которые уже выполняются, продолжают работать со своим набором + self._tools = ToolRegistry(tools) + def _get_client(self) -> LLMClient: if self._client is None: self._client = LLMClient( @@ -267,6 +392,7 @@ class Assistant(QObject): generation: int, client: LLMClient, messages: list[dict[str, Any]], + registry: ToolRegistry, use_tools: bool, model_key: tuple[str, str], ) -> None: @@ -274,7 +400,7 @@ class Assistant(QObject): try: for _round in range(MAX_TOOL_ROUNDS + 1): # На последнем круге инструменты не предлагаем — модель обязана ответить текстом - tools = self._tools.schemas() if use_tools and _round < MAX_TOOL_ROUNDS else None + tools = registry.schemas() if use_tools and _round < MAX_TOOL_ROUNDS else None try: calls, text = self._stream_round(generation, client, messages, tools) except ToolsNotSupportedError as exc: @@ -288,7 +414,7 @@ class Assistant(QObject): break messages.append(_assistant_tool_message(text, calls)) for call in calls: - result = self._tools.execute(call.name, call.arguments) + result = self._execute_tool_call(generation, registry, call) log.info("Инструмент %s(%s): %s", call.name, call.arguments, result.content) if generation != self._generation: return @@ -304,6 +430,21 @@ class Assistant(QObject): else: self._worker_done.emit(generation) + def _execute_tool_call( + self, generation: int, registry: ToolRegistry, call: ToolCall + ) -> ToolResult: + prepared = registry.prepare(call.name, call.arguments) + if isinstance(prepared, ToolResult): + return prepared + tool, arguments = prepared + if not tool.confirm: + return registry.run(tool, arguments) + summary = tool.summary(arguments) + self._worker_confirm.emit(generation, tool.name, json.dumps(arguments, ensure_ascii=False)) + return ToolResult( + True, _CONFIRMATION_RESULT.format(summary=summary), f"Ждёт подтверждения: {summary}" + ) + def _stream_round( self, generation: int, @@ -335,10 +476,28 @@ class Assistant(QObject): @Slot(str) def _on_voice_recognized(self, text: str) -> None: if text.strip(): - self.send(text) + self.send(text, by_voice=True) else: self.error_occurred.emit("Не удалось разобрать речь") + @Slot() + def _on_commands_changed(self) -> None: + self._rebuild_tools() + assert self._commands is not None + for error in self._commands.errors: + self.error_occurred.emit(error) + + @Slot(int, str, str) + def _on_worker_confirm(self, generation: int, name: str, arguments_json: str) -> None: + if generation == self._generation and (tool := self._tools.get(name)) is not None: + self._ask_confirmation(tool, json.loads(arguments_json)) + + @Slot() + def _listen_for_confirmation(self) -> None: + voice = self._voice + if voice is not None and self._pending is not None and self._state is AssistantState.IDLE: + voice.start() + @Slot(int, str) def _on_worker_chunk(self, generation: int, piece: str) -> None: if generation == self._generation: @@ -396,6 +555,9 @@ class Assistant(QObject): if state is not self._state: self._state = state self.state_changed.emit(state) + if state is AssistantState.IDLE and self._listen_after_reply: + self._listen_after_reply = False + QTimer.singleShot(0, self._listen_for_confirmation) self._sync_wake_word() def _sync_wake_word(self) -> None: @@ -421,3 +583,7 @@ def _assistant_tool_message(text: str, calls: list[ToolCall]) -> dict[str, Any]: for call in calls ], } + + +def _confirmation_question(summary: str) -> str: + return f"Выполнить: {summary}? Скажите «да» или «нет»." diff --git a/src/agr_assistent/default_commands.yaml b/src/agr_assistent/default_commands.yaml new file mode 100644 index 0000000..1ba3fd8 --- /dev/null +++ b/src/agr_assistent/default_commands.yaml @@ -0,0 +1,153 @@ +# Команды ассистента. Файл перечитывается автоматически после сохранения. +# +# Модель выбирает команду по описанию (description) и сама подставляет параметры. +# Фразы (phrases) выполняются мгновенно, без обращения к модели, если запрос совпал +# с ними целиком: регистр, «ё» и знаки препинания не важны. Во фразе можно указать +# параметр: «громкость {level}». +# +# Поля команды: +# name — латиница, цифры, _ и - +# description — что делает команда (по-русски, для модели) +# phrases — необязательно: точные фразы без LLM +# parameters — необязательно: имя -> {type: string|integer|number|boolean, +# description, enum, minimum, maximum, default, optional} +# confirm — true: спросить «да/нет» перед выполнением +# reply — необязательно: что ответить после выполнения по точной фразе +# action — что сделать, type одно из: +# run — program, args (список), wait (дождаться и вернуть вывод модели), +# timeout_seconds, hidden (без окна), cwd +# open — target: ссылка, файл или папка +# http — url, method, headers, json, timeout_seconds +# keys — keys: сочетание (ctrl+shift+esc) или volume_up, volume_down, volume_mute, +# media_play_pause, media_next, media_previous, media_stop; repeat +# +# В строках действий {параметр} заменяется значением, ${ПЕРЕМЕННАЯ} и %ПЕРЕМЕННАЯ% — +# переменными окружения (удобно для токенов: они не попадают в модель). +# Программы запускаются без командной оболочки: параметр не может превратиться в +# отдельную команду. cmd, PowerShell и .bat/.cmd сами разбирают аргументы, поэтому +# значения параметров с их спецсимволами (& | ; $ кавычки скобки и т. п.) отклоняются. + +commands: + - name: media_play_pause + description: Поставить воспроизведение музыки или видео на паузу либо продолжить + phrases: [пауза, продолжи, поставь на паузу, продолжи воспроизведение] + reply: Готово. + action: + type: keys + keys: media_play_pause + + - name: media_next + description: Включить следующий трек + phrases: [следующий трек, дальше, следующая песня] + reply: Следующий. + action: + type: keys + keys: media_next + + - name: media_previous + description: Включить предыдущий трек + phrases: [предыдущий трек, предыдущая песня] + reply: Предыдущий. + action: + type: keys + keys: media_previous + + - name: volume_up + description: Сделать звук громче + phrases: [громче, погромче] + parameters: + steps: + type: integer + description: На сколько шагов (один шаг — 2%) + minimum: 1 + maximum: 50 + default: 5 + reply: Громче. + action: + type: keys + keys: volume_up + repeat: "{steps}" + + - name: volume_down + description: Сделать звук тише + phrases: [тише, потише] + parameters: + steps: + type: integer + description: На сколько шагов (один шаг — 2%) + minimum: 1 + maximum: 50 + default: 5 + reply: Тише. + action: + type: keys + keys: volume_down + repeat: "{steps}" + + - name: volume_mute + description: Выключить или снова включить звук + phrases: [выключи звук, включи звук] + action: + type: keys + keys: volume_mute + + - name: web_search + description: Найти что-нибудь в интернете — открывает поиск в браузере + parameters: + query: + type: string + description: Поисковый запрос + action: + type: open + target: https://www.google.com/search?q={query} + + - name: open_downloads + description: Открыть папку «Загрузки» + phrases: [открой загрузки] + reply: Открываю. + action: + type: open + target: "%USERPROFILE%\\Downloads" + + - name: lock_computer + description: Заблокировать компьютер + phrases: [заблокируй компьютер] + action: + type: run + program: rundll32.exe + args: [user32.dll,LockWorkStation] + + # --- Примеры: раскомментируйте и поправьте под себя + + # - name: shutdown_computer + # description: Выключить компьютер через минуту + # phrases: [выключи компьютер] + # confirm: true + # action: + # type: run + # program: shutdown + # args: [/s, /t, "60"] + + # - name: disk_space + # description: Узнать, сколько свободного места на дисках + # action: + # type: run + # program: powershell + # args: [-NoProfile, -Command, "Get-PSDrive -PSProvider FileSystem | Format-Table Name, Free, Used"] + # wait: true + + # - name: room_light + # description: Включить или выключить свет в комнате (Home Assistant) + # parameters: + # state: + # type: string + # enum: ["on", "off"] + # description: on — включить, off — выключить + # action: + # type: http + # method: POST + # url: http://homeassistant.local:8123/api/services/light/turn_{state} + # headers: + # Authorization: Bearer ${HA_TOKEN} + # json: + # entity_id: light.room diff --git a/src/agr_assistent/hotkey.py b/src/agr_assistent/hotkey.py index 1319b9d..6bab622 100644 --- a/src/agr_assistent/hotkey.py +++ b/src/agr_assistent/hotkey.py @@ -70,17 +70,24 @@ def parse_hotkey(spec: str) -> tuple[int, int]: raise HotkeyError(f"Неизвестный модификатор '{name}' в горячей клавише '{spec}'") modifiers |= _MODIFIERS[name] - if len(key) == 1 and key.isascii() and key.isalnum(): - virtual_key = ord(key.upper()) - elif key in _NAMED_KEYS: - virtual_key = _NAMED_KEYS[key] - elif match := _FUNCTION_KEY.fullmatch(key): - virtual_key = 0x70 + int(match.group(1)) - 1 - else: + virtual_key = key_code(key) + if virtual_key is None: raise HotkeyError(f"Неизвестная клавиша '{key}' в горячей клавише '{spec}'") return modifiers, virtual_key +def key_code(key: str) -> int | None: + """Виртуальный код обычной клавиши: буква, цифра, f1–f24 или имя (space, enter…).""" + key = key.strip().lower() + if len(key) == 1 and key.isascii() and key.isalnum(): + return ord(key.upper()) + if key in _NAMED_KEYS: + return _NAMED_KEYS[key] + if match := _FUNCTION_KEY.fullmatch(key): + return 0x70 + int(match.group(1)) - 1 + return None + + class GlobalHotkey(QObject): """Слушает клавишу в отдельном потоке с собственной очередью сообщений Windows.""" diff --git a/src/agr_assistent/llm/tools.py b/src/agr_assistent/llm/tools.py index db656cb..e30cac1 100644 --- a/src/agr_assistent/llm/tools.py +++ b/src/agr_assistent/llm/tools.py @@ -28,6 +28,13 @@ class Tool: description: str parameters: dict[str, Any] # JSON Schema объекта аргументов handler: Callable[[dict[str, Any]], ToolResult] + # Выполнять только после явного «да» пользователя + confirm: bool = False + # Человекочитаемое описание вызова для вопроса-подтверждения и журнала + describe: Callable[[dict[str, Any]], str] | None = None + + def summary(self, arguments: dict[str, Any]) -> str: + return self.describe(arguments) if self.describe else self.description def schema(self) -> dict[str, Any]: return { @@ -55,7 +62,17 @@ class ToolRegistry: def schemas(self) -> list[dict[str, Any]]: return [tool.schema() for tool in self._tools.values()] + def get(self, name: str) -> Tool | None: + return self._tools.get(name) + def execute(self, name: str, arguments_json: str) -> ToolResult: + prepared = self.prepare(name, arguments_json) + if isinstance(prepared, ToolResult): + return prepared + return self.run(*prepared) + + def prepare(self, name: str, arguments_json: str) -> tuple[Tool, dict[str, Any]] | ToolResult: + """Находит инструмент и проверяет аргументы; ToolResult — описание ошибки.""" tool = self._tools.get(name) if tool is None: return ToolResult(False, f"Инструмента {name} не существует", f"Неизвестный инструмент {name}") @@ -70,6 +87,11 @@ class ToolRegistry: if errors: message = "; ".join(errors) return ToolResult(False, f"Некорректные аргументы: {message}", f"{name}: {message}") + return tool, arguments + + @staticmethod + def run(tool: Tool, arguments: dict[str, Any]) -> ToolResult: + name = tool.name try: return tool.handler(arguments) except ToolError as exc: @@ -104,6 +126,12 @@ def validate_arguments( allowed = ", ".join(str(option) for option in spec["enum"]) errors.append(f"{name}: недопустимое значение {converted!r} (допустимы: {allowed})") continue + if "minimum" in spec and converted < spec["minimum"]: + errors.append(f"{name}: значение меньше {spec['minimum']}") + continue + if "maximum" in spec and converted > spec["maximum"]: + errors.append(f"{name}: значение больше {spec['maximum']}") + continue result[name] = converted return result, errors diff --git a/src/agr_assistent/ui/settings_dialog.py b/src/agr_assistent/ui/settings_dialog.py index 3433227..d2ccdcb 100644 --- a/src/agr_assistent/ui/settings_dialog.py +++ b/src/agr_assistent/ui/settings_dialog.py @@ -2,6 +2,7 @@ from __future__ import annotations +import html import logging import threading from pathlib import Path @@ -28,12 +29,14 @@ from PySide6.QtWidgets import ( QPushButton, QSpinBox, QTabWidget, + QTextBrowser, QToolButton, QVBoxLayout, QWidget, ) from agr_assistent import APP_NAME, system +from agr_assistent.commands.catalog import CommandCatalog from agr_assistent.config import ConfigError, data_dir, expand_env, get_value from agr_assistent.core.memory import SOURCE_USER, MemoryStore from agr_assistent.core.settings import Settings, needs_restart @@ -41,6 +44,7 @@ from agr_assistent.core.settings import Settings, needs_restart log = logging.getLogger(__name__) _AUTO_FACT_COLOR = "#757575" +_ERROR_COLOR = "#e53935" _TTS_SPEAKERS = ["xenia", "baya", "kseniya", "aidar", "eugene"] _STT_MODELS = ["large-v3-turbo", "large-v3", "medium", "small", "base", "tiny"] @@ -70,11 +74,16 @@ class _ModelListLoader(QObject): class SettingsDialog(QDialog): def __init__( - self, settings: Settings, memory: MemoryStore | None = None, parent: QWidget | None = None + self, + settings: Settings, + memory: MemoryStore | None = None, + commands: CommandCatalog | None = None, + parent: QWidget | None = None, ) -> None: super().__init__(parent) self._settings = settings self._memory = memory + self._commands = commands self._raw = settings.raw() self._provider_edits: dict[str, dict[str, str]] = { name: {field: str(values.get(field) or "") for field in ("base_url", "api_key", "model")} @@ -90,6 +99,7 @@ class SettingsDialog(QDialog): tabs.addTab(self._build_speech_tab(), "Озвучка") tabs.addTab(self._build_voice_tab(), "Голосовой ввод") tabs.addTab(self._build_memory_tab(), "Память") + tabs.addTab(self._build_commands_tab(), "Команды") tabs.addTab(self._build_general_tab(), "Общие") buttons = QDialogButtonBox( @@ -255,6 +265,62 @@ class SettingsDialog(QDialog): self._reload_facts() return page + def _build_commands_tab(self) -> QWidget: + self._command_list = QTextBrowser() + self._command_errors = QLabel() + self._command_errors.setWordWrap(True) + self._command_errors.setStyleSheet(f"color: {_ERROR_COLOR}") + self._command_errors.setTextInteractionFlags(Qt.TextInteractionFlag.TextSelectableByMouse) + + open_file = QPushButton("Открыть commands.yaml") + reload_button = QPushButton("Перечитать") + buttons = QHBoxLayout() + buttons.addWidget(open_file) + buttons.addWidget(reload_button) + buttons.addStretch() + + hint = QLabel( + "Команды описываются в файле commands.yaml — в нём есть примеры и описание всех полей. " + "После сохранения файл перечитывается автоматически." + ) + hint.setWordWrap(True) + + page = QWidget() + layout = QVBoxLayout(page) + layout.addWidget(hint) + layout.addWidget(self._command_list, 1) + layout.addWidget(self._command_errors) + layout.addLayout(buttons) + + if self._commands is None: + page.setEnabled(False) + return page + catalog = self._commands + open_file.clicked.connect(lambda: _open_path(catalog.path)) + reload_button.clicked.connect(catalog.reload) + catalog.changed.connect(self._show_commands) + self._show_commands() + return page + + def _show_commands(self) -> None: + assert self._commands is not None + blocks = [] + for command in self._commands.commands: + details = [f"{command.name}, {command.action_type}"] + if command.confirm: + details.append("с подтверждением") + lines = [ + f"{html.escape(command.description)} " + f"({html.escape(', '.join(details))})" + ] + if command.phrases: + phrases = ", ".join(f"«{phrase}»" for phrase in command.phrases) + lines.append(f"фразы: {html.escape(phrases)}") + blocks.append(f"

{'
'.join(lines)}

") + self._command_list.setHtml("".join(blocks) or "

Команд пока нет.

") + self._command_errors.setText("\n".join(self._commands.errors)) + self._command_errors.setVisible(bool(self._commands.errors)) + def _build_general_tab(self) -> QWidget: self._start_minimized = QCheckBox("Запускаться свёрнутым в трей") self._start_minimized.setChecked(bool(get_value(self._raw, "ui.start_minimized"))) diff --git a/tests/test_assistant_flow.py b/tests/test_assistant_flow.py index 68cf3f0..589f524 100644 --- a/tests/test_assistant_flow.py +++ b/tests/test_assistant_flow.py @@ -6,9 +6,11 @@ from pathlib import Path import numpy as np import pytest -from PySide6.QtCore import QCoreApplication +from PySide6.QtCore import QCoreApplication, QObject, Signal from PySide6.QtWidgets import QApplication +from agr_assistent.commands import actions +from agr_assistent.commands.catalog import CommandCatalog from agr_assistent.config import LLMConfig, ProviderConfig from agr_assistent.core.assistant import Assistant from agr_assistent.core.memory import MemoryStore @@ -52,7 +54,14 @@ def memory(tmp_path: Path) -> MemoryStore: store.close() -def _assistant(server: FakeLLMServer, memory: MemoryStore, *, auto_save: bool = True) -> Assistant: +def _assistant( + server: FakeLLMServer, + memory: MemoryStore, + *, + auto_save: bool = True, + commands: CommandCatalog | None = None, + voice: object = None, +) -> Assistant: config = LLMConfig( provider="fake", providers={"fake": ProviderConfig("fake", server.base_url, "key", "fake-model")}, @@ -62,7 +71,14 @@ def _assistant(server: FakeLLMServer, memory: MemoryStore, *, auto_save: bool = timeout_seconds=10, ) speaker = Speaker(_SilentEngine(), _SilentPlayer(), enabled=False) - return Assistant(config, speaker, None, memory=memory, memory_auto_save=auto_save) + return Assistant( + config, + speaker, + voice, # type: ignore[arg-type] + memory=memory, + memory_auto_save=auto_save, + commands=commands, + ) def test_remember_request_runs_tool_and_answers( @@ -139,3 +155,190 @@ def test_auto_save_can_be_disabled( _wait_until(lambda: bool(finished)) assert "requested_by_user=false" not in fake_llm.requests[0]["messages"][0]["content"] + + +# --- команды + +_COMMANDS_YAML = """ +commands: + - name: open_downloads + description: Открыть папку «Загрузки» + phrases: [открой загрузки] + reply: Открываю. + action: {type: open, target: "C:/Downloads"} + - name: web_search + description: Найти в интернете + parameters: + query: {type: string} + action: {type: open, target: "https://example.com/?q={query}"} + - name: shutdown_computer + description: Выключить компьютер + phrases: [выключи компьютер] + confirm: true + action: {type: open, target: "shutdown://now"} +""" + + +class _FakeVoice(QObject): + listening_started = Signal() + recognizing_started = Signal() + finished = Signal() + recognized = Signal(str) + error_occurred = Signal(str) + + def __init__(self) -> None: + super().__init__() + self.started = 0 + self.is_listening = False + self.is_recognizing = False + self.is_active = False + + def start(self) -> None: + self.started += 1 + + def stop(self) -> None: + pass + + def cancel(self) -> None: + pass + + +@pytest.fixture +def opened(monkeypatch: pytest.MonkeyPatch) -> list[str]: + targets: list[str] = [] + monkeypatch.setattr(actions, "open_target", targets.append) + return targets + + +def _commands_assistant( + server: FakeLLMServer, memory: MemoryStore, tmp_path: Path, voice: _FakeVoice | None = None +) -> tuple[Assistant, CommandCatalog]: + path = tmp_path / "commands.yaml" + path.write_text(_COMMANDS_YAML, encoding="utf-8") + catalog = CommandCatalog(path) + return _assistant(server, memory, commands=catalog, voice=voice), catalog + + +def _collect(assistant: Assistant) -> dict[str, list]: + events: dict[str, list] = {"tools": [], "replies": [], "errors": []} + assistant.tool_executed.connect(lambda display, ok: events["tools"].append((display, ok))) + assistant.reply_finished.connect(events["replies"].append) + assistant.error_occurred.connect(events["errors"].append) + return events + + +def test_exact_phrase_runs_without_model( + qapp: QApplication, fake_llm: FakeLLMServer, memory: MemoryStore, tmp_path: Path, opened: list[str] +) -> None: + assistant, _catalog = _commands_assistant(fake_llm, memory, tmp_path) + events = _collect(assistant) + + assistant.send("Открой загрузки!") + _wait_until(lambda: bool(events["replies"])) + + assert opened == ["C:/Downloads"] + assert events["replies"] == ["Открываю."] + assert events["tools"] == [("Открыл: Открыть папку «Загрузки»", True)] + assert fake_llm.requests == [] + + +def test_model_calls_command_tool( + qapp: QApplication, fake_llm: FakeLLMServer, memory: MemoryStore, tmp_path: Path, opened: list[str] +) -> None: + fake_llm.replies += [ + Reply(tool_calls=[("web_search", '{"query": "погода в Казани"}')]), + Reply(text="Открыл поиск."), + ] + assistant, _catalog = _commands_assistant(fake_llm, memory, tmp_path) + events = _collect(assistant) + + assistant.send("Поищи погоду в Казани") + _wait_until(lambda: bool(events["replies"])) + + assert opened == ["https://example.com/?q=%D0%BF%D0%BE%D0%B3%D0%BE%D0%B4%D0%B0%20%D0%B2%20%D0%9A%D0%B0%D0%B7%D0%B0%D0%BD%D0%B8"] + tool_names = {tool["function"]["name"] for tool in fake_llm.requests[0]["tools"]} + assert {"web_search", "open_downloads", "remember"} <= tool_names + assert "команды для управления компьютером" in fake_llm.requests[0]["messages"][0]["content"] + assert events["replies"] == ["Открыл поиск."] + + +def test_model_command_with_confirmation_waits_for_yes( + qapp: QApplication, fake_llm: FakeLLMServer, memory: MemoryStore, tmp_path: Path, opened: list[str] +) -> None: + fake_llm.replies += [ + Reply(tool_calls=[("shutdown_computer", "{}")]), + Reply(text="Выключить компьютер?"), + ] + assistant, _catalog = _commands_assistant(fake_llm, memory, tmp_path) + events = _collect(assistant) + + assistant.send("Выключи комп, пожалуйста") + _wait_until(lambda: len(events["replies"]) == 1) + assert opened == [] + assert events["tools"] == [("Ждёт подтверждения: Выключить компьютер", True)] + assert "НЕ выполнена" in fake_llm.requests[1]["messages"][-1]["content"] + + assistant.send("Да") + _wait_until(lambda: len(events["replies"]) == 2) + assert opened == ["shutdown://now"] + assert events["replies"][1] == "Готово." + assert len(fake_llm.requests) == 2 # «да» обработано без модели + + +def test_phrase_confirmation_can_be_declined_or_dropped( + qapp: QApplication, fake_llm: FakeLLMServer, memory: MemoryStore, tmp_path: Path, opened: list[str] +) -> None: + fake_llm.replies.append(Reply(text="Не понял.")) + assistant, _catalog = _commands_assistant(fake_llm, memory, tmp_path) + events = _collect(assistant) + + assistant.send("Выключи компьютер") + _wait_until(lambda: len(events["replies"]) == 1) + assert events["replies"][0] == "Выполнить: Выключить компьютер? Скажите «да» или «нет»." + + assistant.send("нет") + _wait_until(lambda: len(events["replies"]) == 2) + assert events["replies"][1] == "Хорошо, не выполняю." + + # После отказа «да» уже ничего не подтверждает и уходит модели + assistant.send("да") + _wait_until(lambda: len(events["replies"]) == 3) + assert opened == [] + assert len(fake_llm.requests) == 1 + + +def test_voice_confirmation_starts_listening( + qapp: QApplication, fake_llm: FakeLLMServer, memory: MemoryStore, tmp_path: Path, opened: list[str] +) -> None: + voice = _FakeVoice() + assistant, _catalog = _commands_assistant(fake_llm, memory, tmp_path, voice) + events = _collect(assistant) + + assistant.send("выключи компьютер", by_voice=True) + _wait_until(lambda: voice.started == 1) + + assistant.send("открой загрузки", by_voice=True) + _wait_until(lambda: len(events["replies"]) == 2) + for _ in range(20): + QCoreApplication.processEvents() + time.sleep(0.005) + assert voice.started == 1 # без вопроса микрофон сам не включается + + +def test_commands_file_changes_are_picked_up( + qapp: QApplication, fake_llm: FakeLLMServer, memory: MemoryStore, tmp_path: Path, opened: list[str] +) -> None: + assistant, catalog = _commands_assistant(fake_llm, memory, tmp_path) + events = _collect(assistant) + + catalog.path.write_text( + _COMMANDS_YAML.replace("[открой загрузки]", "[покажи загрузки]") + + " - name: broken\n description: x\n action: {type: nope}\n", + encoding="utf-8", + ) + catalog.reload() + assistant.send("покажи загрузки") + _wait_until(lambda: bool(events["replies"])) + + assert opened == ["C:/Downloads"] + assert any("broken" in error for error in events["errors"]) diff --git a/tests/test_commands_actions.py b/tests/test_commands_actions.py new file mode 100644 index 0000000..f908541 --- /dev/null +++ b/tests/test_commands_actions.py @@ -0,0 +1,218 @@ +import json +import sys +import threading +from collections.abc import Iterator +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path +from typing import Any + +import pytest + +from agr_assistent.commands import actions +from agr_assistent.commands.model import parse_command + + +def _command(action: dict[str, Any], parameters: dict[str, Any] | None = None): # type: ignore[no-untyped-def] + return parse_command( + {"name": "test", "description": "Тест", "parameters": parameters, "action": action} + ) + + +def test_substitution_encodes_urls_and_keeps_json_types() -> None: + values = {"query": "кофе & чай", "level": 30, "on": True} + + assert actions.substitute("q={query}", values, url_encode=True) == ( + "q=%D0%BA%D0%BE%D1%84%D0%B5%20%26%20%D1%87%D0%B0%D0%B9" + ) + assert actions.substitute("{missing}-{level}", values) == "-30" + assert actions.substitute_json( + {"level": "{level}", "text": "уровень {level}", "flags": ["{on}"]}, values + ) == {"level": 30, "text": "уровень 30", "flags": [True]} + + +def test_parameter_values_cannot_reach_environment(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("TEST_SECRET_TOKEN", "секрет") + values = {"value": "${TEST_SECRET_TOKEN}"} + + assert actions.substitute_json("Bearer ${TEST_SECRET_TOKEN} {value}", values) == ( + "Bearer секрет ${TEST_SECRET_TOKEN}" + ) + + +def test_run_waits_and_returns_output() -> None: + command = _command( + { + "type": "run", + "program": sys.executable, + "args": ["-c", "import sys; print('привет', sys.argv[1])", "{name}"], + "wait": True, + }, + {"name": {"type": "string"}}, + ) + + outcome = actions.execute(command, {"name": "мир"}) + + assert outcome.ok + assert "привет мир" in outcome.content + + +def test_run_reports_exit_code_and_timeout() -> None: + failing = _command( + {"type": "run", "program": sys.executable, "args": ["-c", "raise SystemExit(3)"], "wait": True} + ) + slow = _command( + { + "type": "run", + "program": sys.executable, + "args": ["-c", "import time; time.sleep(5)"], + "wait": True, + "timeout_seconds": 0.5, + } + ) + + failed = actions.execute(failing, {}) + assert not failed.ok and "Код завершения 3" in failed.content + timed_out = actions.execute(slow, {}) + assert not timed_out.ok and "не завершилась" in timed_out.content + assert not actions.execute(_command({"type": "run", "program": "no-such-program-xyz"}), {}).ok + + +@pytest.mark.skipif(sys.platform != "win32", reason="bat-файлы есть только в Windows") +def test_batch_file_rejects_cmd_metacharacters(tmp_path: Path) -> None: + marker = tmp_path / "ran.txt" + script = tmp_path / "echo.bat" + script.write_text(f'@echo %1 > "{marker}"\n', encoding="utf-8") + command = _command( + {"type": "run", "program": str(script), "args": ["{text}"], "wait": True}, + {"text": {"type": "string"}}, + ) + + refused = actions.execute(command, {"text": "hi & calc"}) + assert not refused.ok and "недопустимые символы" in refused.content + assert not marker.exists() + + assert actions.execute(command, {"text": "hello"}).ok + assert marker.exists() + + +@pytest.mark.skipif(sys.platform != "win32", reason="cmd и powershell есть только в Windows") +def test_shell_interpreters_reject_injection_but_run_plain_values() -> None: + cmd = _command( + {"type": "run", "program": "cmd", "args": ["/c", "echo", "{text}"], "wait": True}, + {"text": {"type": "string"}}, + ) + powershell = _command( + { + "type": "run", + "program": "powershell", + "args": ["-NoProfile", "-Command", "Write-Output '{text}'"], + "wait": True, + }, + {"text": {"type": "string"}}, + ) + + assert not actions.execute(cmd, {"text": "hi & calc"}).ok + assert not actions.execute(powershell, {"text": "x'; Remove-Item C:\\temp"}).ok + + echoed = actions.execute(cmd, {"text": "Привет"}) + assert echoed.ok and "Привет" in echoed.content # вывод cmd в OEM-кодировке декодируется + written = actions.execute(powershell, {"text": "Мир"}) + assert written.ok and "Мир" in written.content + + +def test_open_uses_encoded_url(monkeypatch: pytest.MonkeyPatch) -> None: + opened: list[str] = [] + monkeypatch.setattr(actions, "open_target", opened.append) + command = _command( + {"type": "open", "target": "https://example.com/search?q={query}"}, + {"query": {"type": "string"}}, + ) + + assert actions.execute(command, {"query": "a b"}).ok + assert opened == ["https://example.com/search?q=a%20b"] + + +@pytest.fixture +def http_server() -> Iterator[tuple[str, list[dict[str, Any]]]]: + received: list[dict[str, Any]] = [] + + class Handler(BaseHTTPRequestHandler): + def log_message(self, *args: object) -> None: + pass + + def _respond(self) -> None: + length = int(self.headers.get("Content-Length") or 0) + received.append( + { + "method": self.command, + "path": self.path, + "auth": self.headers.get("Authorization"), + "body": json.loads(self.rfile.read(length)) if length else None, + } + ) + status = 500 if self.path.startswith("/fail") else 200 + payload = json.dumps({"state": "ok"}).encode() + self.send_response(status) + self.send_header("Content-Length", str(len(payload))) + self.end_headers() + self.wfile.write(payload) + + do_GET = do_POST = _respond + + server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) + threading.Thread(target=server.serve_forever, daemon=True).start() + yield f"http://127.0.0.1:{server.server_port}", received + server.shutdown() + server.server_close() + + +def test_http_sends_json_and_secret_headers( + http_server: tuple[str, list[dict[str, Any]]], monkeypatch: pytest.MonkeyPatch +) -> None: + base_url, received = http_server + monkeypatch.setenv("TEST_HA_TOKEN", "token123") + command = _command( + { + "type": "http", + "method": "POST", + "url": base_url + "/light/turn_{state}", + "headers": {"Authorization": "Bearer ${TEST_HA_TOKEN}"}, + "json": {"entity_id": "light.room", "brightness": "{level}"}, + }, + {"state": {"type": "string", "enum": ["on", "off"]}, "level": {"type": "integer"}}, + ) + + outcome = actions.execute(command, {"state": "on", "level": 80}) + + assert outcome.ok and "HTTP 200" in outcome.content + assert received == [ + { + "method": "POST", + "path": "/light/turn_on", + "auth": "Bearer token123", + "body": {"entity_id": "light.room", "brightness": 80}, + } + ] + assert "token123" not in outcome.content + outcome.display + + +def test_http_error_status_is_failure(http_server: tuple[str, list[dict[str, Any]]]) -> None: + base_url, _received = http_server + + outcome = actions.execute(_command({"type": "http", "url": base_url + "/fail"}), {}) + + assert not outcome.ok and "HTTP 500" in outcome.content + assert not actions.execute(_command({"type": "http", "url": "http://127.0.0.1:9/"}), {}).ok + + +@pytest.mark.skipif(sys.platform != "win32", reason="клавиши эмулируются только в Windows") +def test_keys_repeat(monkeypatch: pytest.MonkeyPatch) -> None: + pressed: list[list[int]] = [] + monkeypatch.setattr(actions, "send_keys", pressed.append) + command = _command( + {"type": "keys", "keys": "volume_up", "repeat": "{steps}"}, + {"steps": {"type": "integer", "default": 3}}, + ) + + assert actions.execute(command, {}).ok + assert pressed == [[0xAF]] * 3 diff --git a/tests/test_commands_model.py b/tests/test_commands_model.py new file mode 100644 index 0000000..34cf5e5 --- /dev/null +++ b/tests/test_commands_model.py @@ -0,0 +1,118 @@ +from pathlib import Path + +import pytest + +from agr_assistent.commands.catalog import default_commands_text +from agr_assistent.commands.model import CommandError, load_commands, parse_command, parse_key_combo + + +def _command(**overrides: object) -> dict: + item = {"name": "test", "description": "Тест", "action": {"type": "open", "target": "https://x.io"}} + item.update(overrides) + return item + + +def test_default_commands_file_is_valid(tmp_path: Path) -> None: + path = tmp_path / "commands.yaml" + path.write_text(default_commands_text(), encoding="utf-8") + + commands, errors = load_commands(path) + + assert errors == [] + assert {command.name for command in commands} >= {"media_play_pause", "volume_up", "web_search"} + + +def test_commented_examples_are_valid_too(tmp_path: Path) -> None: + """Примеры в конце файла закомментированы — раскомментированные, они тоже должны загружаться.""" + lines = default_commands_text().splitlines() + start = next(i for i, line in enumerate(lines) if "--- Примеры" in line) + uncommented = lines[: start + 1] + [ + line.replace(" # ", " ", 1) if line.startswith(" # ") else line + for line in lines[start + 1 :] + ] + path = tmp_path / "commands.yaml" + path.write_text("\n".join(uncommented), encoding="utf-8") + + commands, errors = load_commands(path) + + assert errors == [] + assert {"shutdown_computer", "disk_space", "room_light"} <= {c.name for c in commands} + + +def test_parameters_schema_defaults_and_summary() -> None: + command = parse_command( + _command( + parameters={ + "level": {"type": "integer", "minimum": 0, "maximum": 100}, + "steps": {"type": "integer", "default": 5}, + "note": "Просто описание", + }, + action={"type": "keys", "keys": "volume_up", "repeat": "{steps}"}, + ) + ) + + assert command.schema() == { + "type": "object", + "properties": { + "level": {"type": "integer", "minimum": 0, "maximum": 100}, + "steps": {"type": "integer"}, + "note": {"type": "string", "description": "Просто описание"}, + }, + "required": ["level", "note"], + } + assert command.with_defaults({"level": 30}) == {"steps": 5, "level": 30} + assert command.summary({"level": 30}) == "Тест (steps: 5, level: 30)" + + +@pytest.mark.parametrize( + ("overrides", "message"), + [ + ({"name": "плохое имя"}, "латиница"), + ({"name": "remember"}, "занято"), + ({"description": ""}, "описание"), + ({"action": {"type": "shell", "command": "rm"}}, "run, open, http или keys"), + ({"action": {"type": "run"}}, "program"), + ({"action": {"type": "http", "url": "ftp://x"}}, "http://"), + ({"action": {"type": "keys", "keys": "ctrl+bogus"}}, "неизвестная клавиша"), + ({"action": {"type": "open", "target": "https://x.io/{query}"}}, "неизвестные параметры"), + ( + {"parameters": {"level": {"type": "integer"}}, "phrases": ["громкость"]}, + "не хватает обязательных", + ), + ], +) +def test_invalid_commands_are_rejected(overrides: dict, message: str) -> None: + with pytest.raises(CommandError, match=message): + parse_command(_command(**overrides)) + + +def test_one_broken_command_does_not_break_others(tmp_path: Path) -> None: + path = tmp_path / "commands.yaml" + path.write_text( + """ +commands: + - name: good + description: Хорошая + action: {type: open, target: "https://x.io"} + - name: bad + description: Плохая + action: {type: nope} + - name: good + description: Дубль + action: {type: open, target: "https://y.io"} +""", + encoding="utf-8", + ) + + commands, errors = load_commands(path) + + assert [command.name for command in commands] == ["good"] + assert len(errors) == 2 + assert "bad" in errors[0] and "уже используется" in errors[1] + + +def test_key_combos() -> None: + assert parse_key_combo("ctrl+shift+esc") == [0x11, 0x10, 0x1B] + assert parse_key_combo("volume_mute") == [0xAD] + with pytest.raises(CommandError): + parse_key_combo("ctrl+") diff --git a/tests/test_matching.py b/tests/test_matching.py new file mode 100644 index 0000000..b797091 --- /dev/null +++ b/tests/test_matching.py @@ -0,0 +1,48 @@ +from agr_assistent.commands.matching import confirmation_decision, match_phrase, normalize +from agr_assistent.commands.model import parse_command + +_COMMANDS = [ + parse_command( + { + "name": "pause", + "description": "Пауза", + "phrases": ["Поставь на паузу", "пауза"], + "action": {"type": "keys", "keys": "media_play_pause"}, + } + ), + parse_command( + { + "name": "volume", + "description": "Громкость", + "phrases": ["громкость {level} процентов", "громкость {level}"], + "parameters": {"level": {"type": "integer", "minimum": 0, "maximum": 100}}, + "action": {"type": "keys", "keys": "volume_up"}, + } + ), +] + + +def test_normalize() -> None: + assert normalize(" Ещё, пожалуйста!! ") == "еще пожалуйста" + + +def test_whole_phrase_must_match() -> None: + assert match_phrase(_COMMANDS, "Пауза.")[0].name == "pause" + assert match_phrase(_COMMANDS, "поставь на паузу")[0].name == "pause" + assert match_phrase(_COMMANDS, "пауза в работе") is None + assert match_phrase(_COMMANDS, "") is None + + +def test_phrase_parameters_are_validated() -> None: + command, arguments = match_phrase(_COMMANDS, "Громкость 30 процентов") + assert command.name == "volume" and arguments == {"level": 30} + + assert match_phrase(_COMMANDS, "громкость 30")[1] == {"level": 30} + assert match_phrase(_COMMANDS, "громкость 300") is None + assert match_phrase(_COMMANDS, "громкость много") is None + + +def test_confirmation_decision() -> None: + assert confirmation_decision("Да!") is True + assert confirmation_decision("не надо") is False + assert confirmation_decision("да, но сначала сохрани файл") is None