# core/books/storage.py import os from supabase import create_client, Client from typing import List, Optional, Dict, Any from schemas.books.sources_schema import DocRaw, DocMetadata SUPABASE_URL = os.getenv("SUPABASE_URL") SUPABASE_SERVICE_ROLE_KEY = os.getenv("SUPABASE_SERVICE_ROLE_KEY") supabase: Client = create_client(SUPABASE_URL, SUPABASE_SERVICE_ROLE_KEY) def init_db(): return def check_db_health() -> bool: try: supabase.table("documents_raw").select("doc_id").limit(1).execute() return True except Exception: return False # ------------------------- # RAW (Phase 1) # ------------------------- def insert_raw_document(raw: DocRaw) -> None: data = raw.model_dump() resp = supabase.table("documents_raw").insert(data).execute() if getattr(resp, "error", None): raise RuntimeError(f"Supabase insert documents_raw failed: {resp.error}") def mark_raw_status(doc_id: str, status: str, error_reason: str = "") -> None: resp = ( supabase.table("documents_raw") .update({"status": status, "error_reason": error_reason}) .eq("doc_id", doc_id) .execute() ) if getattr(resp, "error", None): raise RuntimeError(f"Supabase update documents_raw failed: {resp.error}") def fetch_pending_raw_docs(limit: int = 25) -> List[Dict[str, Any]]: resp = ( supabase.table("documents_raw") .select("*") .eq("status", "pending") .limit(limit) .execute() ) if getattr(resp, "error", None): raise RuntimeError(f"Supabase select documents_raw failed: {resp.error}") return resp.data or [] def fetch_raw_doc(doc_id: str) -> Optional[Dict[str, Any]]: resp = ( supabase.table("documents_raw") .select("*") .eq("doc_id", doc_id) .limit(1) .execute() ) if getattr(resp, "error", None): raise RuntimeError(f"Supabase select documents_raw failed: {resp.error}") data = resp.data or [] return data[0] if data else None def fetch_raw_docs_for_user( user_id: str, book_id: str | None = None, status: str = "pending" ) -> List[Dict[str, Any]]: q = ( supabase.table("documents_raw") .select("*") .eq("user_id", user_id) .eq("status", status) ) if book_id: q = q.eq("book_id", book_id) resp = q.execute() if getattr(resp, "error", None): raise RuntimeError(f"Supabase select documents_raw failed: {resp.error}") return resp.data or [] # ------------------------- # METADATA (Phase 2) # ------------------------- def upsert_document_metadata(md: DocMetadata) -> None: data = { "doc_id": md.doc_id, "title": md.title, "authors": ", ".join(md.authors), "year": md.year, "publisher_or_journal": md.publisher_or_journal or "", "normalized_source_type": md.normalized_source_type or "pdf", "apa7": md.apa7 or "", "metadata": md.metadata or {}, } resp = ( supabase.table("documents_metadata") .upsert(data, on_conflict="doc_id") .execute() ) if getattr(resp, "error", None): raise RuntimeError(f"Supabase upsert documents_metadata failed: {resp.error}") # ------------------------- # METADATA (Phase 2) - READ # ------------------------- def fetch_document_metadata(doc_id: str) -> Optional[Dict[str, Any]]: """ Fetch processed metadata for a document. """ resp = ( supabase.table("documents_metadata") .select("*") .eq("doc_id", doc_id) .limit(1) .execute() ) if getattr(resp, "error", None): raise RuntimeError(f"Supabase select documents_metadata failed: {resp.error}") data = resp.data or [] return data[0] if data else None def delete_book_docs(user_id: str, book_id: str): # يمسح raw وبالتالي metadata تتشال cascade resp = ( supabase.table("documents_raw") .delete() .eq("user_id", user_id) .eq("book_id", book_id) .execute() ) if getattr(resp, "error", None): raise RuntimeError(f"Supabase delete documents_raw failed: {resp.error}") def delete_raw_doc(doc_id: str) -> None: resp = supabase.table("documents_raw").delete().eq("doc_id", doc_id).execute() if getattr(resp, "error", None): raise RuntimeError(f"Supabase delete documents_raw failed: {resp.error}") def delete_metadata(doc_id: str) -> None: resp = supabase.table("documents_metadata").delete().eq("doc_id", doc_id).execute() if getattr(resp, "error", None): raise RuntimeError(f"Supabase delete documents_metadata failed: {resp.error}") def delete_chunks(doc_id: str) -> None: resp = ( supabase.table("chunks") # أو اسم التابل عندك .delete() .eq("doc_id", doc_id) .execute() ) if getattr(resp, "error", None): raise RuntimeError(f"Supabase delete chunks failed: {resp.error}")