Spaces:
Sleeping
Sleeping
| """Local usage-analytics logging for the IPA Translator app. | |
| Tracks how often each key feature is used (typed-text analysis vs. | |
| self-recording vs. uploaded recording, IPA keyboard opens, alternate | |
| options, Allosaurus comparisons) and what words/phrases are analyzed, | |
| so the Usage tab can show adoption over time. | |
| Best-effort throughout: any logging, read, or sync failure is caught and | |
| never raised, so a stats problem can never break the app people actually | |
| rely on. Data lives in a local SQLite file next to this module, which by | |
| itself does NOT survive a container restart (e.g. a redeploy or a wake | |
| from sleep). If an HF_TOKEN Space secret is configured, the DB is also | |
| restored from and periodically pushed to a private HF dataset repo, so | |
| history survives restarts; without that secret, everything still works, | |
| just local-only -- use the "Export CSV" button on the Usage tab as a | |
| manual backup in that case. | |
| """ | |
| import csv | |
| import os | |
| import sqlite3 | |
| import tempfile | |
| import time | |
| from collections import Counter | |
| from contextlib import closing | |
| from datetime import datetime, timezone | |
| from typing import Optional | |
| _DB_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "data", "usage_stats.db") | |
| # Durable backup: a private HF dataset repo the DB is restored from (once, | |
| # at startup) and pushed to (throttled, after each new event). Entirely | |
| # optional -- every function below no-ops if HF_TOKEN isn't set, so local | |
| # logging/reporting works the same with or without it. | |
| _DATASET_REPO_ID = os.environ.get("USAGE_STATS_DATASET_REPO", "ZacDLX1915/ipa-translator-usage-stats") | |
| _DATASET_FILENAME = "usage_stats.db" | |
| _SYNC_INTERVAL_SECONDS = 300 # push at most once per 5 minutes of activity | |
| _restore_attempted = False | |
| _last_sync_attempt = 0.0 | |
| # Event types tracked, in the order they should be displayed. | |
| EVENT_LABELS = { | |
| "analyze_text": "Analyze — typed text", | |
| "analyze_record": "Analyze — self-recording", | |
| "analyze_upload": "Analyze — uploaded recording", | |
| "keyboard_open": "IPA keyboard opened", | |
| "alt_options": "Alternate options requested", | |
| "allosaurus": "Allosaurus comparison run", | |
| } | |
| EVENT_ORDER = list(EVENT_LABELS.keys()) | |
| def _ensure_restored() -> None: | |
| """On first DB access each process, best-effort pull any prior history | |
| down from the dataset repo -- only when there's no local DB yet (a | |
| genuine cold start), so a mid-process retry can't clobber activity | |
| already logged this run. No-op without HF_TOKEN, or if nothing's been | |
| synced there yet.""" | |
| global _restore_attempted | |
| if _restore_attempted: | |
| return | |
| _restore_attempted = True | |
| if os.path.exists(_DB_PATH) or not os.environ.get("HF_TOKEN"): | |
| return | |
| try: | |
| from huggingface_hub import hf_hub_download | |
| import shutil | |
| downloaded = hf_hub_download( | |
| repo_id=_DATASET_REPO_ID, repo_type="dataset", filename=_DATASET_FILENAME, | |
| token=os.environ["HF_TOKEN"], | |
| ) | |
| os.makedirs(os.path.dirname(_DB_PATH), exist_ok=True) | |
| shutil.copy(downloaded, _DB_PATH) | |
| print(f"[usage-tracking] restored history from {_DATASET_REPO_ID}", flush=True) | |
| except Exception as exc: | |
| print(f"[usage-tracking] no prior history restored ({exc})", flush=True) | |
| def sync_to_dataset(force: bool = False) -> str: | |
| """Best-effort push of the local DB to the dataset repo. Throttled to | |
| at most once per _SYNC_INTERVAL_SECONDS unless force=True. Returns a | |
| status string -- "no_token" | "throttled" | "synced" | "failed" -- | |
| so callers (e.g. the manual Sync button) can give accurate feedback.""" | |
| global _last_sync_attempt | |
| if not os.environ.get("HF_TOKEN") or not os.path.exists(_DB_PATH): | |
| return "no_token" | |
| now = time.time() | |
| if not force and (now - _last_sync_attempt) < _SYNC_INTERVAL_SECONDS: | |
| return "throttled" | |
| _last_sync_attempt = now | |
| try: | |
| from huggingface_hub import HfApi | |
| HfApi().upload_file( | |
| path_or_fileobj=_DB_PATH, | |
| path_in_repo=_DATASET_FILENAME, | |
| repo_id=_DATASET_REPO_ID, | |
| repo_type="dataset", | |
| token=os.environ["HF_TOKEN"], | |
| commit_message="usage stats sync", | |
| ) | |
| return "synced" | |
| except Exception as exc: | |
| print(f"[usage-tracking] sync to {_DATASET_REPO_ID} failed: {exc}", flush=True) | |
| return "failed" | |
| def _connect() -> sqlite3.Connection: | |
| _ensure_restored() | |
| os.makedirs(os.path.dirname(_DB_PATH), exist_ok=True) | |
| conn = sqlite3.connect(_DB_PATH) | |
| conn.execute( | |
| "CREATE TABLE IF NOT EXISTS events (" | |
| " id INTEGER PRIMARY KEY AUTOINCREMENT," | |
| " ts TEXT NOT NULL," | |
| " event_type TEXT NOT NULL," | |
| " text TEXT," | |
| " uid TEXT," | |
| " lang TEXT" | |
| ")" | |
| ) | |
| # Migrate DBs created before the uid/lang columns existed (including ones | |
| # restored from an older dataset-repo snapshot). | |
| cols = {row[1] for row in conn.execute("PRAGMA table_info(events)")} | |
| if "uid" not in cols: | |
| conn.execute("ALTER TABLE events ADD COLUMN uid TEXT") | |
| if "lang" not in cols: | |
| conn.execute("ALTER TABLE events ADD COLUMN lang TEXT") | |
| return conn | |
| def log_event( | |
| event_type: str, text: Optional[str] = None, uid: Optional[str] = None, | |
| lang: Optional[str] = None, | |
| ) -> None: | |
| """Record one feature-use event, then opportunistically sync (throttled, | |
| best-effort) so history isn't lost if the Space restarts soon after. | |
| uid is an anonymous, browser-generated identifier (see app.py's page-load | |
| JS) used only to count distinct users -- it's not tied to any real | |
| identity, and is None whenever the browser couldn't produce one. | |
| lang is the human-readable language label selected in the dropdown | |
| (e.g. "Japanese"), not the internal eSpeak code -- stored as-is since | |
| it's only ever used for display on the Usage tab.""" | |
| try: | |
| clean_text = (text or "").strip() or None | |
| clean_uid = (uid or "").strip() or None | |
| clean_lang = (lang or "").strip() or None | |
| with closing(_connect()) as conn: | |
| conn.execute( | |
| "INSERT INTO events (ts, event_type, text, uid, lang) VALUES (?, ?, ?, ?, ?)", | |
| (datetime.now(timezone.utc).isoformat(timespec="seconds"), event_type, | |
| clean_text, clean_uid, clean_lang), | |
| ) | |
| conn.commit() | |
| except Exception as exc: | |
| print(f"[usage-tracking] failed to log {event_type!r}: {exc}", flush=True) | |
| return | |
| sync_to_dataset() | |
| def get_summary(top_words: int = 25, days: int = 14) -> dict: | |
| """Counts by event type, a day-by-day activity table (last `days` | |
| calendar days with any activity), and the most frequently tried | |
| words/phrases. Never raises -- returns an empty/error summary on | |
| failure so the Usage tab can't crash the app.""" | |
| empty = { | |
| "counts": {et: 0 for et in EVENT_ORDER}, "total": 0, "daily": [], | |
| "top_words": [], "languages": [], "first_ts": None, "last_ts": None, | |
| "error": None, "unique_users": 0, | |
| } | |
| try: | |
| with closing(_connect()) as conn: | |
| rows = conn.execute("SELECT ts, event_type, text, uid, lang FROM events").fetchall() | |
| except Exception as exc: | |
| empty["error"] = str(exc) | |
| return empty | |
| if not rows: | |
| return empty | |
| counts = Counter(r[1] for r in rows) | |
| words = Counter((r[2] or "").strip().lower() for r in rows if r[2]) | |
| languages = Counter(r[4] for r in rows if r[4]) | |
| daily = Counter((r[0][:10], r[1]) for r in rows) # (date, event_type) -> count | |
| daily_uids: dict = {} | |
| for ts, _et, _text, uid, _lang in rows: | |
| if uid: | |
| daily_uids.setdefault(ts[:10], set()).add(uid) | |
| recent_dates = sorted({d for d, _ in daily.keys()}, reverse=True)[:days] | |
| daily_table = [] | |
| for d in recent_dates: | |
| row = {"date": d, **{et: daily.get((d, et), 0) for et in EVENT_ORDER}} | |
| row["total"] = sum(row[et] for et in EVENT_ORDER) | |
| row["unique_users"] = len(daily_uids.get(d, ())) | |
| daily_table.append(row) | |
| return { | |
| "counts": {et: counts.get(et, 0) for et in EVENT_ORDER}, | |
| "total": sum(counts.values()), | |
| "daily": daily_table, | |
| "top_words": words.most_common(top_words), | |
| "languages": languages.most_common(), | |
| "first_ts": min(r[0] for r in rows), | |
| "last_ts": max(r[0] for r in rows), | |
| "unique_users": len({r[3] for r in rows if r[3]}), | |
| "error": None, | |
| } | |
| def export_csv() -> Optional[str]: | |
| """Write every logged event to a temp CSV and return its path (or | |
| None on failure) -- the manual backup path since the DB itself | |
| doesn't survive a container restart.""" | |
| try: | |
| with closing(_connect()) as conn: | |
| rows = conn.execute("SELECT ts, event_type, text, uid, lang FROM events ORDER BY id").fetchall() | |
| fd, path = tempfile.mkstemp(suffix="_ipa_usage_export.csv") | |
| with os.fdopen(fd, "w", newline="") as f: | |
| writer = csv.writer(f) | |
| writer.writerow(["timestamp_utc", "event_type", "text", "uid", "lang"]) | |
| writer.writerows(rows) | |
| return path | |
| except Exception as exc: | |
| print(f"[usage-tracking] export failed: {exc}", flush=True) | |
| return None | |