Metadata-Version: 2.3
Name: divparser
Version: 0.4.0
Summary: Python SDK for DivParser API - Web scraping and HTML parsing with AI-powered extraction
Author: solomon344
Author-email: solomon344 <willamssolomon672@gmail.com>
Requires-Dist: requests>=2.28.0
Requires-Dist: openpyxl>=3.1.0
Requires-Dist: html2text>=2024.2.26
Requires-Python: >=3.8
Description-Content-Type: text/markdown

# DivParser Python SDK

A Python SDK for [DivParser](https://www.divparser.com/) - AI-powered web scraping and HTML parsing.

## Features

- **Web Scraping**: Extract structured data from web pages
- **Fetch-then-Extract**: Fetch a page first, decide on the extraction schema afterwards
- **HTML Parsing**: Parse raw HTML content directly
- **Schedules**: Create and manage recurring scrapes (cron or interval cadence)
- **Async Job Handling**: Non-blocking job submission with status polling
- **Pagination Support**: Scrape multiple URLs in a single batch
- **Export**: Save results as JSON, CSV, or XLSX, and fetched HTML as-is or converted to Markdown
- **Delivery**: Deliver results directly to a connected S3/Google Drive/Dropbox Storage
- **Simple API**: Pythonic interface to the DivParser REST API

## Installation

```bash
pip install divparser
```

Or if using `uv`:

```bash
uv pip install divparser
```

## Quick Start

### Setup

```python
from divparser import DivParser

# Initialize the client with your API key
client = DivParser(api_key="your_api_key_here")
```

Get your API key from [DivParser Console](https://divparser.com/dashboard/settings).

### Scraping a Web Page

```python
# Scrape a single page and wait for results
result = client.scrape_and_parse(
    url="https://example.com/products",
    schema="Extract product name, price, and rating from each item"
)

# Access the extracted data
for item in result["results"][0]["data"]:
    print(item)
```

### Parsing HTML Content

```python
# Parse HTML content directly
html_content = "<html><body><h1>Title</h1><p>Content</p></body></html>"

result = client.parse_and_wait(
    html=html_content,
    schema="Extract all headings and paragraphs"
)

# Get the parsed data
data = result["results"][0]["data"]
print(data)
```

### Paginated Scraping

```python
# Scrape multiple URLs
urls = [
    "https://example.com/page/1",
    "https://example.com/page/2",
    "https://example.com/page/3"
]

result = client.scrape_paginated(
    urls=urls,
    schema="Extract product name and price",
    wait=True
)

# Combine results from all pages
from divparser.utils import flatten_results
all_items = flatten_results(result["results"])
```

### Fetch First, Extract Later

```python
# Fetch the page now, decide on a schema afterwards
fetch_result = client.fetch(url="https://example.com/products")
client.wait_for_completion(fetch_result["jobId"])

extracted = client.extract_fetch(
    scrape_id=fetch_result["scrapeId"],
    schema="Extract product name and price"
)
```

### Recurring Schedules

```python
schedule = client.create_schedule(
    name="Daily product check",
    project_id="YOUR_PROJECT_ID",
    schedule={"type": "cron", "pattern": "0 9 * * *"},
    scrape={
        "url": "https://example.com/products",
        "schema": "Extract product name, price, and availability"
    }
)

# List the scrapes this schedule has generated so far
runs = client.list_schedule_runs(schedule["scheduleId"])

# Pause it, resume it, or delete it entirely
client.pause_schedule(schedule["scheduleId"])
client.resume_schedule(schedule["scheduleId"])
client.delete_schedule(schedule["scheduleId"])
```

### Saving Results to Disk

```python
from divparser import save_scrape_as, save_fetch_as

# Save a scrape/parse result as CSV, JSON, or XLSX — usable rows (SUCCESS
# and REQUIRES_ATTENTION) are combined into one flat export automatically.
result = client.get_scrape(scrape_id)
save_scrape_as(result["results"], "csv", "products.csv")

# Save a fetch's raw HTML, or convert it to Markdown first.
fetch_html = client.get_fetch_html(scrape_id)
save_fetch_as(fetch_html["html"], "markdown", "page.md")
```

## API Reference

### Scraping

#### `scrape(url, schema, name=None, page_type="LISTING", wait=False, timeout=300, delivery_config=None)`

Create a scrape job for a single URL.

**Parameters:**
- `url` (str): Target page URL
- `schema` (str): Extraction instructions (plain English or Nestlang)
- `name` (str, optional): Friendly label for this scrape
- `page_type` (str): "LISTING" (default) or "DETAIL"
- `wait` (bool): Wait for completion before returning
- `timeout` (int): Max seconds to wait (only if wait=True)
- `delivery_config` (dict, optional): See [Delivering Results to a Storage](#delivering-results-to-a-storage)

**Returns:** Dictionary with `scrapeId`, `jobId`, and optionally `results`

#### `scrape_paginated(urls, schema, name=None, page_type="LISTING", wait=False, timeout=300, delivery_config=None)`

Create a scrape job for multiple URLs.

**Parameters:**
- `urls` (List[str]): Array of URLs to scrape
- `schema` (str): Extraction instructions
- `name` (str, optional): Friendly label
- `page_type` (str): "LISTING" or "DETAIL"
- `wait` (bool): Wait for completion
- `timeout` (int): Max seconds to wait
- `delivery_config` (dict, optional): See [Delivering Results to a Storage](#delivering-results-to-a-storage)

**Returns:** Dictionary with `scrapeId`, `jobId`, and optionally `results`

#### `list_scrapes(limit=20, cursor=None)`

List all scrapes for the authenticated user.

**Returns:** Dictionary with list of scrapes and pagination info

#### `get_scrape(scrape_id)`

Retrieve a scrape and its results by ID.

**Parameters:**
- `scrape_id` (str): The scrapeId from creation

**Returns:** Dictionary with scrape details and results

### Parsing

#### `parse(html, schema, name=None, wait=False, timeout=300)`

Submit raw HTML for structured extraction.

**Parameters:**
- `html` (str): Full HTML content to parse
- `schema` (str): Extraction instructions
- `name` (str, optional): Friendly label
- `wait` (bool): Wait for completion
- `timeout` (int): Max seconds to wait

**Returns:** Dictionary with `scrapeId`, `jobId`, and optionally `results`

#### `get_parse(parse_id)`

Retrieve results for a completed parse job.

**Parameters:**
- `parse_id` (str): The scrapeId from parse creation

**Returns:** Dictionary with parse details and results

### Fetching

#### `fetch(url, name=None, project_id=None, proxy_mode=None, delivery_config=None)`

Fetch a URL only, with no extraction.

**Parameters:**
- `url` (str): Target page URL
- `name` (str, optional): Friendly label
- `project_id` (str, optional): Project to attach this fetch to
- `proxy_mode` (str, optional): `"residential"`, `"unblocker"`, or `"http"`
- `delivery_config` (dict, optional): See [Delivering Results to a Storage](#delivering-results-to-a-storage) — fetch content is raw HTML/Markdown, never tabular

**Returns:** Dictionary with `scrapeId`, `jobId`, and `message`

**Note:** this endpoint reports every failure as a 402 (not just insufficient credits) —
check the response's `message`, not just the status code, to tell failures apart.

#### `get_fetch_html(scrape_id)`

Retrieve the raw HTML for a fetched scrape. Only meaningful for a scrape created via `fetch()` —
`get_scrape()` never includes this (it's not an extraction result).

**Parameters:**
- `scrape_id` (str): The scrapeId returned from `fetch()`

**Returns:** Dictionary with `id` and `html`

#### `extract_fetch(scrape_id, schema, delivery_config=None)`

Attach a schema to a previously-fetched scrape and run instant (AI) extraction against its
already-stored HTML.

**Parameters:**
- `scrape_id` (str): The scrapeId returned from `fetch()`
- `schema` (str): Extraction instructions
- `delivery_config` (dict, optional): See [Delivering Results to a Storage](#delivering-results-to-a-storage) — the extracted result is tabular, so the full destination set applies

**Returns:** Dictionary with `scrapeId`, `jobId`, and `message`

### Scheduling

#### `create_schedule(name, project_id, schedule, scrape, iterations=None, delivery_config=None)`

Create a recurring schedule that repeats a template scrape.

**Parameters:**
- `name` (str): Friendly label for this schedule
- `project_id` (str): Project to attach this schedule to
- `schedule` (dict): `{"type": "cron", "pattern": "0 9 * * *"}` or `{"type": "interval", "every": <ms>}`
- `scrape` (dict): Template scrape, e.g. `{"url": ..., "schema": ..., "name": ..., "pageType": ...}`
- `iterations` (int, optional): Cap on how many runs this schedule performs
- `delivery_config` (dict, optional): See [Delivering Results to a Storage](#delivering-results-to-a-storage) — applies to every future run this schedule generates, not just the template

**Returns:** Dictionary with `scheduleId`, `templateScrapeId`, `status`, and `message`

#### `list_schedules()`

List all schedules for the authenticated user. Not paginated.

#### `get_schedule(schedule_id)`

Retrieve a single schedule by ID.

#### `pause_schedule(schedule_id)` / `resume_schedule(schedule_id)`

Pause or resume a schedule's recurring runs.

#### `delete_schedule(schedule_id)`

Stop and permanently delete a schedule.

#### `list_schedule_runs(schedule_id, limit=20, cursor=None)`

List the scrapes a schedule has generated so far, cursor-paginated (same shape as `list_scrapes`).

### Utilities

#### `check_status(job_id)`

Poll the status of a job.

**Parameters:**
- `job_id` (str): The jobId returned from creation

**Returns:** Dictionary with `completed` (bool) and `state` (str)

#### `wait_for_completion(job_id, timeout=300, poll_interval=1.0)`

Wait for a job to complete.

**Parameters:**
- `job_id` (str): The jobId to poll
- `timeout` (int): Max seconds to wait
- `poll_interval` (float): Seconds between polls

**Returns:** Status dictionary when completed

**Raises:** `TimeoutError` if job doesn't complete

## Delivering Results to a Storage

`scrape()`, `scrape_paginated()`, `fetch()`, `extract_fetch()`, and `create_schedule()` all accept
an optional `delivery_config`, which uploads the completed run's data to a destination you've
already connected via the dashboard's Storages page — the API can't accept raw S3/Google
Drive/Dropbox credentials inline, only reference an already-connected one. A destination that
isn't connected (or isn't active) is silently dropped, not an error.

```python
from divparser.constants import DESTINATION_S3, DESTINATION_GOOGLE_DRIVE, DESTINATION_DROPBOX

client.scrape(
    url, schema,
    delivery_config={
        "destinations": [DESTINATION_S3, DESTINATION_GOOGLE_DRIVE],
        "format": "csv"  # "json" (default) | "csv" | "xlsx"
    }
)

# create_schedule's delivery_config applies to every future run, not just the template.
client.create_schedule(
    name="Daily check",
    project_id=project_id,
    schedule={"type": "cron", "pattern": "0 9 * * *"},
    scrape={"url": url, "schema": schema},
    delivery_config={"destinations": [DESTINATION_DROPBOX]}
)
```

`fetch()`'s content is raw HTML/Markdown, never tabular — use `format: "html"` or `"markdown"`
instead.

## Utility Functions

The `divparser.utils` module provides helper functions for working with results. Every result
has one of three statuses: `SUCCESS`, `FAILED`, or `REQUIRES_ATTENTION` (the long-term parser's
self-healing selectors came back under-confident — data still came back, it's just worth a second
look). `extract_data_from_results()` and `flatten_results()` include both `SUCCESS` and
`REQUIRES_ATTENTION` rows since both carry real data; only `FAILED` rows are skipped.

```python
from divparser.utils import (
    extract_data_from_results,
    flatten_results,
    filter_results_by_status,
    get_results_by_url,
    count_requires_attention_results,
    get_result_stats
)

# Flatten nested results (SUCCESS + REQUIRES_ATTENTION rows only)
all_items = flatten_results(results)

# Get statistics
stats = get_result_stats(results)
print(f"Success rate: {stats['success_rate']:.1f}%")
print(f"Usable rate (including needs-review rows): {stats['usable_rate']:.1f}%")
if stats["requires_attention"]:
    print(f"{stats['requires_attention']} result(s) need a second look")

# Group by URL
by_url = get_results_by_url(results)
```

## Saving & Converting Data

The `divparser` package (and `divparser.convert` for the lower-level primitives) provide
export helpers:

```python
from divparser import save_scrape_as, save_fetch_as, to_json, to_csv, to_xlsx, html_to_markdown
```

### `save_scrape_as(results, format, path)`

Save a scrape/parse result to disk as `"json"`, `"csv"`, or `"xlsx"`. `results` is the `results`
list from `get_scrape()`/`get_parse()` — rows with usable data (`SUCCESS` and
`REQUIRES_ATTENTION`) are combined into one flat export via `flatten_results()`; `FAILED` rows
are skipped.

### `save_fetch_as(html, format, path)`

Save a fetch's raw HTML to disk, either `"html"` as-is or converted to `"markdown"`.

### `to_json(data)` / `to_csv(data)` / `to_xlsx(data, sheet_name="ScrapedData")` / `html_to_markdown(html)`

Lower-level conversion primitives that return the converted string/bytes instead of writing a
file — use these if you want to stream, upload, or otherwise handle the converted data yourself,
or are working with a custom dataset instead of a full `results` list.

## Examples

### Example 1: Extract Job Listings

```python
from divparser import DivParser

client = DivParser(api_key="your_api_key")

result = client.scrape_and_parse(
    url="https://example-jobs.com/listings",
    schema="""
    Extract the following for each job:
    - job title
    - company name
    - location
    - salary range (if available)
    """,
    name="Job Listings Scrape"
)

for job in result["results"][0]["data"]:
    print(f"{job['title']} at {job['company']} in {job['location']}")
```

### Example 2: Parse Product Information from HTML

```python
html_content = """
<html>
<body>
    <div class="product">
        <h2>Widget Pro</h2>
        <p class="price">$49.99</p>
        <p class="rating">4.5 stars</p>
    </div>
    <div class="product">
        <h2>Widget Lite</h2>
        <p class="price">$19.99</p>
        <p class="rating">4.2 stars</p>
    </div>
</body>
</html>
"""

result = client.parse_and_wait(
    html=html_content,
    schema="Extract product name, price, and rating"
)

for product in result["results"][0]["data"]:
    print(f"{product['name']}: {product['price']} ({product['rating']})")
```

### Example 3: Batch Scraping Multiple Pages

```python
from divparser.utils import flatten_results

pages = [f"https://example.com/products?page={i}" for i in range(1, 4)]

result = client.scrape_paginated(
    urls=pages,
    schema="Extract product ID, name, and price"
)

# Get all products from all pages
all_products = flatten_results(result["results"])
print(f"Total products: {len(all_products)}")
```

## Error Handling

```python
from divparser import DivParser
import requests

client = DivParser(api_key="your_api_key")

try:
    result = client.scrape_and_parse(
        url="https://example.com",
        schema="Extract content"
    )
except requests.exceptions.HTTPError as e:
    print(f"API Error: {e}")
except TimeoutError as e:
    print(f"Job timed out: {e}")
```

## Best Practices

1. **Use Descriptive Schemas**: Clear instructions in your schema lead to better extraction
2. **Set Appropriate Timeouts**: Complex extractions may need longer timeouts
3. **Batch Operations**: Use `scrape_paginated` for multiple URLs instead of individual requests
4. **Handle Errors**: Always catch exceptions for production code
5. **Reuse Clients**: Create one client instance and reuse it

## API Documentation

For more detailed information, visit [DivParser API Reference](https://www.divparser.com/docs?p=API+Reference).

## License

MIT

## Support

For issues, questions, or feature requests, visit [DivParser Support](https://www.divparser.com/).
