| """Private project-document extraction and optional Qdrant vector publishing.""" |
|
|
| from __future__ import annotations |
|
|
| import hashlib |
| import json |
| import os |
| import re |
| import uuid |
| from pathlib import Path |
| from urllib.parse import quote |
|
|
| import numpy as np |
| import requests |
|
|
| from document_text import extract_document |
| from project_store import ProjectStore |
|
|
|
|
| class KnowledgeServiceError(Exception): |
| pass |
|
|
|
|
| def _chunks(text: str, *, size: int = 1_400, overlap: int = 180, limit: int = 64) -> list[str]: |
| paragraphs = [re.sub(r"\s+", " ", value).strip() for value in re.split(r"\n\s*\n", text) if value.strip()] |
| chunks: list[str] = [] |
| current = "" |
| for paragraph in paragraphs: |
| pending = paragraph |
| while pending: |
| room = size - len(current) |
| if room <= 80: |
| chunks.append(current.strip()) |
| current = current[-overlap:].lstrip() |
| room = size - len(current) |
| take = pending[:room] |
| split = take.rfind(" ") if len(pending) > room else len(take) |
| if split < max(80, room // 2): |
| split = len(take) |
| current = (current + " " + pending[:split]).strip() |
| pending = pending[split:].lstrip() |
| if len(chunks) >= limit: |
| return chunks[:limit] |
| if current and len(chunks) < limit: |
| chunks.append(current.strip()) |
| return chunks[:limit] |
|
|
|
|
| class KnowledgeService: |
| def __init__(self, projects: ProjectStore, corpus): |
| self.projects = projects |
| self.corpus = corpus |
| self.qdrant_url = os.environ.get("QDRANT_URL", "").rstrip("/") |
| self.qdrant_key = os.environ.get("QDRANT_API_KEY", "") |
| self.collection = os.environ.get("QDRANT_KNOWLEDGE_COLLECTION", "moonley_tenant_knowledge") |
| self.supabase_url = os.environ.get("SUPABASE_URL", "").rstrip("/") |
| self.supabase_key = os.environ.get("SUPABASE_SERVICE_ROLE_KEY", "") |
| self.supabase_bucket = os.environ.get("SUPABASE_STORAGE_BUCKET", "") |
|
|
| @property |
| def qdrant_configured(self) -> bool: |
| return bool(self.qdrant_url and self.qdrant_key) |
|
|
| @property |
| def supabase_configured(self) -> bool: |
| return bool(self.supabase_url and self.supabase_key and self.supabase_bucket) |
|
|
| def status(self) -> dict: |
| return { |
| "source_provider": "supabase_private" if self.supabase_configured else ("mounted_volume" if self.projects.configured else "unconfigured"), |
| "vector_provider": "qdrant" if self.qdrant_configured else "private_mounted_volume", |
| "qdrant_configured": self.qdrant_configured, |
| "supabase_configured": self.supabase_configured, |
| "ocr": "tesseract", |
| "chat_embedding_default": False, |
| } |
|
|
| @staticmethod |
| def _owner_key(owner_id: str) -> str: |
| return hashlib.sha256(owner_id.encode("utf-8")).hexdigest() |
|
|
| def source_text(self, owner_id: str, project_id: str, document_id: str) -> tuple[str, dict]: |
| knowledge_dir = self.projects.knowledge_dir(owner_id, project_id, document_id) |
| cache_path = knowledge_dir / "content.json" |
| if cache_path.exists(): |
| payload = json.loads(cache_path.read_text(encoding="utf-8")) |
| return str(payload.get("text") or ""), dict(payload.get("extraction") or {}) |
| document = self.projects.document_record(owner_id, project_id, document_id) |
| result = extract_document( |
| self.projects.document_path(owner_id, project_id, document_id), |
| str(document.get("media_type") or ""), |
| ) |
| extraction = result.public_dict() |
| temp = knowledge_dir / f".content-{uuid.uuid4().hex}.tmp" |
| temp.write_text( |
| json.dumps({"version": 1, "text": result.text, "extraction": extraction}, ensure_ascii=False), |
| encoding="utf-8", |
| ) |
| os.replace(temp, cache_path) |
| self.projects.record_extraction(owner_id, project_id, document_id, extraction, status="extracted") |
| return result.text, extraction |
|
|
| def ingest(self, owner_id: str, project_id: str, document_id: str) -> None: |
| try: |
| if self.supabase_configured: |
| self._publish_supabase(owner_id, project_id, document_id) |
| text, extraction = self.source_text(owner_id, project_id, document_id) |
| chunks = _chunks(text) |
| if not chunks: |
| raise ValueError("No readable text was found in the document.") |
| vectors = self.corpus.encode_documents(chunks) |
| knowledge_dir = self.projects.knowledge_dir(owner_id, project_id, document_id) |
| chunk_payload = { |
| "version": 1, |
| "chunks": [{"id": index, "text": value} for index, value in enumerate(chunks)], |
| } |
| chunks_temp = knowledge_dir / f".chunks-{uuid.uuid4().hex}.tmp" |
| chunks_temp.write_text(json.dumps(chunk_payload, ensure_ascii=False), encoding="utf-8") |
| os.replace(chunks_temp, knowledge_dir / "chunks.json") |
| provider = "qdrant" if self.qdrant_configured else "private_mounted_volume" |
| if self.qdrant_configured: |
| self._publish_qdrant(owner_id, project_id, document_id, vectors) |
| else: |
| vector_temp = knowledge_dir / f".vectors-{uuid.uuid4().hex}.npy" |
| np.save(vector_temp, vectors) |
| os.replace(vector_temp, knowledge_dir / "vectors.npy") |
| self.projects.record_extraction( |
| owner_id, |
| project_id, |
| document_id, |
| { |
| **extraction, |
| "chunk_count": len(chunks), |
| "vector_provider": provider, |
| "source_provider": "supabase_private" if self.supabase_configured else "mounted_volume", |
| }, |
| status="ready", |
| ) |
| except Exception as exc: |
| try: |
| self.projects.record_extraction( |
| owner_id, |
| project_id, |
| document_id, |
| {"method": "failed", "text_chars": 0}, |
| status="failed", |
| ) |
| except Exception: |
| pass |
| print(f"[knowledge] ingestion failed document={document_id}: {type(exc).__name__}", flush=True) |
|
|
| def _supabase_headers(self, media_type: str | None = None) -> dict[str, str]: |
| headers = { |
| "apikey": self.supabase_key, |
| "Authorization": f"Bearer {self.supabase_key}", |
| "x-upsert": "true", |
| } |
| if media_type: |
| headers["Content-Type"] = media_type |
| return headers |
|
|
| def _source_object_path(self, owner_id: str, project_id: str, document: dict) -> str: |
| suffix = Path(str(document.get("stored_name") or "")).suffix.lower() |
| return f"users/{self._owner_key(owner_id)}/projects/{project_id}/documents/{document['id']}{suffix}" |
|
|
| def _publish_supabase(self, owner_id: str, project_id: str, document_id: str) -> None: |
| document = self.projects.document_record(owner_id, project_id, document_id) |
| object_path = self._source_object_path(owner_id, project_id, document) |
| endpoint = ( |
| f"{self.supabase_url}/storage/v1/object/{quote(self.supabase_bucket, safe='')}/" |
| f"{quote(object_path, safe='/')}" |
| ) |
| response = requests.post( |
| endpoint, |
| headers=self._supabase_headers(str(document.get("media_type") or "application/octet-stream")), |
| data=self.projects.document_path(owner_id, project_id, document_id).read_bytes(), |
| timeout=90, |
| ) |
| if response.status_code not in {200, 201}: |
| raise KnowledgeServiceError(f"Private Supabase upload failed ({response.status_code}).") |
|
|
| def delete(self, owner_id: str, project_id: str, document_id: str) -> None: |
| document = self.projects.document_record(owner_id, project_id, document_id) |
| if self.qdrant_configured: |
| tenant_id = self._owner_key(owner_id) |
| response = requests.post( |
| f"{self.qdrant_url}/collections/{self.collection}/points/delete?wait=true", |
| headers=self._headers(), |
| timeout=45, |
| json={ |
| "filter": { |
| "must": [ |
| {"key": "tenant_id", "match": {"value": tenant_id}}, |
| {"key": "project_id", "match": {"value": project_id}}, |
| {"key": "document_id", "match": {"value": document_id}}, |
| ] |
| } |
| }, |
| ) |
| if response.status_code not in {200, 404}: |
| raise KnowledgeServiceError(f"Qdrant deletion failed ({response.status_code}).") |
| if self.supabase_configured: |
| object_path = self._source_object_path(owner_id, project_id, document) |
| endpoint = ( |
| f"{self.supabase_url}/storage/v1/object/{quote(self.supabase_bucket, safe='')}/" |
| f"{quote(object_path, safe='/')}" |
| ) |
| response = requests.delete(endpoint, headers=self._supabase_headers(), timeout=45) |
| if response.status_code not in {200, 404}: |
| raise KnowledgeServiceError(f"Private Supabase deletion failed ({response.status_code}).") |
|
|
| def delete_project(self, owner_id: str, project_id: str) -> None: |
| project = self.projects.get_project(owner_id, project_id) |
| for document in project.get("documents") or []: |
| self.delete(owner_id, project_id, str(document.get("id") or "")) |
|
|
| def _headers(self) -> dict[str, str]: |
| return {"api-key": self.qdrant_key, "Content-Type": "application/json"} |
|
|
| def _publish_qdrant(self, owner_id: str, project_id: str, document_id: str, vectors: np.ndarray) -> None: |
| collection_url = f"{self.qdrant_url}/collections/{self.collection}" |
| response = requests.get(collection_url, headers=self._headers(), timeout=15) |
| if response.status_code == 404: |
| response = requests.put( |
| collection_url, |
| headers=self._headers(), |
| timeout=30, |
| json={"vectors": {"size": int(vectors.shape[1]), "distance": "Cosine"}, "on_disk_payload": True}, |
| ) |
| response.raise_for_status() |
| tenant_id = self._owner_key(owner_id) |
| namespace = uuid.UUID("db853e8b-aeb1-47c8-a7fc-680c662ba8ee") |
| points = [ |
| { |
| "id": str(uuid.uuid5(namespace, f"{tenant_id}:{project_id}:{document_id}:{index}")), |
| "vector": vector.tolist(), |
| "payload": { |
| "tenant_id": tenant_id, |
| "project_id": project_id, |
| "document_id": document_id, |
| "chunk_id": index, |
| }, |
| } |
| for index, vector in enumerate(vectors) |
| ] |
| for start in range(0, len(points), 32): |
| result = requests.put( |
| f"{collection_url}/points?wait=true", |
| headers=self._headers(), |
| timeout=90, |
| json={"points": points[start:start + 32]}, |
| ) |
| result.raise_for_status() |
|
|
