lapa-llm/quality-estimation

1import os2import uuid3import datetime as dt4import sys5from pathlib import Path6from typing import Optional7 8from supabase import create_client, Client9 10 11def _utc_now_iso() -> str:12    return dt.datetime.now(dt.timezone.utc).isoformat()13 14 15class AnalyticsLogger:16    """17    Simple Supabase logger for:18      - Sessions (id: uuid, created_at: timestamptz)19      - Chats (id: uuid, session_id: uuid, timestamp: timestamptz, user: text, answer: text)20    """21 22    def __init__(self):23        url = os.getenv("SUPABASE_URL")24        key = os.getenv("SUPABASE_KEY")25        if not url or not key:26            raise RuntimeError("Missing SUPABASE_URL or SUPABASE_KEY env var.")27        self.client: Client = create_client(url, key)28        self.session_id: Optional[str] = None29 30    def start_session(self, model_id: str) -> str:31        """32        Creates a session row and returns the session UUID (string).33        """34        sid = str(uuid.uuid4())35        payload = {"id": sid, "created_at": _utc_now_iso(), "model_id": model_id}36        try:37            self.client.table("Sessions").insert(payload).execute()38            self.session_id = sid39            return sid40        except Exception as e:41            print(f"[AnalyticsLogger] Failed to start session: {e}", file=sys.stderr)42            raise e43 44    def _upload_image(self, image_path: str) -> Optional[str]:45        try:46            with open(image_path, "rb") as img_file:47                image_name = f'{uuid.uuid4()}{Path(image_path).suffix}'48                response = self.client.storage.from_("Images").upload(image_name, img_file, {"cacheControl": "3600", "upsert": "true"})49 50                return response.full_path51        except:52            print(f"[AnalyticsLogger] Failed to upload image: {response['error']}", file=sys.stderr)53            return None54 55    def log_interaction(self, user: str | tuple[str, str], answer: str, ts_iso: Optional[str] = None) -> None:56        """57        Inserts a single chat interaction.58        """59        if not self.session_id:60            raise ValueError("Session not started. Call start_session() first.")61        session_id = self.session_id62 63        image_handle: str | None = None64 65        if isinstance(user, tuple): # (image_path, user_name)66            image, user = user67 68            image_handle = self._upload_image(image)69 70        chat_payload = {71            "id": str(uuid.uuid4()),72            "session_id": session_id,73            "timestamp": ts_iso or _utc_now_iso(),74            "user": user,75            "answer": answer,76            "user_image_path": image_handle,77        }78        try:79            self.client.table("Chats").insert(chat_payload).execute()80        except Exception as e:81            print(f"[AnalyticsLogger] Failed to log interaction: {e}", file=sys.stderr)82