ingestion changes

app.py

@@ -25,6 +25,7 @@ from pages.sources import (
     handle_file_upload,
     handle_url_add,
     handle_source_delete,
+    run_ingestion_pipeline,
 )
 from pages.artifacts import (
     render_no_sources_gate,
@@ -431,6 +432,11 @@ with gr.Blocks(css=CUSTOM_CSS, theme=dark_theme, title="NotebookLM") as demo:
         inputs=[file_uploader, user_state],
         outputs=[user_state, source_list_html, source_header, source_selector],
         api_name=False,
+    ).then(
+        fn=run_ingestion_pipeline,
+        inputs=[user_state],
+        outputs=[user_state, source_list_html, source_header, source_selector],
+        api_name=False,
     ).then(
         fn=refresh_all,
         inputs=[user_state],
@@ -444,6 +450,11 @@ with gr.Blocks(css=CUSTOM_CSS, theme=dark_theme, title="NotebookLM") as demo:
         inputs=[url_input, user_state],
         outputs=[user_state, source_list_html, source_header, url_input, source_selector],
         api_name=False,
+    ).then(
+        fn=run_ingestion_pipeline,
+        inputs=[user_state],
+        outputs=[user_state, source_list_html, source_header, source_selector],
+        api_name=False,
     ).then(
         fn=refresh_all,
         inputs=[user_state],

ingestion_engine/__init__.py

@@ -0,0 +1,5 @@
+"""Ingestion Engine — document processing pipeline."""
+
+from ingestion_engine.ingestion_manager import IngestionManager
+
+__all__ = ["IngestionManager"]

ingestion_engine/chunker.py

@@ -0,0 +1,79 @@
+"""Recursive character text splitter with overlap."""
+
+
+def chunk_text(
+    text: str,
+    chunk_size: int = 500,
+    chunk_overlap: int = 50,
+) -> list[dict]:
+    """
+    Split text into overlapping chunks.
+
+    Uses character-count heuristic: 1 token ~ 4 characters.
+    Splits recursively on paragraph, newline, sentence, then word boundaries.
+
+    Returns list of {"text": str, "chunk_index": int}.
+    """
+    separators = ["\n\n", "\n", ". ", " "]
+    char_size = chunk_size * 4
+    char_overlap = chunk_overlap * 4
+
+    raw_chunks = _recursive_split(text, char_size, separators)
+    merged = _merge_small_chunks(raw_chunks, char_size)
+    overlapped = _apply_overlap(merged, char_overlap)
+
+    return [
+        {"text": chunk.strip(), "chunk_index": i}
+        for i, chunk in enumerate(overlapped)
+        if chunk.strip()
+    ]
+
+
+def _recursive_split(text: str, max_chars: int, separators: list[str]) -> list[str]:
+    """Split text using the first separator that produces sub-max chunks."""
+    if len(text) <= max_chars:
+        return [text]
+
+    for sep in separators:
+        parts = text.split(sep)
+        if len(parts) > 1:
+            chunks = []
+            for part in parts:
+                if len(part) <= max_chars:
+                    chunks.append(part)
+                else:
+                    remaining_seps = separators[separators.index(sep) + 1 :]
+                    if remaining_seps:
+                        chunks.extend(_recursive_split(part, max_chars, remaining_seps))
+                    else:
+                        for i in range(0, len(part), max_chars):
+                            chunks.append(part[i : i + max_chars])
+            return chunks
+
+    return [text[i : i + max_chars] for i in range(0, len(text), max_chars)]
+
+
+def _merge_small_chunks(chunks: list[str], max_chars: int) -> list[str]:
+    """Merge consecutive small chunks that together fit within max_chars."""
+    merged = []
+    buffer = ""
+    for chunk in chunks:
+        if buffer and len(buffer) + len(chunk) + 1 > max_chars:
+            merged.append(buffer)
+            buffer = chunk
+        else:
+            buffer = (buffer + " " + chunk) if buffer else chunk
+    if buffer:
+        merged.append(buffer)
+    return merged
+
+
+def _apply_overlap(chunks: list[str], overlap_chars: int) -> list[str]:
+    """Prepend the tail of the previous chunk to each subsequent chunk."""
+    if len(chunks) <= 1 or overlap_chars <= 0:
+        return chunks
+    result = [chunks[0]]
+    for i in range(1, len(chunks)):
+        prev_tail = chunks[i - 1][-overlap_chars:]
+        result.append(prev_tail + " " + chunks[i])
+    return result

ingestion_engine/embedding_generator.py

@@ -0,0 +1,37 @@
+"""Sentence-transformer embedding generation with model caching."""
+
+import logging
+from functools import lru_cache
+
+logger = logging.getLogger(__name__)
+
+MODEL_NAME = "all-MiniLM-L6-v2"
+EMBEDDING_DIM = 384
+
+
+@lru_cache(maxsize=1)
+def get_model():
+    """Load and cache the sentence-transformers model (loaded once per process)."""
+    from sentence_transformers import SentenceTransformer
+
+    logger.info("Loading embedding model: %s", MODEL_NAME)
+    model = SentenceTransformer(MODEL_NAME)
+    logger.info("Embedding model loaded. Dimension: %d", EMBEDDING_DIM)
+    return model
+
+
+def generate(texts: list[str], batch_size: int = 64) -> list[list[float]]:
+    """Encode a list of texts into normalized embedding vectors."""
+    model = get_model()
+    embeddings = model.encode(
+        texts,
+        batch_size=batch_size,
+        show_progress_bar=False,
+        normalize_embeddings=True,
+    )
+    return embeddings.tolist()
+
+
+def generate_query(query: str) -> list[float]:
+    """Embed a single query string (convenience wrapper for future RAG Engine)."""
+    return generate([query])[0]

ingestion_engine/ingestion_manager.py

@@ -0,0 +1,101 @@
+"""Ingestion Manager — orchestrates the full source processing pipeline."""
+
+import logging
+
+from ingestion_engine import pdf_extractor, text_extractor, url_scrapper, transcripter
+from ingestion_engine.chunker import chunk_text
+from ingestion_engine.embedding_generator import generate
+from persistence.vector_store import VectorStore
+
+logger = logging.getLogger(__name__)
+
+EXTRACTORS = {
+    "pdf": lambda fp, _url: pdf_extractor.extract(fp),
+    "txt": lambda fp, _url: text_extractor.extract(fp),
+    "pptx": lambda fp, _url: _extract_pptx(fp),
+    "url": lambda _fp, url: url_scrapper.extract(url),
+    "youtube": lambda _fp, url: transcripter.extract(url),
+}
+
+
+def _extract_pptx(file_path: str) -> str:
+    """Extract text from a PPTX file (lazy import to keep python-pptx optional)."""
+    from pptx import Presentation
+
+    prs = Presentation(file_path)
+    texts = []
+    for slide in prs.slides:
+        for shape in slide.shapes:
+            if shape.has_text_frame:
+                for paragraph in shape.text_frame.paragraphs:
+                    text = paragraph.text.strip()
+                    if text:
+                        texts.append(text)
+    return "\n\n".join(texts)
+
+
+class IngestionManager:
+    """Orchestrates: extract -> chunk -> embed -> store in Pinecone."""
+
+    def __init__(self):
+        self.vector_store = VectorStore()
+
+    def process_source(self, source, file_path: str | None, notebook_id: str) -> tuple[int, str | None]:
+        """
+        Run the full ingestion pipeline for a single source.
+
+        Args:
+            source: state.Source object
+            file_path: local file path (None for URL/YouTube sources)
+            notebook_id: used as Pinecone namespace
+
+        Returns:
+            (chunk_count, error_message) — error_message is None on success
+        """
+        try:
+            # Step 1: Extract text
+            extractor = EXTRACTORS.get(source.file_type)
+            if not extractor:
+                return 0, f"Unsupported file type: {source.file_type}"
+
+            raw_text = extractor(file_path, source.source_url)
+            if not raw_text or not raw_text.strip():
+                return 0, "No text could be extracted from this source."
+
+            logger.info("Extracted %d characters from %s", len(raw_text), source.filename)
+
+            # Step 2: Chunk
+            chunks = chunk_text(raw_text, chunk_size=500, chunk_overlap=50)
+            if not chunks:
+                return 0, "Text was extracted but produced no usable chunks."
+
+            logger.info("Created %d chunks from %s", len(chunks), source.filename)
+
+            # Step 3: Embed
+            chunk_texts = [c["text"] for c in chunks]
+            vectors = generate(chunk_texts)
+
+            logger.info("Generated %d embeddings for %s", len(vectors), source.filename)
+
+            # Step 4: Prepare records and upsert to Pinecone
+            records = []
+            for chunk, vector in zip(chunks, vectors):
+                records.append({
+                    "id": f"{source.id}_{chunk['chunk_index']}",
+                    "values": vector,
+                    "metadata": {
+                        "source_id": source.id,
+                        "source_filename": source.filename,
+                        "chunk_index": chunk["chunk_index"],
+                        "text": chunk["text"][:1000],
+                    },
+                })
+
+            self.vector_store.upsert(records, namespace=notebook_id)
+
+            logger.info("Stored %d vectors for %s in namespace %s", len(records), source.filename, notebook_id)
+            return len(chunks), None
+
+        except Exception as e:
+            logger.error("Ingestion failed for %s: %s", source.filename, e)
+            return 0, f"Ingestion error: {str(e)}"

ingestion_engine/pdf_extractor.py

@@ -0,0 +1,15 @@
+"""PDF text extraction using PyMuPDF."""
+
+import fitz
+
+
+def extract(file_path: str) -> str:
+    """Extract text from all pages of a PDF file."""
+    doc = fitz.open(file_path)
+    pages = []
+    for page in doc:
+        text = page.get_text()
+        if text.strip():
+            pages.append(text)
+    doc.close()
+    return "\n\n".join(pages)

ingestion_engine/text_extractor.py

@@ -0,0 +1,7 @@
+"""Plain text file extraction."""
+
+
+def extract(file_path: str) -> str:
+    """Read a plain text file as UTF-8."""
+    with open(file_path, "r", encoding="utf-8", errors="replace") as f:
+        return f.read()

ingestion_engine/transcripter.py

@@ -0,0 +1,25 @@
+"""YouTube video transcript extraction."""
+
+from urllib.parse import urlparse, parse_qs
+from youtube_transcript_api import YouTubeTranscriptApi
+
+
+def extract(url: str) -> str:
+    """Fetch the transcript for a YouTube video."""
+    video_id = _parse_video_id(url)
+    if not video_id:
+        raise ValueError(f"Could not parse YouTube video ID from: {url}")
+
+    transcript_list = YouTubeTranscriptApi.get_transcript(video_id)
+    return " ".join(entry["text"] for entry in transcript_list)
+
+
+def _parse_video_id(url: str) -> str | None:
+    """Extract video ID from youtube.com/watch?v=... or youtu.be/... URLs."""
+    parsed = urlparse(url)
+    hostname = parsed.hostname or ""
+    if "youtu.be" in hostname:
+        return parsed.path.lstrip("/")
+    if "youtube.com" in hostname:
+        return parse_qs(parsed.query).get("v", [None])[0]
+    return None

ingestion_engine/url_scrapper.py

@@ -0,0 +1,21 @@
+"""Web page text extraction via requests + BeautifulSoup."""
+
+import requests
+from bs4 import BeautifulSoup
+
+
+def extract(url: str) -> str:
+    """Fetch a web page and extract its main text content."""
+    headers = {"User-Agent": "Mozilla/5.0 (NotebookLM Bot)"}
+    response = requests.get(url, headers=headers, timeout=15)
+    response.raise_for_status()
+
+    soup = BeautifulSoup(response.text, "html.parser")
+
+    # Remove non-content elements
+    for tag in soup(["script", "style", "nav", "footer", "header", "aside"]):
+        tag.decompose()
+
+    text = soup.get_text(separator="\n", strip=True)
+    lines = [line.strip() for line in text.splitlines() if line.strip()]
+    return "\n".join(lines)

pages/sources.py

@@ -59,6 +59,14 @@ def render_source_list(state: UserData) -> str:
             meta_parts.append(f"{source.chunk_count} chunks")
         meta_str = " · ".join(meta_parts)
 
+        if source.status == "processing":
+            badge = '<span class="source-badge processing">Processing...</span>'
+        elif source.status == "failed":
+            err = source.error_message or "Unknown error"
+            badge = f'<span class="source-badge failed" title="{err}">Failed</span>'
+        else:
+            badge = '<span class="source-badge ready">Ready</span>'
+
         html += (
             f'<div class="source-card">'
             f'<div class="source-icon {ft}">{cfg["icon"]}</div>'
@@ -66,7 +74,7 @@ def render_source_list(state: UserData) -> str:
             f'<div class="name">{source.filename}</div>'
             f'<div class="meta">{meta_str}</div>'
             f'</div>'
-            f'<span class="source-badge ready">Ready</span>'
+            f'{badge}'
             f'</div>'
         )
     return html
@@ -100,13 +108,14 @@ def handle_file_upload(files, state: UserData) -> tuple[UserData, str, str, list
         if file_ext not in ALLOWED_TYPES:
             continue
 
-        # Get file size
+        # Get file size and path
         try:
             import os
             file_path = f.name if hasattr(f, 'name') else str(f)
             size_bytes = os.path.getsize(file_path)
             size_mb = round(size_bytes / (1024 * 1024), 2)
         except Exception:
+            file_path = None
             size_mb = 0
 
         if size_mb > MAX_FILE_SIZE_MB:
@@ -119,9 +128,10 @@ def handle_file_upload(files, state: UserData) -> tuple[UserData, str, str, list
             size_mb=size_mb,
             source_url=None,
             chunk_count=0,
-            status="ready",
+            status="processing",
             error_message=None,
             created_at=datetime.now().isoformat(),
+            file_path=file_path,
         )
         nb.sources.append(source)
 
@@ -153,9 +163,10 @@ def handle_url_add(url: str, state: UserData) -> tuple[UserData, str, str, str,
         size_mb=None,
         source_url=url,
         chunk_count=0,
-        status="ready",
+        status="processing",
         error_message=None,
         created_at=datetime.now().isoformat(),
+        file_path=None,
     )
     nb.sources.append(source)
 
@@ -163,10 +174,47 @@ def handle_url_add(url: str, state: UserData) -> tuple[UserData, str, str, str,
 
 
 def handle_source_delete(source_name: str, state: UserData) -> tuple[UserData, str, str, list[str]]:
-    """Delete a source by filename. Returns (state, source_list_html, header_html, source_choices)."""
+    """Delete a source by filename and remove its vectors from Pinecone."""
     nb = get_active_notebook(state)
     if not nb or not source_name:
         return state, render_source_list(state), render_source_header(state), get_source_choices(state)
 
+    # Delete vectors from Pinecone before removing from state
+    source_to_delete = next((s for s in nb.sources if s.filename == source_name), None)
+    if source_to_delete:
+        try:
+            from persistence.vector_store import VectorStore
+            VectorStore().delete_by_source(source_to_delete.id, nb.id)
+        except Exception:
+            pass  # Best-effort; source removed from UI regardless
+
     nb.sources = [s for s in nb.sources if s.filename != source_name]
     return state, render_source_list(state), render_source_header(state), get_source_choices(state)
+
+
+def run_ingestion_pipeline(state: UserData) -> tuple[UserData, str, str, list[str]]:
+    """Process all sources with status='processing' through the ingestion engine."""
+    from ingestion_engine import IngestionManager
+
+    nb = get_active_notebook(state)
+    if not nb:
+        return state, render_source_list(state), render_source_header(state), get_source_choices(state)
+
+    manager = IngestionManager()
+
+    for source in nb.sources:
+        if source.status != "processing":
+            continue
+
+        chunk_count, error = manager.process_source(source, source.file_path, nb.id)
+
+        if error:
+            source.status = "failed"
+            source.error_message = error
+            source.chunk_count = 0
+        else:
+            source.status = "ready"
+            source.error_message = None
+            source.chunk_count = chunk_count
+
+    return state, render_source_list(state), render_source_header(state), get_source_choices(state)

persistence/__init__.py

@@ -0,0 +1,5 @@
+"""Persistence Layer — storage services."""
+
+from persistence.vector_store import VectorStore
+
+__all__ = ["VectorStore"]

persistence/vector_store.py

@@ -0,0 +1,107 @@
+"""Pinecone vector store operations."""
+
+import os
+import logging
+
+logger = logging.getLogger(__name__)
+
+INDEX_NAME = "notebooklm"
+UPSERT_BATCH_SIZE = 100
+
+
+class VectorStore:
+    """Pinecone client for upserting, deleting, and querying vectors."""
+
+    def __init__(self):
+        self._index = None
+
+    def _get_index(self):
+        """Lazy-initialize the Pinecone index connection."""
+        if self._index is not None:
+            return self._index
+
+        from pinecone import Pinecone
+
+        api_key = os.environ.get("PINECONE_API_KEY")
+        if not api_key:
+            raise RuntimeError(
+                "PINECONE_API_KEY not found in environment. "
+                "Add it as a Secret in your HF Space settings."
+            )
+
+        pc = Pinecone(api_key=api_key)
+        self._index = pc.Index(INDEX_NAME)
+        logger.info("Connected to Pinecone index: %s", INDEX_NAME)
+        return self._index
+
+    def upsert(self, records: list[dict], namespace: str) -> int:
+        """
+        Upsert embedding records into Pinecone in batches.
+
+        Args:
+            records: List of {"id": str, "values": list[float], "metadata": dict}
+            namespace: Pinecone namespace (notebook_id)
+
+        Returns:
+            Number of vectors upserted
+        """
+        index = self._get_index()
+        total = 0
+        for i in range(0, len(records), UPSERT_BATCH_SIZE):
+            batch = records[i : i + UPSERT_BATCH_SIZE]
+            index.upsert(vectors=batch, namespace=namespace)
+            total += len(batch)
+        logger.info("Upserted %d vectors to namespace '%s'", total, namespace)
+        return total
+
+    def delete_by_source(self, source_id: str, namespace: str) -> None:
+        """Delete all vectors belonging to a specific source."""
+        try:
+            index = self._get_index()
+            index.delete(
+                filter={"source_id": {"$eq": source_id}},
+                namespace=namespace,
+            )
+            logger.info("Deleted vectors for source '%s' from namespace '%s'", source_id, namespace)
+        except Exception as e:
+            logger.error("Failed to delete vectors from Pinecone: %s", e)
+
+    def delete_namespace(self, namespace: str) -> None:
+        """Delete all vectors in a namespace (when a notebook is deleted)."""
+        try:
+            index = self._get_index()
+            index.delete(delete_all=True, namespace=namespace)
+            logger.info("Deleted entire namespace '%s'", namespace)
+        except Exception as e:
+            logger.error("Failed to delete namespace from Pinecone: %s", e)
+
+    def query(self, query_vector: list[float], namespace: str, top_k: int = 5) -> list[dict]:
+        """
+        Query Pinecone for the most similar chunks.
+
+        Returns list of {"text", "source_id", "source_filename", "chunk_index", "score"}.
+        """
+        try:
+            index = self._get_index()
+            results = index.query(
+                vector=query_vector,
+                namespace=namespace,
+                top_k=top_k,
+                include_metadata=True,
+            )
+
+            matches = []
+            for match in results.get("matches", []):
+                meta = match.get("metadata", {})
+                matches.append({
+                    "text": meta.get("text", ""),
+                    "source_id": meta.get("source_id", ""),
+                    "source_filename": meta.get("source_filename", ""),
+                    "chunk_index": meta.get("chunk_index", 0),
+                    "score": match.get("score", 0.0),
+                })
+            return matches
+
+        except Exception as e:
+            logger.error("Pinecone query failed: %s", e)
+            return []

requirements.txt

@@ -1 +1,8 @@
 gradio>=5.0.0
+sentence-transformers>=2.2.0
+pinecone-client>=3.0.0
+PyMuPDF>=1.23.0
+python-pptx>=0.6.21
+beautifulsoup4>=4.12.0
+requests>=2.31.0
+youtube-transcript-api>=0.6.0

state.py

@@ -5,7 +5,7 @@ import uuid
 
 
 class Source:
-    def __init__(self, id, filename, file_type, size_mb, source_url, chunk_count, status, error_message, created_at):
+    def __init__(self, id, filename, file_type, size_mb, source_url, chunk_count, status, error_message, created_at, file_path=None):
         self.id = id
         self.filename = filename
         self.file_type = file_type  # "pdf", "pptx", "txt", "url", "youtube"
@@ -15,6 +15,7 @@ class Source:
         self.status = status  # "ready", "processing", "failed"
         self.error_message = error_message
         self.created_at = created_at
+        self.file_path = file_path
 
 
 class Message:
@@ -88,6 +89,12 @@ def create_notebook(state, title):
 
 def delete_notebook(state, nb_id):
     if nb_id in state.notebooks:
+        # Clean up Pinecone vectors for this notebook
+        try:
+            from persistence.vector_store import VectorStore
+            VectorStore().delete_namespace(nb_id)
+        except Exception:
+            pass  # Best-effort cleanup
         del state.notebooks[nb_id]
         remaining = list(state.notebooks.keys())
         state.active_notebook_id = remaining[0] if remaining else None

theme.py

@@ -220,6 +220,15 @@ CUSTOM_CSS = """
     font-weight: 600; letter-spacing: 0.3px;
 }
 .source-badge.ready { background: rgba(34,197,94,0.15); color: #22c55e; }
+.source-badge.processing {
+    background: rgba(234,179,8,0.15); color: #eab308;
+    animation: pulse-badge 1.5s ease-in-out infinite;
+}
+.source-badge.failed { background: rgba(239,68,68,0.15); color: #ef4444; cursor: help; }
+@keyframes pulse-badge {
+    0%, 100% { opacity: 1; }
+    50% { opacity: 0.5; }
+}
 
 /* ── Welcome hero ── */
 .welcome-hero {