diff --git a/commands.yaml b/commands.yaml index 83779ca..dc99f0a 100644 --- a/commands.yaml +++ b/commands.yaml @@ -1307,3 +1307,31 @@ llm_status: - ты сейчас где работаешь - что за модель action: {type: llm_status} + +now_playing: + phrases: + - что играет + - что сейчас играет + - что за музыка + - что за песня + - какой трек + - какая песня + - что слушаю + action: {type: now_playing} + +world_clock: + phrases: + - сколько времени в + - который час в + - сколько часов в + - время в + action: {type: world_clock} + +daily_quote: + phrases: + - цитата дня + - мотивирующая цитата + - скажи цитату + - вдохнови меня + - цитата + action: {type: daily_quote} diff --git a/extensions.py b/extensions.py index 25e77b8..ba0ed9d 100644 --- a/extensions.py +++ b/extensions.py @@ -913,6 +913,160 @@ def do_llm_status(action, voice): _speak("LLM не настроен.") +# ── Now Playing (Windows SMTC) ────────────────────────────────────────────── + +_NOW_PLAYING_PS = """ +[Windows.Media.Control.GlobalSystemMediaTransportControlsSessionManager,Windows.Media.Control,ContentType=WindowsRuntime] | Out-Null; +$mgr = [Windows.Media.Control.GlobalSystemMediaTransportControlsSessionManager]::RequestAsync(); +while ($mgr.Status -eq 0) { Start-Sleep -Milliseconds 50 }; +$session = $mgr.GetResults().GetCurrentSession(); +if (-not $session) { Write-Output 'NONE'; exit }; +$propsTask = $session.TryGetMediaPropertiesAsync(); +while ($propsTask.Status -eq 0) { Start-Sleep -Milliseconds 50 }; +$props = $propsTask.GetResults(); +Write-Output ("{0}|{1}" -f $props.Artist, $props.Title) +""".strip().replace('\n', '; ') + + +def do_now_playing(action, voice): + try: + res = subprocess.run( + ['powershell', '-NoProfile', '-ExecutionPolicy', 'Bypass', '-Command', _NOW_PLAYING_PS], + capture_output=True, text=True, encoding='utf-8', timeout=8, + ) + except Exception as exc: + print(f"[ext] now_playing: {exc}") + _speak("Не получилось получить данные.") + return + + out = (res.stdout or '').strip() + if out == 'NONE' or not out: + _speak("Сейчас ничего не играет.") + return + if '|' not in out: + _speak("Не понял ответ системы.") + return + artist, _, title = out.partition('|') + artist, title = artist.strip(), title.strip() + if not title: + _speak("Трек без названия.") + return + if artist: + _speak(f"Сейчас играет: {artist} — {title}.") + else: + _speak(f"Сейчас играет: {title}.") + + +# ── World Clock ───────────────────────────────────────────────────────────── + +_TZ_MAP = { + 'москва': 'Europe/Moscow', + 'петербург': 'Europe/Moscow', 'спб': 'Europe/Moscow', + 'новосибирск': 'Asia/Novosibirsk', + 'екатеринбург': 'Asia/Yekaterinburg', + 'владивосток': 'Asia/Vladivostok', + 'омск': 'Asia/Omsk', + 'красноярск': 'Asia/Krasnoyarsk', + 'лондон': 'Europe/London', + 'нью-йорк': 'America/New_York', 'нью йорк': 'America/New_York', + 'токио': 'Asia/Tokyo', + 'пекин': 'Asia/Shanghai', + 'сидней': 'Australia/Sydney', + 'дубай': 'Asia/Dubai', + 'париж': 'Europe/Paris', + 'берлин': 'Europe/Berlin', + 'киев': 'Europe/Kyiv', + 'минск': 'Europe/Minsk', + 'алматы': 'Asia/Almaty', + 'ташкент': 'Asia/Tashkent', + 'сеул': 'Asia/Seoul', +} + + +def do_world_clock(action, voice): + import urllib.request + low = (voice or '').lower() + city = low + for trig in ('сколько времени в', 'который час в', 'сколько часов в', 'время в'): + idx = city.find(trig) + if idx >= 0: + city = city[idx + len(trig):].strip(' ,.:') + break + if not city: + _speak("В каком городе?") + return + + tz = None + for stem, zone in _TZ_MAP.items(): + if stem in city: + tz = zone + break + if not tz: + _speak("Не знаю такого города.") + return + + try: + with urllib.request.urlopen(f"https://worldtimeapi.org/api/timezone/{tz}", + timeout=8) as r: + import json as _json + data = _json.loads(r.read().decode()) + except Exception as exc: + print(f"[ext] world_clock: {exc}") + _speak("Не получилось узнать время.") + return + + dt = data.get('datetime', '') + m = re.search(r'T(\d{2}):(\d{2})', dt) + if not m: + _speak("Не понял ответ сервиса.") + return + _speak(f"В {city} сейчас {m.group(1)}:{m.group(2)}.") + + +# ── Daily quote ───────────────────────────────────────────────────────────── + +def do_daily_quote(action, voice): + import urllib.request + import json as _json + try: + with urllib.request.urlopen("https://zenquotes.io/api/random", timeout=8) as r: + data = _json.loads(r.read().decode()) + except Exception as exc: + print(f"[ext] daily_quote: {exc}") + _speak("Не получилось получить цитату.") + return + + if not data or not isinstance(data, list): + _speak("Пустой ответ.") + return + q = data[0] + quote_en = q.get('q', '') + author = q.get('a', 'Аноним') + if not quote_en: + _speak("Пустая цитата.") + return + + client = llm_backend.current_client() + text = quote_en + if client: + try: + resp = client.chat.completions.create( + model=llm_backend.current_model(), + messages=[ + {'role': 'system', 'content': 'Переведи английскую цитату на русский. Только перевод.'}, + {'role': 'user', 'content': quote_en}, + ], + max_tokens=200, temperature=0.2, timeout=20, + ) + translated = resp.choices[0].message.content.strip() + if translated: + text = translated + except Exception as exc: + print(f"[ext] daily_quote translate: {exc}") + + _speak(f"{text} — {author}.") + + # ── codebase Q&A (proxies to dev_handlers) ────────────────────────────────── def do_codebase_set(action, voice): diff --git a/main.py b/main.py index ceab4eb..d15fe41 100644 --- a/main.py +++ b/main.py @@ -1630,6 +1630,12 @@ def run_action(action): extensions.do_llm_switch(action, _current_voice or '') elif t == 'llm_status': extensions.do_llm_status(action, _current_voice or '') + elif t == 'now_playing': + extensions.do_now_playing(action, _current_voice or '') + elif t == 'world_clock': + extensions.do_world_clock(action, _current_voice or '') + elif t == 'daily_quote': + extensions.do_daily_quote(action, _current_voice or '') _current_voice: str = '' diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..ce595a0 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,45 @@ +"""Pytest config: temp-dir isolation for state modules. + +Each test gets a fresh dir for memory.json / schedule.json / etc so concurrent +runs don't pollute the real user data. +""" + +import os +import sys +import importlib +import pytest + +# Make parent package importable. +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + + +@pytest.fixture +def isolated_state(tmp_path, monkeypatch): + """Patch state modules to use a tmp_path so we don't touch real state files.""" + + def patch_module(mod_name, path_attrs): + """Reimport a module and rewire its `_PATH` and friends to tmp_path.""" + if mod_name in sys.modules: + del sys.modules[mod_name] + mod = importlib.import_module(mod_name) + for attr in path_attrs: + if not hasattr(mod, attr): + continue + original = getattr(mod, attr) + new_val = os.path.join(str(tmp_path), os.path.basename(original)) + monkeypatch.setattr(mod, attr, new_val) + # Also reset any cached _loaded / _initialized flags + for flag in ('_loaded', '_initialized', '_thread_started'): + if hasattr(mod, flag): + monkeypatch.setattr(mod, flag, False) + # Reset stores + for store_attr in ('_store', '_tasks'): + if hasattr(mod, store_attr): + cur = getattr(mod, store_attr) + if isinstance(cur, dict): + monkeypatch.setattr(mod, store_attr, {}) + elif isinstance(cur, list): + monkeypatch.setattr(mod, store_attr, []) + return mod + + yield patch_module diff --git a/tests/test_llm_backend.py b/tests/test_llm_backend.py new file mode 100644 index 0000000..f98cc25 --- /dev/null +++ b/tests/test_llm_backend.py @@ -0,0 +1,68 @@ +"""Tests for llm_backend.parse_backend + state machine.""" + +import os +import importlib +import sys + + +def test_parse_backend_groq_aliases(): + if 'llm_backend' in sys.modules: + del sys.modules['llm_backend'] + m = importlib.import_module('llm_backend') + for alias in ('groq', 'GROQ', 'cloud', 'облако', 'клауд', ' Groq '): + assert m.parse_backend(alias) == 'groq', f"failed: {alias!r}" + + +def test_parse_backend_ollama_aliases(): + if 'llm_backend' in sys.modules: + del sys.modules['llm_backend'] + m = importlib.import_module('llm_backend') + for alias in ('ollama', 'OLLAMA', 'local', 'локальный', 'локал', 'локальн'): + assert m.parse_backend(alias) == 'ollama', f"failed: {alias!r}" + + +def test_parse_backend_unknown_returns_none(): + if 'llm_backend' in sys.modules: + del sys.modules['llm_backend'] + m = importlib.import_module('llm_backend') + assert m.parse_backend('openai') is None + assert m.parse_backend('claude') is None + assert m.parse_backend('') is None + assert m.parse_backend(None) is None + + +def test_persist_round_trip(tmp_path, monkeypatch): + if 'llm_backend' in sys.modules: + del sys.modules['llm_backend'] + m = importlib.import_module('llm_backend') + monkeypatch.setattr(m, '_CHOICE_FILE', str(tmp_path / 'llm_backend.txt')) + m._persist('ollama') + assert m._read_persisted() == 'ollama' + m._persist('groq') + assert m._read_persisted() == 'groq' + + +def test_auto_detect_fallback_to_ollama(tmp_path, monkeypatch): + """When no GROQ_TOKEN in config, auto-detect should pick ollama.""" + if 'llm_backend' in sys.modules: + del sys.modules['llm_backend'] + m = importlib.import_module('llm_backend') + + # Mock config module with no GROQ_TOKEN + class FakeConfig: + GROQ_TOKEN = '' + monkeypatch.setitem(sys.modules, 'config', FakeConfig()) + + assert m._auto_detect_backend() == 'ollama' + + +def test_auto_detect_picks_groq_when_token_present(tmp_path, monkeypatch): + if 'llm_backend' in sys.modules: + del sys.modules['llm_backend'] + m = importlib.import_module('llm_backend') + + class FakeConfig: + GROQ_TOKEN = 'gsk_fake_test_token' + monkeypatch.setitem(sys.modules, 'config', FakeConfig()) + + assert m._auto_detect_backend() == 'groq' diff --git a/tests/test_macros_store.py b/tests/test_macros_store.py new file mode 100644 index 0000000..ff53efc --- /dev/null +++ b/tests/test_macros_store.py @@ -0,0 +1,87 @@ +"""Tests for macros_store: record/save/cancel/list/delete + control-phrase filter.""" + + +def test_is_macro_control_filters_record_phrases(isolated_state): + m = isolated_state('macros_store', ['_PATH']) + assert m._is_control("Запиши макрос работа") + assert m._is_control("сохрани макрос") + assert m._is_control("запусти макрос работа") + assert m._is_control("Удали макрос работа") + # Regular commands NOT filtered + assert not m._is_control("открой браузер") + assert not m._is_control("установи громкость 50") + + +def test_start_save_round_trip(isolated_state): + m = isolated_state('macros_store', ['_PATH']) + m.start("test") + assert m.is_recording() + assert m.recording_name() == 'test' + + m.record_step("открой браузер") + m.record_step("режим работа") + m.record_step("установи громкость 30") + # Control phrase should be filtered + m.record_step("запиши макрос новый") + + count = m.save() + assert count == 3 # 3 + 1 filtered = 3 + assert not m.is_recording() + + mac = m.get('test') + assert mac is not None + assert mac['steps'] == [ + 'открой браузер', + 'режим работа', + 'установи громкость 30', + ] + + +def test_save_without_recording_raises(isolated_state): + import pytest + m = isolated_state('macros_store', ['_PATH']) + with pytest.raises(RuntimeError): + m.save() + + +def test_save_empty_recording_raises(isolated_state): + import pytest + m = isolated_state('macros_store', ['_PATH']) + m.start("empty") + with pytest.raises(RuntimeError): + m.save() + + +def test_cancel_aborts_without_saving(isolated_state): + m = isolated_state('macros_store', ['_PATH']) + m.start("test") + m.record_step("foo") + assert m.cancel() is True + assert not m.is_recording() + assert m.get('test') is None + + +def test_delete(isolated_state): + m = isolated_state('macros_store', ['_PATH']) + m.start("test") + m.record_step("x") + m.save() + assert m.delete("test") is True + assert m.get('test') is None + assert m.delete("test") is False # idempotent + + +def test_list_names_sorted(isolated_state): + m = isolated_state('macros_store', ['_PATH']) + for name in ('zeta', 'alpha', 'beta'): + m.start(name) + m.record_step("step") + m.save() + assert m.list_names() == ['alpha', 'beta', 'zeta'] + + +def test_replay_unknown_raises(isolated_state): + import pytest + m = isolated_state('macros_store', ['_PATH']) + with pytest.raises(KeyError): + m.replay("nonexistent", lambda s: None) diff --git a/tests/test_memory_store.py b/tests/test_memory_store.py new file mode 100644 index 0000000..7ed055f --- /dev/null +++ b/tests/test_memory_store.py @@ -0,0 +1,79 @@ +"""Tests for memory_store — port of long_term_memory tests in rust.""" + +import json +import os +import pytest + + +def test_remember_recall_round_trip(isolated_state): + m = isolated_state('memory_store', ['_PATH']) + m.remember("любимый чай", "улун") + assert m.recall("любимый чай") == "улун" + assert m.recall("Любимый Чай") == "улун" # case-insensitive + + +def test_recall_returns_none_for_missing(isolated_state): + m = isolated_state('memory_store', ['_PATH']) + assert m.recall("nonexistent") is None + + +def test_forget_removes(isolated_state): + m = isolated_state('memory_store', ['_PATH']) + m.remember("foo", "bar") + assert m.forget("foo") is True + assert m.recall("foo") is None + assert m.forget("foo") is False # idempotent + + +def test_search_by_key_and_value(isolated_state): + m = isolated_state('memory_store', ['_PATH']) + m.remember("любимый чай", "улун") + m.remember("любимый кофе", "арабика") + m.remember("питомец", "собака бася") + + hits = m.search("чай", 5) + assert len(hits) == 1 + assert hits[0]['key'] == "любимый чай" + + hits = m.search("бася", 5) + assert len(hits) == 1 + assert hits[0]['key'] == "питомец" + + +def test_search_respects_limit(isolated_state): + m = isolated_state('memory_store', ['_PATH']) + for i in range(10): + m.remember(f"key{i}", "value") + hits = m.search("value", 3) + assert len(hits) == 3 + + +def test_search_empty_query_returns_nothing(isolated_state): + m = isolated_state('memory_store', ['_PATH']) + m.remember("foo", "bar") + assert m.search("", 5) == [] + assert m.search(" ", 5) == [] + + +def test_persists_across_module_reload(isolated_state, tmp_path): + m = isolated_state('memory_store', ['_PATH']) + m.remember("persistent", "yes") + json_file = m._PATH + assert os.path.isfile(json_file) + with open(json_file, encoding='utf-8') as f: + data = json.load(f) + assert data['entries']['persistent']['value'] == 'yes' + + +def test_build_llm_context_format(isolated_state): + m = isolated_state('memory_store', ['_PATH']) + m.remember("любимый чай", "улун") + # Query "чай" is substring of key "любимый чай" → matches. + ctx = m.build_llm_context("чай", 5) + assert "любимый чай = улун" in ctx + assert ctx.startswith("Известные факты") + + +def test_build_llm_context_empty(isolated_state): + m = isolated_state('memory_store', ['_PATH']) + assert m.build_llm_context("anything", 5) == "" diff --git a/tests/test_profiles_store.py b/tests/test_profiles_store.py new file mode 100644 index 0000000..c9815aa --- /dev/null +++ b/tests/test_profiles_store.py @@ -0,0 +1,61 @@ +"""Tests for profiles_store.""" + +import os +import json + + +def test_default_seeded_on_init(isolated_state, tmp_path): + p = isolated_state('profiles_store', ['_PROFILES_DIR', '_ACTIVE_FILE']) + p._init() + names = p.list_names() + # 5 defaults: default, work, game, sleep, driving + for expected in ('default', 'work', 'game', 'sleep', 'driving'): + assert expected in names, f"missing seeded profile: {expected}" + + +def test_active_defaults_to_default(isolated_state): + p = isolated_state('profiles_store', ['_PROFILES_DIR', '_ACTIVE_FILE']) + a = p.active() + assert a['name'] == 'default' + + +def test_set_active_persists(isolated_state): + p = isolated_state('profiles_store', ['_PROFILES_DIR', '_ACTIVE_FILE']) + p.set_active('work') + assert p.active_name() == 'work' + assert os.path.isfile(p._ACTIVE_FILE) + with open(p._ACTIVE_FILE, encoding='utf-8') as f: + assert f.read().strip() == 'work' + + +def test_set_active_unknown_raises(isolated_state): + p = isolated_state('profiles_store', ['_PROFILES_DIR', '_ACTIVE_FILE']) + try: + p.set_active('nonexistent') + assert False, "should have raised" + except ValueError: + pass + + +def test_allows_command_default_allows_all(isolated_state): + p = isolated_state('profiles_store', ['_PROFILES_DIR', '_ACTIVE_FILE']) + p.set_active('default') + assert p.allows_command('anything') + assert p.allows_command('games.steam.launch') + + +def test_allows_command_work_disables_games(isolated_state): + p = isolated_state('profiles_store', ['_PROFILES_DIR', '_ACTIVE_FILE']) + p.set_active('work') + assert not p.allows_command('games.steam.launch') + assert not p.allows_command('fun.joke') + assert p.allows_command('time.current') + + +def test_allows_command_driving_whitelist(isolated_state): + p = isolated_state('profiles_store', ['_PROFILES_DIR', '_ACTIVE_FILE']) + p.set_active('driving') + assert p.allows_command('music.play') + assert p.allows_command('weather.now') + assert not p.allows_command('windows.minimize') + assert not p.allows_command('screen.brightness') diff --git a/tests/test_scheduler_store.py b/tests/test_scheduler_store.py new file mode 100644 index 0000000..b59257c --- /dev/null +++ b/tests/test_scheduler_store.py @@ -0,0 +1,109 @@ +"""Tests for scheduler_store: parser + add/remove + serde.""" + +import time +import pytest + + +def test_parse_daily(isolated_state): + s = isolated_state('scheduler_store', ['_PATH']) + schedule = s.parse_schedule("daily 09:00") + assert schedule == {'kind': 'daily', 'hour': 9, 'minute': 0} + + +def test_parse_at_today_or_tomorrow(isolated_state): + s = isolated_state('scheduler_store', ['_PATH']) + schedule = s.parse_schedule("at 23:59") + assert schedule['kind'] == 'once' + assert schedule['at'] > time.time() + + +def test_parse_every_minutes(isolated_state): + s = isolated_state('scheduler_store', ['_PATH']) + schedule = s.parse_schedule("every 30 minutes") + assert schedule == {'kind': 'interval', 'seconds': 1800} + + +def test_parse_every_russian(isolated_state): + s = isolated_state('scheduler_store', ['_PATH']) + schedule = s.parse_schedule("every 2 часа") + assert schedule == {'kind': 'interval', 'seconds': 7200} + + +def test_parse_in_minutes_is_once(isolated_state): + s = isolated_state('scheduler_store', ['_PATH']) + schedule = s.parse_schedule("in 5 minutes") + assert schedule['kind'] == 'once' + # Should be roughly 300 seconds in the future + delta = schedule['at'] - time.time() + assert 290 <= delta <= 310 + + +def test_parse_bad_unit_raises(isolated_state): + s = isolated_state('scheduler_store', ['_PATH']) + with pytest.raises(ValueError): + s.parse_schedule("every 5 weeks") + + +def test_parse_bad_time_raises(isolated_state): + s = isolated_state('scheduler_store', ['_PATH']) + with pytest.raises(ValueError): + s.parse_schedule("daily 25:00") + with pytest.raises(ValueError): + s.parse_schedule("at 12:99") + + +def test_add_remove(isolated_state): + s = isolated_state('scheduler_store', ['_PATH']) + tid = s.add({ + 'name': 'Test', + 'schedule': 'in 1 hour', + 'action': {'type': 'speak', 'text': 'hello'}, + }) + assert tid + tasks = s.list_all() + assert len(tasks) == 1 + assert tasks[0]['name'] == 'Test' + + assert s.remove(tid) is True + assert s.list_all() == [] + assert s.remove(tid) is False # idempotent + + +def test_remove_by_text(isolated_state): + s = isolated_state('scheduler_store', ['_PATH']) + s.add({ + 'name': 'Water', + 'schedule': 'every 2 hours', + 'action': {'type': 'speak', 'text': 'Попей воды.'}, + }) + s.add({ + 'name': 'Stretch', + 'schedule': 'every 50 minutes', + 'action': {'type': 'speak', 'text': 'Размяться.'}, + }) + + removed = s.remove_by_text("воды") + assert removed == 1 + assert len(s.list_all()) == 1 + + +def test_clear(isolated_state): + s = isolated_state('scheduler_store', ['_PATH']) + s.add({'name': 'a', 'schedule': 'in 1 hour', 'action': {'type': 'speak', 'text': 'x'}}) + s.add({'name': 'b', 'schedule': 'in 2 hours', 'action': {'type': 'speak', 'text': 'y'}}) + assert s.clear() == 2 + assert s.list_all() == [] + + +def test_next_fire_once(isolated_state): + s = isolated_state('scheduler_store', ['_PATH']) + schedule = {'kind': 'once', 'at': 500} + assert s._next_fire(schedule, None, 1000) == 500 + # Already fired + assert s._next_fire(schedule, 500, 1000) is None + + +def test_next_fire_interval(isolated_state): + s = isolated_state('scheduler_store', ['_PATH']) + schedule = {'kind': 'interval', 'seconds': 60} + assert s._next_fire(schedule, 100, 1000) == 160