Python SDK

pip install geonode-scraper-sdk

import os

from geonode_scraper_sdk import Configuration

configuration = Configuration(api_key={"ApiKeyAuth": os.environ["GEONODE_SCRAPER_API_KEY"]})

from geonode_scraper_sdk import ApiClient, ExtractionApi, ExtractRequest, OutputFormat

with ApiClient(configuration) as api_client:
    api = ExtractionApi(api_client)
    response = api.extract_v1_extract_post(
        ExtractRequest(url="https://example.com", formats=[OutputFormat.MARKDOWN])
    )
    print(response.data.markdown)
    print(response.tokens_charged)  # what this request cost

from geonode_scraper_sdk import ExtractRequest, OutputFormat, ProxySettings, ProxyType, WaitConfig, WaitUntil

request = ExtractRequest(
    url="https://example.com/products",
    formats=[OutputFormat.MARKDOWN],
    render_js=True,                                   # use a headless browser
    proxy=ProxySettings(country="DE", type=ProxyType.RESIDENTIAL),
    wait_config=WaitConfig(wait_until=WaitUntil.NETWORKIDLE, wait_for="#content", wait_timeout=2000),
    timeout=60000,                                    # give up after 60 s
)

import time

from geonode_scraper_sdk import ApiClient, ExtractionApi, ExtractRequest, JobStatus, ProcessingMode

DONE = {JobStatus.COMPLETED, JobStatus.FAILED, JobStatus.CANCELLED}

with ApiClient(configuration) as api_client:
    api = ExtractionApi(api_client)
    accepted = api.extract_v1_extract_post(
        ExtractRequest(url="https://example.com", processing_mode=ProcessingMode.ASYNC)
    )
    job = api.get_job_result_v1_extract_job_id_get(accepted.job_id)
    while job.status not in DONE:
        time.sleep(2)
        job = api.get_job_result_v1_extract_job_id_get(accepted.job_id)

    if job.status == JobStatus.COMPLETED:
        print(job.data.html)
    else:
        print(job.status, job.error)

from geonode_scraper_sdk import ApiClient, BatchApi, BatchRequest, OutputFormat

with ApiClient(configuration) as api_client:
    api = BatchApi(api_client)
    accepted = api.create_batch_v1_batch_post(
        BatchRequest(urls=["https://example.com", "https://example.org"], formats=[OutputFormat.MARKDOWN])
    )
    status = api.get_batch_status_v1_batch_job_id_get(job_id=accepted.job_id, page=1, page_size=50)
    print(status.status, f"{status.completed_urls}/{status.total_urls}")
    for item in status.results or []:
        print(item.url, item.status)

from geonode_scraper_sdk import ApiClient, CrawlApi, CrawlRequest, OutputFormat

with ApiClient(configuration) as api_client:
    api = CrawlApi(api_client)
    accepted = api.create_crawl_v1_crawl_post(
        CrawlRequest(url="https://example.com", depth=2, limit=50, formats=[OutputFormat.MARKDOWN])
    )
    print(accepted.job_id, accepted.estimated_pages)
    status = api.get_crawl_status_v1_crawl_job_id_get(job_id=accepted.job_id, page=1, page_size=50)
    print(status.status, f"{status.completed_pages}/{status.total_pages}")

from geonode_scraper_sdk import ApiClient, MapApi, MapRequest

with ApiClient(configuration) as api_client:
    result = MapApi(api_client).map_urls_v1_map_post(MapRequest(url="https://example.com"))
    for link in result.links or []:
        print(link.url, link.source)

from geonode_scraper_sdk import ApiClient, SearchApi, SearchRequest

with ApiClient(configuration) as api_client:
    result = SearchApi(api_client).search_v1_search_post(SearchRequest(query="geonode scraper api"))
    for hit in result.results:
        print(hit.position, hit.title, hit.url)

from geonode_scraper_sdk import ApiClient, WebhookCreate, WebhookEventType, WebhooksApi

with ApiClient(configuration) as api_client:
    webhook = WebhooksApi(api_client).create_webhook_v1_webhooks_post(
        WebhookCreate(url="https://your-app.example/hooks/geonode", event_type=WebhookEventType.EXTRACT_COMPLETED)
    )
    print(webhook.id, webhook.secret)  # store the secret: it signs every delivery

import hashlib
import hmac


def is_from_geonode(raw_body: bytes, signature_header: str, secret: str) -> bool:
    expected = "sha256=" + hmac.new(secret.encode(), raw_body, hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, signature_header)

from datetime import datetime, timedelta, timezone

from geonode_scraper_sdk import ApiClient, StatisticsApi, UsageApi

with ApiClient(configuration) as api_client:
    usage = UsageApi(api_client).get_concurrency_usage_v1_usage_concurrency_get()
    print(f"{usage.work_concurrency_in_use} of {usage.work_concurrency_limit} concurrent requests in use")

    week_ago = datetime.now(timezone.utc) - timedelta(days=7)
    stats = StatisticsApi(api_client).get_statistics_v1_statistics_get(start_date=week_ago)
    print(stats.extraction_count, stats.success_rate, stats.period_tokens_used)

from geonode_scraper_sdk import ApiClient, ApiException, ExtractionApi, ExtractRequest

with ApiClient(configuration) as api_client:
    try:
        ExtractionApi(api_client).extract_v1_extract_post(ExtractRequest(url="https://example.com"))
    except ApiException as exc:
        print(exc.status, exc.body)
        if exc.status == 429:
            print("retry after", exc.headers.get("Retry-After"), "seconds")