| 11 | from datetime import datetime |
| 12 | |
| 13 | class SimplePDFProcessor: |
| 14 | def __init__(self, db_path: str = "chat_data.db"): |
| 15 | """Initialize simple PDF processor with SQLite storage""" |
| 16 | self.db_path = db_path |
| 17 | self.init_database() |
| 18 | print("✅ Simple PDF processor initialized") |
| 19 | |
| 20 | def init_database(self): |
| 21 | """Initialize SQLite database for storing PDF content""" |
| 22 | conn = sqlite3.connect(self.db_path) |
| 23 | conn.execute(''' |
| 24 | CREATE TABLE IF NOT EXISTS pdf_documents ( |
| 25 | id TEXT PRIMARY KEY, |
| 26 | session_id TEXT NOT NULL, |
| 27 | filename TEXT NOT NULL, |
| 28 | content TEXT NOT NULL, |
| 29 | created_at TEXT NOT NULL |
| 30 | ) |
| 31 | ''') |
| 32 | |
| 33 | conn.commit() |
| 34 | conn.close() |
| 35 | |
| 36 | def extract_text_from_pdf(self, pdf_bytes: bytes) -> str: |
| 37 | """Extract text from PDF bytes""" |
| 38 | try: |
| 39 | print(f"📄 Starting PDF text extraction ({len(pdf_bytes)} bytes)") |
| 40 | pdf_file = BytesIO(pdf_bytes) |
| 41 | pdf_reader = PyPDF2.PdfReader(pdf_file) |
| 42 | |
| 43 | print(f"📖 PDF has {len(pdf_reader.pages)} pages") |
| 44 | |
| 45 | text = "" |
| 46 | for page_num, page in enumerate(pdf_reader.pages): |
| 47 | print(f"📄 Processing page {page_num + 1}") |
| 48 | try: |
| 49 | page_text = page.extract_text() |
| 50 | if page_text.strip(): |
| 51 | text += f"\n--- Page {page_num + 1} ---\n" |
| 52 | text += page_text + "\n" |
| 53 | print(f"✅ Page {page_num + 1}: extracted {len(page_text)} characters") |
| 54 | except Exception as page_error: |
| 55 | print(f"❌ Error on page {page_num + 1}: {str(page_error)}") |
| 56 | continue |
| 57 | |
| 58 | print(f"📄 Total extracted text: {len(text)} characters") |
| 59 | return text.strip() |
| 60 | |
| 61 | except Exception as e: |
| 62 | print(f"❌ Error extracting text from PDF: {str(e)}") |
| 63 | print(f"❌ Error type: {type(e).__name__}") |
| 64 | return "" |
| 65 | |
| 66 | def process_pdf(self, pdf_bytes: bytes, filename: str, session_id: str) -> Dict[str, Any]: |
| 67 | """Process a PDF file and store in database""" |
| 68 | print(f"📄 Processing PDF: {filename}") |
| 69 | |
| 70 | # Extract text |
no outgoing calls
no test coverage detected