Upload folder using huggingface_hub

api.py

@@ -95,6 +95,33 @@ class EligibilityRequest(BaseModel):
     occupation: str = "Unknown"
 
 
+class GazetteSearchQuery(BaseModel):
+    """Search query for the Gazette Vault hybrid search pipeline."""
+    query: str = Field(..., min_length=2, max_length=500)
+    gazette_type: Optional[str] = Field(default=None, description="Filter: central, state, extraordinary")
+    state: Optional[str] = Field(default=None, description="Filter by state applicability")
+    limit: int = Field(default=10, ge=1, le=50)
+
+    @field_validator('query')
+    @classmethod
+    def clean_gazette_query(cls, v: str) -> str:
+        v = re.sub(r'<[^>]+>', '', v)
+        v = re.sub(r'\s+', ' ', v).strip()
+        if not v:
+            raise ValueError('Query is empty after cleaning')
+        return v
+
+    @field_validator('gazette_type')
+    @classmethod
+    def validate_gazette_type(cls, v: Optional[str]) -> Optional[str]:
+        if v is None:
+            return None
+        valid = ["central", "state", "extraordinary"]
+        if v.lower() not in valid:
+            return None
+        return v.lower()
+
+
 # --- INITIALIZE AI (LAZY LOADED) ---
 embedding_model = None
 reranker_model = None
@@ -159,6 +186,34 @@ def rerank_chunks(query: str, chunks: list) -> list:
     candidates.sort(key=lambda x: x['rerank_score'], reverse=True)
     return candidates[:5]
 
+
+def rerank_gazette_chunks(query: str, chunks: list) -> list:
+    """
+    Cross-encoder re-ranking for gazette search results.
+    Takes top 30 RRF candidates (gazette chunks are denser than scheme chunks),
+    runs them through Ettin reranker, returns top results.
+
+    Unlike rerank_chunks() which caps at 20, gazette documents
+    require a wider initial pool due to dense legislative language.
+    """
+    if not chunks:
+        return []
+
+    # Gazette-specific: wider pool (30) because legislative text
+    # has higher semantic density than scheme descriptions
+    candidates = chunks[:30]
+
+    reranker = get_reranker_model()
+    pairs = [[query, c.get('chunk_text', '')] for c in candidates]
+
+    scores = reranker.predict(pairs, batch_size=32, show_progress_bar=False)
+
+    for i, score in enumerate(scores):
+        candidates[i]['cross_encoder_score'] = float(score)
+
+    candidates.sort(key=lambda x: x['cross_encoder_score'], reverse=True)
+    return candidates[:10]  # Return top 10 for gazette (more results than scheme search)
+
 # --- ENDPOINTS ---
 @app.get("/health")
 async def health_check():
@@ -452,4 +507,93 @@ async def get_coverage_gaps(request: Request, limit: int = 20):
         "total_gaps": len(result.data),
         "queries": result.data
     }
+
+
+@app.post("/api/gazette/search")
+@limiter.limit("15/minute")
+async def gazette_search(request: Request, query: GazetteSearchQuery):
+    """
+    Gazette Vault Hybrid Search Endpoint (Sprint 28).
+
+    Pipeline:
+    1. Embed query with Nomic (768-dim) using search_query: prefix
+    2. Call gazette_hybrid_search RPC (BM25 + Vector + RRF)
+    3. Cross-encoder re-rank with Ettin (top 30 → top 10)
+    4. Return page-pinned results to frontend GazetteViewer
+    """
+    print(f"📜 Gazette search: '{query.query}' | Type: {query.gazette_type} | State: {query.state}")
+
+    try:
+        # Step 1: Embed query
+        model = get_embedding_model()
+        query_embedding = model.encode(
+            f"search_query: {query.query}",
+            normalize_embeddings=True
+        ).tolist()
+
+        # Step 2: Call Supabase RPC
+        rpc_params: Dict[str, Any] = {
+            "query_text": query.query,
+            "query_embedding": query_embedding,
+            "match_count": 50,
+            "k_smoothing": 60,
+        }
+
+        # Add optional filters (pass None to disable filter in SQL)
+        if query.gazette_type:
+            rpc_params["filter_gazette_type"] = query.gazette_type
+        if query.state:
+            rpc_params["filter_state"] = query.state
+
+        result = supabase.rpc("gazette_hybrid_search", rpc_params).execute()
+        raw_results = cast(List[Dict[str, Any]], result.data or [])
+
+        if not raw_results:
+            return {
+                "query": query.query,
+                "results": [],
+                "total": 0,
+                "pipeline": "gazette_hybrid_search + ettin_reranker"
+            }
+
+        # Step 3: Cross-encoder re-rank
+        reranked = rerank_gazette_chunks(query.query, raw_results)
+
+        # Step 4: Trim to requested limit
+        reranked = reranked[:query.limit]
+
+        # Step 5: Format response with page-pinned results
+        results = []
+        for chunk in reranked:
+            results.append({
+                "id": chunk.get("id"),
+                "document_title": chunk.get("document_title", "Unknown Gazette"),
+                "source_url": chunk.get("source_url"),
+                "page_number": chunk.get("page_number", 1),
+                "snippet_text": chunk.get("chunk_text", ""),
+                "gazette_type": chunk.get("gazette_type", "central"),
+                "issuing_authority": chunk.get("issuing_authority"),
+                "notification_date": str(chunk.get("notification_date", "")),
+                "cross_encoder_score": round(chunk.get("cross_encoder_score", 0.0), 4),
+                "rrf_score": float(chunk.get("rrf_score", 0.0)),
+            })
+
+        print(f"   ✅ Returning {len(results)} gazette results (best score: {results[0]['cross_encoder_score']:.4f})")
+
+        return {
+            "query": query.query,
+            "results": results,
+            "total": len(results),
+            "pipeline": "gazette_hybrid_search + ettin_reranker"
+        }
+
+    except Exception as e:
+        print(f"❌ Gazette search error: {e}")
+        return {
+            "query": query.query,
+            "results": [],
+            "total": 0,
+            "error": str(e),
+            "pipeline": "gazette_hybrid_search + ettin_reranker"
+        }
         

