app/persistence.py

"""Anonymous data collection via append-only JSONL + throttled HF Dataset push.

Modeled on the clanker-pet Snapshotter: buffer rows in memory, append to a local
JSONL on disk, and (throttled, background, fail-soft) push the JSONL to the PRIVATE
dataset deucebucket/vadugwi-data using huggingface_hub + os.environ HF_TOKEN.

Nothing here EVER raises into the caller. If HF_TOKEN is absent (local dev) we just
write the local JSONL and skip the push (logged once). A dataset hiccup must never
break /api/submit.

PRIVACY (HARD, design spec 11.3): rows carry ONLY {ts, engine_version, nonce, ...
rating fields}. No usernames, device ids, cookies, IPs, headers, or raw probe text.
"""
from __future__ import annotations

import json
import os
import threading
import time

STATE_FILE = "vadugwi-data.jsonl"


def _token() -> str | None:
    return os.environ.get("HF_TOKEN")


class DataSink:
    """Append-only anonymous row sink: local JSONL + throttled fail-soft HF push."""

    def __init__(
        self,
        jsonl_path: str,
        repo_id: str,
        min_interval: float = 60.0,
        now=time.monotonic,
    ):
        self._path = jsonl_path
        self._repo = repo_id
        self._min = min_interval
        self._now = now
        self._last = 0.0
        self._lock = threading.Lock()
        self._thread: threading.Thread | None = None
        self._warned_no_token = False
        os.makedirs(os.path.dirname(self._path) or ".", exist_ok=True)

    def append(self, rows: list[dict]) -> None:
        """Append rows to the local JSONL (always), then maybe push. Fail-soft."""
        if not rows:
            return
        try:
            with open(self._path, "a", encoding="utf-8") as f:
                for row in rows:
                    f.write(json.dumps(row, ensure_ascii=False) + "\n")
        except Exception:
            return
        self._maybe_push()

    def _maybe_push(self) -> None:
        if not _token():
            if not self._warned_no_token:
                self._warned_no_token = True
                print("[persistence] HF_TOKEN absent — writing local JSONL only, "
                      "skipping dataset push (local dev mode).")
            return
        with self._lock:
            t = self._now()
            if t - self._last < self._min:
                return
            self._last = t
        self._thread = threading.Thread(target=self._push, daemon=True)
        self._thread.start()

    def _push(self) -> None:
        token = _token()
        if not token or not os.path.exists(self._path):
            return
        try:
            from huggingface_hub import HfApi

            HfApi().upload_file(
                path_or_fileobj=self._path,
                path_in_repo=STATE_FILE,
                repo_id=self._repo,
                repo_type="dataset",
                token=token,
            )
        except Exception:
            # dataset hiccup must never surface; next append retries (throttled)
            return

    def restore(self) -> bool:
        """Optional restore-on-boot: pull the existing JSONL so we keep appending
        to the full history instead of starting empty. Fail-soft / best-effort."""
        token = _token()
        if not token:
            return False
        try:
            from huggingface_hub import hf_hub_download

            cached = hf_hub_download(
                repo_id=self._repo,
                repo_type="dataset",
                filename=STATE_FILE,
                token=token,
            )
            import shutil

            shutil.copyfile(cached, self._path)
            return True
        except Exception:
            return False