"""All database and storage access, in one place. Runs with the SERVICE ROLE key, which bypasses RLS and the column grants. That is the whole security model in one sentence: the browser physically cannot write `confidence`, `verdict` or `signals` because those columns are not granted to `authenticated`, and this service is the only thing holding a key that can. The service key must never reach the frontend. It lives only in the Hugging Face Space's secrets. """ from __future__ import annotations import io import math import os from datetime import datetime, timezone import numpy as np from PIL import Image from supabase import Client, create_client BUCKET = "tree-photos" def _client() -> Client: url = os.environ.get("SUPABASE_URL") key = os.environ.get("SUPABASE_SERVICE_KEY") if not url or not key: raise RuntimeError( "SUPABASE_URL and SUPABASE_SERVICE_KEY must be set. " "On Hugging Face Spaces these go in Settings -> Variables and secrets." ) return create_client(url, key) _db: Client | None = None def db() -> Client: global _db if _db is None: _db = _client() return _db def parse_ts(value: str) -> datetime: """Postgres timestamptz -> aware datetime. Postgres emits '+00:00' or a 6-digit fractional second; Python's fromisoformat is fussy about 'Z' on older versions. Normalising here stops a timezone bug from silently turning the travel-speed gate into nonsense. """ v = value.replace("Z", "+00:00") dt = datetime.fromisoformat(v) return dt if dt.tzinfo else dt.replace(tzinfo=timezone.utc) def vector_literal(v: np.ndarray) -> str: """pgvector wants '[0.1,0.2,...]' as text, not a JSON array.""" return "[" + ",".join(f"{x:.6f}" for x in v.tolist()) + "]" # --------------------------------------------------------------------------- # Reads # --------------------------------------------------------------------------- def get_checkin(checkin_id: str) -> dict | None: r = db().table("checkins").select("*").eq("id", checkin_id).limit(1).execute() return r.data[0] if r.data else None def get_tree(tree_id: str) -> dict | None: r = db().table("trees").select("*").eq("id", tree_id).limit(1).execute() return r.data[0] if r.data else None def list_pending(limit: int = 50) -> list[dict]: r = ( db() .table("checkins") .select("id") .eq("verdict", "pending") .order("server_received_at", desc=False) .limit(limit) .execute() ) return r.data or [] def other_phashes(exclude_checkin_id: str) -> list[tuple[str, str]]: """Every other check-in's pHash, for the duplicate gate. Deliberately global, not scoped to this tree: submitting tree 7's photo as tree 12's check-in is one of the attacks, and scoping the search to tree 12 would miss it entirely. """ r = ( db() .table("checkins") .select("id, phash") .not_.is_("phash", "null") .neq("id", exclude_checkin_id) .execute() ) return [(row["id"], row["phash"]) for row in (r.data or []) if row.get("phash")] def trees_near_registered_before( tree_id: str, lat: float, lng: float, before_iso: str, radius_m: float, ) -> list[tuple[str, float, float]]: """Trees within `radius_m` that were registered BEFORE this one. Across all owners: one physical tree registered by two different people is the double-payment case, and scoping this to the owner would miss exactly that. A bounding box does the elimination in Postgres and the true distance is measured here. PostGIS is available and `trees.location` is indexed, but reaching it needs a SQL function; the box uses the plain lat/lng columns and is correct because the caller re-measures. At pilot scale the difference is unmeasurable — the box is there so this does not become a full table scan when the grove is 40,000 trees rather than 40. """ d_lat = radius_m / 111_320.0 d_lng = radius_m / max(111_320.0 * math.cos(math.radians(lat)), 1e-6) r = ( db() .table("trees") .select("id, lat, lng") .gte("lat", lat - d_lat) .lte("lat", lat + d_lat) .gte("lng", lng - d_lng) .lte("lng", lng + d_lng) .lt("created_at", before_iso) .neq("id", tree_id) .execute() ) return [(row["id"], row["lat"], row["lng"]) for row in (r.data or [])] def previous_checkins(tree_id: str, before_iso: str) -> list[dict]: """Earlier visits to this tree, newest first, with embeddings.""" r = ( db() .table("checkins") .select( "id, wide_photo, close_photo, captured_at, wide_embedding, " "close_embedding, signals, verdict" ) .eq("tree_id", tree_id) .lt("captured_at", before_iso) .order("captured_at", desc=True) .execute() ) return r.data or [] def previous_by_submitter( submitter_id: str, before_iso: str, exclude_checkin_id: str ) -> dict | None: """The submitter's most recent check-in anywhere — for the travel gate.""" r = ( db() .table("checkins") .select("id, lat, lng, captured_at") .eq("submitted_by", submitter_id) .lt("captured_at", before_iso) .neq("id", exclude_checkin_id) .order("captured_at", desc=True) .limit(1) .execute() ) return r.data[0] if r.data else None def download_image(path: str) -> Image.Image: raw = db().storage.from_(BUCKET).download(path) return Image.open(io.BytesIO(raw)) def parse_embedding(value) -> np.ndarray | None: """pgvector comes back as either a string literal or a list, depending on the PostgREST version. Handle both rather than discovering it at 2am.""" if value is None: return None if isinstance(value, str): value = value.strip().strip("[]") if not value: return None return np.fromstring(value, sep=",", dtype=np.float32) return np.asarray(value, dtype=np.float32) # --------------------------------------------------------------------------- # Writes — only this process may perform them # --------------------------------------------------------------------------- def write_result( checkin_id: str, *, phash: str, wide_embedding: np.ndarray, close_embedding: np.ndarray, confidence: int, verdict: str, signals: dict, audit_sample: bool = False, ) -> None: db().table("checkins").update( { "phash": phash, "wide_embedding": vector_literal(wide_embedding), "close_embedding": vector_literal(close_embedding), "confidence": confidence, "verdict": verdict, "signals": signals, "audit_sample": audit_sample, } ).eq("id", checkin_id).execute() def write_advice(checkin_id: str, species_guess: dict, advice: dict) -> None: """Store the advisory assessment. Writes ONLY the two advisory columns. It does not touch `confidence`, `verdict` or `signals`, and that separation is the whole point: advice is generated by a component whose error rate we have not measured, so it must not be able to move a number we publish. Note this is a separate write from write_result rather than a parameter on it. Scoring is replayable and re-run whenever a threshold changes; advice is a paid external call that should survive a re-score untouched. Folding the two together would either destroy advice on every recalibration or make recalibration cost money. """ db().table("checkins").update( {"species_guess": species_guess, "advice": advice} ).eq("id", checkin_id).execute() def list_checkins_without_advice(limit: int = 100) -> list[dict]: """Check-ins that have never been assessed, oldest first.""" r = ( db() .table("checkins") .select("id") .is_("advice", "null") .order("server_received_at", desc=False) .limit(limit) .execute() ) return r.data or [] def update_tree_status(tree_id: str, confidence: int, verdict: str) -> None: """Roll the latest verdict up onto the tree. 'alive' only on a verified check-in. A tree never becomes 'dead' automatically — that is a human decision, because being wrong about it stops someone's payment. """ status = { "verified": "alive", "review": "pending", "flagged": "flagged", }.get(verdict, "pending") db().table("trees").update( {"status": status, "current_confidence": confidence} ).eq("id", tree_id).execute()