AI Data Extraction and Processing

import openai, os, json
from PyPDF2 import PdfReader

client = openai.OpenAI(api_key=os.environ["OPENAI_API_KEY"])

def extract_from_pdf(pdf_path: str, extraction_schema: dict) -> dict:
    # 1. Extract text from PDF
    reader = PdfReader(pdf_path)
    text = "\n".join(page.extract_text() or "" for page in reader.pages)

    # 2. Use AI to extract structured data
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": f"""Extract data from this document into the following JSON schema:
{json.dumps(extraction_schema, indent=2)}
Rules:
- Use null for fields that cannot be determined
- Dates in ISO 8601 format
- Numbers as plain floats
- Reply with valid JSON only"""}, {"role": "user", "content": text[:8000]}],
        response_format={"type": "json_object"},
        temperature=0.1
    )
    return json.loads(response.choices[0].message.content)

# Example: Extract invoice data
invoice_schema = {
    "vendor": "string", "invoice_number": "string",
    "date": "ISO date", "total": "number", "currency": "string",
    "line_items": [{"description": "string", "amount": "number"}]
}
# result = extract_from_pdf("invoice.pdf", invoice_schema)

import requests
from bs4 import BeautifulSoup

def scrape_and_extract(url: str, extract_what: str) -> dict:
    # 1. Fetch and clean the page
    resp = requests.get(url, timeout=15, headers={"User-Agent": "Mozilla/5.0"})
    soup = BeautifulSoup(resp.text, "html.parser")

    # Remove scripts, styles, nav, footer
    for tag in soup.find_all(["script", "style", "nav", "footer", "header"]):
        tag.decompose()
    text = soup.get_text(separator="\n", strip=True)[:6000]

    # 2. AI extraction
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": f"Extract the following from this webpage: {extract_what}. Return JSON."},
            {"role": "user", "content": text}
        ],
        response_format={"type": "json_object"},
        temperature=0.1
    )
    return json.loads(response.choices[0].message.content)

# Example: extract product info from a landing page
# product = scrape_and_extract("https://example.com/product",
#     "product name, price, features (array), description")

import base64

def extract_from_image(image_path: str, instructions: str) -> dict:
    with open(image_path, "rb") as f:
        b64 = base64.b64encode(f.read()).decode()

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": [
            {"type": "text", "text": f"{instructions}\nReply with JSON only."},
            {"type": "image_url", "image_url": {"url": f"data:image/png;base64,{b64}"}}
        ]}],
        response_format={"type": "json_object"},
        max_tokens=1000
    )
    return json.loads(response.choices[0].message.content)

# Extract data from a scanned receipt
# receipt = extract_from_image("receipt.png",
#     "Extract: store name, date, items with prices, total")

import os, json, time
from pathlib import Path
from concurrent.futures import ThreadPoolExecutor, as_completed

def process_document_batch(folder: str, schema: dict, max_workers: int = 3) -> list:
    results = []
    files = list(Path(folder).glob("*.pdf"))
    print(f"Processing {len(files)} documents...")

    def process_one(pdf_path):
        try:
            data = extract_from_pdf(str(pdf_path), schema)
            return {"file": pdf_path.name, "status": "success", "data": data}
        except Exception as e:
            return {"file": pdf_path.name, "status": "error", "error": str(e)}

    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = {executor.submit(process_one, f): f for f in files}
        for future in as_completed(futures):
            result = future.result()
            results.append(result)
            print(f"  {result['file']}: {result['status']}")
            time.sleep(0.5)  # rate limit courtesy

    # Save results
    with open("extraction_results.json", "w") as f:
        json.dump(results, f, indent=2, default=str)

    success = sum(1 for r in results if r['status'] == 'success')
    print(f"Done: {success}/{len(results)} successful")
    return results