data_processing.py

import os
import uuid
import json
import pickle
import concurrent.futures
from typing import List, Dict
from unstructured.partition.pdf import partition_pdf
from unstructured.chunking.title import chunk_by_title
from langchain_core.documents import Document
from langchain_pinecone import PineconeVectorStore
from langchain_core.messages import HumanMessage
from pinecone import Pinecone

from config import *

# Initialize Pinecone Client
if PINECONE_API_KEY:
    pc = Pinecone(api_key=PINECONE_API_KEY)
else:
    pc = None

class SessionDocStore:
    """
    In-memory storage for heavy content (Images/Tables) tied to a session/document.
    """
    def __init__(self):
        self.store = {}

    def save_chunk(self, doc_id: str, data: Dict):
        self.store[doc_id] = data

    def get_chunk(self, doc_id: str):
        return self.store.get(doc_id, {})
    
    def clear(self):
        self.store = {}

    def save_to_disk(self, file_path: str):
        with open(file_path, 'wb') as f:
            pickle.dump(self.store, f)
            
    def load_from_disk(self, file_path: str):
        if os.path.exists(file_path):
            with open(file_path, 'rb') as f:
                self.store = pickle.load(f)

def create_multimodal_summary(text, tables, images, api_key: str):
    llm = get_llm(api_key=api_key)
    prompt_text = f"Analyze content. TEXT: {text[:1000]}. INSTRUCTIONS: Summarize text and describe tables in detail without missing out /skipping any information."
    
    message_content = [{"type": "text", "text": prompt_text}]
    if images:
        for b64_str in images:
            if "," in b64_str: b64_str = b64_str.split(",")[1]
            message_content.append({
                "type": "image_url",
                "image_url": {"url": f"data:image/jpeg;base64,{b64_str}", "detail": "low"}
            })
            
    response = llm.invoke([HumanMessage(content=message_content)])
    return response.content

def process_single_chunk(i, chunk, doc_store, api_key: str):
    """
    Worker function to process a single chunk in a separate thread.
    """
    content = {'text': chunk.text, 'tables': [], 'images': []}
    
    # Extract visual data
    if hasattr(chunk, 'metadata') and hasattr(chunk.metadata, 'orig_elements'):
        for element in chunk.metadata.orig_elements:
            el_type = type(element).__name__
            if el_type == 'Table':
                content['tables'].append(getattr(element.metadata, 'text_as_html', element.text))
            elif el_type == 'Image' and hasattr(element.metadata, 'image_base64'):
                content['images'].append(element.metadata.image_base64)

    # Deciding whether to call LLM (Slow) or just use text (Fast)
    if content['images'] or content['tables']:
        enhanced_text = create_multimodal_summary(content['text'], content['tables'], content['images'], api_key)
    else:
        enhanced_text = content['text']

    doc_id = str(uuid.uuid4())
    
    # Save heavy data to local store
    doc_store.save_chunk(doc_id, {
        "raw_text": content['text'],
        "tables": content['tables'],
        "images": content['images']
    })
    
    # Return the processed Document
    return Document(
        page_content=enhanced_text,
        metadata={"doc_id": doc_id, "chunk_index": i}
    )

def process_and_ingest(file_path: str, doc_store: SessionDocStore, api_key: str):
    if not pc:
        raise ValueError("PINECONE_API_KEY is not set.")
    
    print(f"📄 Partitioning: {file_path}")
    
    elements = partition_pdf(
        filename=file_path, 
        strategy="auto", 
        infer_table_structure=True,
        extract_image_block_types=["Table"], 
        extract_image_block_to_payload=True
    )
    
    chunks = chunk_by_title(elements, max_characters=2000, new_after_n_chars=1500, combine_text_under_n_chars=300)
    
    documents = []
    print(f"🔄 Processing {len(chunks)} chunks in parallel...")
    
    with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
        future_to_chunk = {
            executor.submit(process_single_chunk, i, chunk, doc_store, api_key): i 
            for i, chunk in enumerate(chunks)
        }
        
        futures_list = list(future_to_chunk.keys())
        
        for future in futures_list:
            try:
                doc = future.result()
                documents.append(doc)
            except Exception as e:
                print(f"❌ Error processing chunk: {e}")

    print(f"🔮 Ingesting {len(documents)} vectors to Pinecone Index: {INDEX_NAME}")
    
    # Clear index first if necessary, or just use a specific namespace. For this single-file app, we can use a hardcoded namespace
    namespace = "single-pdf-mcp"
    
    PineconeVectorStore.from_documents(
        documents=documents,
        index_name=INDEX_NAME,
        embedding=get_embeddings(),
        namespace=namespace 
    )
    
    return documents