"""Proactive scheduler — port of `crates/jarvis-core/src/scheduler.rs`. JSON store at /schedule.json. Background daemon thread ticks every 30s and fires due tasks via the speak callback. Schedule formats (parsed by `parse_schedule`): "daily HH:MM" "at HH:MM" "every N minutes" / "every N hours" / "every N seconds" "in N minutes" / "in N hours" / "in N seconds" Public API: init() # start the background thread, load state add(task_dict) -> str # returns id remove(id) -> bool list_all() -> list[dict] remove_by_text(query) -> int clear() -> int """ import json import os import re import threading import time _HERE = os.path.dirname(os.path.abspath(__file__)) _PATH = os.path.join(_HERE, 'schedule.json') _TICK_SECS = 30 _speak_fn = print _tasks: list[dict] = [] _lock = threading.Lock() _thread_started = False def set_speak(fn): global _speak_fn _speak_fn = fn def _speak(text): try: _speak_fn(text) except Exception as exc: print(f"[scheduler] speak: {exc}") # ── persistence ─────────────────────────────────────────────────────────── def _load(): global _tasks if os.path.isfile(_PATH): try: with open(_PATH, encoding='utf-8') as f: data = json.load(f) _tasks = data.get('tasks', []) if isinstance(data, dict) else [] except (OSError, json.JSONDecodeError) as exc: print(f"[scheduler] corrupt: {exc}") _tasks = [] def _save(): tmp = _PATH + '.tmp' try: with open(tmp, 'w', encoding='utf-8') as f: json.dump({'tasks': _tasks}, f, ensure_ascii=False, indent=2) os.replace(tmp, _PATH) except OSError as exc: print(f"[scheduler] save: {exc}") # ── parsing ─────────────────────────────────────────────────────────────── _RU_UNITS = { 'минут': 60, 'минуту': 60, 'минуты': 60, 'час': 3600, 'часа': 3600, 'часов': 3600, 'секунд': 1, 'секунду': 1, 'секунды': 1, } _EN_UNITS = { 'minute': 60, 'minutes': 60, 'hour': 3600, 'hours': 3600, 'second': 1, 'seconds': 1, } def _unit_to_secs(token: str) -> int | None: """Map a unit word to seconds. Tries Russian stems first.""" low = token.lower() for stem, n in _RU_UNITS.items(): if low.startswith(stem): return n for stem, n in _EN_UNITS.items(): if low.startswith(stem): return n return None def parse_schedule(spec: str) -> dict: """Returns a task-schedule dict, raises ValueError on bad input. Schedule shape: {'kind': 'daily', 'hour': H, 'minute': M} {'kind': 'interval', 'seconds': S} {'kind': 'once', 'at': ts} """ s = spec.strip().lower() now = int(time.time()) # daily HH:MM m = re.match(r'^daily\s+(\d{1,2}):(\d{2})$', s) if m: h, mn = int(m.group(1)), int(m.group(2)) if not (0 <= h <= 23 and 0 <= mn <= 59): raise ValueError("hour/minute out of range") return {'kind': 'daily', 'hour': h, 'minute': mn} # at HH:MM m = re.match(r'^at\s+(\d{1,2}):(\d{2})$', s) if m: h, mn = int(m.group(1)), int(m.group(2)) if not (0 <= h <= 23 and 0 <= mn <= 59): raise ValueError("hour/minute out of range") import datetime as dt target = dt.datetime.now().replace(hour=h, minute=mn, second=0, microsecond=0) if target.timestamp() <= now: target = target + dt.timedelta(days=1) return {'kind': 'once', 'at': int(target.timestamp())} # every N m = re.match(r'^every\s+(\d+)\s+(\S+)$', s) if m: n = int(m.group(1)) secs_per = _unit_to_secs(m.group(2)) if secs_per is None: raise ValueError(f"unknown unit: {m.group(2)}") return {'kind': 'interval', 'seconds': n * secs_per} # in N m = re.match(r'^in\s+(\d+)\s+(\S+)$', s) if m: n = int(m.group(1)) secs_per = _unit_to_secs(m.group(2)) if secs_per is None: raise ValueError(f"unknown unit: {m.group(2)}") return {'kind': 'once', 'at': now + n * secs_per} raise ValueError(f"unknown schedule spec: {spec}") def _next_fire(schedule: dict, last_fired: int | None, now: int) -> int | None: kind = schedule.get('kind') if kind == 'once': if last_fired is not None: return None return schedule.get('at') if kind == 'interval': anchor = last_fired if last_fired is not None else now return anchor + int(schedule['seconds']) if kind == 'daily': import datetime as dt h, mn = int(schedule['hour']), int(schedule['minute']) today = dt.datetime.now().replace(hour=h, minute=mn, second=0, microsecond=0) today_ts = int(today.timestamp()) if today_ts > now and (last_fired is None or dt.datetime.fromtimestamp(last_fired).date() != today.date()): return today_ts return today_ts + 86400 return None # ── public API ──────────────────────────────────────────────────────────── def _gen_id() -> str: import uuid as uuidlib return f"task-{uuidlib.uuid4().hex[:12]}" def add(task: dict) -> str: """Add task. Required keys: name, schedule (dict or string), action. Action is a dict with at least {'type': 'speak', 'text': '...'}. """ sched = task.get('schedule') if isinstance(sched, str): sched = parse_schedule(sched) if not isinstance(sched, dict): raise ValueError("missing or invalid schedule") tid = task.get('id') or _gen_id() record = { 'id': tid, 'name': task.get('name', 'task'), 'schedule': sched, 'action': task.get('action', {'type': 'speak', 'text': 'Напоминание.'}), 'last_fired': None, 'enabled': task.get('enabled', True), 'created_at': int(time.time()), } with _lock: _tasks[:] = [t for t in _tasks if t['id'] != tid] _tasks.append(record) _save() return tid def remove(task_id: str) -> bool: with _lock: before = len(_tasks) _tasks[:] = [t for t in _tasks if t['id'] != task_id] if len(_tasks) != before: _save() return True return False def list_all() -> list[dict]: with _lock: return [dict(t) for t in _tasks] def clear() -> int: with _lock: n = len(_tasks) _tasks.clear() _save() return n def remove_by_text(query: str) -> int: nq = query.strip().lower() if not nq: return 0 with _lock: before = len(_tasks) def matches(t): if nq in t['name'].lower(): return True action = t.get('action', {}) if action.get('type') == 'speak': if nq in (action.get('text') or '').lower(): return True return False _tasks[:] = [t for t in _tasks if not matches(t)] removed = before - len(_tasks) if removed: _save() return removed # ── background tick ──────────────────────────────────────────────────────── def _fire_action(action: dict) -> None: atype = action.get('type', 'speak') if atype == 'speak': _speak(action.get('text', '...')) else: print(f"[scheduler] unknown action type: {atype}") def _tick(): now = int(time.time()) fired_ids = [] expired_once = [] with _lock: for task in list(_tasks): if not task.get('enabled', True): continue nf = _next_fire(task['schedule'], task.get('last_fired'), now) if nf is not None and nf <= now: fired_ids.append(task['id']) if task['schedule'].get('kind') == 'once': expired_once.append(task['id']) # Fire outside the lock — speak() can block. for task in list_all(): if task['id'] in fired_ids: print(f"[scheduler] firing {task['id']} '{task['name']}'") _fire_action(task['action']) if fired_ids: with _lock: for task in _tasks: if task['id'] in fired_ids: task['last_fired'] = now _tasks[:] = [t for t in _tasks if t['id'] not in expired_once] _save() def _tick_loop(): while True: try: time.sleep(_TICK_SECS) _tick() except Exception as exc: print(f"[scheduler] tick err: {exc}") def init(): """Load tasks + start the background thread (idempotent).""" global _thread_started _load() if _thread_started: return _thread_started = True threading.Thread(target=_tick_loop, name='jarvis-scheduler', daemon=True).start() print(f"[scheduler] thread started ({len(_tasks)} tasks loaded)")