ingestion/gazette_ingester.py

@@ -0,0 +1,344 @@
+"""
+GovBridge India — Gazette Document Ingestion Pipeline
+Sprint 28: Semantic Boundary Chunking with Page-Level Granularity
+
+Usage:
+    python -m ingestion.gazette_ingester \
+        --url "https://example.gov.in/gazette.pdf" \
+        --title "Gazette of India Extraordinary Part II" \
+        --gazette-type central \
+        --authority "Ministry of Finance" \
+        --state National
+"""
+import os
+import sys
+import hashlib
+import argparse
+import tempfile
+from typing import Optional
+from datetime import date
+
+import httpx
+
+# Ensure parent is on path for config import
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+from config import settings
+
+# Lazy imports — only loaded when main() runs
+fitz = None  # PyMuPDF
+SentenceTransformer = None
+supabase_client = None
+
+
+def _get_fitz():
+    """Lazy import PyMuPDF."""
+    global fitz
+    if fitz is None:
+        import fitz as _fitz
+        fitz = _fitz
+    return fitz
+
+
+def _get_supabase():
+    """Lazy Supabase client singleton."""
+    global supabase_client
+    if supabase_client is None:
+        from supabase import create_client
+        supabase_client = create_client(settings.SUPABASE_URL, settings.SUPABASE_KEY)
+    return supabase_client
+
+
+def _get_embedding_model():
+    """Lazy load Nomic embedding model (768-dim)."""
+    global SentenceTransformer
+    if SentenceTransformer is None:
+        from sentence_transformers import SentenceTransformer as _ST
+        SentenceTransformer = _ST
+    # Use a module-level cache variable
+    if not hasattr(_get_embedding_model, '_model'):
+        print("⏳ Loading Nomic Embedding Model (768-dim)...")
+        _get_embedding_model._model = SentenceTransformer(
+            'nomic-ai/nomic-embed-text-v1',
+            trust_remote_code=True
+        )
+        print("✅ Nomic Model loaded")
+    return _get_embedding_model._model
+
+
+# ═══════════════════════════════════════════════════════════════
+# CORE: Page-Aware Text Extraction
+# ═══════════════════════════════════════════════════════════════
+
+def extract_pages_from_pdf(pdf_path: str) -> list[dict]:
+    """
+    Extract text from each page of a PDF using PyMuPDF.
+    Returns a list of dicts: [{"page": 1, "text": "..."}, ...]
+    Pages with no extractable text are skipped (image-only scans).
+    """
+    pdf = _get_fitz()
+    doc = pdf.open(pdf_path)
+    pages = []
+    for i, page in enumerate(doc):
+        text = page.get_text("text").strip()
+        if text:
+            pages.append({"page": i + 1, "text": text})
+    doc.close()
+    return pages
+
+
+# ═══════════════════════════════════════════════════════════════
+# CORE: Semantic Boundary Chunking (Page-Aware)
+# ═══════════════════════════════════════════════════════════════
+
+def chunk_gazette_pages(
+    pages: list[dict],
+    chunk_size: int = 500,
+    overlap: int = 50
+) -> list[dict]:
+    """
+    Chunk gazette text using paragraph-boundary splitting with overlap.
+    Each chunk retains its source page_number for the frontend viewer
+    to scroll to the exact location.
+
+    Args:
+        pages: List of {"page": int, "text": str} from extract_pages_from_pdf
+        chunk_size: Target character count per chunk
+        overlap: Character overlap between consecutive chunks
+
+    Returns:
+        List of {"page_number": int, "chunk_index": int, "chunk_text": str}
+    """
+    all_chunks = []
+    global_index = 0
+
+    for page_data in pages:
+        page_num = page_data["page"]
+        text = page_data["text"]
+
+        # Split by paragraph boundaries
+        paragraphs = [p.strip() for p in text.split('\n\n') if p.strip()]
+
+        current = ""
+        page_chunks = []
+
+        for para in paragraphs:
+            if len(current) + len(para) < chunk_size:
+                current += " " + para if current else para
+            else:
+                if current.strip():
+                    page_chunks.append(current.strip())
+                current = para
+
+        if current.strip():
+            page_chunks.append(current.strip())
+
+        # Apply overlap between consecutive chunks on the same page
+        if len(page_chunks) > 1:
+            overlapped = [page_chunks[0]]
+            for i in range(1, len(page_chunks)):
+                tail = page_chunks[i - 1][-overlap:] if len(page_chunks[i - 1]) > overlap else page_chunks[i - 1]
+                overlapped.append(tail + " " + page_chunks[i])
+            page_chunks = overlapped
+
+        for chunk_text in page_chunks:
+            all_chunks.append({
+                "page_number": page_num,
+                "chunk_index": global_index,
+                "chunk_text": chunk_text,
+            })
+            global_index += 1
+
+    return all_chunks
+
+
+# ═══════════════════════════════════════════════════════════════
+# CORE: Embedding Generation
+# ═══════════════════════════════════════════════════════════════
+
+def generate_embeddings(chunks: list[dict]) -> list[dict]:
+    """
+    Generate 768-dim Nomic embeddings for each chunk.
+    Uses the `search_document:` prefix required by Nomic models.
+
+    Modifies chunks in-place by adding 'embedding' key.
+    Processes in batches of 32 to control memory on 16GB HF containers.
+    """
+    model = _get_embedding_model()
+    texts = [f"search_document: {c['chunk_text']}" for c in chunks]
+
+    batch_size = 32
+    all_embeddings = []
+
+    for i in range(0, len(texts), batch_size):
+        batch = texts[i:i + batch_size]
+        embeddings = model.encode(
+            batch,
+            normalize_embeddings=True,
+            show_progress_bar=False
+        ).tolist()
+        all_embeddings.extend(embeddings)
+
+    for i, emb in enumerate(all_embeddings):
+        chunks[i]["embedding"] = emb
+
+    return chunks
+
+
+# ═══════════════════════════════════════════════════════════════
+# CORE: Supabase Upsert
+# ═══════════════════════════════════════════════════════════════
+
+def upsert_gazette_chunks(
+    chunks: list[dict],
+    metadata: dict
+) -> int:
+    """
+    Upsert gazette chunks to the gazette_chunks table.
+    Uses content_hash + chunk_index as the conflict resolution key.
+
+    Args:
+        chunks: List of dicts with page_number, chunk_index, chunk_text, embedding
+        metadata: Dict with document_title, source_url, gazette_type,
+                  issuing_authority, notification_date, state
+
+    Returns:
+        Number of chunks upserted
+    """
+    sb = _get_supabase()
+    source_url = metadata.get("source_url", "")
+    content_hash_base = hashlib.sha256(source_url.encode()).hexdigest()[:16]
+
+    rows = []
+    for chunk in chunks:
+        row = {
+            "document_title": metadata["document_title"],
+            "source_url": source_url,
+            "gazette_type": metadata.get("gazette_type", "central"),
+            "issuing_authority": metadata.get("issuing_authority"),
+            "notification_date": metadata.get("notification_date"),
+            "state": metadata.get("state", "National"),
+            "page_number": chunk["page_number"],
+            "chunk_index": chunk["chunk_index"],
+            "chunk_text": chunk["chunk_text"],
+            "embedding": chunk["embedding"],
+            "content_hash": content_hash_base,
+            "is_active": True,
+        }
+        rows.append(row)
+
+    # Batch upsert in groups of 50 to avoid payload size limits
+    upserted = 0
+    for i in range(0, len(rows), 50):
+        batch = rows[i:i + 50]
+        sb.table("gazette_chunks").upsert(
+            batch,
+            on_conflict="content_hash,chunk_index"
+        ).execute()
+        upserted += len(batch)
+        print(f"  📦 Upserted batch {i // 50 + 1}: {len(batch)} chunks")
+
+    return upserted
+
+
+# ═══════════════════════════════════════════════════════════════
+# MAIN: CLI Entry Point
+# ═══════════════════════════════════════════════════════════════
+
+HEADERS = {
+    "User-Agent": "GovBridge-Gazette-Ingester/2.0 (+https://govbridge-india.pages.dev)",
+    "Accept": "application/pdf,*/*;q=0.8"
+}
+
+
+def download_pdf(url: str) -> str:
+    """Download PDF to a temporary file. Returns path."""
+    with tempfile.NamedTemporaryFile(suffix='.pdf', delete=False) as tmp:
+        with httpx.stream(
+            "GET", url,
+            headers=HEADERS,
+            timeout=120.0,
+            follow_redirects=True
+        ) as r:
+            r.raise_for_status()
+            for data in r.iter_bytes():
+                tmp.write(data)
+        return tmp.name
+
+
+def main():
+    parser = argparse.ArgumentParser(
+        description="GovBridge Gazette Ingestion Pipeline (Sprint 28)"
+    )
+    parser.add_argument("--url", required=True, help="URL of the gazette PDF")
+    parser.add_argument("--title", required=True, help="Document title")
+    parser.add_argument("--gazette-type", default="central",
+                        choices=["central", "state", "extraordinary"],
+                        help="Type of gazette")
+    parser.add_argument("--authority", default=None,
+                        help="Issuing authority (e.g., 'Ministry of Finance')")
+    parser.add_argument("--state", default="National",
+                        help="State applicability")
+    parser.add_argument("--date", default=None,
+                        help="Notification date (YYYY-MM-DD)")
+    args = parser.parse_args()
+
+    notification_date = args.date
+    if notification_date:
+        # Validate date format
+        try:
+            date.fromisoformat(notification_date)
+        except ValueError:
+            print(f"❌ Invalid date format: {notification_date}. Use YYYY-MM-DD.")
+            sys.exit(1)
+
+    print(f"🚀 Starting Gazette Ingestion: {args.url}")
+    print(f"   Title: {args.title}")
+    print(f"   Type: {args.gazette_type}")
+
+    # Step 1: Download
+    print("📥 Downloading PDF...")
+    pdf_path = download_pdf(args.url)
+
+    try:
+        # Step 2: Extract pages
+        print("📄 Extracting pages...")
+        pages = extract_pages_from_pdf(pdf_path)
+        print(f"   ✅ Extracted text from {len(pages)} pages")
+
+        if not pages:
+            print("❌ No text extracted — PDF may be a pure image scan.")
+            print("   Image-only PDFs require PaddleOCR (see ocr_processor_runner.py)")
+            sys.exit(1)
+
+        # Step 3: Chunk with page awareness
+        print("✂️ Chunking with semantic boundaries...")
+        chunks = chunk_gazette_pages(pages, chunk_size=500, overlap=50)
+        print(f"   ✅ Generated {len(chunks)} chunks across {len(pages)} pages")
+
+        # Step 4: Generate embeddings
+        print("🧠 Generating 768-dim Nomic embeddings...")
+        chunks = generate_embeddings(chunks)
+        print(f"   ✅ All {len(chunks)} chunks embedded")
+
+        # Step 5: Upsert to Supabase
+        print("💾 Upserting to gazette_chunks table...")
+        metadata = {
+            "document_title": args.title,
+            "source_url": args.url,
+            "gazette_type": args.gazette_type,
+            "issuing_authority": args.authority,
+            "notification_date": notification_date,
+            "state": args.state,
+        }
+        count = upsert_gazette_chunks(chunks, metadata)
+        print(f"   ✅ {count} chunks upserted successfully")
+
+        print(f"\n🎉 Gazette ingestion complete: {args.title}")
+        print(f"   Chunks: {count} | Pages: {len(pages)} | Embedding dim: 768")
+
+    finally:
+        os.unlink(pdf_path)
+
+
+if __name__ == "__main__":
+    main()

requirements.txt

@@ -40,4 +40,7 @@ pydantic-settings
 psutil
 
 # --- Deterministic Rules Engine (Sprint 19) ---
-openfisca-core>=44.0.0
+openfisca-core>=44.0.0
+
+# --- PDF Processing (Sprint 28) ---
+pymupdf