schedules.py

"""Schedule store — recurring prompts fired without human intervention."""

import json
import re
import time
import threading
import uuid
from pathlib import Path


# Interval parsing: "every 30m", "every 1h", "every 2h", "daily at 09:00"
_INTERVAL_RE = re.compile(
    r"every\s+(\d+)\s*(m|min|h|hr|d|day)s?\b",
    re.IGNORECASE
)
_DAILY_RE = re.compile(
    r"daily\s+at\s+(\d{1,2}):(\d{2})\b",
    re.IGNORECASE
)


def parse_schedule_spec(spec: str) -> tuple[int | None, str | None]:
    """Parse natural-language schedule spec.
    Returns (interval_seconds, None) for intervals, or (None, cron_expr) for daily.
    Returns (None, None) if unparseable.
    """
    spec = spec.strip()
    m = _INTERVAL_RE.search(spec)
    if m:
        val = int(m.group(1))
        unit = m.group(2).lower()
        if unit in ("m", "min"):
            secs = val * 60
        elif unit in ("h", "hr"):
            secs = val * 3600
        elif unit in ("d", "day"):
            secs = val * 86400
        else:
            return (None, None)
        if secs < 60:
            secs = 60  # minimum 1 minute
        return (secs, None)

    dm = _DAILY_RE.search(spec)
    if dm:
        hour = int(dm.group(1)) % 24
        minute = int(dm.group(2)) % 60
        # Store as "HH:MM" for daily; we compute next_run from current time
        return (86400, f"{hour:02d}:{minute:02d}")  # interval=1 day, time hint

    return (None, None)


def compute_next_run(
    interval_seconds: int,
    last_run: float | None,
    daily_at: str | None = None,
) -> float:
    """Compute next run timestamp. daily_at is "HH:MM" for daily schedules."""
    now = time.time()
    if last_run is None:
        if daily_at:
            # First run: today at HH:MM, or tomorrow if already past
            parts = daily_at.split(":")
            hour, minute = int(parts[0]), int(parts[1])
            from datetime import datetime
            today = datetime.fromtimestamp(now)
            target = today.replace(hour=hour, minute=minute, second=0, microsecond=0)
            ts = target.timestamp()
            if ts <= now:
                from datetime import timedelta
                target = target + timedelta(days=1)
                ts = target.timestamp()
            return ts
        return now + interval_seconds
    if daily_at:
        from datetime import datetime, timedelta
        last_dt = datetime.fromtimestamp(last_run)
        parts = daily_at.split(":")
        hour, minute = int(parts[0]), int(parts[1])
        next_dt = last_dt.replace(hour=hour, minute=minute, second=0, microsecond=0)
        if next_dt <= last_dt:
            next_dt = next_dt + timedelta(days=1)
        return next_dt.timestamp()
    return last_run + interval_seconds


class ScheduleStore:
    def __init__(self, path: str):
        self._path = Path(path)
        self._path.parent.mkdir(parents=True, exist_ok=True)
        self._schedules: list[dict] = []
        self._lock = threading.Lock()
        self._callbacks: list = []
        self._load()

    def _load(self):
        if not self._path.exists():
            return
        try:
            raw = json.loads(self._path.read_text("utf-8"))
            if isinstance(raw, list):
                self._schedules = raw
        except (json.JSONDecodeError, TypeError):
            self._schedules = []

    def _save(self):
        self._path.write_text(
            json.dumps(self._schedules, indent=2, ensure_ascii=False) + "\n",
            "utf-8",
        )

    def on_change(self, callback):
        """Register callback(action, schedule) on any change."""
        self._callbacks.append(callback)

    def _fire(self, action: str, schedule: dict):
        for cb in self._callbacks:
            try:
                cb(action, schedule)
            except Exception:
                pass

    def list_all(self, active_only: bool = False) -> list[dict]:
        with self._lock:
            result = list(self._schedules)
        if active_only:
            result = [s for s in result if s.get("active", True)]
        return result

    def get(self, schedule_id: str) -> dict | None:
        with self._lock:
            for s in self._schedules:
                if s.get("id") == schedule_id:
                    return dict(s)
            return None

    def create(
        self,
        prompt: str,
        targets: list[str],
        channel: str = "general",
        interval_seconds: int | None = None,
        daily_at: str | None = None,
        one_shot: bool = False,
        send_at: float | None = None,
        created_by: str = "user",
    ) -> dict:
        """Create a schedule. Either interval_seconds or daily_at must be set.
        If one_shot=True, the schedule auto-deletes after firing once.
        If send_at is provided (epoch), use it as next_run directly.
        """
        schedule_id = str(uuid.uuid4())[:8]
        now = time.time()
        last_run = None
        if daily_at:
            interval_seconds = 86400
        if send_at:
            next_run = send_at
        else:
            next_run = compute_next_run(
                interval_seconds or 86400,
                last_run,
                daily_at=daily_at,
            )
        with self._lock:
            s = {
                "id": schedule_id,
                "prompt": prompt.strip()[:500],
                "targets": [t.strip().lstrip("@") for t in targets if t.strip()],
                "channel": channel or "general",
                "interval_seconds": interval_seconds or 86400,
                "daily_at": daily_at,
                "next_run": next_run,
                "created_at": now,
                "last_run": None,
                "active": True,
                "one_shot": one_shot,
                "created_by": created_by,
            }
            self._schedules.append(s)
            self._save()
        self._fire("create", s)
        return dict(s)

    def run_due(self) -> list[dict]:
        """Return list of schedules that are due (next_run <= now). Does not update."""
        now = time.time()
        with self._lock:
            due = [s for s in self._schedules if s.get("active") and s.get("next_run", 0) <= now]
        return [dict(s) for s in due]

    def mark_run(self, schedule_id: str) -> dict | None:
        """Mark schedule as run, advance next_run. Returns updated schedule."""
        with self._lock:
            for s in self._schedules:
                if s.get("id") != schedule_id:
                    continue
                now = time.time()
                s["last_run"] = now
                s["next_run"] = compute_next_run(
                    s.get("interval_seconds", 3600),
                    now,
                    daily_at=s.get("daily_at"),
                )
                self._save()
                result = dict(s)
                break
            else:
                return None
        self._fire("update", result)
        return result

    def delete(self, schedule_id: str) -> dict | None:
        with self._lock:
            for i, s in enumerate(self._schedules):
                if s.get("id") == schedule_id:
                    removed = self._schedules.pop(i)
                    self._save()
                    result = dict(removed)
                    break
            else:
                return None
        self._fire("delete", result)
        return result

    def toggle(self, schedule_id: str) -> dict | None:
        with self._lock:
            for s in self._schedules:
                if s.get("id") == schedule_id:
                    s["active"] = not s.get("active", True)
                    self._save()
                    result = dict(s)
                    break
            else:
                return None
        self._fire("update", result)
        return result