Data Extractor

from unstructured.partition.auto import partition
# Automatically detect and process any documentelements = partition("document.pdf")
# Access extracted elementsfor element in elements:    print(f"Type: {type(element).__name__}")    print(f"Text: {element.text}")    print(f"Metadata: {element.metadata}")

from unstructured.documents.elements import (    Title,    NarrativeText,    Text,    ListItem,    Table,    Image,    Header,    Footer,    PageBreak,    Address,    EmailAddress,)
# Elements have consistent structureelement.text           # Raw text contentelement.metadata       # Rich metadataelement.category       # Element typeelement.id            # Unique identifier

from unstructured.partition.auto import partition
# Process any file typeelements = partition(    filename="document.pdf",    strategy="auto",          # or "fast", "hi_res", "ocr_only"    include_metadata=True,    include_page_breaks=True,)
# Filter by typetitles = [e for e in elements if isinstance(e, Title)]tables = [e for e in elements if isinstance(e, Table)]

# PDF with optionsfrom unstructured.partition.pdf import partition_pdf
elements = partition_pdf(    filename="document.pdf",    strategy="hi_res",              # High quality extraction    infer_table_structure=True,     # Detect tables    include_page_breaks=True,    languages=["en"],               # OCR language)
# Word documentsfrom unstructured.partition.docx import partition_docx
elements = partition_docx(    filename="document.docx",    include_metadata=True,)
# HTMLfrom unstructured.partition.html import partition_html
elements = partition_html(    filename="page.html",    include_metadata=True,)

from unstructured.partition.auto import partition
elements = partition("report.pdf", infer_table_structure=True)
# Extract tablesfor element in elements:    if element.category == "Table":        print("Table found:")        print(element.text)                # Access structured table data        if hasattr(element, 'metadata') and element.metadata.text_as_html:            print("HTML:", element.metadata.text_as_html)

from unstructured.partition.auto import partition
elements = partition("document.pdf")
for element in elements:    meta = element.metadata        # Common metadata fields    print(f"Page: {meta.page_number}")    print(f"Filename: {meta.filename}")    print(f"Filetype: {meta.filetype}")    print(f"Coordinates: {meta.coordinates}")    print(f"Languages: {meta.languages}")

from unstructured.partition.auto import partitionfrom unstructured.chunking.title import chunk_by_titlefrom unstructured.chunking.basic import chunk_elements
# Partition documentelements = partition("document.pdf")
# Chunk by title (semantic chunks)chunks = chunk_by_title(    elements,    max_characters=1000,    combine_text_under_n_chars=200,)
# Or basic chunkingchunks = chunk_elements(    elements,    max_characters=500,    overlap=50,)
for chunk in chunks:    print(f"Chunk ({len(chunk.text)} chars):")    print(chunk.text[:100] + "...")

from unstructured.partition.auto import partitionfrom pathlib import Pathfrom concurrent.futures import ThreadPoolExecutor
def process_document(file_path):    """Process single document."""    try:        elements = partition(str(file_path))        return {            'file': str(file_path),            'status': 'success',            'elements': len(elements),            'text': '\n\n'.join([e.text for e in elements])        }    except Exception as e:        return {            'file': str(file_path),            'status': 'error',            'error': str(e)        }
def batch_process(input_dir, max_workers=4):    """Process all documents in directory."""    input_path = Path(input_dir)    files = list(input_path.glob('*'))        with ThreadPoolExecutor(max_workers=max_workers) as executor:        results = list(executor.map(process_document, files))        return results

from unstructured.partition.auto import partitionfrom unstructured.staging.base import elements_to_json, elements_to_dicts
elements = partition("document.pdf")
# To JSON stringjson_str = elements_to_json(elements)
# To list of dictsdicts = elements_to_dicts(elements)
# To DataFrameimport pandas as pddf = pd.DataFrame(dicts)

def document_to_json(file_path, output_path=None):    """Convert document to structured JSON."""    from unstructured.partition.auto import partition    from unstructured.staging.base import elements_to_json    import json        elements = partition(file_path)        # Create structured output    output = {        'source': file_path,        'elements': []    }        for element in elements:        output['elements'].append({            'type': type(element).__name__,            'text': element.text,            'metadata': {                'page': element.metadata.page_number,                'coordinates': element.metadata.coordinates.to_dict() if element.metadata.coordinates else None            }        })        if output_path:        with open(output_path, 'w') as f:            json.dump(output, f, indent=2)        return output

from unstructured.partition.email import partition_email
def parse_email(email_path):    """Extract structured data from email."""        elements = partition_email(email_path)        email_data = {        'subject': None,        'from': None,        'to': [],        'date': None,        'body': [],        'attachments': []    }        for element in elements:        meta = element.metadata                # Extract headers from metadata        if meta.subject:            email_data['subject'] = meta.subject        if meta.sent_from:            email_data['from'] = meta.sent_from        if meta.sent_to:            email_data['to'] = meta.sent_to                # Body content        email_data['body'].append({            'type': type(element).__name__,            'text': element.text        })        return email_data

from unstructured.partition.pdf import partition_pdffrom unstructured.chunking.title import chunk_by_title
def extract_paper(pdf_path):    """Extract structured data from research paper."""        elements = partition_pdf(        filename=pdf_path,        strategy="hi_res",        infer_table_structure=True,        include_page_breaks=True    )        paper = {        'title': None,        'abstract': None,        'sections': [],        'tables': [],        'references': []    }        # Find title (usually first Title element)    for element in elements:        if element.category == "Title" and not paper['title']:            paper['title'] = element.text            break        # Extract tables    for element in elements:        if element.category == "Table":            paper['tables'].append({                'page': element.metadata.page_number,                'content': element.text,                'html': element.metadata.text_as_html if hasattr(element.metadata, 'text_as_html') else None            })        # Chunk into sections    chunks = chunk_by_title(elements, max_characters=2000)        current_section = None    for chunk in chunks:        if chunk.category == "Title":            paper['sections'].append({                'title': chunk.text,                'content': ''            })        elif paper['sections']:            paper['sections'][-1]['content'] += chunk.text + '\n'        return paper
paper = extract_paper('research_paper.pdf')print(f"Title: {paper['title']}")print(f"Tables: {len(paper['tables'])}")print(f"Sections: {len(paper['sections'])}")

from unstructured.partition.auto import partitionimport re
def extract_invoice_data(file_path):    """Extract key data from invoice."""        elements = partition(file_path, strategy="hi_res")        # Combine all text    full_text = '\n'.join([e.text for e in elements])        invoice = {        'invoice_number': None,        'date': None,        'total': None,        'vendor': None,        'line_items': [],        'tables': []    }        # Extract patterns    inv_match = re.search(r'Invoice\s*#?\s*:?\s*(\w+[-\w]*)', full_text, re.I)    if inv_match:        invoice['invoice_number'] = inv_match.group(1)        date_match = re.search(r'Date\s*:?\s*(\d{1,2}[-/]\d{1,2}[-/]\d{2,4})', full_text, re.I)    if date_match:        invoice['date'] = date_match.group(1)        total_match = re.search(r'Total\s*:?\s*\$?([\d,]+\.?\d*)', full_text, re.I)    if total_match:        invoice['total'] = float(total_match.group(1).replace(',', ''))        # Extract tables    for element in elements:        if element.category == "Table":            invoice['tables'].append(element.text)        return invoice
invoice = extract_invoice_data('invoice.pdf')print(f"Invoice #: {invoice['invoice_number']}")print(f"Total: ${invoice['total']}")

from unstructured.partition.auto import partitionfrom unstructured.chunking.title import chunk_by_titlefrom pathlib import Pathimport json
def build_corpus(input_dir, output_path):    """Build searchable corpus from document collection."""        input_path = Path(input_dir)    corpus = []        # Support multiple formats    patterns = ['*.pdf', '*.docx', '*.html', '*.txt', '*.md']    files = []    for pattern in patterns:        files.extend(input_path.glob(pattern))        for file in files:        print(f"Processing: {file.name}")                try:            elements = partition(str(file))            chunks = chunk_by_title(elements, max_characters=1000)                        for i, chunk in enumerate(chunks):                corpus.append({                    'id': f"{file.stem}_{i}",                    'source': str(file),                    'type': type(chunk).__name__,                    'text': chunk.text,                    'page': chunk.metadata.page_number if chunk.metadata.page_number else None                })                except Exception as e:            print(f"  Error: {e}")        # Save corpus    with open(output_path, 'w') as f:        json.dump(corpus, f, indent=2)        print(f"Corpus built: {len(corpus)} chunks from {len(files)} files")    return corpus
corpus = build_corpus('./documents', 'corpus.json')