Upload supabase_client.py

supabase_client.py

@@ -0,0 +1,264 @@
+"""
+REST client for Supabase/PostgREST.
+Uses the service key directly and avoids the Python SDK dependency tree.
+"""
+
+from __future__ import annotations
+
+from typing import Dict, List, Optional
+import logging
+
+import httpx
+
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+
+class SupabaseClient:
+    """Minimal async client for the `fipi_tasks` table."""
+
+    ALLOWED_TASK_FIELDS = {
+        "title",
+        "content",
+        "source_url",
+        "task_type",
+        "images",
+        "variants",
+        "task_number",
+        "source_kind",
+        "task_guid",
+        "rubert_analysis",
+        "scraped_at",
+    }
+
+    def __init__(self, url: str, key: str):
+        self.base_url = f"{url.rstrip('/')}/rest/v1"
+        self.table_name = "fipi_tasks"
+        self.headers = {
+            "apikey": key,
+            "Authorization": f"Bearer {key}",
+            "Content-Type": "application/json",
+        }
+
+    async def _request(
+        self,
+        method: str,
+        path: str,
+        *,
+        params: Optional[Dict[str, str]] = None,
+        json: Optional[Dict | List[Dict]] = None,
+        prefer: Optional[str] = None,
+    ) -> List[Dict]:
+        headers = dict(self.headers)
+        if prefer:
+            headers["Prefer"] = prefer
+
+        async with httpx.AsyncClient(timeout=30.0) as client:
+            response = await client.request(
+                method,
+                f"{self.base_url}/{path}",
+                params=params,
+                json=json,
+                headers=headers,
+            )
+            response.raise_for_status()
+
+        if not response.content:
+            return []
+
+        data = response.json()
+        return data if isinstance(data, list) else [data]
+
+    def _prepare_task_payload(self, task: Dict) -> Dict:
+        payload = {
+            key: value
+            for key, value in task.items()
+            if key in self.ALLOWED_TASK_FIELDS
+        }
+        payload.setdefault("task_type", "other")
+        payload.setdefault("images", [])
+        payload.setdefault("variants", [])
+        return payload
+
+    def _has_enrichment_changes(self, existing: Dict, incoming: Dict) -> bool:
+        for field in ("task_number", "source_kind", "task_guid"):
+            if incoming.get(field) and not existing.get(field):
+                return True
+
+        incoming_variants = incoming.get("variants") or []
+        existing_variants = existing.get("variants") or []
+        if incoming_variants and not existing_variants:
+            return True
+
+        incoming_images = incoming.get("images") or []
+        existing_images = existing.get("images") or []
+        if incoming_images and not existing_images:
+            return True
+
+        return False
+
+    async def is_available(self) -> bool:
+        try:
+            await self._request(
+                "GET",
+                self.table_name,
+                params={"select": "id", "limit": "1"},
+            )
+            return True
+        except Exception as e:
+            logger.error("Supabase availability check failed: %s", e)
+            return False
+
+    async def insert_task(self, task: Dict) -> Optional[Dict]:
+        try:
+            existing = await self.get_task_by_url(task.get("source_url", ""))
+            if existing:
+                if self._has_enrichment_changes(existing, task):
+                    logger.info("Updating existing task metadata: %s", task.get("source_url"))
+                    return await self.update_task(existing["id"], task)
+
+                logger.info("Task already exists: %s", task.get("source_url"))
+                return None
+
+            result = await self._request(
+                "POST",
+                self.table_name,
+                json=self._prepare_task_payload(task),
+                prefer="return=representation",
+            )
+            return result[0] if result else None
+        except httpx.HTTPStatusError as e:
+            detail = e.response.text if e.response is not None else str(e)
+            logger.error("Error inserting task: %s", detail)
+            return None
+        except Exception as e:
+            logger.error("Error inserting task: %s", e)
+            return None
+
+    async def insert_tasks_batch(self, tasks: List[Dict]) -> List[Dict]:
+        saved = []
+        for task in tasks:
+            result = await self.insert_task(task)
+            if result:
+                saved.append(result)
+        logger.info("Saved %s of %s tasks", len(saved), len(tasks))
+        return saved
+
+    async def get_task_by_id(self, task_id: int) -> Optional[Dict]:
+        try:
+            result = await self._request(
+                "GET",
+                self.table_name,
+                params={"select": "*", "id": f"eq.{task_id}"},
+            )
+            return result[0] if result else None
+        except Exception as e:
+            logger.error("Error getting task by id: %s", e)
+            return None
+
+    async def get_task_by_url(self, url: str) -> Optional[Dict]:
+        if not url:
+            return None
+        try:
+            result = await self._request(
+                "GET",
+                self.table_name,
+                params={"select": "*", "source_url": f"eq.{url}"},
+            )
+            return result[0] if result else None
+        except Exception as e:
+            logger.error("Error getting task by url: %s", e)
+            return None
+
+    async def get_latest_tasks(self, limit: int = 10) -> List[Dict]:
+        try:
+            return await self._request(
+                "GET",
+                self.table_name,
+                params={
+                    "select": "*",
+                    "order": "scraped_at.desc",
+                    "limit": str(limit),
+                },
+            )
+        except Exception as e:
+            logger.error("Error getting latest tasks: %s", e)
+            return []
+
+    async def get_all_tasks(self) -> List[Dict]:
+        try:
+            return await self._request(
+                "GET",
+                self.table_name,
+                params={"select": "*", "order": "scraped_at.desc"},
+            )
+        except Exception as e:
+            logger.error("Error getting all tasks: %s", e)
+            return []
+
+    async def search_tasks(self, query: str) -> List[Dict]:
+        try:
+            escaped_query = query.replace(",", " ").replace("(", " ").replace(")", " ")
+            pattern = f"*{escaped_query}*"
+            or_filter = f"(title.ilike.{pattern},content.ilike.{pattern})"
+            return await self._request(
+                "GET",
+                self.table_name,
+                params={
+                    "select": "*",
+                    "or": or_filter,
+                },
+            )
+        except Exception as e:
+            logger.error("Error searching tasks: %s", e)
+            return []
+
+    async def get_tasks_by_type(self, task_type: str) -> List[Dict]:
+        try:
+            return await self._request(
+                "GET",
+                self.table_name,
+                params={"select": "*", "task_type": f"eq.{task_type}"},
+            )
+        except Exception as e:
+            logger.error("Error getting tasks by type: %s", e)
+            return []
+
+    async def update_task(self, task_id: int, updates: Dict) -> Optional[Dict]:
+        try:
+            result = await self._request(
+                "PATCH",
+                self.table_name,
+                params={"id": f"eq.{task_id}"},
+                json=self._prepare_task_payload(updates),
+                prefer="return=representation",
+            )
+            return result[0] if result else None
+        except Exception as e:
+            logger.error("Error updating task: %s", e)
+            return None
+
+    async def delete_task(self, task_id: int) -> bool:
+        try:
+            await self._request(
+                "DELETE",
+                self.table_name,
+                params={"id": f"eq.{task_id}"},
+                prefer="return=representation",
+            )
+            return True
+        except Exception as e:
+            logger.error("Error deleting task: %s", e)
+            return False
+
+    async def get_stats(self) -> Dict:
+        try:
+            all_tasks = await self.get_all_tasks()
+            stats = {"total": len(all_tasks), "by_type": {}}
+            for task in all_tasks:
+                task_type = task.get("task_type", "unknown")
+                stats["by_type"][task_type] = stats["by_type"].get(task_type, 0) + 1
+            return stats
+        except Exception as e:
+            logger.error("Error getting stats: %s", e)
+            return {"total": 0, "by_type": {}}