diff --git a/bright_data_scrape/README.md b/bright_data_scrape/README.md new file mode 100644 index 0000000..a9f74ad --- /dev/null +++ b/bright_data_scrape/README.md @@ -0,0 +1,121 @@ +# Bright Data Web Scraper Connector Example + +## Connector overview + +This connector syncs web scraping data from Bright Data's Web Scraper API to your Fivetran destination. Bright Data provides a scalable web scraping platform that allows you to extract data from websites while handling proxies, CAPTCHAs, and other web scraping challenges. The connector retrieves scraped page content, metadata, and extracted data from configured URLs, enabling you to analyze web data in your data warehouse. + +## Requirements + +- [Supported Python versions](https://github.com/fivetran/fivetran_connector_sdk/blob/main/README.md#requirements) +- Operating system: + - Windows: 10 or later (64-bit only) + - macOS: 13 (Ventura) or later (Apple Silicon [arm64] or Intel [x86_64]) + - Linux: Distributions such as Ubuntu 20.04 or later, Debian 10 or later, or Amazon Linux 2 or later (arm64 or x86_64) + +## Getting started + +Refer to the [Connector SDK Setup Guide](https://fivetran.com/docs/connector-sdk/setup-guide) to get started. + +To initialize a new Connector SDK project using this connector as a starting point, run: + +```bash +fivetran init --template connectors/bright_data_serp +``` +```fivetran init``` initializes a new Connector SDK project by setting up the project structure, configuration files, and a connector you can run immediately with ```fivetran debug```. For more information on ```fivetran init```, refer to the [Connector SDK init documentation](https://fivetran.com/docs/connector-sdk/connector-development-and-configuration/connector-sdk-commands#fivetraninit). + +Note: Ensure you have updated the `configuration.json` file with the necessary parameters before running `fivetran debug`. See the [Configuration file](https://github.com/fivetran/community_connectors/pull/41#configuration-file) section for details on the required configuration parameters. + +## Features + +- Syncs scraped web page data including content, metadata, and extracted fields +- Supports multiple URL input formats (single URL, comma-separated, newline-separated, JSON array) +- Processes URLs in batch for efficient data collection +- Handles asynchronous scraping jobs with automatic polling for completion +- Dynamically discovers schema fields from scraped data +- Flattens nested JSON structures for easy analysis +- Supports incremental syncing with state management +- Includes retry logic with exponential backoff for API rate limits and transient errors + +## Configuration file + +```json +{ + "api_token": "", + "dataset_id": "", + "scrape_url": "" +} +``` + +Configuration parameters: + +- `api_token` (required): Your Bright Data API token (Bearer token format, obtained from Bright Data dashboard) +- `dataset_id` (required): The ID of the Bright Data dataset to use for scraping (e.g., `"gd_lyy3tktm25m4avu764"`) +- `scrape_url` (required): URL(s) to scrape. Supports multiple formats: + - Single URL: `"https://www.example.com"` + - Comma-separated: `"https://www.example.com,https://www.example2.com"` + - Newline-separated: `"https://www.example.com\nhttps://www.example2.com"` + - JSON array string: `"[\"https://www.example.com\",\"https://www.example2.com\"]"` + + +> Note: When submitting connector code as a [Community Connector](https://github.com/fivetran/community_connectors/tree/main) in the open-source [Connector SDK repository](https://github.com/fivetran/community_connectors/tree/main), ensure the `configuration.json` file has placeholder values. When adding the connector to your production repository, ensure that the `configuration.json` file is not checked into version control to protect sensitive information. + + +## Requirements file + +This connector does not require any additional Python packages beyond what is pre-installed in the Fivetran environment. + +> Note: [Some packages](https://fivetran.com/docs/connector-sdk/technical-reference#preinstalledpackages) are pre-installed in the Connector SDK runtime environment. To avoid dependency conflicts, do not declare them in your `requirements.txt`. + + +## Authentication + +The Bright Data API uses Bearer token authentication. To obtain your API token: + +1. Visit the [Bright Data website](https://brightdata.com). +2. Create an account or log in to your existing account. +3. Navigate to **Settings > Users** or visit https://brightdata.com/cp/setting/users. +4. Generate and make a note of your API token. +5. Add the API token to your `configuration.json` file as the value for the `api_token` parameter. + +The API token is included in all API requests as a Bearer token in the Authorization header (refer to the `perform_scrape()` function in `helpers/scrape.py`). + +## Data handling + +The connector processes data in the following order: + +1. URL Parsing - Normalizes the `scrape_url` input into a list of URLs, supporting multiple input formats (refer to the `parse_scrape_urls()` function) +2. Job Triggering - Sends a POST request to `/datasets/v3/trigger` with the URLs to scrape (refer to the `_trigger_scrape()` function in `helpers/scrape.py`) +3. Snapshot Polling - Polls the `/datasets/v3/snapshot/{snapshot_id}` endpoint until results are ready (refer to the `_poll_snapshot()` function in `helpers/scrape.py`) +4. Result Processing - Flattens nested JSON structures and processes each result (refer to the `process_scrape_results()` and `process_and_upsert_results()` functions) +5. Schema Discovery - Dynamically discovers fields from scraped data to build consistent upsert rows (refer to the `collect_all_fields()` function in `helpers/data_processing.py`) +6. Data Upsertion - Upserts processed records to the destination table (refer to the `process_and_upsert_results()` function) + +All data is upserted to the destination, allowing for incremental updates. Records are uniquely identified by the combination of `url` and `result_index`, which form the primary key. The connector tracks state for synced URLs to enable efficient incremental syncs. + +## Error handling + +The connector implements comprehensive error handling: + +- Retry logic - Transient errors (408, 429, 500, 502, 503, 504) trigger exponential backoff retries up to 3 attempts (refer to the retry logic in `helpers/scrape.py`) +- Timeout handling - Request timeouts are handled with configurable timeout values (default: 120 seconds) +- HTTP error codes - 400/422 errors fail immediately with detailed error messages, 404 errors raise errors for invalid snapshots, other errors are logged and raised (refer to error handling in `helpers/scrape.py`) +- Request exceptions - Network errors trigger retries with exponential backoff +- Snapshot polling - Handles 202 (Accepted) status codes by waiting and retrying, continues polling indefinitely until snapshot is ready or failed + +## Tables created + +| Table Name | Primary Key | Description | +|-------------------|--------------------------------------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| +| `SCRAPE_RESULTS` | `url`, `result_index` | Contains scraped web page data including content, metadata, and extracted fields. Each row represents a single result from a scraped URL. The `url` field identifies the source URL, and `result_index` distinguishes multiple results from the same URL. Nested JSON structures are flattened with underscore separators (e.g., `user_name`, `user_details_age`).| + +## Additional files + +The connector uses the following additional files: + +- `helpers/validation.py` - Configuration parameter validation +- `helpers/scrape.py` - Bright Data API interaction, job triggering, and snapshot polling +- `helpers/data_processing.py` - Data flattening, field discovery, and result processing utilities + +## Additional considerations + +The examples provided are intended to help you effectively use Fivetran's Connector SDK. While we've tested the code, Fivetran cannot be held responsible for any unexpected or negative consequences that may arise from using these examples. For inquiries, please reach out to our Support team. diff --git a/bright_data_scrape/configuration.json b/bright_data_scrape/configuration.json new file mode 100644 index 0000000..74a18ee --- /dev/null +++ b/bright_data_scrape/configuration.json @@ -0,0 +1,6 @@ +{ + "api_token": "", + "dataset_id": "", + "scrape_url": "" +} + diff --git a/bright_data_scrape/connector.py b/bright_data_scrape/connector.py new file mode 100644 index 0000000..c6c9b2a --- /dev/null +++ b/bright_data_scrape/connector.py @@ -0,0 +1,339 @@ +"""This connector syncs web scraping data from Bright Data's Web Scraper API to Fivetran destination. +See the Technical Reference documentation +(https://fivetran.com/docs/connectors/connector-sdk/technical-reference#update) +and the Best Practices documentation +(https://fivetran.com/docs/connectors/connector-sdk/best-practices) for details +""" + +# For reading configuration from a JSON file +import json + +# For parsing URLs +from urllib.parse import urlparse + +# Import required classes from fivetran_connector_sdk +from fivetran_connector_sdk import Connector + +# For enabling Logs in your connector code +from fivetran_connector_sdk import Logging as log + +# For supporting Data operations like upsert(), update(), delete() and checkpoint() +from fivetran_connector_sdk import Operations as op + +# Helper functions for data processing and validation +from helpers import ( + collect_all_fields, + perform_scrape, + process_scrape_result, + validate_configuration, +) + +# Table name constant +__SCRAPE_TABLE = "scrape_results" + +# Linkedin Post By URL dataset ids +LINKEDIN_POST_BY_URL_DATASET_ID = "gd_d85r5d60186q96c883" + + +def schema(configuration: dict): + """ + Define the schema function which lets you configure the schema your connector delivers. + See the technical reference documentation for more details on the schema function: + https://fivetran.com/docs/connectors/connector-sdk/technical-reference#schema + Args: + configuration: a dictionary that holds the configuration settings for the connector. + """ + return [ + { + "table": __SCRAPE_TABLE, + "primary_key": [ + "url", + "result_index", + ], + "columns": { + "url": "STRING", + "result_index": "INT", + }, + } + ] + + +def update(configuration: dict, state: dict): + """ + Define the update function which lets you configure how your connector fetches data. + See the technical reference documentation for more details on the update function: + https://fivetran.com/docs/connectors/connector-sdk/technical-reference#update + Args: + configuration: a dictionary that holds the configuration settings for the connector. + state: a dictionary that holds the state of the connector. + """ + # Validate the configuration to ensure it contains all required values + validate_configuration(configuration=configuration) + + api_token = configuration.get("api_token") + dataset_id = configuration.get("dataset_id") + scrape_url_input = configuration.get("scrape_url", "") + + urls = parse_scrape_urls(scrape_url_input) + + if not urls: + message = ( + f"No URLs provided in configuration; scrape_url input: {scrape_url_input}" + ) + log.error(message) + raise RuntimeError(message) + sync_scrape_urls(api_token, dataset_id, urls, state) + + +def parse_scrape_urls(scrape_url_input): + """ + Parse URLs from configuration input, supporting multiple formats. + Args: + scrape_url_input: The scrape_url configuration value (various formats supported). + Returns: + list: List of URL strings. + """ + if not scrape_url_input: + return [] + + if isinstance(scrape_url_input, list): + return [ + item.strip() + for item in scrape_url_input + if isinstance(item, str) and item.strip() + ] + + if isinstance(scrape_url_input, str): + # Try parsing as JSON first (e.g. '["https://..."]' or '"https://..."') + try: + parsed = json.loads(scrape_url_input) + if isinstance(parsed, list): + return [ + item.strip() + for item in parsed + if isinstance(item, str) and item.strip() + ] + if isinstance(parsed, str) and parsed.strip(): + return [parsed.strip()] + except (json.JSONDecodeError, TypeError): + # Not valid JSON – treat as plain string (single URL or delimited list) + pass + + # Try comma-separated format + if "," in scrape_url_input: + return [ + item.strip() for item in scrape_url_input.split(",") if item.strip() + ] + + # Try newline-separated format + if "\n" in scrape_url_input: + return [ + item.strip() for item in scrape_url_input.split("\n") if item.strip() + ] + + # Single URL (or invalid string – downstream validation can filter) + return [scrape_url_input.strip()] if scrape_url_input.strip() else [] + + return [] + + +def _is_valid_url(url: str) -> bool: + """Return True if the string has a valid URL structure (scheme and netloc).""" + if not url or not isinstance(url, str) or not url.strip(): + return False + parsed = urlparse(url.strip()) + return bool(parsed.scheme and parsed.netloc) + + +def sync_scrape_urls(api_token, dataset_id, urls, state): + """ + Sync scrape results for the requested URLs. + Args: + api_token: Bright Data API token. + dataset_id: ID of the dataset to use for scraping. + urls: List of URLs to scrape (processed in batch by API). + state: State dictionary for tracking sync progress. + """ + valid_urls = [] + for url in urls: + if _is_valid_url(url): + valid_urls.append(url.strip()) + else: + log.warning(f"Skipping invalid URL: {url}") + + if not valid_urls: + log.warning("No valid URLs to sync after filtering invalid entries") + raise RuntimeError("No valid URLs configured for sync") + + log.info(f"Starting scrape sync for {len(valid_urls)} URL(s)") + + # Fetch scrape results for all URLs + # The Bright Data REST API processes URLs and returns results in order + # Apply dataset-specific query parameters when needed + if dataset_id == LINKEDIN_POST_BY_URL_DATASET_ID: + scrape_results = perform_scrape( + api_token=api_token, + dataset_id=dataset_id, + url=valid_urls, + extra_query_params={"discover_by": "profile_url", "type": "discover_new"}, + ) + else: + scrape_results = perform_scrape( + api_token=api_token, + dataset_id=dataset_id, + url=valid_urls, + ) + + # Normalize results to always be a list + if not isinstance(scrape_results, list): + scrape_results = [scrape_results] + + if not scrape_results: + log.warning("No scrape results returned from API") + return + + # Process and flatten results + processed_results = process_scrape_results(scrape_results, valid_urls) + + if not processed_results: + log.warning("No processed results to upsert") + return + + log.info(f"Upserting {len(processed_results)} scrape results to Fivetran") + + all_fields = collect_all_fields(processed_results) + + # Upsert each result + process_and_upsert_results(processed_results, all_fields) + + # Update state with sync information + state["last_scrape_urls"] = valid_urls + state["last_scrape_count"] = len(processed_results) + + op.checkpoint(state) + + log.info(f"Completed scrape sync. Total synced: {len(processed_results)} results") + + +def process_scrape_results(scrape_results, urls): + """ + Process and flatten scrape results. + Args: + scrape_results: List of scrape results from API. + urls: List of URLs that were scraped. + Returns: + list: List of processed result dictionaries. + """ + processed_results = [] + + if len(urls) == 1 and len(scrape_results) > 1: + # Single URL with multiple results - process all results + url = urls[0] + log.info( + f"Processing {len(scrape_results)} results from single URL. " + f"Each result will get a unique result_index (0 to {len(scrape_results) - 1})" + ) + for result_idx, result in enumerate(scrape_results): + if isinstance(result, dict): + result_url = ( + result.get("input", {}).get("url") or result.get("url") or url + ) + processed_results.append( + process_scrape_result(result, result_url, result_idx) + ) + elif isinstance(result, list): + for item_idx, item in enumerate(result): + result_url = ( + item.get("input", {}).get("url") + if isinstance(item, dict) + else url + ) + processed_results.append( + process_scrape_result(item, result_url or url, item_idx) + ) + else: + # Multiple URLs or one-to-one mapping - match by index + missing_results = [] + for url_idx, url in enumerate(urls): + if url_idx < len(scrape_results): + result = scrape_results[url_idx] + if isinstance(result, list): + for item_idx, item in enumerate(result): + processed_results.append( + process_scrape_result(item, url, item_idx) + ) + else: + processed_results.append(process_scrape_result(result, url, 0)) + else: + missing_results.append((url_idx, url)) + # Log missing results once after processing + if missing_results: + log.warning( + f"No result found for {len(missing_results)} URL(s) at indices: " + f"{', '.join(str(idx) for idx, _ in missing_results[:5])}" + f"{' (and more)' if len(missing_results) > 5 else ''}" + ) + + return processed_results + + +def process_and_upsert_results(processed_results, all_fields): + """ + Process and upsert scrape result records. + Args: + processed_results: List of processed result dictionaries. + all_fields: List of all field names discovered from results. + """ + primary_keys = {"url": str, "result_index": int} + primary_key_errors = [] + for result in processed_results: + # Ensure primary keys are always present with correct types + for pk, pk_type in primary_keys.items(): + if pk not in result: + primary_key_errors.append(f"Primary key '{pk}' missing from result") + result[pk] = pk_type() if pk_type == str else 0 + else: + current_value = result[pk] + if not isinstance(current_value, pk_type): + try: + if pk_type == str: + result[pk] = str(current_value) + elif pk_type == int: + if isinstance(current_value, str): + cleaned = current_value.strip().strip("[]\"'") + result[pk] = int(cleaned) if cleaned.isdigit() else 0 + else: + result[pk] = int(current_value) + except (ValueError, TypeError): + primary_key_errors.append( + f"Could not convert primary key '{pk}' to {pk_type.__name__}" + ) + result[pk] = pk_type() if pk_type == str else 0 + row = {} + for field in all_fields: + row[field] = result.get(field) + + # The 'upsert' operation is used to insert or update data in the destination table. + # The first argument is the name of the destination table. + # The second argument is a dictionary containing the record to be upserted. + op.upsert(table=__SCRAPE_TABLE, data=row) + + # Log primary key errors once after processing all results + if primary_key_errors: + unique_errors = list(set(primary_key_errors)) + log.warning( + f"Primary key validation issues: {', '.join(unique_errors[:3])}" + f"{' (and more)' if len(unique_errors) > 3 else ''}" + ) + + +connector = Connector(update=update, schema=schema) + + +if __name__ == "__main__": + # Open the configuration.json file and load its contents + with open("configuration.json", "r") as f: + configuration = json.load(f) + + # Test the connector locally + connector.debug(configuration=configuration) diff --git a/bright_data_scrape/helpers/__init__.py b/bright_data_scrape/helpers/__init__.py new file mode 100644 index 0000000..1253102 --- /dev/null +++ b/bright_data_scrape/helpers/__init__.py @@ -0,0 +1,12 @@ +"""Helper module exports for easy importing.""" + +from .data_processing import collect_all_fields, process_scrape_result +from .scrape import perform_scrape +from .validation import validate_configuration + +__all__ = [ + "collect_all_fields", + "process_scrape_result", + "perform_scrape", + "validate_configuration", +] diff --git a/bright_data_scrape/helpers/data_processing.py b/bright_data_scrape/helpers/data_processing.py new file mode 100644 index 0000000..39ca48f --- /dev/null +++ b/bright_data_scrape/helpers/data_processing.py @@ -0,0 +1,98 @@ +"""Utilities for flattening Bright Data scrape results.""" + +# For serializing nested data structures (lists, dicts) to JSON strings +import json + +# For type hints in function signatures +from typing import Any, Dict, List, Set + + +def flatten_dict( + data: Any, parent_key: str = "", separator: str = "_", max_depth: int = 10 +) -> Dict[str, Any]: + """ + Flatten nested dictionaries and lists into a single level. + + Args: + data: The data structure to flatten (dict, list, or primitive). + parent_key: The parent key prefix for nested keys. + separator: The separator to use between nested key levels. + max_depth: Maximum nesting depth to flatten before serializing to JSON. + + Returns: + A flattened dictionary with all nested keys combined using the separator. + """ + if max_depth <= 0: + return {parent_key: json.dumps(data) if data else None} + + items: List[tuple] = [] + + if isinstance(data, dict): + for key, value in data.items(): + new_key = f"{parent_key}{separator}{key}" if parent_key else key + if isinstance(value, dict): + items.extend(flatten_dict(value, new_key, separator, max_depth - 1).items()) + elif isinstance(value, list): + items.append((new_key, json.dumps(value) if value else "[]")) + else: + items.append((new_key, value)) + elif isinstance(data, list): + return {parent_key: json.dumps(data) if data else "[]"} + else: + return {parent_key: data} + + return dict(items) + + +def collect_all_fields(results: List[Dict[str, Any]]) -> Set[str]: + """Return the union of keys that appear across processed results. + + Args: + results: List of processed result dictionaries. + + Returns: + Set of all unique field names found across all results.""" + all_fields: Set[str] = set() + for result in results: + all_fields.update(result.keys()) + return all_fields + + +def process_scrape_result(result: Any, url: str, result_index: int) -> Dict[str, Any]: + """ + Flatten an individual scrape result and add metadata columns. + + Primary key fields (url, result_index) are always preserved and never overwritten + by values from the flattened API response, even if the response contains fields + with the same names. + + Args: + result: The scrape result to process (dict or other type). + url: The URL that was scraped. + result_index: The index of this result for the given URL. + + Returns: + A flattened dictionary with metadata fields and primary keys. + """ + base_fields = { + "url": url, + "result_index": result_index, + "position": result_index + 1, + } + + if not isinstance(result, dict): + base_fields["raw_data"] = str(result) + return base_fields + + flattened = flatten_dict(result) + + # Remove exact primary key matches from flattened data so API fields named + for pk_field in ("url", "result_index"): + flattened.pop(pk_field, None) + + final_result = {**flattened, **base_fields} + + final_result["result_index"] = int(result_index) + final_result["position"] = int(result_index + 1) + + return final_result diff --git a/bright_data_scrape/helpers/scrape.py b/bright_data_scrape/helpers/scrape.py new file mode 100644 index 0000000..f29f3cf --- /dev/null +++ b/bright_data_scrape/helpers/scrape.py @@ -0,0 +1,720 @@ +"""Bright Data Web Scraper helper functions.""" + +# For parsing JSON responses from the Bright Data API +import json + +# For adding delays between retry attempts (exponential backoff) +import time + +# For type hints in function signatures +from typing import Any, Dict, List, Optional, Union + +# For making HTTP requests to the Bright Data API +import requests + +# For catching network and HTTP request exceptions +from requests import RequestException, Response + +# For logging connector operations, errors, and status updates +from fivetran_connector_sdk import Logging as log + +BRIGHT_DATA_BASE_URL = "https://api.brightdata.com" +DEFAULT_TIMEOUT_SECONDS = 120 +RETRY_STATUS_CODES = {408, 429, 500, 502, 503, 504} + +MAX_RESPONSE_SIZE_BYTES = 100 * 1024 * 1024 +CHUNK_SIZE_BYTES = 8192 + + +def _parse_response_payload(response: Response) -> Any: + """Return JSON payload when available, otherwise raw text.""" + try: + return response.json() + except ValueError: + return response.text + + +def _extract_error_detail(response: Response) -> str: + """Extract error detail from a failed Bright Data response.""" + try: + payload = response.json() + if isinstance(payload, dict): + # Check for validation_errors array (400 Bad Request) + if "validation_errors" in payload and isinstance(payload["validation_errors"], list): + errors = ", ".join(str(err) for err in payload["validation_errors"]) + return f"Validation errors: {errors}" + return str(payload) + except ValueError: + return response.text + + +def _parse_large_json_array_streaming(response, max_size_bytes: int = MAX_RESPONSE_SIZE_BYTES): + """ + Parse a large JSON array response using chunked reading to manage memory. + + For very large responses (>100MB), this reads the response in chunks and parses + incrementally. However, JSON arrays still require the full response for parsing. + + Note: For truly large datasets, JSONL format is recommended as it allows + line-by-line streaming without loading the entire response. + + Args: + response: requests.Response object to parse + max_size_bytes: Threshold for using chunked reading (default 100MB) + + Returns: + List of parsed JSON objects, or None if not applicable/parse failed + """ + content_length = response.headers.get("Content-Length") + + # Only attempt chunked parsing for responses larger than threshold + if not content_length or int(content_length) < max_size_bytes: + return None + + try: + # Read response in chunks to monitor progress and manage memory pressure + buffer = "" + chunk_count = 0 + for chunk in response.iter_content(chunk_size=CHUNK_SIZE_BYTES, decode_unicode=True): + if chunk: + buffer += chunk + chunk_count += 1 + # Log progress for very large responses + if chunk_count % 1000 == 0: + log.info( + f"Reading large response: {len(buffer)} bytes read " + f"({chunk_count} chunks)" + ) + + # Verify it looks like a JSON array + buffer_stripped = buffer.strip() + if not buffer_stripped.startswith("["): + # Not a JSON array, fall back to standard parsing + return None + + # Parse the accumulated JSON array + # Note: We still need to load it all for JSON array parsing + # JSONL format would be better for true streaming + parsed_data = json.loads(buffer_stripped) + + if isinstance(parsed_data, list): + log.info( + f"Successfully parsed large JSON array with {len(parsed_data)} records " + f"({len(buffer)} bytes)" + ) + return parsed_data + + except (json.JSONDecodeError, ValueError, MemoryError) as e: + log.warning( + f"Failed to parse large JSON response: {str(e)}. " + f"Consider using JSONL format for very large datasets." + ) + return None + except Exception as e: + log.info(f"Chunked parse encountered error: {str(e)}, falling back to standard parsing") + return None + + return None + + +def perform_scrape( + api_token: str, + dataset_id: str, + url: Union[str, List[str]], + poll_interval: int = 30, + timeout: int = DEFAULT_TIMEOUT_SECONDS, + retries: int = 3, + backoff_factor: float = 1.5, + extra_query_params: Dict[str, Any] = None, +) -> List[Dict[str, Any]]: + """ + Scrape URLs using Bright Data's Web Scraper REST API. + + This function triggers a scrape job via POST /datasets/v3/trigger, receives a snapshot_id, + and polls GET /datasets/v3/snapshot/{snapshot_id} until the status is "ready" or "failed". + The endpoint handles polling internally, so we poll indefinitely until we receive a ready/failed status. + + Args: + api_token: Bright Data API token + dataset_id: ID of the dataset to use for scraping + url: Single URL or list of URLs to scrape + poll_interval: Interval between polling attempts in seconds + timeout: Request timeout in seconds + retries: Number of retries for failed requests + backoff_factor: Backoff factor for exponential backoff + extra_query_params: Optional dictionary of additional query parameters to include + in the scrape trigger request (e.g., {"discover_by": "profile_url", "type": "discover_new"}) + + Returns: + List of scraped results (each result is a dictionary) + + Raises: + ValueError: If API token, dataset_id, or URL is invalid + RuntimeError: If scrape trigger or polling fails + """ + if not api_token or not isinstance(api_token, str): + raise ValueError("A valid Bright Data API token is required") + + if not dataset_id or not isinstance(dataset_id, str): + raise ValueError("dataset_id must be a non-empty string") + + if not url: + raise ValueError("URL cannot be empty") + + if not isinstance(url, (str, list)): + raise TypeError("URL must be a string or list of strings") + + # Normalize URLs to list + urls: List[str] + if isinstance(url, list): + urls = [item.strip() for item in url if isinstance(item, str) and item.strip()] + else: + urls = [url.strip()] + + if not urls: + raise ValueError("At least one non-empty URL must be provided") + + # Build payload for trigger request + # Convert list of URL strings to API-required format: + # [{"url": "https://example.com/1"}, {"url": "https://example.com/2"}, ...] + # This matches the Bright Data API endpoint structure which accepts a list of objects + payload = [{"url": single_url} for single_url in urls] + + # Trigger scrape job + snapshot_id = _trigger_scrape( + api_token=api_token, + dataset_id=dataset_id, + payload=payload, + timeout=timeout, + retries=retries, + backoff_factor=backoff_factor, + extra_query_params=extra_query_params or {}, + ) + + if not snapshot_id: + raise RuntimeError("Failed to trigger scrape job - no snapshot_id returned") + + # Poll snapshot until ready or failed + results = _poll_snapshot( + api_token=api_token, + snapshot_id=snapshot_id, + poll_interval=poll_interval, + timeout=timeout, + ) + + # Normalize results to always be a list + if not isinstance(results, list): + results = [results] if results else [] + + result_count = len(results) + log.info(f"Scrape completed successfully. Retrieved {result_count} result(s)") + + return results + + +def _trigger_scrape( + api_token: str, + dataset_id: str, + payload: List[Dict[str, Any]], + timeout: int, + retries: int, + backoff_factor: float, + extra_query_params: Dict[str, Any] = None, +) -> str: + """ + Trigger a scrape job via POST /datasets/v3/trigger. + + Returns: + snapshot_id string from the response + + Raises: + ValueError: For 400/422 errors (invalid input) + RuntimeError: For other API errors or network failures + """ + headers = { + "Authorization": f"Bearer {api_token}", + "Content-Type": "application/json", + } + + # Build query parameters + # According to Bright Data API docs, trigger endpoint accepts: + # dataset_id, format, and include_errors as query parameters + params: Dict[str, Any] = { + "dataset_id": dataset_id, + "format": "json", + "include_errors": "true", # Include errors report with the results + } + + # Add any additional query parameters from configuration + # This allows dataset-specific parameters to be configured rather than hardcoded + if extra_query_params: + params.update(extra_query_params) + + # Build request body - according to Bright Data API docs, body should be: + # [{"url": "https://..."}, {"url": "https://..."}, ...] + # The API documentation shows that the body should only contain URL objects + # as shown in the API examples + body_payload = payload.copy() + + url_count = len(payload) + log.info( + f"Triggering Bright Data scrape for {url_count} URL{'s' if url_count > 1 else ''} " + f"using dataset_id: {dataset_id}" + ) + + attempt = 0 + backoff = backoff_factor + + while attempt <= retries: + try: + # Log request details for debugging (without sensitive data) + log.info( + f"Sending POST request to {BRIGHT_DATA_BASE_URL}/datasets/v3/trigger " + f"with params: {params}, payload containing {url_count} URL(s)" + ) + + response = requests.post( + f"{BRIGHT_DATA_BASE_URL}/datasets/v3/trigger", + headers=headers, + params=params, + json=body_payload, + timeout=timeout, + ) + + if response.status_code == 200: + response_data = _parse_response_payload(response) + if isinstance(response_data, dict): + snapshot_id = response_data.get("snapshot_id") + if snapshot_id: + log.info( + f"Scrape job triggered successfully. Snapshot ID: {snapshot_id[:8]}..." + ) + return str(snapshot_id) + raise RuntimeError("Trigger response missing snapshot_id field") + raise RuntimeError(f"Unexpected trigger response format: {response_data}") + + # Handle 400/422 as ValueError (invalid input) + if response.status_code in (400, 422): + error_detail = _extract_error_detail(response) + # Log the full response for debugging + try: + response_json = response.json() + log.info( + f"Invalid scrape request (status {response.status_code}): {error_detail}. " + f"Request URL: {response.url}, " + f"Request payload (first URL): {body_payload[0] if body_payload else 'empty'}, " + f"Full response: {response_json}" + ) + except ValueError: + log.info( + f"Invalid scrape request (status {response.status_code}): {error_detail}. " + f"Request URL: {response.url}, " + f"Request payload (first URL): {body_payload[0] if body_payload else 'empty'}, " + f"Response text: {response.text[:500]}" + ) + raise ValueError(f"Invalid scrape request: {error_detail}") + + # Retry on certain status codes + if response.status_code in RETRY_STATUS_CODES and attempt < retries: + log.info( + f"Bright Data scrape trigger retry {attempt + 1}/{retries} " + f"(status code: {response.status_code})" + ) + attempt += 1 + time.sleep(backoff) + backoff *= backoff_factor + continue + + # For other errors, raise with details + error_detail = _extract_error_detail(response) + log.info( + f"Bright Data scrape trigger failed (status {response.status_code}): " + f"{error_detail}" + ) + response.raise_for_status() + + except RequestException as exc: + if attempt < retries: + log.info( + f"Error contacting Bright Data Scraper API: {str(exc)}. " + f"Retrying ({attempt + 1}/{retries})" + ) + attempt += 1 + time.sleep(backoff) + backoff *= backoff_factor + continue + raise RuntimeError( + f"Failed to trigger Bright Data scrape after {retries} retries: {str(exc)}" + ) from exc + + +def _poll_snapshot( + api_token: str, + snapshot_id: str, + poll_interval: int, + timeout: int, + max_attempts: int = 100, +) -> Union[Dict[str, Any], List[Dict[str, Any]]]: + """ + Poll the snapshot endpoint until status is "ready" or "failed". + + According to API docs: GET /datasets/v3/snapshot/{snapshot_id}?format=json + + Args: + api_token: Bright Data API token + snapshot_id: Snapshot ID to poll + poll_interval: Interval between polling attempts in seconds + timeout: Request timeout in seconds + max_attempts: Maximum number of polling attempts before raising an error (default: 1000) + + Returns: + Snapshot data when ready (dict or list) + + Raises: + RuntimeError: If snapshot fails or request errors occur, or if max_attempts is exceeded + """ + headers = { + "Authorization": f"Bearer {api_token}", + } + + params: Dict[str, Any] = { + "format": "json", + "include_errors": "true", + } + + attempt = 0 + while attempt < max_attempts: + attempt += 1 + + try: + response = requests.get( + f"{BRIGHT_DATA_BASE_URL}/datasets/v3/snapshot/{snapshot_id}", + headers=headers, + params=params, + timeout=timeout, + ) + + if response.status_code == 200: + # Log response details only on first attempt + if attempt == 1: + _log_initial_response_info(response, snapshot_id) + + # Parse and handle the response + snapshot_data = _parse_snapshot_response(response, snapshot_id, attempt) + + if snapshot_data is not None: + # Snapshot is ready, return the data + log.info(f"Snapshot {snapshot_id[:8]}... ready after {attempt} attempt(s)") + return snapshot_data + else: + # Snapshot is still processing, continue polling + _log_polling_progress(snapshot_id, attempt, response) + time.sleep(poll_interval) + continue + + elif response.status_code == 404: + error_msg = f"Snapshot {snapshot_id[:8]}... not found" + log.info(error_msg) + raise RuntimeError(error_msg) + else: + _handle_non_200_response(response, snapshot_id, attempt) + + except RequestException as exc: + # Log network errors every 5 attempts to reduce log volume + if attempt % 5 == 0: + log.info( + f"Error polling snapshot {snapshot_id[:8]}...: {str(exc)}. " + f"Retrying (attempt {attempt})" + ) + + time.sleep(poll_interval) + + raise RuntimeError( + f"Snapshot {snapshot_id[:8]} polling timed out after {max_attempts} attempts" + ) + + +def _log_initial_response_info(response: Response, snapshot_id: str) -> None: + """Log initial response details.""" + content_length = response.headers.get("Content-Length") + content_type = response.headers.get("Content-Type", "").lower() + + size_info = f", size: {content_length} bytes" if content_length else "" + log.info( + f"Snapshot {snapshot_id[:8]}... poll response " + f"Content-Type: {content_type or 'unknown'}{size_info}" + ) + + +def _parse_snapshot_response( + response: Response, snapshot_id: str, attempt: int +) -> Optional[Union[Dict[str, Any], List[Dict[str, Any]]]]: + """ + Parse snapshot response and return data if ready, None if still processing. + + Returns: + Snapshot data if ready, None if still processing. + """ + content_type = response.headers.get("Content-Type", "").lower() + + # Handle JSONL format + if "jsonl" in content_type or "json-lines" in content_type: + return _parse_jsonl_response(response, snapshot_id, attempt) + + # Check for large responses + content_length = response.headers.get("Content-Length") + max_size_bytes = MAX_RESPONSE_SIZE_BYTES + if content_length and int(content_length) > max_size_bytes: + log.warning( + f"Snapshot {snapshot_id[:8]}... response size ({content_length} bytes) exceeds " + f"maximum recommended size ({max_size_bytes} bytes). " + f"Consider using JSONL format for large datasets." + ) + + # Try chunked parsing for large JSON arrays + chunked_data = _parse_large_json_array_streaming(response, max_size_bytes) + if chunked_data is not None: + log.info( + f"Snapshot {snapshot_id[:8]}... ready (chunked parse with {len(chunked_data)} records) " + f"after {attempt} attempt(s)" + ) + return chunked_data + + # Parse standard JSON response + return _parse_json_response(response, snapshot_id, attempt) + + +def _parse_jsonl_response( + response: Response, snapshot_id: str, attempt: int +) -> Optional[List[Dict[str, Any]]]: + """ + Parse JSON Lines format response. + + Returns: + List of parsed objects if data is ready (may be empty), None if still processing. + """ + results = [] + parse_errors = [] + + for line_num, line in enumerate(response.iter_lines(decode_unicode=True), 1): + line = line.strip() + if not line: + continue + try: + parsed_obj = json.loads(line) + results.append(parsed_obj) + except json.JSONDecodeError as e: + parse_errors.append((line_num, str(e), line[:200])) + + if parse_errors: + log.warning( + f"Snapshot {snapshot_id[:8]}... JSONL parse errors on {len(parse_errors)} line(s). " + f"First error: line {parse_errors[0][0]}: {parse_errors[0][1]}" + ) + + log.info( + f"Snapshot {snapshot_id[:8]}... ready (JSONL format with {len(results)} records) " + f"after {attempt} attempt(s)" + ) + return results + + +def _parse_json_response( + response: Response, snapshot_id: str, attempt: int +) -> Optional[Union[Dict[str, Any], List[Dict[str, Any]]]]: + """ + Parse standard JSON response (array, object, or string). + + Returns: + Snapshot data if ready, None if still processing. + """ + # Try to parse response as JSON + try: + response_data = response.json() + except (ValueError, json.JSONDecodeError): + # If standard parsing fails, try raw text + return _parse_text_response(response.text, snapshot_id, attempt) + + # Handle based on response type + if isinstance(response_data, list): + log.info( + f"Snapshot {snapshot_id[:8]}... ready (array response with {len(response_data)} records) " + f"after {attempt} attempt(s)" + ) + return response_data + + elif isinstance(response_data, dict): + return _parse_dict_response(response_data, snapshot_id, attempt) + + elif isinstance(response_data, str): + return _parse_text_response(response_data, snapshot_id, attempt) + + # Unknown type - treat as data + return [response_data] if response_data else [] + + +def _parse_text_response( + text: str, snapshot_id: str, attempt: int +) -> Optional[Union[Dict[str, Any], List[Dict[str, Any]]]]: + """ + Parse text response (could be JSON string or plain text status). + + Returns: + Snapshot data if ready, None if still processing. + """ + text_lower = text.lower() + + # Check if it's a "not ready" status message + if any(phrase in text_lower for phrase in ("not ready", "processing", "pending", "try again")): + return None + + # Try parsing as JSON + trimmed_text = text.strip() + if not (trimmed_text.startswith(("{", "["))): + # Doesn't look like JSON + log.info( + f"Snapshot {snapshot_id[:8]}... string response (not JSON-like): " f"{text[:200]}..." + ) + return None + + try: + parsed_data = json.loads(text) + # Normalize to list format + if isinstance(parsed_data, list): + return parsed_data + elif isinstance(parsed_data, dict): + return [parsed_data] + else: + return [parsed_data] if parsed_data else [] + except json.JSONDecodeError as e: + # Try parsing as JSONL format + log.warning(f"Snapshot {snapshot_id[:8]}... JSON decode error: {str(e)}") + return _parse_text_as_jsonl(text, snapshot_id, attempt) + + +def _parse_text_as_jsonl( + text: str, snapshot_id: str, attempt: int +) -> Optional[List[Dict[str, Any]]]: + """ + Attempt to parse text as JSONL format. + """ + results = [] + for line in text.split("\n"): + line = line.strip() + if not line: + continue + try: + parsed_obj = json.loads(line) + results.append(parsed_obj) + except json.JSONDecodeError: + continue + + if results: + log.info( + f"Snapshot {snapshot_id[:8]}... ready (detected JSONL format " + f"with {len(results)} records) after {attempt} attempt(s)" + ) + return results + + return None + + +def _parse_dict_response( + data: Dict[str, Any], snapshot_id: str, attempt: int +) -> Optional[Union[Dict[str, Any], List[Dict[str, Any]]]]: + """ + Parse dictionary response, checking status and extracting data. + + Returns: + Snapshot data if ready, None if still processing. + """ + status = data.get("status", "").lower() + + if status == "ready": + return _extract_data_from_dict(data) + + elif status == "failed": + error_msg = _extract_error_message(data) + log.info(f"Snapshot {snapshot_id[:8]}... failed: {error_msg}") + raise RuntimeError(f"Snapshot {snapshot_id[:8]}... failed: {error_msg}") + + elif status in ("running", "pending", "processing", "scheduled"): + # Still processing + return None + + else: + # No status field or unknown status + if "status" not in data: + # Could be data - return it wrapped in list + return [data] if data else [] + else: + # Unknown status - continue polling + return None + + +def _extract_data_from_dict(data: Dict[str, Any]) -> List[Dict[str, Any]]: + """ + Extract snapshot data from response dictionary. + """ + snapshot_data = None + for key in ("data", "records", "results"): + if key in data: + snapshot_data = data[key] + break + + # If no specific data key, remove metadata fields from the response itself + if snapshot_data is None: + metadata_keys = ( + "status", + "id", + "snapshot_id", + "created", + "dataset_id", + "customer_id", + "cost", + "initiation_type", + ) + snapshot_data = {k: v for k, v in data.items() if k not in metadata_keys} + + # Normalize to list format + if isinstance(snapshot_data, list): + return snapshot_data + elif isinstance(snapshot_data, dict): + return [snapshot_data] + else: + return [snapshot_data] if snapshot_data is not None else [] + + +def _extract_error_message(data: Dict[str, Any]) -> str: + """Extract error message from response dictionary.""" + for key in ("error", "warning", "message", "detail", "details"): + if key in data: + error_value = data[key] + if error_value: + return str(error_value) + return "Unknown error" + + +def _log_polling_progress(snapshot_id: str, attempt: int, response: Response) -> None: + """Log polling progress based on attempt count.""" + if attempt % 5 == 0: + try: + response_data = response.json() + if isinstance(response_data, dict): + status = response_data.get("status", "") + if status: + log.info( + f"Snapshot {snapshot_id[:8]}... status: {status} " f"(attempt {attempt})" + ) + except (ValueError, json.JSONDecodeError): + log.info(f"Snapshot {snapshot_id[:8]}... still processing (attempt {attempt})") + + +def _handle_non_200_response(response: Response, snapshot_id: str, attempt: int) -> None: + """Handle non-200 HTTP responses.""" + error_detail = _extract_error_detail(response) + log.info( + f"Error polling snapshot {snapshot_id[:8]}... " + f"(status {response.status_code}): {error_detail}" + ) + response.raise_for_status() diff --git a/bright_data_scrape/helpers/validation.py b/bright_data_scrape/helpers/validation.py new file mode 100644 index 0000000..640cc7e --- /dev/null +++ b/bright_data_scrape/helpers/validation.py @@ -0,0 +1,20 @@ +"""Configuration validation utilities.""" + + +def validate_configuration(configuration: dict) -> None: + """ + Validate the configuration dictionary to ensure it contains all required parameters. + + This function is called at the start of the update method to ensure that the connector + has all necessary configuration values. + + Args: + configuration: A dictionary that holds the configuration settings for the connector. + + Raises: + ValueError: If any required configuration parameter is missing. + """ + required_configs = ["api_token", "dataset_id", "scrape_url"] + for key in required_configs: + if key not in configuration: + raise ValueError(f"Missing required configuration value: {key}")