Element-Based Output¶

from kreuzberg import extract_file_sync, ExtractionConfig, PageConfig

config = ExtractionConfig(
    output_format="element_based",  # Element-based output
    pages=PageConfig(
        extract_pages=True,           # Also extract per-page content
        insert_page_markers=False
    ),
    ocr=OcrConfig(
        force_ocr=False,              # Enable OCR for scanned pages
        language="eng"
    )
)

result = extract_file_sync("document.pdf", config=config)

# Element-based processing
for element in result.elements:
    if element.element_type == "title":
        process_heading(element)

# Page-based processing (also available)
for page in result.pages:
    print(f"Page {page.page_number}: {len(page.content)} characters")

from kreuzberg import extract_file_sync, ExtractionConfig

config = ExtractionConfig(output_format="element_based")
result = extract_file_sync("research-paper.pdf", config=config)

for element in result.elements:
    print(f"Type: {element.element_type}")
    print(f"Text: {element.text[:100]}...")
    print(f"Page: {element.metadata.page_number}")
    print(f"Index: {element.metadata.element_index}")
    print("---")

from kreuzberg import extract_file_sync, ExtractionConfig

config = ExtractionConfig(output_format="element_based")
result = extract_file_sync("document.pdf", config=config)

# Extract all titles
titles = [elem for elem in result.elements if elem.element_type == "title"]
for title in titles:
    level = title.metadata.additional.get("level", "unknown")
    print(f"{level}: {title.text}")

# Extract all tables
tables = [elem for elem in result.elements if elem.element_type == "table"]
for table in tables:
    row_count = table.metadata.additional.get("row_count", 0)
    col_count = table.metadata.additional.get("column_count", 0)
    print(f"Table ({row_count}x{col_count}): {table.text[:50]}...")

# Extract all list items
list_items = [elem for elem in result.elements if elem.element_type == "list_item"]
for item in list_items:
    marker = item.metadata.additional.get("list_marker", "")
    print(f"{marker} {item.text}")

from kreuzberg import extract_file_sync, ExtractionConfig

config = ExtractionConfig(output_format="element_based")
result = extract_file_sync("document.pdf", config=config)

# Elements with bounding boxes (PDFs)
for element in result.elements:
    if element.metadata.coordinates:
        coords = element.metadata.coordinates
        print(f"Element at ({coords.x0}, {coords.y0}) - ({coords.x1}, {coords.y1})")
        print(f"  Type: {element.element_type}")
        print(f"  Text: {element.text[:30]}...")

# Elements by page
from collections import defaultdict
elements_by_page = defaultdict(list)
for element in result.elements:
    page = element.metadata.page_number or 0
    elements_by_page[page].append(element)

for page_num, page_elements in sorted(elements_by_page.items()):
    print(f"Page {page_num}: {len(page_elements)} elements")

from kreuzberg import extract_file_sync, ExtractionConfig

config = ExtractionConfig(output_format="element_based")
result = extract_file_sync("report.pdf", config=config)

# Build table of contents from titles and headings
toc = []
for element in result.elements:
    if element.element_type == "title" or element.element_type == "heading":
        level = element.metadata.additional.get("level", "h1")
        page = element.metadata.page_number or "?"
        toc.append({
            "level": level,
            "text": element.text,
            "page": page
        })

# Print table of contents
for entry in toc:
    indent = "  " * (int(entry["level"][1]) - 1)  # h1=0, h2=1, h3=2, etc.
    print(f"{indent}{entry['text']} (p. {entry['page']})")

from kreuzberg import extract_file_sync, ExtractionConfig
import json

config = ExtractionConfig(output_format="element_based")
result = extract_file_sync("article.pdf", config=config)

# Build JSON-LD structure
jsonld = {
    "@context": "http://schema.org",
    "@type": "ScholarlyArticle",
    "headline": None,
    "author": [],
    "articleBody": []
}

for element in result.elements:
    if element.element_type == "title" and not jsonld["headline"]:
        jsonld["headline"] = element.text
    elif element.element_type == "narrative_text":
        jsonld["articleBody"].append({
            "@type": "TextObject",
            "text": element.text,
            "position": element.metadata.element_index
        })

print(json.dumps(jsonld, indent=2))

from kreuzberg import extract_file_sync, ExtractionConfig

config = ExtractionConfig(output_format="element_based")
result = extract_file_sync("book.pdf", config=config)

# Chunk by sections (title + following paragraphs)
chunks = []
current_chunk = {"title": None, "content": []}

for element in result.elements:
    if element.element_type == "title":
        # Save previous chunk if it has content
        if current_chunk["title"] or current_chunk["content"]:
            chunks.append(current_chunk)
        # Start new chunk
        current_chunk = {"title": element.text, "content": []}
    elif element.element_type == "narrative_text":
        current_chunk["content"].append(element.text)

# Save last chunk
if current_chunk["title"] or current_chunk["content"]:
    chunks.append(current_chunk)

