import os import re import pickle import json import traceback import fitz from langchain_core.documents import Document FOLDER = "./sermons_all" OUTPUT_FILE = "sermon_chunks.pkl" DLQ_FILE = "chunking_dlq.json" def clean_filename(name): name = re.sub(r'[\\/*?:"<>|]', '', name) name = re.sub(r'\s+', ' ', name).strip() return name def extract_sermon_info(filename): name = filename.replace(".pdf", "").replace(".PDF", "") parts = name.split(" ", 1) if len(parts) == 2: date_code = parts[0] title = parts[1] else: date_code = "00-0000" title = name return date_code, title def extract_location(first_page_text, title): # E.g. "Faith Is The SubstanceOakland, California, USA" or "Jeffersonville, Indiana, USA" lines = [line.strip() for line in first_page_text.split("\n") if line.strip()][:6] location = "Unknown" for line in lines: if "usa" in line.lower() or "u.s.a." in line.lower() or "unknown" in line.lower(): # If the line starts with the title, strip the title prefix if line.lower().startswith(title.lower()): loc_candidate = line[len(title):].strip() if loc_candidate: location = loc_candidate break # Spacing/punctuation fallback match = re.search(re.escape(title), line, re.IGNORECASE) if match: loc_candidate = line[match.end():].strip() if loc_candidate: location = loc_candidate break # On its own line location = line break location = re.sub(r'^[^\w]+', '', location).strip() return location def clean_line(line, clean_title): cleaned = line.strip() if not cleaned: return None # Ignore page boundaries if cleaned.startswith("--- PAGE") and cleaned.endswith("---"): return None # Dynamic title header matching # Strips leading and trailing page numbers/digits to isolate the clean title alphanum_line = re.sub(r'[^a-zA-Z0-9]', '', cleaned).lower() stripped_alphanum = re.sub(r'^\d+|\d+$', '', alphanum_line) if stripped_alphanum == clean_title: return None # Ignore common sermon locations / dates at startup if cleaned in ["Jeffersonville, Indiana, USA", "Chicago, Illinois, USA", "Macon, Georgia, USA", "Spindale, North Carolina, USA"]: return None return line def is_adjacent_to_header(lines, idx, clean_title_key): # Check previous non-empty lines for i in range(idx - 1, -1, -1): stripped = lines[i].strip() if stripped: if clean_line(stripped, clean_title_key) is None: return True break # Check next non-empty lines for i in range(idx + 1, len(lines)): stripped = lines[i].strip() if stripped: if clean_line(stripped, clean_title_key) is None: return True break return False def parse_pdf_generator(filepath, title): """ Generator that parses a single PDF page-by-page and yields parsed paragraph dictionaries with location. """ doc = fitz.open(filepath) clean_title_key = re.sub(r'[^a-zA-Z0-9]', '', title).lower() # Extract location from first page location = "Unknown" if len(doc) > 0: first_page_text = doc[0].get_text() if first_page_text: location = extract_location(first_page_text, title) current_para_num = 1 current_para_text = [] current_para_start_page = 1 expected_para = 1 for page_idx, page in enumerate(doc): page_num = page_idx + 1 text = page.get_text() if not text: continue lines = text.split("\n") for line_idx, line in enumerate(lines): line_cleaned = clean_line(line, clean_title_key) if line_cleaned is None: continue line_stripped = line_cleaned.strip() if not line_stripped: continue # Strip standalone page number if adjacent to title header if line_stripped.isdigit() and is_adjacent_to_header(lines, line_idx, clean_title_key): continue # 1. Check if line is exactly a paragraph number if line_stripped.isdigit(): val = int(line_stripped) if expected_para <= val <= expected_para + 5: if val == 1: # Merge pre-intro text into paragraph 1 current_para_num = 1 expected_para = 2 continue # Yield previous paragraph if current_para_text: yield { "paragraph": str(current_para_num), "text": " ".join(current_para_text), "page_start": current_para_start_page, "page_end": page_num, "location": location } current_para_text = [] current_para_num = val expected_para = val + 1 current_para_start_page = page_num continue # 2. Check if line ends with a paragraph number matched_end = False for candidate in range(expected_para, expected_para + 3): pattern = rf"^(?P.*?)(?.*)$" match = re.match(pattern, line_stripped) if match: text_after = match.group("text").strip() # Yield previous paragraph if current_para_text: yield { "paragraph": str(current_para_num), "text": " ".join(current_para_text), "page_start": current_para_start_page, "page_end": page_num, "location": location } current_para_text = [] current_para_num = candidate expected_para = candidate + 1 current_para_start_page = page_num if text_after: current_para_text.append(text_after) matched_start = True break if matched_start: continue current_para_text.append(line_stripped) # Yield the final paragraph if current_para_text: yield { "paragraph": str(current_para_num), "text": " ".join(current_para_text), "page_start": current_para_start_page, "page_end": len(doc), "location": location } def scan_all_sermons_generator(folder): """ Generator that processes all PDFs in the folder one by one. Yields tuple of (Document, None) on success, or (None, err_dict) on failure. """ files = sorted([f for f in os.listdir(folder) if f.lower().endswith(".pdf")]) print(f"Discovered {len(files)} PDFs in '{folder}' to chunk.") for filename in files: filepath = os.path.join(folder, filename) date_code, title = extract_sermon_info(filename) try: for chunk in parse_pdf_generator(filepath, title): text_clean = re.sub(r"\s+", " ", chunk["text"]).strip() doc = Document( page_content=text_clean, metadata={ "source": filename, "title": title, "date_code": date_code, "paragraph": chunk["paragraph"], "page_start": chunk["page_start"], "page_end": chunk["page_end"], "location": chunk["location"] } ) yield doc, None except Exception as e: err_info = { "filename": filename, "error": str(e), "traceback": traceback.format_exc() } yield None, err_info def main(): if not os.path.exists(FOLDER): print(f"Error: Folder '{FOLDER}' does not exist.") return all_chunks = [] dlq_list = [] processed_count = 0 failed_count = 0 current_file = None print("\nStarting ingestion pipeline...") # Process using the generator for doc, err in scan_all_sermons_generator(FOLDER): if err: print(f" [DLQ] Failed to parse: {err['filename']} - Error: {err['error']}") dlq_list.append(err) failed_count += 1 else: all_chunks.append(doc) file_source = doc.metadata["source"] if file_source != current_file: current_file = file_source processed_count += 1 if processed_count % 50 == 0: print(f"Processed {processed_count} sermons... ({len(all_chunks)} chunks total)") # Save the Dead Letter Queue (DLQ) if any errors occurred if dlq_list: with open(DLQ_FILE, "w", encoding="utf-8") as f: json.dump(dlq_list, f, indent=2) print(f"\n[WARNING] Dead Letter Queue created with {len(dlq_list)} failures at '{DLQ_FILE}'.") else: if os.path.exists(DLQ_FILE): os.remove(DLQ_FILE) # Save output to pickle file print(f"\nIngested {processed_count} sermons with {len(all_chunks)} total chunks.") if failed_count > 0: print(f"Failed to ingest {failed_count} files (logged to DLQ).") print(f"Saving chunks to '{OUTPUT_FILE}'...") with open(OUTPUT_FILE, "wb") as f: pickle.dump(all_chunks, f) print("Ingestion complete and successfully serialized!") if __name__ == "__main__": main()