Spaces:
Sleeping
Sleeping
| import os | |
| import uuid | |
| import datetime as dt | |
| import sys | |
| from pathlib import Path | |
| from typing import Optional | |
| from supabase import create_client, Client | |
| def _utc_now_iso() -> str: | |
| return dt.datetime.now(dt.timezone.utc).isoformat() | |
| class AnalyticsLogger: | |
| """ | |
| Simple Supabase logger for: | |
| - Sessions (id: uuid, created_at: timestamptz) | |
| - Chats (id: uuid, session_id: uuid, timestamp: timestamptz, user: text, answer: text) | |
| """ | |
| def __init__(self): | |
| url = os.getenv("SUPABASE_URL") | |
| key = os.getenv("SUPABASE_KEY") | |
| if not url or not key: | |
| raise RuntimeError("Missing SUPABASE_URL or SUPABASE_KEY env var.") | |
| self.client: Client = create_client(url, key) | |
| self.session_id: Optional[str] = None | |
| def start_session(self, model_id: str) -> str: | |
| """ | |
| Creates a session row and returns the session UUID (string). | |
| """ | |
| sid = str(uuid.uuid4()) | |
| payload = {"id": sid, "created_at": _utc_now_iso(), "model_id": model_id} | |
| try: | |
| self.client.table("Sessions").insert(payload).execute() | |
| self.session_id = sid | |
| return sid | |
| except Exception as e: | |
| print(f"[AnalyticsLogger] Failed to start session: {e}", file=sys.stderr) | |
| raise e | |
| def _upload_image(self, image_path: str) -> Optional[str]: | |
| try: | |
| with open(image_path, "rb") as img_file: | |
| image_name = f'{uuid.uuid4()}{Path(image_path).suffix}' | |
| response = self.client.storage.from_("Images").upload(image_name, img_file, {"cacheControl": "3600", "upsert": "true"}) | |
| return response.full_path | |
| except: | |
| print(f"[AnalyticsLogger] Failed to upload image: {response['error']}", file=sys.stderr) | |
| return None | |
| def log_interaction(self, user: str | tuple[str, str], answer: str, ts_iso: Optional[str] = None) -> None: | |
| """ | |
| Inserts a single chat interaction. | |
| """ | |
| if not self.session_id: | |
| raise ValueError("Session not started. Call start_session() first.") | |
| session_id = self.session_id | |
| image_handle: str | None = None | |
| if isinstance(user, tuple): # (image_path, user_name) | |
| image, user = user | |
| image_handle = self._upload_image(image) | |
| chat_payload = { | |
| "id": str(uuid.uuid4()), | |
| "session_id": session_id, | |
| "timestamp": ts_iso or _utc_now_iso(), | |
| "user": user, | |
| "answer": answer, | |
| "user_image_path": image_handle, | |
| } | |
| try: | |
| self.client.table("Chats").insert(chat_payload).execute() | |
| except Exception as e: | |
| print(f"[AnalyticsLogger] Failed to log interaction: {e}", file=sys.stderr) | |