# Print chunks
for i, chunk in enumerate(chunks):
    title = chunk["title"] or "[No Title]"
    content_preview = " ".join(chunk["content"])[:100]
    print(f"Chunk {i}: {title}")
    print(f"  Content: {content_preview}...")

from kreuzberg import extract_file_sync, ExtractionConfig
from typing import List, Dict

def index_elements(pdf_path: str) -> List[Dict]:
    """Extract and index all elements from a PDF."""
    config = ExtractionConfig(output_format="element_based")
    result = extract_file_sync(pdf_path, config=config)

    index = []
    for element in result.elements:
        index.append({
            "id": element.element_id,
            "type": element.element_type,
            "text": element.text,
            "page": element.metadata.page_number,
            "searchable": element.text.lower()  # For simple search
        })

    return index

def search_elements(index: List[Dict], query: str) -> List[Dict]:
    """Search elements by text content."""
    query_lower = query.lower()
    return [elem for elem in index if query_lower in elem["searchable"]]

# Usage
index = index_elements("knowledge-base.pdf")
results = search_elements(index, "machine learning")

for result in results:
    print(f"[{result['type']}] (Page {result['page']}): {result['text'][:80]}...")

from kreuzberg import extract_file_sync, ExtractionConfig
from typing import Dict, List

def extract_research_paper_structure(pdf_path: str) -> Dict:
    """Extract structured data from a research paper."""
    config = ExtractionConfig(output_format="element_based")
    result = extract_file_sync(pdf_path, config=config)

    structure = {
        "title": None,
        "authors": [],
        "abstract": None,
        "sections": [],
        "tables": [],
        "images": [],
        "references": []
    }

    current_section = None
    in_abstract = False
    in_references = False

    for element in result.elements:
        if element.element_type == "title":
            level = element.metadata.additional.get("level")
            if not structure["title"] and level == "h1":
                structure["title"] = element.text
            elif "abstract" in element.text.lower():
                in_abstract = True
                in_references = False
            elif "references" in element.text.lower():
                in_abstract = False
                in_references = True
            else:
                current_section = {
                    "heading": element.text,
                    "level": level,
                    "content": []
                }
                structure["sections"].append(current_section)
                in_abstract = False
                in_references = False

        elif element.element_type == "narrative_text":
            if in_abstract:
                structure["abstract"] = element.text
            elif in_references:
                structure["references"].append(element.text)
            elif current_section:
                current_section["content"].append(element.text)

        elif element.element_type == "table":
            structure["tables"].append({
                "text": element.text,
                "page": element.metadata.page_number,
                "dimensions": (
                    element.metadata.additional.get("row_count"),
                    element.metadata.additional.get("column_count")
                )
            })

        elif element.element_type == "image":
            structure["images"].append({
                "format": element.metadata.additional.get("format"),
                "dimensions": (
                    element.metadata.additional.get("width"),
                    element.metadata.additional.get("height")
                ),
                "page": element.metadata.page_number
            })

    return structure

# Usage
paper = extract_research_paper_structure("research-paper.pdf")
print(f"Title: {paper['title']}")
print(f"Sections: {len(paper['sections'])}")
print(f"Tables: {len(paper['tables'])}")
print(f"Images: {len(paper['images'])}")

from kreuzberg import extract_file_sync, ExtractionConfig
from typing import List, Dict

def build_document_hierarchy(pdf_path: str) -> Dict:
    """Build hierarchical structure from flat element list."""
    config = ExtractionConfig(output_format="element_based")
    result = extract_file_sync(pdf_path, config=config)

    root = {"type": "root", "children": []}
    stack = [root]  # Stack of current hierarchy

    for element in result.elements:
        if element.element_type in ["title", "heading"]:
            level_str = element.metadata.additional.get("level", "h1")
            level = int(level_str[1])  # h1 → 1, h2 → 2, etc.

            # Pop stack until we find appropriate parent
            while len(stack) > level:
                stack.pop()

            node = {
                "type": "section",
                "level": level,
                "title": element.text,
                "children": []
            }
            stack[-1]["children"].append(node)
            stack.append(node)

        elif element.element_type == "narrative_text":
            node = {
                "type": "paragraph",
                "text": element.text
            }
            stack[-1]["children"].append(node)

        elif element.element_type in ["table", "image", "code_block"]:
            node = {
                "type": element.element_type,
                "text": element.text,
                "metadata": element.metadata.additional
            }
            stack[-1]["children"].append(node)

    return root

# Usage
hierarchy = build_document_hierarchy("documentation.pdf")

def print_hierarchy(node, indent=0):
    """Print hierarchical structure."""
    prefix = "  " * indent
    if node["type"] == "section":
        print(f"{prefix}[{node['level']}] {node['title']}")
    elif node["type"] == "paragraph":
        print(f"{prefix}• {node['text'][:50]}...")
    elif node["type"] in ["table", "image", "code_block"]:
        print(f"{prefix}[{node['type'].upper()}]")

    for child in node.get("children", []):
        print_hierarchy(child, indent + 1)

print_hierarchy(hierarchy)