J.A.R.V.I.S-py/macros_store.py

"""Voice-macro recorder — port of `crates/jarvis-core/src/macros.rs`.

Records successful command phrases, replays them through the dispatch later.
JSON file at `<here>/macros.json`.

Public API:
    is_recording() / recording_name()
    start(name) / save() / cancel()
    list_names() / get(name) / delete(name)
    record_step(phrase)        # called by main.py on each successful command
    replay(name, dispatch_fn)  # dispatch_fn fires one phrase
"""

import json
import os
import time
import threading

_HERE = os.path.dirname(os.path.abspath(__file__))
_PATH = os.path.join(_HERE, 'macros.json')

_recording: dict | None = None  # {name, steps: list[str]}
_store: dict[str, dict] = {}
_loaded = False
_recording_lock = threading.Lock()


def _load():
    global _store, _loaded
    if _loaded:
        return
    if os.path.isfile(_PATH):
        try:
            with open(_PATH, encoding='utf-8') as f:
                data = json.load(f)
            _store = data.get('macros', {}) if isinstance(data, dict) else {}
        except (OSError, json.JSONDecodeError) as exc:
            print(f"[macros] corrupt: {exc}")
            _store = {}
    _loaded = True


def _save():
    tmp = _PATH + '.tmp'
    try:
        with open(tmp, 'w', encoding='utf-8') as f:
            json.dump({'macros': _store}, f, ensure_ascii=False, indent=2)
        os.replace(tmp, _PATH)
    except OSError as exc:
        print(f"[macros] save: {exc}")


def _normalize(s: str) -> str:
    return (s or '').strip().lower()


_CONTROL_TOKENS = (
    'запиши макрос', 'сохрани макрос', 'отмени макрос',
    'прекрати макрос', 'отмени запись', 'запусти макрос',
    'воспроизведи макрос', 'удали макрос',
)


def _is_control(phrase: str) -> bool:
    low = (phrase or '').lower()
    return any(tok in low for tok in _CONTROL_TOKENS)


def is_recording() -> bool:
    with _recording_lock:
        return _recording is not None


def recording_name() -> str | None:
    with _recording_lock:
        return _recording['name'] if _recording else None


def start(name: str) -> None:
    nname = _normalize(name)
    if not nname:
        raise ValueError("empty macro name")
    with _recording_lock:
        global _recording
        _recording = {'name': nname, 'steps': []}


def cancel() -> bool:
    with _recording_lock:
        global _recording
        if _recording is None:
            return False
        _recording = None
        return True


def save() -> int:
    """Stop recording, persist, return step count. Raises if nothing recorded."""
    with _recording_lock:
        global _recording
        if _recording is None:
            raise RuntimeError("не было активной записи")
        if not _recording['steps']:
            _recording = None
            raise RuntimeError("макрос пустой")

        _load()
        name = _recording['name']
        steps = list(_recording['steps'])
        _recording = None

    _store[name] = {
        'name': name,
        'steps': steps,
        'created_at': int(time.time()),
        'last_run': None,
    }
    _save()
    return len(steps)


def record_step(phrase: str) -> None:
    """Called by main.py after every successful command execution."""
    if not phrase:
        return
    if _is_control(phrase):
        return
    with _recording_lock:
        if _recording is None:
            return
        _recording['steps'].append(phrase.strip())


def list_names() -> list[str]:
    _load()
    return sorted(_store.keys())


def get(name: str) -> dict | None:
    _load()
    return _store.get(_normalize(name))


def delete(name: str) -> bool:
    _load()
    nname = _normalize(name)
    if nname in _store:
        del _store[nname]
        _save()
        return True
    return False


def replay(name: str, dispatch_fn) -> int:
    """Replay macro by name. dispatch_fn(phrase) fires a synthetic command.
    Returns step count. Raises if macro missing.
    Runs in a background thread with 800ms inter-step delay."""
    _load()
    nname = _normalize(name)
    m = _store.get(nname)
    if not m:
        raise KeyError(f"Макрос '{name}' не найден.")

    steps = list(m['steps'])

    def runner():
        for i, step in enumerate(steps):
            try:
                dispatch_fn(step)
            except Exception as exc:
                print(f"[macros] replay step {i}: {exc}")
            if i + 1 < len(steps):
                time.sleep(0.8)
        m['last_run'] = int(time.time())
        _save()

    threading.Thread(target=runner, name=f"macro-replay-{nname}",
                     daemon=True).start()
    return len(steps)