"""Personalization store + watch/topic toggles + a minimal dev auth (spec §5). The persistence is a small JSON file (documented in DASHBOARD.md) holding one preferences blob per user_id, matching API_CONTRACT section 3.3. It is intentionally simple and swappable: replacing PrefsStore with a DB-backed store is a local change; the preferences *shape* (and therefore the whole contract) is auth-method-independent, which is the point of Open Decision D1. A minimal LOCAL account (LocalAuth) is provided as the D1a default so the dashboard is runnable end-to-end without an IdP. It is explicitly a development placeholder (no password hashing, single-host JSON): swap it for a real auth service in production. Auth is NOT modelled in the data contract. """ from __future__ import annotations import hashlib import json import os import uuid from typing import Any, Dict, List, Optional # --------------------------------------------------------------------------- # Preferences store (API_CONTRACT section 3.3) # --------------------------------------------------------------------------- class PrefsStore: """File-backed per-user preferences. One blob per user_id. On-disk shape is `{"u_001": {...}, ...}` so the store is human-readable and diff-stable. All mutating ops are idempotent: re-adding an existing watch/topic is a no-op, and removing an absent one does not error. """ def __init__(self, path): self.path = path self._data = self._load() # -- persistence -------------------------------------------------- def _load(self): if not os.path.exists(self.path): return {} try: with open(self.path, encoding="utf-8") as fh: return json.load(fh) or {} except (ValueError, OSError): return {} def _save(self): os.makedirs(os.path.dirname(os.path.abspath(self.path)) or ".", exist_ok=True) tmp = self.path + ".tmp" with open(tmp, "w", encoding="utf-8") as fh: json.dump(self._data, fh, indent=2, sort_keys=True) os.replace(tmp, self.path) # -- reads -------------------------------------------------------- def get(self, user_id): return self._data.get(user_id) or {} def has(self, user_id): return user_id in self._data and bool(self._data[user_id]) def list_users(self): return sorted(k for k in self._data.keys() if self._data[k]) # -- writes ------------------------------------------------------- def put(self, user_id, prefs): merged = self.get(user_id) merged.update(prefs or {}) self._clean(merged) self._data[user_id] = merged self._save() return merged def add_watch(self, user_id, legislator_id): prefs = self.get(user_id) watch = list(prefs.get("watched_legislators") or []) if legislator_id not in watch: watch.append(legislator_id) return self._set(user_id, "watched_legislators", watch) def remove_watch(self, user_id, legislator_id): prefs = self.get(user_id) watch = [x for x in (prefs.get("watched_legislators") or []) if x != legislator_id] return self._set(user_id, "watched_legislators", watch) def add_topic(self, user_id, tag): prefs = self.get(user_id) topics = list(prefs.get("opted_in_tags") or []) if tag not in topics: topics.append(tag) return self._set(user_id, "opted_in_tags", topics) def remove_topic(self, user_id, tag): prefs = self.get(user_id) topics = [t for t in (prefs.get("opted_in_tags") or []) if t != tag] return self._set(user_id, "opted_in_tags", topics) def _set(self, user_id, key, value): prefs = self.get(user_id) prefs[key] = value self._clean(prefs) self._data[user_id] = prefs self._save() return prefs @staticmethod def _clean(prefs): for key in ("watched_legislators", "opted_in_tags", "districts"): prefs.setdefault(key, []) # --------------------------------------------------------------------------- # Minimal local account (Open Decision D1: D1a default — dev placeholder) # --------------------------------------------------------------------------- class LocalAuth: """A development local-account auth (email + a derived bearer token). This is the D1a default so the dashboard runs with no external IdP. It is deliberately NOT production-grade: no password hashing, no rotation, single host. Production should swap in a real auth service or the D1b OAuth path; the preferences shape is identical, so nothing else in the system changes. """ def __init__(self, path): self.path = path self._data = self._load() def _load(self): if not os.path.exists(self.path): return {} try: with open(self.path, encoding="utf-8") as fh: return json.load(fh) or {} except (ValueError, OSError): return {} def _save(self): os.makedirs(os.path.dirname(os.path.abspath(self.path)) or ".", exist_ok=True) tmp = self.path + ".tmp" with open(tmp, "w", encoding="utf-8") as fh: json.dump(self._data, fh, indent=2, sort_keys=True) os.replace(tmp, self.path) @staticmethod def _new_user_id(): return "u_" + uuid.uuid4().hex[:8] @staticmethod def _token_for(user_id, email): h = hashlib.sha256(("local:" + user_id + ":" + email).encode()).hexdigest() return "dev." + h[:24] def register(self, email, user_id=None): email = (email or "").strip().lower() for uid, rec in self._data.items(): if rec.get("email") == email: rec["token"] = self._token_for(uid, email) self._save() return {"user_id": uid, "token": rec["token"], "email": email} uid = user_id or self._new_user_id() self._data[uid] = {"email": email, "token": self._token_for(uid, email)} self._save() return {"user_id": uid, "token": self._data[uid]["token"], "email": email} def verify_token(self, token): if not token: return None for uid, rec in self._data.items(): t = rec.get("token") or "" if token == t or token == t[4:] or token == uid: return uid return None def user_by_id(self, user_id): return self._data.get(user_id)