Spaces:
Sleeping
Sleeping
| """Supabase client and helper functions for Project Memo. | |
| Tables expected in Supabase (create via the SQL editor): | |
| CREATE TABLE products ( | |
| id UUID PRIMARY KEY DEFAULT gen_random_uuid(), | |
| created_at TIMESTAMPTZ NOT NULL DEFAULT now(), | |
| updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), | |
| title TEXT NOT NULL, | |
| subtitle TEXT, | |
| description TEXT, | |
| bullet_points JSONB DEFAULT '[]'::jsonb, | |
| images JSONB DEFAULT '[]'::jsonb, | |
| category TEXT, | |
| tags JSONB DEFAULT '[]'::jsonb, | |
| aliexpress_id TEXT, | |
| aliexpress_url TEXT, | |
| base_cost NUMERIC(10,2) NOT NULL DEFAULT 0, | |
| shipping_cost NUMERIC(10,2) NOT NULL DEFAULT 0, | |
| current_price NUMERIC(10,2) NOT NULL DEFAULT 0, | |
| markup_multiplier NUMERIC(4,2) NOT NULL DEFAULT 2.5, | |
| demand_multiplier NUMERIC(4,2) NOT NULL DEFAULT 1.0, | |
| ai_multiplier NUMERIC(4,2) NOT NULL DEFAULT 1.0, | |
| stock_status TEXT NOT NULL DEFAULT 'in_stock', | |
| rating NUMERIC(3,2) DEFAULT 0, | |
| total_orders INTEGER DEFAULT 0, | |
| is_active BOOLEAN NOT NULL DEFAULT TRUE, | |
| variants JSONB DEFAULT '[]'::jsonb, | |
| seo_keywords JSONB DEFAULT '[]'::jsonb, | |
| score INTEGER DEFAULT 0 | |
| ); | |
| CREATE TABLE orders ( | |
| id UUID PRIMARY KEY DEFAULT gen_random_uuid(), | |
| created_at TIMESTAMPTZ NOT NULL DEFAULT now(), | |
| updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), | |
| customer_name TEXT NOT NULL, | |
| customer_email TEXT NOT NULL, | |
| shipping_address JSONB, | |
| items JSONB NOT NULL, | |
| subtotal NUMERIC(10,2) NOT NULL, | |
| shipping_total NUMERIC(10,2) NOT NULL DEFAULT 0, | |
| paypal_fee NUMERIC(10,2) NOT NULL DEFAULT 0, | |
| total NUMERIC(10,2) NOT NULL, | |
| status TEXT NOT NULL DEFAULT 'pending', | |
| paypal_order_id TEXT, | |
| tracking_number TEXT, | |
| notes TEXT | |
| ); | |
| CREATE TABLE analytics ( | |
| id BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY, | |
| created_at TIMESTAMPTZ NOT NULL DEFAULT now(), | |
| product_id UUID REFERENCES products(id) ON DELETE SET NULL, | |
| event_type TEXT NOT NULL, | |
| source TEXT DEFAULT 'direct', | |
| metadata JSONB DEFAULT '{}'::jsonb | |
| ); | |
| CREATE TABLE agent_status ( | |
| id UUID PRIMARY KEY DEFAULT gen_random_uuid(), | |
| updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), | |
| agent_name TEXT NOT NULL UNIQUE, | |
| status TEXT NOT NULL DEFAULT 'idle', | |
| last_run TIMESTAMPTZ, | |
| runs_today INTEGER DEFAULT 0, | |
| errors_today INTEGER DEFAULT 0, | |
| metadata JSONB DEFAULT '{}'::jsonb | |
| ); | |
| """ | |
| from __future__ import annotations | |
| import logging | |
| import re | |
| import uuid | |
| from datetime import datetime, timezone | |
| from typing import Any | |
| from config import get_settings | |
| logger = logging.getLogger("memo.db") | |
| # ── Optional Supabase import ───────────────────────────────────── | |
| # Gracefully handle missing supabase package (e.g. during UI-only preview) | |
| _SUPABASE_AVAILABLE = False | |
| _Client = None | |
| _create_client = None | |
| try: | |
| from supabase import create_client as _create_client, Client as _Client | |
| _SUPABASE_AVAILABLE = True | |
| except ImportError: | |
| logger.warning("supabase package not installed — database calls will return empty results") | |
| # ── Singleton client ───────────────────────────────────────────── | |
| _settings = get_settings() | |
| _client: Any = None | |
| def get_supabase() -> Any: | |
| """Return a lazily-initialised Supabase client. | |
| Raises RuntimeError if the supabase package is not installed or | |
| credentials are not configured. | |
| """ | |
| global _client | |
| if not _SUPABASE_AVAILABLE or _create_client is None: | |
| raise RuntimeError("Supabase package not installed — run: pip install supabase") | |
| if _client is None: | |
| _client = _create_client(_settings.supabase_url, _settings.supabase_key) | |
| return _client | |
| def _is_db_available() -> bool: | |
| """Check if Supabase is configured and reachable.""" | |
| if not _SUPABASE_AVAILABLE: | |
| return False | |
| if not _settings.supabase_url or not _settings.supabase_key: | |
| return False | |
| if "mock" in _settings.supabase_url.lower(): | |
| return False | |
| return True | |
| def _now_iso() -> str: | |
| return datetime.now(timezone.utc).isoformat() | |
| # ── Input Validation Guards ────────────────────────────────────── | |
| _UUID_RE = re.compile( | |
| r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$", | |
| re.IGNORECASE, | |
| ) | |
| def _validate_uuid(value: str, field_name: str = "id") -> str: | |
| """Validate that a string is a well-formed UUID. | |
| Raises ValueError if not a valid UUID, preventing injection | |
| through ID fields passed to Supabase query filters. | |
| """ | |
| if not value or not _UUID_RE.match(value): | |
| raise ValueError(f"Invalid {field_name}: expected UUID format") | |
| return value | |
| def _validate_limit(limit: int, max_val: int = 500) -> int: | |
| """Clamp a limit parameter to a safe range.""" | |
| return max(1, min(int(limit), max_val)) | |
| def _sanitize_str(value: str, max_length: int = 1000) -> str: | |
| """Truncate and strip a string field to prevent oversized payloads.""" | |
| return value.strip()[:max_length] if value else value | |
| # ── Products ───────────────────────────────────────────────────── | |
| # ── Allowed sort columns (prevent injection via order_by) ─────── | |
| _ALLOWED_ORDER_COLUMNS = {"created_at", "updated_at", "current_price", "rating", "total_orders", "score", "title"} | |
| def fetch_products( | |
| *, | |
| active_only: bool = True, | |
| category: str | None = None, | |
| limit: int = 50, | |
| order_by: str = "created_at", | |
| ascending: bool = False, | |
| ) -> list[dict[str, Any]]: | |
| """Return a list of products, optionally filtered.""" | |
| if not _is_db_available(): | |
| return [] | |
| # Validate order_by against allowlist to prevent column injection | |
| if order_by not in _ALLOWED_ORDER_COLUMNS: | |
| order_by = "created_at" | |
| limit = _validate_limit(limit) | |
| db = get_supabase() | |
| query = db.table("products").select("*") | |
| if active_only: | |
| query = query.eq("is_active", True) | |
| if category: | |
| query = query.eq("category", _sanitize_str(category, 100)) | |
| query = query.order(order_by, desc=not ascending).limit(limit) | |
| resp = query.execute() | |
| return resp.data or [] | |
| def fetch_product(product_id: str) -> dict[str, Any] | None: | |
| """Return a single product by its UUID.""" | |
| if not _is_db_available(): | |
| return None | |
| _validate_uuid(product_id, "product_id") | |
| db = get_supabase() | |
| resp = db.table("products").select("*").eq("id", product_id).single().execute() | |
| return resp.data | |
| def insert_product(data: dict[str, Any]) -> dict[str, Any]: | |
| """Insert a new product row.""" | |
| if not _is_db_available(): | |
| return data | |
| db = get_supabase() | |
| resp = db.table("products").insert(data).execute() | |
| return resp.data[0] if resp.data else {} | |
| def update_product(product_id: str, data: dict[str, Any]) -> dict[str, Any] | None: | |
| """Update fields on an existing product.""" | |
| if not _is_db_available(): | |
| return data | |
| _validate_uuid(product_id, "product_id") | |
| db = get_supabase() | |
| data["updated_at"] = _now_iso() | |
| resp = db.table("products").update(data).eq("id", product_id).execute() | |
| return resp.data[0] if resp.data else None | |
| def upsert_product(data: dict[str, Any]) -> dict[str, Any]: | |
| """Insert or update a product (matched by aliexpress_id).""" | |
| if not _is_db_available(): | |
| return data | |
| existing = None | |
| if data.get("aliexpress_id"): | |
| db = get_supabase() | |
| resp = ( | |
| db.table("products") | |
| .select("id") | |
| .eq("aliexpress_id", data["aliexpress_id"]) | |
| .limit(1) | |
| .execute() | |
| ) | |
| if resp.data: | |
| existing = resp.data[0] | |
| if existing: | |
| return update_product(existing["id"], data) # type: ignore[arg-type] | |
| return insert_product(data) | |
| def count_products() -> int: | |
| if not _is_db_available(): | |
| return 0 | |
| db = get_supabase() | |
| resp = db.table("products").select("id", count="exact").execute() | |
| return resp.count or 0 # type: ignore[return-value] | |
| # ── Orders ─────────────────────────────────────────────────────── | |
| def fetch_orders(*, limit: int = 100, status: str | None = None) -> list[dict[str, Any]]: | |
| if not _is_db_available(): | |
| return [] | |
| limit = _validate_limit(limit) | |
| db = get_supabase() | |
| query = db.table("orders").select("*") | |
| if status: | |
| query = query.eq("status", _sanitize_str(status, 50)) | |
| query = query.order("created_at", desc=True).limit(limit) | |
| resp = query.execute() | |
| return resp.data or [] | |
| def fetch_order(order_id: str) -> dict[str, Any] | None: | |
| if not _is_db_available(): | |
| return None | |
| _validate_uuid(order_id, "order_id") | |
| db = get_supabase() | |
| resp = db.table("orders").select("*").eq("id", order_id).single().execute() | |
| return resp.data | |
| def insert_order(data: dict[str, Any]) -> dict[str, Any]: | |
| if not _is_db_available(): | |
| return data | |
| db = get_supabase() | |
| resp = db.table("orders").insert(data).execute() | |
| return resp.data[0] if resp.data else {} | |
| def update_order(order_id: str, data: dict[str, Any]) -> dict[str, Any] | None: | |
| if not _is_db_available(): | |
| return data | |
| _validate_uuid(order_id, "order_id") | |
| db = get_supabase() | |
| data["updated_at"] = _now_iso() | |
| resp = db.table("orders").update(data).eq("id", order_id).execute() | |
| return resp.data[0] if resp.data else None | |
| def update_order_by_paypal_id(paypal_id: str, data: dict[str, Any]) -> dict[str, Any] | None: | |
| """Update an order matched by its PayPal order ID.""" | |
| if not _is_db_available(): | |
| return data | |
| db = get_supabase() | |
| data["updated_at"] = _now_iso() | |
| resp = db.table("orders").update(data).eq("paypal_order_id", paypal_id).execute() | |
| return resp.data[0] if resp.data else None | |
| def fetch_order_by_paypal_id(paypal_id: str) -> dict[str, Any] | None: | |
| """Return a single order matched by its PayPal order ID.""" | |
| if not _is_db_available(): | |
| return None | |
| db = get_supabase() | |
| resp = db.table("orders").select("*").eq("paypal_order_id", paypal_id).limit(1).execute() | |
| return resp.data[0] if resp.data else None | |
| def count_orders() -> int: | |
| if not _is_db_available(): | |
| return 0 | |
| db = get_supabase() | |
| resp = db.table("orders").select("id", count="exact").execute() | |
| return resp.count or 0 # type: ignore[return-value] | |
| def total_revenue() -> float: | |
| if not _is_db_available(): | |
| return 0.0 | |
| db = get_supabase() | |
| resp = db.table("orders").select("total").in_("status", ["paid", "shipped", "delivered"]).execute() | |
| return sum(float(o["total"]) for o in (resp.data or [])) | |
| # ── Analytics ──────────────────────────────────────────────────── | |
| def track_event( | |
| product_id: str | None, | |
| event_type: str, | |
| source: str = "direct", | |
| metadata: dict | None = None, | |
| ) -> None: | |
| """Insert a lightweight analytics event.""" | |
| if not _is_db_available(): | |
| return | |
| db = get_supabase() | |
| db.table("analytics").insert( | |
| { | |
| "product_id": product_id, | |
| "event_type": event_type, | |
| "source": source, | |
| "metadata": metadata or {}, | |
| } | |
| ).execute() | |
| def get_product_events(product_id: str, event_type: str, hours: int = 6) -> int: | |
| """Count events for a product within a time window.""" | |
| if not _is_db_available(): | |
| return 0 | |
| _validate_uuid(product_id, "product_id") | |
| db = get_supabase() | |
| resp = ( | |
| db.table("analytics") | |
| .select("id", count="exact") | |
| .eq("product_id", product_id) | |
| .eq("event_type", _sanitize_str(event_type, 50)) | |
| .execute() | |
| ) | |
| return resp.count or 0 # type: ignore[return-value] | |
| def analytics_summary(hours: int = 24) -> dict[str, Any]: | |
| """Aggregate analytics for the admin dashboard.""" | |
| if not _is_db_available(): | |
| return { | |
| "total_events": 0, "views": 0, "carts": 0, "purchases": 0, | |
| "cart_rate": 0, "conversion_rate": 0, "sources": {}, | |
| } | |
| db = get_supabase() | |
| resp = db.table("analytics").select("*").execute() | |
| events = resp.data or [] | |
| views = sum(1 for e in events if e["event_type"] == "view") | |
| carts = sum(1 for e in events if e["event_type"] == "add_to_cart") | |
| purchases = sum(1 for e in events if e["event_type"] == "purchase") | |
| sources: dict[str, int] = {} | |
| for e in events: | |
| src = e.get("source", "direct") | |
| sources[src] = sources.get(src, 0) + 1 | |
| return { | |
| "total_events": len(events), | |
| "views": views, | |
| "carts": carts, | |
| "purchases": purchases, | |
| "cart_rate": round(carts / views * 100, 2) if views else 0, | |
| "conversion_rate": round(purchases / views * 100, 2) if views else 0, | |
| "sources": sources, | |
| } | |
| # ── Agent Status ───────────────────────────────────────────────── | |
| def fetch_agent_statuses() -> list[dict[str, Any]]: | |
| if not _is_db_available(): | |
| return [] | |
| db = get_supabase() | |
| resp = db.table("agent_status").select("*").order("agent_name").execute() | |
| return resp.data or [] | |
| def update_agent_status(agent_name: str, status: str, metadata: dict | None = None) -> None: | |
| if not _is_db_available(): | |
| return | |
| db = get_supabase() | |
| db.table("agent_status").upsert( | |
| { | |
| "agent_name": agent_name, | |
| "status": status, | |
| "updated_at": _now_iso(), | |
| "last_run": _now_iso(), | |
| "metadata": metadata or {}, | |
| }, | |
| on_conflict="agent_name", | |
| ).execute() | |
| # ── Recent Purchases (Social Proof) ────────────────────────────── | |
| _SIMULATED_CITIES = [ | |
| "New York", "London", "Paris", "Berlin", "Tokyo", "Sydney", "Toronto", | |
| "Amsterdam", "Stockholm", "Dubai", "Singapore", "Los Angeles", "Madrid", | |
| "Rome", "Vienna", "Copenhagen", "Munich", "Barcelona", "Melbourne", "Zurich", | |
| ] | |
| _SIMULATED_FIRST_NAMES = [ | |
| "Sarah", "James", "Emma", "Michael", "Olivia", "William", "Ava", "Oliver", | |
| "Sophia", "Elijah", "Isabella", "Lucas", "Mia", "Mason", "Charlotte", "Ethan", | |
| "Amelia", "Logan", "Harper", "Benjamin", "Evelyn", "Alexander", "Abigail", "Daniel", | |
| ] | |
| def fetch_recent_purchases(*, limit: int = 10, fallback_to_simulated: bool = True) -> list[dict[str, Any]]: | |
| """Return recent successful orders formatted for social proof notifications. | |
| If no real paid orders exist and fallback_to_simulated is True, | |
| generates realistic simulated purchases from active products. | |
| """ | |
| if not _is_db_available(): | |
| return [] | |
| db = get_supabase() | |
| resp = ( | |
| db.table("orders") | |
| .select("*") | |
| .in_("status", ["paid", "shipped", "delivered"]) | |
| .order("created_at", desc=True) | |
| .limit(_validate_limit(limit, max_val=50)) | |
| .execute() | |
| ) | |
| orders = resp.data or [] | |
| purchases: list[dict[str, Any]] = [] | |
| for order in orders: | |
| items = order.get("items", []) | |
| if not items: | |
| continue | |
| # Use first item for the notification | |
| item = items[0] | |
| # Anonymize name | |
| raw_name = order.get("customer_name", "Someone") | |
| first_name = raw_name.split()[0] if raw_name and raw_name != "Guest" else "Someone" | |
| if first_name in {"Guest", "guest", ""}: | |
| first_name = "Someone" | |
| purchases.append({ | |
| "name": first_name, | |
| "product_title": item.get("title", "a product"), | |
| "location": "", | |
| "time_ago": "recently", | |
| "image": "", | |
| "is_simulated": False, | |
| }) | |
| if purchases or not fallback_to_simulated: | |
| return purchases[:limit] | |
| # ── Fallback: generate simulated purchases from active products ─ | |
| try: | |
| product_resp = db.table("products").select("title, images, current_price").eq("is_active", True).limit(limit * 2).execute() | |
| products = product_resp.data or [] | |
| except Exception: | |
| products = [] | |
| import random | |
| rng = random.Random(42) # deterministic seed for stability | |
| rng.shuffle(products) | |
| for i, product in enumerate(products[:limit]): | |
| images = product.get("images", []) if isinstance(product.get("images"), list) else [] | |
| purchases.append({ | |
| "name": rng.choice(_SIMULATED_FIRST_NAMES), | |
| "product_title": product.get("title", "a product"), | |
| "location": rng.choice(_SIMULATED_CITIES), | |
| "time_ago": f"{rng.randint(1, 45)} minutes ago", | |
| "image": images[0] if images else "", | |
| "is_simulated": True, | |
| }) | |
| return purchases | |