feat: unified SSOT architecture and scaled vector database ingestion

ingestion/ocr_processor_runner.py

@@ -2,19 +2,15 @@ import os
 import sys
 import hashlib
 import httpx
-from supabase import create_client
 
 sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
 from ingestion.ocr_processor import download_and_extract
 
-SUPABASE_URL = os.environ["SUPABASE_URL"]
-SUPABASE_KEY = os.environ["SUPABASE_KEY"]
+# --- SSOT INGESTION CONFIGURATION ---
 HF_SPACE_URL = os.environ.get("HF_SPACE_URL", "https://harshrawat18-govbridge-api.hf.space")
 ADMIN_SECRET = os.environ.get("ADMIN_SECRET", "")
 PDF_URL = os.environ.get("PDF_URL", "")
 
-supabase = create_client(SUPABASE_URL, SUPABASE_KEY)
-
 def chunk_text(text: str, chunk_size: int = 300) -> list:
     """Split text into word-based chunks with overlap."""
     words = text.split()
@@ -27,8 +23,8 @@ def chunk_text(text: str, chunk_size: int = 300) -> list:
             chunks.append(chunk)
     return chunks
 
-def ingest_to_supabase(chunks: list, metadata: dict):
-    """Send chunks to the HF Space ingest endpoint."""
+def ingest_to_hf(chunks: list, metadata: dict):
+    """Send chunks straight to the AI Vector Database (HF Space)."""
     for i, chunk in enumerate(chunks):
         content_hash = hashlib.sha256(
             f"{metadata['source_url']}_{i}".encode()
@@ -39,8 +35,8 @@ def ingest_to_supabase(chunks: list, metadata: dict):
             "text": chunk,
             "source_url": metadata["source_url"],
             "ministry": metadata.get("ministry", "Government of India"),
-            "state": metadata.get("state", "Central"),
-            "doc_type": metadata.get("doc_type", "gazette"),
+            "state": metadata.get("state", "National"),
+            "doc_type": metadata.get("doc_type", "document"),
             "content_hash": content_hash,
             "chunk_index": i
         }
@@ -54,7 +50,7 @@ def ingest_to_supabase(chunks: list, metadata: dict):
                 timeout=30.0
             )
             if response.status_code == 200:
-                print(f"✅ Chunk {i+1}/{len(chunks)} ingested")
+                print(f"✅ Chunk {i+1}/{len(chunks)} injected into AI Brain")
             else:
                 print(f"⚠️ Chunk {i+1} failed: {response.status_code} {response.text}")
         except Exception as e:
@@ -62,34 +58,38 @@ def ingest_to_supabase(chunks: list, metadata: dict):
 
 def main():
     if not PDF_URL:
-        print("❌ No PDF_URL provided")
+        print("❌ No PDF_URL provided in terminal")
+        sys.exit(1)
+
+    if not ADMIN_SECRET:
+        print("❌ 🔒 SECURITY HALT: ADMIN_SECRET environment variable is missing!")
         sys.exit(1)
 
-    print(f"🚀 Starting PDF extraction: {PDF_URL}")
+    print(f"🚀 Starting PDF memory extraction: {PDF_URL}")
 
     try:
         result = download_and_extract(PDF_URL)
         text = result["text"]
         word_count = len(text.split())
-        print(f"✅ Extracted {word_count} words via {result['extraction_method']}")
-        print(f"📄 Preview: {text[:200]}...")
-
+        print(f"✅ Extracted {word_count} words in memory via {result['extraction_method']}")
+        
         chunks = chunk_text(text, chunk_size=300)
-        print(f"📦 Created {len(chunks)} chunks")
+        print(f"📦 Sliced into {len(chunks)} high-density AI chunks")
 
