File size: 5,036 Bytes
325b94c | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 | # 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}")
|