src/chunk_sermons.py

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<text>.*?)(?<!\d){candidate}$"
                match = re.match(pattern, line_stripped)
                if match:
                    text_before = match.group("text").strip()
                    if text_before:
                        current_para_text.append(text_before)
                    # 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
                    matched_end = True
                    break
                    
            if matched_end:
                continue
                
            # 3. Check if line starts with a paragraph number
            matched_start = False
            for candidate in range(expected_para, expected_para + 3):
                pattern = rf"^{candidate}(?!\d)\s*(?P<text>.*)$"
                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()