-        ingest_to_supabase(chunks, {
+        # Dynamically injecting NEP Metadata for this specific run
+        ingest_to_hf(chunks, {
             "source_url": PDF_URL,
-            "title": "Government PDF Document",
-            "ministry": "Government of India",
-            "state": "Central",
-            "doc_type": "gazette"
+            "title": "National Education Policy 2020",
+            "ministry": "Ministry of Education",
+            "state": "National",
+            "doc_type": "document"
         })
 
-        print(f"🎉 Done. {len(chunks)} chunks ingested to Supabase.")
+        print(f"🎉 MASSIVE W! {len(chunks)} chunks ingested to Vector DB.")
 
     except Exception as e:
         print(f"❌ Critical failure: {e}")
         sys.exit(1)
 
 if __name__ == "__main__":
-    main()
+    main()

scraper.py

@@ -2,20 +2,17 @@ import os
 import asyncio
 import hashlib
 import random
+import httpx
 from playwright.async_api import async_playwright
-from supabase import create_client, Client
 import logging
 
 # Setup logging
 logging.basicConfig(level=logging.INFO, format='%(levelname)s: %(message)s')
 logger = logging.getLogger(__name__)
 
-# --- 1. SETUP YOUR SUPABASE VAULT HERE ---
-SUPABASE_URL = os.environ.get("SUPABASE_URL")
-SUPABASE_KEY = os.environ.get("SUPABASE_KEY")
-
-# Connect to the database
-supabase: Client = create_client(SUPABASE_URL, SUPABASE_KEY)
+# --- UNIFIED INGESTION CONFIGURATION (OPTION A) ---
+HF_SPACE_URL = os.environ.get("HF_SPACE_URL", "https://harshrawat18-govbridge-api.hf.space")
+ADMIN_SECRET = os.environ.get("ADMIN_SECRET", "") 
 
 # --- STEALTH SCRIPT COLLECTION ---
 STEALTH_SCRIPTS = [
@@ -148,26 +145,56 @@ GOVERNMENT_SITES = [
         "url": "https://www.myscheme.gov.in/search",
         "source": "MyScheme Portal",
         "selector": "h2",
-        "extract_type": "myscheme"
+        "extract_type": "myscheme",
+        "doc_type": "scheme"
     },
     {
         "name": "PM-KISAN Scheme",
         "url": "https://pmkisan.gov.in/",
         "source": "Ministry of Agriculture",
         "selector": "h2, h3, div.content",
-        "extract_type": "generic"
+        "extract_type": "generic",
+        "doc_type": "scheme"
     },
-    # Truncated list for efficiency; add additional sites as needed
 ]
 
+async def push_to_hf_api(title: str, text: str, source_url: str, ministry: str, doc_type: str):
+    """Pushes scraped data directly to the Hugging Face Ingestion Engine"""
+    payload = {
+        "title": title,
+        "text": text,
+        "ministry": ministry,
+        "state": "National", 
+        "source_url": source_url,
+        "doc_type": doc_type
+    }
+    
+    async with httpx.AsyncClient() as client:
+        try:
+            resp = await client.post(
+                f"{HF_SPACE_URL}/api/admin/ingest",
+                json=payload,
+                params={"admin_key": ADMIN_SECRET},
+                headers={"X-Admin-Secret": ADMIN_SECRET},
+                timeout=30.0
+            )
+            if resp.status_code == 200:
+                logger.info(f"✅ Ingested to Vector DB: {title[:40]}...")
+                return True
+            else:
+                logger.error(f"❌ HF API Error: {resp.status_code} - {resp.text}")
+                return False
+        except Exception as e:
+            logger.error(f"⚠️ Network Error pushing {title[:30]}: {e}")
+            return False
+
 async def scrape_myscheme(page, site_config):
     """Specialized scraper for MyScheme portal"""
     try:
         logger.info(f"🎯 Scraping {site_config['name']}...")
-        await asyncio.sleep(random.uniform(1.5, 3.5)) # Human-like delay
-        await page.goto(site_config["url"], timeout=30000, wait_until="networkidle")
+        await asyncio.sleep(random.uniform(1.5, 3.5)) 
+        await page.goto(site_config["url"], timeout=60000, wait_until="domcontentloaded")
         
-        # Human-like scrolling
         await page.evaluate("window.scrollTo(0, document.body.scrollHeight / 3)")
         await asyncio.sleep(random.uniform(0.8, 1.8))
         
@@ -178,25 +205,18 @@ async def scrape_myscheme(page, site_config):
         if len(schemes) >= 2:
             for i in range(0, len(schemes) - 1, 2):
                 try:
-                    inner_text = schemes[i]
-                    content_hash = hashlib.sha256(inner_text.encode('utf-8')).hexdigest()
+                    title = schemes[i]
+                    department = schemes[i + 1] if i + 1 < len(schemes) else "Unknown"
                     
-                    existing = supabase.table("official_documents").select("id").eq("content_hash", content_hash).execute()
-                    if existing.data:
-                        print('Duplicate detected, skipping')
-                        continue
-
-                    data_to_save = {
-                        "title": inner_text,
-                        "department": schemes[i + 1] if i + 1 < len(schemes) else "Unknown",
-                        "document_type": "Government Scheme",
-                        "file_url": site_config["url"],
-                        "authentic_source_url": site_config["source"],
-                        "content_hash": content_hash
-                    }
-                    supabase.table("official_documents").insert(data_to_save).execute()
-                    logger.info(f"✅ Uploaded: {schemes[i]}")
-                    results.append(True)
+                    # Push directly to HF Engine
+                    success = await push_to_hf_api(
+                        title=title,
+                        text=f"Details regarding {title} provided by {department}.",
+                        source_url=site_config["url"],
+                        ministry=department,
+                        doc_type=site_config["doc_type"]
+                    )
+                    results.append(success)
                 except Exception as e:
                     logger.error(f"❌ Failed to upload {schemes[i]}: {str(e)}")
                     results.append(False)
@@ -211,7 +231,7 @@ async def scrape_generic(page, site_config):
     try:
         logger.info(f"🎯 Scraping {site_config['name']}...")
         await asyncio.sleep(random.uniform(1.5, 3.5))
-        await page.goto(site_config["url"], timeout=30000, wait_until="networkidle")
+        await page.goto(site_config["url"], timeout=60000, wait_until="domcontentloaded")
         
         await page.evaluate("window.scrollTo(0, document.body.scrollHeight / 3)")
         await asyncio.sleep(random.uniform(0.8, 1.8))
@@ -229,25 +249,17 @@ async def scrape_generic(page, site_config):
         if all_text:
             for i, text in enumerate(all_text[:10]):
                 try:
-                    content_hash = hashlib.sha256(text.encode('utf-8')).hexdigest()
-                    existing = supabase.table("official_documents").select("id").eq("content_hash", content_hash).execute()
-                    if existing.data:
-                        print('Duplicate detected, skipping')
-                        continue
-
-                    data_to_save = {
-                        "title": text,
-                        "department": site_config["source"],
-                        "document_type": "Government Content",
-                        "file_url": site_config["url"],
-                        "authentic_source_url": site_config["source"],
-                        "content_hash": content_hash
-                    }
-                    supabase.table("official_documents").insert(data_to_save).execute()
-                    logger.info(f"✅ Uploaded from {site_config['name']}: {text[:50]}...")
-                    results.append(True)
+                    # Push directly to HF Engine
+                    success = await push_to_hf_api(
+                        title=text,
+                        text=text,
+                        source_url=site_config["url"],
+                        ministry=site_config["source"],
+                        doc_type=site_config["doc_type"]
+                    )
+                    results.append(success)
                 except Exception as e:
-                    logger.warning(f"⚠️ Skipped duplicate or error: {str(e)[:50]}")
+                    logger.warning(f"⚠️ Error: {str(e)[:50]}")
                     results.append(False)
             return results
         return []
@@ -256,10 +268,10 @@ async def scrape_generic(page, site_config):
         return []
 
 async def scrape_and_upload():
-    print("🥷 Booting up GovBridge Multi-Site Bot (Zero-Dependency Stealth Mode)...\n")
+    print("🥷 Booting up GovBridge UNIFIED Scraper (Sending to HF API)...\n")
     
     async with async_playwright() as p:
-        # Launch hardened browser per Claude's specifications
+        # Launch hardened browser
         browser = await p.chromium.launch(
             headless=True,
             args=[
@@ -314,7 +326,7 @@ async def scrape_and_upload():
             await asyncio.sleep(random.uniform(2.0, 4.0)) # Delay between sites
         
         print(f"\n--- 📊 FINAL REPORT ---")
-        print(f"✅ Total items uploaded: {total_uploaded}")
+        print(f"✅ Total items pushed to Vector Engine: {total_uploaded}")
         print("\nShutting down bot.")
         await browser.close()