DEV Community: coreclaw

Web Data Normalization Pipeline: How to Turn Raw Scraped JSON into Consistent Records with Python

coreclaw — Sat, 12 Sep 2026 01:05:11 +0000

Web data normalization is the process of transforming inconsistent raw JSON from different scraping sources into a single, validated schema with predictable field names, data types, and value formats. For teams collecting data from Google Maps, Amazon, LinkedIn, or social platforms through APIs like CoreClaw's scraper store, each source returns its own field names, nesting depth, and value conventions — business_name vs name, phone_number vs phone vs tel, prices as strings with currency symbols vs raw floats. A reusable Python normalization pipeline with field mapping, type coercion, validation rules, and deduplication handles this at scale without writing a custom transformer for every new source.

TL;DR

Build a DataNormalizer class that accepts a field-mapping config (source field → target field), a type-coercion config (target field → expected Python type), and a validation config (required fields, value ranges). Feed raw scraped JSON through the pipeline to get clean, consistent records ready for storage, analytics, or AI-agent context. Use the CoreClaw console to configure and test your data sources, then normalize the structured output downstream in your own pipeline.

Why Raw Scraped Data Is Inconsistent

Even when a web data API returns structured JSON, the schema varies across sources for three reasons:

Source-specific field names. A Google Maps business record uses displayed_name; an Amazon product uses product_title; a LinkedIn company page uses name. All three map to the same concept — a human-readable entity name — but the keys differ.
Inconsistent value formats. Phone numbers may include country codes, extensions, or formatting characters. Prices may be strings like "$29.99", integers in cents like 2999, or floats like 29.99. Dates may be ISO-8601, Unix timestamps, or human-readable strings.
Variable nesting and optional fields. One source nests address fields under location.address; another flattens them at the top level. Optional fields may be null, missing entirely, or an empty string.

Without normalization, every downstream consumer — your database, analytics dashboard, CRM, or AI agent — needs to handle every variant. That creates brittle code that breaks when a source changes its response shape.

The Normalization Pipeline Architecture

A robust normalization pipeline has five stages:

Stage	Input	Output	Key Operations
Field mapping	Raw JSON with source-specific keys	Flat dict with target field names	Key rename, nested-field extraction
Type coercion	String-typed values	Correctly typed values	Parse float, int, datetime, bool
Validation	Coerced record	Clean record or error list	Required fields, range checks, regex
Deduplication	List of validated records	Deduplicated list	Hash-based, fuzzy match, or key-based
Enrichment	Deduplicated record	Enriched record	Derived fields, source provenance

Each stage should be independently testable and configurable. Hard-coding transformations inside one monolithic function makes the pipeline impossible to maintain when you add a new source.

Runnable Python Pipeline

import os
import json
import re
import hashlib
from datetime import datetime, timezone
from typing import Any, Callable

# Configuration is loaded from environment or a config file.
# Endpoint and API key values are conceptual — copy current
# values from your CoreClaw console or product documentation.
CORECLAW_ENDPOINT = os.environ.get("CORECLAW_ENDPOINT", "")
CORECLAW_API_KEY = os.environ.get("CORECLAW_API_KEY", "")

class DataNormalizer:
    """Reusable normalization pipeline for multi-source web data."""

    def __init__(
        self,
        field_map: dict[str, str],
        type_map: dict[str, Callable],
        required_fields: list[str],
        dedup_key: str | None = None,
    ):
        self.field_map = field_map
        self.type_map = type_map
        self.required_fields = required_fields
        self.dedup_key = dedup_key

    # --- Stage 1: Field mapping ---
    def _map_fields(self, raw: dict) -> dict:
        mapped = {}
        for source_key, target_key in self.field_map.items():
            value = self._extract_nested(raw, source_key)
            if value is not None:
                mapped[target_key] = value
        return mapped

    @staticmethod
    def _extract_nested(record: dict, dotted_key: str) -> Any:
        """Support dot-notation for nested fields: 'location.address.city'."""
        keys = dotted_key.split(".")
        value = record
        for k in keys:
            if isinstance(value, dict) and k in value:
                value = value[k]
            else:
                return None
        return value

    # --- Stage 2: Type coercion ---
    @staticmethod
    def _coerce(value: Any, coercer: Callable) -> Any:
        try:
            return coercer(value)
        except (ValueError, TypeError, AttributeError):
            return None

    def _coerce_fields(self, record: dict) -> dict:
        coerced = {}
        for key, value in record.items():
            if key in self.type_map and value is not None:
                coerced[key] = self._coerce(value, self.type_map[key])
            else:
                coerced[key] = value
        return coerced

    # --- Stage 3: Validation ---
    def _validate(self, record: dict) -> tuple[dict | None, list[str]]:
        errors = []
        for field in self.required_fields:
            if field not in record or record[field] in (None, "", []):
                errors.append(f"Missing required field: {field}")
        if errors:
            return None, errors
        return record, []

    # --- Stage 4: Deduplication ---
    @staticmethod
    def _record_hash(record: dict, key: str) -> str:
        """Deterministic hash for deduplication based on a key field."""
        raw = str(record.get(key, ""))
        return hashlib.sha256(raw.encode()).hexdigest()

    def _deduplicate(self, records: list[dict]) -> list[dict]:
        if not self.dedup_key:
            return records
        seen: set[str] = set()
        unique = []
        for r in records:
            h = self._record_hash(r, self.dedup_key)
            if h not in seen:
                seen.add(h)
                unique.append(r)
        return unique

    # --- Stage 5: Enrichment ---
    @staticmethod
    def _enrich(record: dict) -> dict:
        record["_normalized_at"] = datetime.now(timezone.utc).isoformat()
        record["_pipeline_version"] = "1.0"
        return record

    # --- Full pipeline ---
    def normalize(self, raw_records: list[dict]) -> list[dict]:
        clean = []
        errors_log = []
        for raw in raw_records:
            mapped = self._map_fields(raw)
            coerced = self._coerce_fields(mapped)
            validated, errs = self._validate(coerced)
            if validated:
                clean.append(self._enrich(validated))
            else:
                errors_log.extend(errs)
        deduped = self._deduplicate(clean)
        if errors_log:
            print(f"[normalizer] {len(errors_log)} validation issues skipped")
        return deduped


# --- Coercion helpers ---
def to_float(value: Any) -> float:
    """Parse price strings like '$29.99' or '2999' (cents) into float."""
    if isinstance(value, (int, float)):
        return float(value)
    cleaned = re.sub(r"[^\d.]", "", str(value))
    return float(cleaned) if cleaned else 0.0

def to_phone(value: Any) -> str:
    """Strip formatting, keep digits and leading +."""
    digits = re.sub(r"[^\d+]", "", str(value))
    return digits if digits else ""

def to_int(value: Any) -> int:
    return int(float(str(value).replace(",", "")))


# --- Example: normalizing Google Maps business records ---
maps_field_map = {
    "displayed_name": "name",
    "phone": "phone",
    "location.address.postal_code": "postal_code",
    "rating": "rating",
    "user_rating_count": "review_count",
    "location.address.country": "country",
}

maps_type_map = {
    "rating": to_float,
    "review_count": to_int,
    "phone": to_phone,
}

maps_required = ["name", "phone"]
maps_dedup_key = "phone"

normalizer = DataNormalizer(
    field_map=maps_field_map,
    type_map=maps_type_map,
    required_fields=maps_required,
    dedup_key=maps_dedup_key,
)

# Sample raw records (representative — not live API output)
raw_businesses = [
    {
        "displayed_name": "Sunset Bakery",
        "phone": "+1 (415) 555-0123",
        "location": {"address": {"postal_code": "94102", "country": "US"}},
        "rating": "4.7",
        "user_rating_count": "312",
    },
    {
        "displayed_name": "Sunset Bakery",  # duplicate
        "phone": "+1 (415) 555-0123",
        "location": {"address": {"postal_code": "94102", "country": "US"}},
        "rating": 4.7,
        "user_rating_count": 312,
    },
    {
        "displayed_name": "Blue Bottle Coffee",
        "phone": "+1 (415) 555-9876",
        "location": {"address": {"postal_code": "94110", "country": "US"}},
        "rating": "4.5",
        "user_rating_count": "1,204",
    },
]

clean_records = normalizer.normalize(raw_businesses)
print(json.dumps(clean_records, indent=2))

Expected Output

[
  {
    "name": "Sunset Bakery",
    "phone": "+14155550123",
    "postal_code": "94102",
    "country": "US",
    "rating": 4.7,
    "review_count": 312,
    "_normalized_at": "2026-09-12T09:00:00+00:00",
    "_pipeline_version": "1.0"
  },
  {
    "name": "Blue Bottle Coffee",
    "phone": "+14155559876",
    "postal_code": "94110",
    "country": "US",
    "rating": 4.5,
    "review_count": 1204,
    "_normalized_at": "2026-09-12T09:00:00+00:00",
    "_pipeline_version": "1.0"
  }
]

Notice: the duplicate "Sunset Bakery" record (second entry with the same phone) was removed by the deduplication stage, and string-typed ratings and review counts were coerced to float and int.

Adapting the Pipeline for Multiple Sources

The same DataNormalizer class works for Amazon product data, LinkedIn company records, or YouTube channel stats — you only change the configuration:

# Amazon product normalization config
amazon_field_map = {
    "product_title": "name",
    "price.value": "price",
    "price.currency": "currency",
    "rating": "rating",
    "review_count": "review_count",
    "asin": "product_id",
}
amazon_type_map = {
    "price": to_float,
    "rating": to_float,
    "review_count": to_int,
}
amazon_required = ["name", "product_id"]
amazon_dedup_key = "product_id"

amazon_normalizer = DataNormalizer(
    field_map=amazon_field_map,
    type_map=amazon_type_map,
    required_fields=amazon_required,
    dedup_key=amazon_dedup_key,
)

This approach means adding a new data source is a configuration task, not a code change. You define the field map, type map, and validation rules, then reuse the same pipeline.

Business Use Cases

Use Case	Sources	Normalized Output	Downstream Consumer
B2B lead enrichment	Google Maps, LinkedIn	Unified company records with name, phone, address, industry	CRM import (HubSpot, Salesforce)
Price intelligence	Amazon, Walmart, eBay	Product records with consistent price-as-float, currency, product_id	Price monitoring dashboard
Creator analytics	YouTube, Instagram, TikTok	Creator records with engagement_rate, follower_count, content_count	Marketing analytics pipeline
Competitive intelligence	Google SERP, Google Maps	Keyword + business ranking records	SEO reporting tool
AI agent context	All of the above	Structured, validated JSON with provenance	RAG / vector database

Each use case benefits from normalization because downstream systems — CRMs, databases, AI models — expect consistent schemas. Without normalization, you spend more time writing adapters than building features.

DIY Normalization vs CoreClaw Structured Data vs Custom Pipeline

Dimension	DIY (raw requests + custom code)	CoreClaw structured data + custom normalizer	Enterprise data platform
Field consistency	Raw HTML/JSON varies widely	API returns structured JSON; you normalize residual differences	Platform handles normalization end-to-end
Setup model	Build parser + normalizer from scratch	Configure field maps; reuse pipeline class	Vendor manages schema and delivery
Maintenance burden	High — parser breaks on layout changes	Medium — update field map when API schema changes	Low — vendor handles it
Data coverage	Limited to what you can parse	Multiple sources through one API (see pricing for current options)	Varies by vendor
Integration path	Custom HTTP + parsing	HTTP API + env-var-configured Python	Vendor SDK or managed pipeline
Cost model	Infrastructure + dev time	Pay-per-result; no proxy or infra costs	Enterprise licensing

The middle option — CoreClaw structured data plus a lightweight custom normalizer — is the sweet spot for most teams. You get consistent JSON from the API, and your normalizer handles the last mile of field mapping and type coercion.

Limitations and Compliance

Schema drift. Even structured APIs occasionally change field names or add nested objects. Your normalizer should log unknown fields rather than silently dropping them, so you can update the field map when schemas change.
Type coercion edge cases. Phone numbers, currencies, and dates have regional variations. Test coercers against representative records from each region you operate in.
Deduplication is not identity resolution. Hashing on a phone number catches exact duplicates. It does not merge records that refer to the same business with slightly different phone formats. For identity resolution, you need fuzzy matching (e.g., Levenshtein distance or TF-IDF similarity) — a separate concern from normalization.
Compliance. Only collect and normalize publicly available data. Respect target-site terms of service, robots directives where relevant, and applicable privacy regulations (GDPR, CCPA). The normalizer does not make data collection lawful — it only structures data you have already lawfully obtained.
Provenance tracking. Always store the source URL and collection timestamp alongside normalized records. This supports auditability and reproducibility, especially when normalized data feeds AI-agent context.

FAQ

Do I need a normalizer if I use a structured web data API?

Yes, if you combine data from multiple sources or feed it into a system that expects a specific schema. A structured API reduces the parsing burden, but field names, value formats, and nesting still vary across sources. A normalizer handles that last mile.

How do I handle fields that exist in one source but not another?

Mark them as optional in your field map. The normalizer skips missing fields rather than erroring. For required fields, the validation stage catches records that lack them and logs an error.

What happens when a source changes its response schema?

Your normalizer will start dropping fields or producing None values. To detect this early, log unknown source fields (keys in the raw record that are not in your field map). Review these logs periodically and update the field map.

Can this pipeline run inside n8n, Zapier, or Make.com?

Yes. The normalizer is a plain Python class. You can run it as a Code node in n8n, a Python step in a Zapier Code action, or an HTTP webhook handler invoked from any no-code platform. The output is clean JSON that flows into the next node.

How do I normalize data for AI-agent context?

Run the normalizer, then add a _source_url and _collected_at field to each record. Package records as a JSON array or JSONL file and load them into a vector database or pass them as structured context to an LLM. The consistent schema makes retrieval and grounding more reliable.

Should I store raw records or only normalized records?

Store both. Raw records are your source of truth — if the normalizer has a bug, you can re-run it on the raw data. Store normalized records in your analytics or application database for fast querying.

How many sources can one normalizer handle?

There is no hard limit, but in practice, 3-5 sources per normalizer instance keeps the configuration manageable. For more sources, group them by domain (e.g., one normalizer for e-commerce sources, another for social media) and merge outputs downstream.

Summary

A web data normalization pipeline turns inconsistent JSON from multiple scraping sources into clean, validated, deduplicated records with a single consistent schema. The pipeline has five stages — field mapping, type coercion, validation, deduplication, and enrichment — and each stage is configurable without code changes. By combining CoreClaw's structured web data API with a lightweight Python normalizer, you get production-ready data without maintaining parsers for every source. Start by configuring your field maps for one source, validate the output, then add more sources as needed. Explore the CoreClaw console to test data sources and review pricing options for your volume.

YouTube Video Scraper API: How to Extract Public Video Metadata and Stats with Python

coreclaw — Fri, 11 Sep 2026 02:05:36 +0000

YouTube Video Scraper API: How to Extract Public Video Metadata and Stats with Python

The fastest way to turn a single public YouTube video URL into a stable, JSON-shaped metadata record — title, description, duration, view count, like count, tags, upload date, channel name, and thumbnail URL — is to pair an open-source video scraper repository with a small Python normalization layer that flattens fields, coerces types, and appends each result to a JSONL file. This article is for research engineers, content analytics teams, dataset curators, digital archivists, and AI-agent builders who need a per-video primitive they can call repeatedly without re-writing parsing logic each time.

The data-scrape/youtube-video-scraper-api repository provides a Python reference for extracting a single public video's metadata. Combine it with data-scrape/youtube-channel-scraper when your workflow also needs to walk every video on a channel and aggregate stats at the creator level.

TL;DR

Use data-scrape/youtube-video-scraper-api to fetch a single public YouTube video's metadata into a structured JSON record.
Define a narrow input: one video URL or ID per call. Batch jobs iterate the URL list externally so the scraper stays a one-record primitive.
Normalize each record into a stable schema with video_url, video_id, title, description, channel_name, channel_id, duration_seconds, view_count, like_count, comment_count, tags, upload_date, thumbnail_url, scraped_at, and source_repo.
Append each normalized record to a JSONL file so the dataset grows by line and can stream into pandas, DuckDB, or a downstream queue.
Switch to data-scrape/youtube-channel-scraper when you need to enumerate every video on a channel before extracting per-video metadata.
Capture one representative sample first, validate the field schema against your assumptions, and only then automate at scale.

Why per-video metadata is the right primitive

Most YouTube analysis eventually distills into one question: what does this specific video say, who posted it, and what is the engagement snapshot right now? A scraper that answers that question for a single URL becomes the building block for every higher-level workflow — a content audit, a competitive benchmarking study, a research dataset, an archival snapshot, or a context bundle for an AI agent.

A handful of problems recur once you start building:

Schema drift. Duration may arrive as an ISO 8601 string, a colon-delimited string, or raw seconds. View and like counts may be locale-formatted. Tags may arrive as a comma-separated string or a list. Raw JSON rarely has the field names a downstream consumer expects.
Missing fields. Not every public video has a description, tags, or a like count. Some videos disable comments. Your schema must handle nulls gracefully instead of crashing on a missing key.
ID format variability. YouTube video IDs are 11 characters, but URLs can arrive as watch?v=, youtu.be/, embed/, or shorts/ variants. A normalizer that extracts the canonical ID up front prevents duplicate records.
Thumbnail resolution. YouTube provides multiple thumbnail resolutions per video. Pick the highest available and store the URL so the record remains self-describing.
Rate of platform change. YouTube ships front-end changes frequently. Anything that consumes raw HTML or relies on selector paths needs periodic re-validation.
Compliance scope. Public video URLs are in scope. Private videos, member-only content, and live chat replays behind access controls are not. Always respect YouTube's Terms of Service and applicable law when collecting and storing public data.

This article takes the middle path: start with an open-source per-video reference, then add a small, auditable Python layer for normalization and JSONL append.

What the verified repositories provide

data-scrape/youtube-video-scraper-api is a Python reference built around extracting a single public video's metadata into a structured document. It exposes a scraper.py entry point, a requirements.txt, an examples/ folder, and a README. Read the README for the current setup steps and supported arguments before running it — the names of CLI flags and the shape of the response may differ from the illustrative examples in this article.

data-scrape/youtube-channel-scraper is a sibling repository that targets channel-level enumeration rather than single-video resolution. When your analysis needs to iterate over every video on a channel before extracting per-video metadata, you keep the same normalization layer and only swap the source adapter.

The data-scrape profile hosts the rest of the open-source repositories that share a similar pattern, including per-record JSONL normalization pipelines for TikTok, Instagram, X, Zillow, and other domains.

Pipeline design

A maintainable per-video extraction pipeline has five stages: capture raw JSON for a single public video URL, normalize the response into a stable field schema, coerce numeric and date fields into typed columns, append the record to a JSONL file, and surface a small validation job that flags schema drift. Keep the raw JSON as an artifact alongside the normalized record so any future schema change can be replayed against the original payload.

Setup and configuration

Clone the repository and install dependencies inside a virtual environment so packages do not collide with system Python:

git clone https://github.com/data-scrape/youtube-video-scraper-api.git
cd youtube-video-scraper-api
python -m venv .venv
source .venv/bin/activate  # Windows: .venv\Scripts\activate
pip install -r requirements.txt

Verify the scraper runs against a single public video URL before building the normalization layer:

python scraper.py --url "https://www.youtube.com/watch?v=dQw4w9WgXcQ"

The command above is illustrative. Check the repository README for the exact CLI flags and output path the current version uses.

Runnable Python: normalize and append

The following script wraps the scraper's raw JSON output with a normalization layer. It accepts a single video URL or a text file of URLs (one per line), extracts the canonical video ID, normalizes fields into a stable schema, and appends each record to a JSONL file:

"""
Normalize YouTube video metadata into a stable JSONL record.
Wraps the raw output from data-scrape/youtube-video-scraper-api.
"""
import json
import re
import sys
import subprocess
from datetime import datetime, timezone
from pathlib import Path

# ── Configuration ──────────────────────────────────────────────
SCRAPER_SCRIPT = "scraper.py"          # entry point in the cloned repo
OUTPUT_DIR = Path("output")
RAW_DIR = OUTPUT_DIR / "raw"
JSONL_PATH = OUTPUT_DIR / "videos.jsonl"
OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
RAW_DIR.mkdir(parents=True, exist_ok=True)

# ── ID extraction ───────────────────────────────────────────────
VIDEO_ID_RE = re.compile(r"(?:watch\?v=|youtu\.be/|embed/|shorts/)([A-Za-z0-9_-]{11})")

def extract_video_id(url: str) -> str | None:
    """Pull the 11-character video ID from any common YouTube URL variant."""
    match = VIDEO_ID_RE.search(url)
    return match.group(1) if match else None

# ── Field coercion ─────────────────────────────────────────────
def coerce_int(value) -> int | None:
    """Convert locale-formatted count strings like '1,234,567' to int."""
    if value is None:
        return None
    if isinstance(value, int):
        return value
    cleaned = re.sub(r"[^\d]", "", str(value))
    return int(cleaned) if cleaned else None

def coerce_duration(value) -> int | None:
    """Accept ISO 8601 duration, colon-delimited string, or raw seconds."""
    if value is None:
        return None
    if isinstance(value, (int, float)):
        return int(value)
    text = str(value).strip()
    # ISO 8601: PT1M30S
    iso_match = re.match(r"PT(?:(\d+)H)?(?:(\d+)M)?(?:(\d+)S)?", text)
    if iso_match:
        h = int(iso_match.group(1) or 0)
        m = int(iso_match.group(2) or 0)
        s = int(iso_match.group(3) or 0)
        return h * 3600 + m * 60 + s
    # Colon-delimited: 1:30 or 1:02:30
    parts = text.split(":")
    if len(parts) in (2, 3) and all(p.isdigit() for p in parts):
        return sum(int(p) * 60 ** (len(parts) - 1 - i) for i, p in enumerate(parts))
    return None

def coerce_tags(value) -> list[str]:
    """Accept a list or a comma-separated string; return a clean list."""
    if value is None:
        return []
    if isinstance(value, list):
        return [str(t).strip() for t in value if t]
    return [t.strip() for t in str(value).split(",") if t.strip()]

def pick_thumbnail(thumbnails) -> str | None:
    """Pick the highest-resolution thumbnail URL from the raw payload."""
    if not thumbnails:
        return None
    if isinstance(thumbnails, dict):
        return thumbnails.get("url") or thumbnails.get("high", {}).get("url")
    if isinstance(thumbnails, list) and thumbnails:
        last = thumbnails[-1]
        if isinstance(last, dict):
            return last.get("url")
    return None

# ── Normalization ──────────────────────────────────────────────
def normalize(raw: dict, video_url: str) -> dict:
    video_id = raw.get("video_id") or extract_video_id(video_url)
    return {
        "video_url": video_url,
        "video_id": video_id,
        "title": raw.get("title"),
        "description": raw.get("description"),
        "channel_name": raw.get("channel_name") or raw.get("author"),
        "channel_id": raw.get("channel_id"),
        "duration_seconds": coerce_duration(
            raw.get("duration") or raw.get("length")
        ),
        "view_count": coerce_int(raw.get("view_count") or raw.get("views")),
        "like_count": coerce_int(raw.get("like_count") or raw.get("likes")),
        "comment_count": coerce_int(raw.get("comment_count") or raw.get("comments")),
        "tags": coerce_tags(raw.get("tags")),
        "upload_date": raw.get("upload_date") or raw.get("publish_date"),
        "thumbnail_url": pick_thumbnail(raw.get("thumbnails")),
        "scraped_at": datetime.now(timezone.utc).isoformat(),
        "source_repo": "data-scrape/youtube-video-scraper-api",
    }

# ── Runner ─────────────────────────────────────────────────────
def scrape_one(url: str) -> dict:
    """Run the scraper for one URL and return raw JSON."""
    result = subprocess.run(
        [sys.executable, SCRAPER_SCRIPT, "--url", url],
        capture_output=True, text=True, timeout=60,
    )
    if result.returncode != 0:
        raise RuntimeError(f"Scraper failed for {url}: {result.stderr[:500]}")
    raw_path = RAW_DIR / f"{extract_video_id(url)}.json"
    raw_path.write_text(result.stdout, encoding="utf-8")
    return json.loads(result.stdout)

def append_jsonl(record: dict) -> None:
    with open(JSONL_PATH, "a", encoding="utf-8") as f:
        f.write(json.dumps(record, ensure_ascii=False) + "\n")

def run(url: str) -> dict:
    raw = scrape_one(url)
    record = normalize(raw, url)
    append_jsonl(record)
    print(f"OK  {record['video_id']}  views={record['view_count']}  "
          f"duration={record['duration_seconds']}s")
    return record

if __name__ == "__main__":
    if len(sys.argv) < 2:
        print("Usage: python normalize.py <video_url | urls_file.txt>")
        sys.exit(1)
    arg = sys.argv[1]
    if arg.endswith(".txt"):
        urls = [l.strip() for l in Path(arg).read_text().splitlines() if l.strip()]
        for u in urls:
            try:
                run(u)
            except Exception as e:
                print(f"ERR {u}: {e}")
    else:
        run(arg)

Run it for a single video:

python normalize.py "https://www.youtube.com/watch?v=dQw4w9WgXcQ"

Or batch-process from a URL list:

python normalize.py video_urls.txt

The script writes raw JSON to output/raw/<video_id>.json and normalized records to output/videos.jsonl. Adjust SCRAPER_SCRIPT to match the actual entry-point filename documented in the repository README.

Representative output

After running the normalizer, each line in videos.jsonl looks like this:

{
  "video_url": "https://www.youtube.com/watch?v=dQw4w9WgXcQ",
  "video_id": "dQw4w9WgXcQ",
  "title": "Rick Astley - Never Gonna Give You Up (Official Video) (4K Remaster)",
  "description": "The official video for Never Gonna Give You Up by Rick Astley...",
  "channel_name": "Rick Astley",
  "channel_id": "UCuAXFkgswmhWqot6PqM4aQ",
  "duration_seconds": 213,
  "view_count": 1600000000,
  "like_count": 1700000,
  "comment_count": 250000,
  "tags": ["Rick Astley", "Never Gonna Give You Up", "pop"],
  "upload_date": "2009-10-25",
  "thumbnail_url": "https://i.ytimg.com/vi/dQw4w9WgXcQ/maxresdefault.jpg",
  "scraped_at": "2026-09-11T10:15:00+00:00",
  "source_repo": "data-scrape/youtube-video-scraper-api"
}

The numbers above are representative of the field shape, not a live snapshot. View, like, and comment counts change continuously. Always treat scraped counts as point-in-time observations and record scraped_at so the observation window is auditable.

Use cases

Content archival. Capture metadata snapshots for videos that may be edited, deleted, or made private. The JSONL file plus raw JSON artifacts form a reproducible audit trail.
Competitive content analysis. Collect per-video stats for a competitor's recent uploads, then compare view velocity, engagement ratios, and tag strategies.
Research datasets. Build a labeled dataset of public video records. Each record carries its source URL and extraction timestamp so the dataset remains traceable.
AI agent context. Feed normalized video metadata as structured context into an LLM pipeline that summarizes or classifies content without re-scraping on every call.
SEO and keyword research. Extract titles, descriptions, and tags from ranking videos for a topic cluster, then feed the structured text into a keyword analysis workflow.
Creator benchmarking. Combine the per-video primitive with data-scrape/youtube-channel-scraper to aggregate stats at the channel level for creator comparison.

Comparison: per-video vs channel-level scraping

Dimension	Per-video (this article)	Channel-level
Primary repo	`youtube-video-scraper-api`	`youtube-channel-scraper`
Input	One video URL or ID	One channel URL or handle
Output	Single metadata record per video	List of video URLs + channel stats
Best for	Archival, point-in-time snapshot, AI context	Creator audits, upload cadence, benchmarking
Scaling pattern	External URL list → loop	Single channel → enumerate → loop
Normalization	Same schema, same JSONL target	Same schema, same JSONL target

The two repositories are complementary. A typical research workflow uses the channel scraper to discover the video URL list, then feeds each URL into the per-video scraper for deep metadata extraction.

Limitations and compliance

Point-in-time data. View, like, and comment counts are live counters. A record scraped at 10:00 will differ from one scraped at 10:05. Always store scraped_at and never present a scraped count as a fixed value.
No private data. This workflow targets public video URLs only. Member-only videos, private videos, and live chat replays behind access controls are out of scope.
Rate and pacing. Scraping at high concurrency risks IP-level blocking and may violate YouTube's Terms of Service. Add delays between requests and cap concurrency. If you need large-scale collection, verify current rate limits and terms before production use.
Layout changes. YouTube's front-end evolves. If the scraper depends on HTML structure or selector paths, a platform update can break extraction silently. Run a validation job that checks field presence on a known sample after each run.
Thumbnail availability. maxresdefault.jpg is not available for all videos. Fall back to hqdefault.jpg or another resolution if the highest-quality thumbnail returns 404.
Description truncation. YouTube may truncate long descriptions in certain views. Compare the scraped description against the full watch page to verify completeness.
Terms and law. Respect YouTube's Terms of Service, applicable data protection laws, and privacy requirements when collecting, storing, and sharing public video metadata. Do not redistribute scraped content in ways that exceed fair use or platform terms.

Maintenance checklist

[ ] Pin Python and dependency versions in the virtual environment.
[ ] Store raw JSON alongside normalized records for replay.
[ ] Run a schema-validation job after each batch — flag records with missing title, video_id, or duration_seconds.
[ ] Monitor scraper exit codes and log failures with the source URL for retry.
[ ] Re-validate against a known sample video after YouTube ships a front-end change.
[ ] Cap concurrency and add inter-request delays to respect platform rate expectations.
[ ] Document the scraped_at observation window in any downstream report or dashboard.

FAQ

Is there an official YouTube Data API?
Yes. YouTube provides an official Data API v3 with quotas for video, channel, and search endpoints. The open-source repository in this article is an independent reference for extracting public metadata without an API key. Evaluate both options against your quota needs, compliance requirements, and maintenance capacity.

What metadata fields can I extract from a public video?
Typical fields include title, description, channel_name, channel_id, duration_seconds, view_count, like_count, comment_count, tags, upload_date, and thumbnail_url. The exact fields depend on what the scraper returns and may vary by video type. Always validate against a sample.

How often should I re-scrape a video?
For archival, a single snapshot is sufficient. For engagement tracking, daily or weekly re-scrapes with a stored scraped_at timestamp let you compute deltas. Avoid high-frequency scraping — it adds load without meaningful signal for most metrics.

What happens when YouTube changes its layout?
If the scraper relies on page structure, a front-end update can break extraction. Keep raw JSON artifacts, run a validation job on a known sample after each run, and monitor for empty or null fields that signal a silent failure.

Can this connect to pandas, DuckDB, or an AI agent pipeline?
Yes. The JSONL output is line-delimited JSON, which loads directly into pandas via pd.read_json(..., lines=True), imports into DuckDB as a table, and streams into LLM context as structured documents. The per-record schema is intentionally flat to keep these integrations simple.

What should I verify before production use?
Confirm that the scraper's current CLI flags and output format match the README, test against a diverse set of video URLs (standard, Shorts, live replays, older uploads), validate field completeness, and verify that your request pacing respects YouTube's Terms of Service and rate expectations.

Can I use this for videos on private channels or member-only content?
No. This workflow targets public video URLs only. Private videos, member-only content, and any resource behind access controls are out of scope. Always respect YouTube's Terms of Service, applicable law, and privacy requirements.

Next steps

Start with a single public video URL, run the normalizer, and inspect the JSONL output. When you are ready to scale to channel-wide collection, use data-scrape/youtube-channel-scraper to enumerate video URLs, then feed each URL back into the per-video pipeline. Browse the data-scrape profile for adjacent repositories that share the same per-record JSONL pattern across TikTok, Instagram, X, Zillow, and other domains.

Zapier Web Scraping: How to Connect CoreClaw Data to Your CRM and Sheets

coreclaw — Fri, 11 Sep 2026 01:13:33 +0000

Zapier Web Scraping: How to Connect CoreClaw Data to Your CRM and Sheets

The fastest way to get live web data into your business tools is to treat the scraper as a webhook sender and Zapier as the router. Instead of running Python scripts on a server and manually copying JSON into Salesforce or Google Sheets, you send a single HTTP request from a managed web data API to Zapier's Webhooks app, then let Zapier format and forward the results to any of the 5,000+ apps in its directory. This article shows how to wire CoreClaw's structured web data into Zapier workflows for lead generation, price monitoring, and rank tracking — without writing custom code or managing proxies.

TL;DR

Use Zapier's Webhooks by Zapier app to receive structured JSON from a web data API, then route it to CRMs, spreadsheets, Slack, Airtable, or email.
The CoreClaw platform returns normalized JSON for sources like Google Maps, Amazon, Google Search, Instagram, YouTube, and TikTok, so the Zapier formatter step rarely needs custom parsing.
A typical workflow is: Trigger (schedule or manual) → Webhooks (GET/POST to CoreClaw endpoint) → Formatter (shape the JSON) → Action (write to Sheet, CRM record, Slack message, or database).
You can browse ready-made scrapers on the CoreClaw Workers store. Review current rates on the CoreClaw pricing page before scaling.
For custom scrapers, deploy directly from the CoreClaw console.

Why Web Data + Zapier Breaks Without a Structured Source

Most teams that try to connect web scraping to Zapier hit the same wall: the data they receive is raw HTML, inconsistent JSON, or a CSV attachment that changes column order every week. Zapier's formatter is powerful, but it expects predictable keys. If your scraper returns "price": "$19.99" on Monday and "current_price": "19.99 USD" on Tuesday, every downstream Zap step breaks.

The other common failure mode is infrastructure. A self-hosted scraper running on a Raspberry Pi or an EC2 instance needs cron jobs, log rotation, error alerts, and IP rotation. When it fails at 2 AM, the Zapier workflow receives nothing and your CRM pipeline goes quiet. The fix is not more Zapier filters; it is a scraper that returns a stable schema from a managed endpoint.

What CoreClaw + Zapier Looks Like

CoreClaw is a web data API platform. You pick a data source (Google Maps business listings, Amazon product pages, Google SERP results, social profiles), send a request with your search parameters, and receive structured JSON. That JSON has consistent keys, normalized types, and predictable nesting — exactly what Zapier's formatter and mapper steps need.

Zapier receives the JSON through its Webhooks by Zapier trigger, which exposes a unique URL. Your scraper or API client sends a POST request to that URL with the JSON payload. Zapier parses the payload and makes every field available to downstream actions.

The separation of concerns is clean:

Layer	Responsibility	Who manages it
Data collection	Fetch and normalize public web data	CoreClaw (managed API)
Routing	Receive JSON, apply filters, forward to apps	Zapier (no-code platform)
Storage / action	CRM, Sheet, Slack, database, email	Your existing tools

Step-by-Step: Build a Google Maps Leads → Zapier → Sheets Workflow

This example walks through a complete workflow: fetch Google Maps business listings for "coffee shops in Austin, TX" and write each result as a new row in a Google Sheet.

Step 1 — Create the Zapier Webhook Trigger

In Zapier, create a new Zap.
Choose Webhooks by Zapier as the trigger app.
Select Catch Hook.
Copy the webhook URL Zapier generates (it looks like https://hooks.zapier.com/hooks/catch/123456/abcdef/).

Step 2 — Test the Webhook Locally

Before wiring up the scraper, send a test payload from your terminal to confirm Zapier receives it:

curl -X POST https://hooks.zapier.com/hooks/catch/123456/abcdef/ \
  -H "Content-Type: application/json" \
  -d '{"name":"Test Coffee","address":"500 Congress Ave, Austin, TX","phone":"+1-512-555-0188","rating":4.6}'

Return to Zapier and click Test Trigger. You should see the JSON fields parsed into Zapier's field list.

Step 3 — Send CoreClaw Data to the Webhook

The script below mimics the request pattern you would use in a production workflow. It reads the endpoint and credentials from environment variables, fetches structured data from a conceptual CoreClaw endpoint, and forwards each record to the Zapier webhook URL. Replace the environment values with your actual credentials.

import json
import os
import time
from typing import Any

import requests

# --- Environment configuration; never hard-code credentials ---
CORECLAW_ENDPOINT = os.environ["CORECLAW_ENDPOINT"]
CORECLAW_API_KEY = os.environ["CORECLAW_API_KEY"]
ZAPIER_WEBHOOK_URL = os.environ["ZAPIER_WEBHOOK_URL"]

# Pacing between webhook calls to respect Zapier rate limits.
WEBHOOK_DELAY_SECONDS = int(os.environ.get("WEBHOOK_DELAY_SECONDS", "1"))


def fetch_records(query: str, location: str) -> list[dict[str, Any]]:
    """Fetch structured records from the CoreClaw endpoint."""
    headers = {
        "Authorization": f"Bearer {CORECLAW_API_KEY}",
        "Content-Type": "application/json",
    }
    payload = {
        "source": "google_maps",
        "query": query,
        "location": location,
        "max_results": 10,
    }
    resp = requests.post(CORECLAW_ENDPOINT, headers=headers, json=payload, timeout=120)
    resp.raise_for_status()
    data = resp.json()
    # The response shape depends on the current scraper version.
    return data.get("results", [])


def send_to_zapier(record: dict[str, Any]) -> None:
    """Forward one normalized record to the Zapier webhook."""
    resp = requests.post(
        ZAPIER_WEBHOOK_URL,
        headers={"Content-Type": "application/json"},
        json=record,
        timeout=30,
    )
    resp.raise_for_status()
    print(f"Sent: {record.get('name', 'unknown')} — {resp.status_code}")


def main() -> None:
    records = fetch_records("coffee shops", "Austin, TX")
    print(f"Fetched {len(records)} records")

    for record in records:
        send_to_zapier(record)
        time.sleep(WEBHOOK_DELAY_SECONDS)

    print("All records forwarded to Zapier.")


if __name__ == "__main__":
    main()

Run it with:

export CORECLAW_ENDPOINT="https://your-scraper.example.com/run"
export CORECLAW_API_KEY="replace-with-real-key"
export ZAPIER_WEBHOOK_URL="https://hooks.zapier.com/hooks/catch/123456/abcdef/"
export WEBHOOK_DELAY_SECONDS="1"
python zapier_bridge.py

In practice, you would schedule this script via cron, GitHub Actions, or a CoreClaw worker so it runs automatically. The script itself is stateless; every run fetches fresh data and forwards it.

Step 4 — Format and Map in Zapier

Once the webhook trigger is receiving data, add a Formatter step if you need to split addresses, convert ratings to numbers, or truncate long descriptions. Then add the action step:

Google Sheets → Create Spreadsheet Row
Salesforce → Create Lead
HubSpot → Create Contact
Slack → Send Channel Message
Airtable → Create Record

Map each JSON field to the corresponding column or field in the target app. Because CoreClaw returns normalized keys (name, address, phone, rating, website), the mapping is usually one-to-one.

Step 5 — Add Error Handling

Zapier's built-in error handling includes:

Filters: Skip records that are missing a phone number or have a rating below 3.0.
Paths: Route high-rated businesses to a "priority" sheet and others to a "standard" sheet.
Auto-replay: Zapier retries failed actions up to five times on paid plans.

For the scraper side, add a simple health check to your script so it alerts you when the endpoint returns an empty result set or a non-200 status:

if not records:
    raise SystemExit("No records returned. Check query parameters and endpoint status.")

Representative Output Shape

Here is what a single record looks like after it is normalized by the scraper and before it reaches Zapier:

{
  "name": "Houndstooth Coffee",
  "address": "401 Congress Ave, Austin, TX 78701",
  "phone": "+1-512-555-0199",
  "website": "https://houndstoothcoffee.com",
  "rating": 4.7,
  "reviews_count": 1240,
  "category": "Coffee shop",
  "hours": {
    "monday": "07:00-19:00",
    "tuesday": "07:00-19:00"
  },
  "source_url": "https://www.google.com/maps/search/coffee+shops+austin",
  "fetched_at": "2026-09-11T08:30:00+00:00"
}

Zapier parses this into flat fields like name, address, phone, website, rating, reviews_count, category, hours__monday, source_url, and fetched_at. The double underscore for nested objects is standard Zapier behavior and is easy to map in the action step.

Business Use Cases

Use case	Data source	Zapier action	Why it matters
Local lead generation	Google Maps	Create Salesforce lead or HubSpot contact	Sales teams need fresh local-business data without manual research
Price monitoring	Amazon	Append to Google Sheet, alert in Slack	E-commerce teams track competitor prices daily
SEO rank tracking	Google SERP	Log to Airtable, notify in email	Marketing teams monitor keyword movement without opening a browser
Influencer outreach	Instagram / YouTube	Create Notion database entry	Creator teams build outreach lists with follower counts and engagement rates
Competitor tracking	TikTok / social	Post to Slack channel	Product teams watch competitor launches and trends

Each use case uses the same three-layer pattern: CoreClaw collects and normalizes, Zapier routes and formats, and your existing tool stores or acts on the data.

Manual Scraping vs CoreClaw + Zapier vs Enterprise Integration

Dimension	Self-hosted scraper + custom code	CoreClaw API + Zapier	Enterprise ETL platform
Setup time	Days to weeks	Minutes to hours	Weeks to months
Code required	Yes — scraper, parser, router, auth	Minimal — env vars and one Python bridge	Yes — platform specialists
Schema stability	You maintain it	Platform maintains it	Vendor maintains it
Proxy management	Your responsibility	Included	Included
App ecosystem	Whatever you build	5,000+ Zapier apps	Varies by vendor
Best for	Engineering teams with DevOps capacity	Ops, marketing, and sales teams that need fast, no-code routing	Large organizations with dedicated data engineering
Pricing verification	Your infrastructure cost	CoreClaw pricing + Zapier plan	Vendor quote

For most small-to-mid-sized teams, the middle path is the right trade: structured data from a managed API, routed through a no-code platform they already use.

Limitations and Compliance

Public data only. CoreClaw returns data from public web pages. Do not attempt to use it for authenticated pages, private accounts, or data behind access controls.
Rate limits. Zapier imposes task limits and webhook rate limits based on your plan. A run that forwards 500 records in under a minute may trigger throttling. Space out requests with the WEBHOOK_DELAY_SECONDS setting.
Data freshness. Web pages change. A business that was open yesterday may be closed today. Add a fetched_at field to every record and expire or re-verify data on a schedule that matches your use case.
Field availability. Not every source returns every field. A Google Maps record may have no website; an Amazon record may have no reviews. Build Zapier filters that handle missing keys gracefully.
Privacy and terms. Even public data has limits. Respect the target site's terms, applicable privacy law, and robots directives. Do not republish scraped records in ways that mislead or violate platform policies.
Webhook security. The Zapier webhook URL is a secret. Do not commit it to version control. Use environment variables or a secrets manager. If a URL is exposed, regenerate it in Zapier and rotate the value immediately.

FAQ

Is there an official CoreClaw Zapier app?
There is no dedicated Zapier app for CoreClaw at the time of writing. The integration works through Zapier's universal Webhooks by Zapier trigger, which accepts any JSON payload. This is actually an advantage: you are not limited to predefined actions and can send any structured data CoreClaw returns.

What data fields does CoreClaw return?
Fields depend on the source and the specific scraper. Google Maps records typically include name, address, phone, website, rating, reviews_count, category, and hours. Amazon records include title, price, currency, availability, rating, and reviews_count. SERP records include rank, url, title, snippet, and query. Verify the current field set in the CoreClaw Workers store before building your Zapier mapper.

How often should the workflow run?
Daily is sufficient for most lead-generation and price-monitoring use cases. Hourly is reasonable for active campaigns or high-volatility products. Avoid sub-hourly runs unless you have a documented reason, because they burn through Zapier task quotas and scraper credits quickly.

What happens when a page layout changes?
Managed scrapers on the CoreClaw platform are maintained by the platform or the worker author. When a target site changes its layout, the scraper is updated and the schema stays stable. If you deploy a custom scraper via the CoreClaw console, you are responsible for updating the selector logic, but the rest of the Zapier workflow remains unchanged.

Can this connect to Salesforce, HubSpot, or a custom CRM?
Yes. Any app in the Zapier directory — including Salesforce, HubSpot, Pipedrive, Copper, and Close — can receive the formatted JSON. For custom CRMs without a Zapier integration, use the Webhooks by Zapier action step to POST to the CRM's REST API.

What should I verify before production use?
Confirm that: (1) the scraper returns the fields your Zapier mapper expects, (2) the Zapier webhook URL is stored securely, (3) your delay setting respects Zapier rate limits, (4) missing fields are handled by filters rather than crashing the Zap, and (5) your data collection respects target-site terms and applicable law.

How do I handle errors if the CoreClaw endpoint is down?
Wrap the fetch call in a retry loop with exponential backoff. If the endpoint is still down after three retries, exit with an error code so your scheduler (cron, GitHub Actions, or monitoring tool) can alert you. Do not forward partial or empty payloads to Zapier, because they create blank rows or incomplete CRM records.

Summary

Zapier is the router. CoreClaw is the structured data source. Together they replace manual copy-paste, fragile spreadsheet imports, and self-hosted scraper maintenance with a pipeline that ops teams can build in an afternoon. Start with a single source and a single destination — Google Maps to Google Sheets is the classic first workflow — then add filters, paths, and additional actions as your needs grow.

Browse ready-made scrapers on the CoreClaw Workers store, deploy custom scrapers from the CoreClaw console, and confirm current rates on the CoreClaw pricing page before scaling to high-volume automation.

TikTok Video Scraper GitHub: How to Extract Public Video Metadata with Python

coreclaw — Thu, 10 Sep 2026 02:11:35 +0000

TikTok Video Scraper GitHub: How to Extract Public Video Metadata with Python

The fastest way to turn a single public TikTok video URL into a stable, JSON-shaped record you can archive, feed into a database, or hand to an AI agent is to combine an open-source TikTok video scraper repository with a small Python normalization layer that flattens fields and appends each result to a JSONL file. This article is for research engineers, social analytics teams, content moderation architects, dataset curators, and AI-agent builders who need a per-video primitive they can call repeatedly without re-writing parsing logic.

The data-scrape/tiktok-video-scraper repository ships a Python reference for extracting a single public video's metadata. Pair it with data-scrape/tiktok-data-scraper-api when your workflow needs to walk many videos behind a hashtag, sound, or creator handle.

TL;DR

Use data-scrape/tiktok-video-scraper to fetch a single public TikTok video's metadata into a JSON file.
Define a narrow input up front: one video URL per run. Batch jobs iterate the URL list externally so the scraper stays a one-record primitive.
Normalize each record into a stable schema with video_url, video_id, creator_handle, caption, hashtags, music_title, music_author, view_count, like_count, comment_count, share_count, published_at, scraped_at, and source_repo.
Append each normalized record to a JSONL file so the dataset grows by line and can be streamed into pandas, DuckDB, or a downstream queue.
Combine the per-video primitive with data-scrape/tiktok-data-scraper-api when you also need a hashtag or creator pipeline that walks many videos.
Capture one representative sample first, validate the field schema against your assumptions, and only then automate at scale.

Why per-video metadata is the right primitive

Most TikTok analysis eventually distills into one question: what does this specific video say, who posted it, and what is the engagement snapshot right now? A scraper that answers that question for a single URL becomes the building block for every higher-level workflow — a creator audit, a hashtag velocity study, a moderation evidence pack, or a context bundle for an AI agent.

A handful of problems recur once you start building:

Schema drift. Caption delimiters, hashtag rendering, music attribution, and creator display names are formatted inconsistently across videos. Raw JSON rarely has the field names a downstream consumer expects.
Mixed video types. In-feed videos, share-link variants, and search-result pages can carry the same video URL but different surrounding metadata.
Numeric counters as strings. Like, view, comment, and share counts are returned as locale-formatted strings such as 15.4K or 1,200. Treat them as strings until you coerce, never as integers in raw form.
Time zone confusion. published_at may be delivered as a local timestamp without a zone suffix. Always normalize to UTC in your output and document the conversion.
Rate of layout change. TikTok ships front-end changes frequently. Anything that consumes raw HTML or relies on selector paths needs a re-validation job.
Compliance scope. Public video URLs are in scope. Private accounts, follower-only live replays, and direct messages are not.

This article takes the middle path: start with an open-source per-video reference, then add a small, auditable Python layer for normalization and JSONL append.

What the verified repositories provide

data-scrape/tiktok-video-scraper is a Python reference built around extracting a single public video into a JSON document. It exposes a scraper.py entry point, a requirements.txt, an examples/ folder, and a README. Read the README for the current setup steps and supported arguments before running it — the names of CLI flags and the shape of the response may differ from the illustrative examples in this article.

data-scrape/tiktok-data-scraper-api is a sibling repository that targets hashtag and creator discovery rather than single-video resolution. When your analysis needs to iterate over hundreds or thousands of videos behind one creator handle or hashtag, you keep the same normalization layer and only swap the source adapter.

The data-scrape profile hosts the rest of the open-source repositories that share a similar pattern, including adjacent per-record → JSONL normalization pipelines for other domains.

Pipeline design

Setup and configuration

Clone the repository and install dependencies inside a virtual environment so packages do not collide with system Python:

git clone https://github.com/data-scrape/tiktok-video-scraper.git
cd tiktok-video-scraper
python -m venv .venv
source .venv/bin/activate  # Windows: .venv\Scripts\activate
pip install -r requirements.txt

Set configuration through environment variables instead of hard-coding URLs or pacing. The example below forwards a single video URL, an output path, and a polite per-request delay to the scraper's CLI. Adjust variable names to match the current version of scraper.py; never commit a real target URL or proxy configuration to source control.

export TIKTOK_TARGET_URL="https://www.tiktok.com/@examplecreator/video/7280000000000000001"
export TIKTOK_OUTPUT_FILE="raw_video.json"
export TIKTOK_REQUEST_DELAY=2.0
python scraper.py \
  --url "$TIKTOK_TARGET_URL" \
  --output "$TIKTOK_OUTPUT_FILE" \
  --delay "$TIKTOK_REQUEST_DELAY"

Treat the first call as a smoke test. If the JSON contains every field your downstream code expects, the same command embedded in a loop is your batch driver. If a field is missing or renamed, update the normalization step rather than chasing one-off branches inside the consumer.

Normalization: From raw JSON to a stable JSONL record

The script below reads raw_video.json, flattens the response into a stable schema, coerces numeric and date fields, and appends one JSON line to videos.jsonl. Re-running it against a new URL grows the same file incrementally, which is the property you want for streaming analytics and audit trails.

import datetime as dt
import json
import os
import pathlib
import re

RAW_PATH = pathlib.Path(os.environ.get("TIKTOK_OUTPUT_FILE", "raw_video.json"))
JSONL_PATH = pathlib.Path(os.environ.get("TIKTOK_JSONL_FILE", "videos.jsonl"))
SOURCE_REPO = "tiktok-video-scraper"

SCHEMA = (
    "video_url", "video_id", "creator_handle", "creator_display_name",
    "caption", "hashtags", "music_title", "music_author",
    "view_count", "like_count", "comment_count", "share_count",
    "published_at", "scraped_at", "source_repo",
)


def normalize_tags(raw_tags):
    """Lowercase, strip leading '#', drop empty strings."""
    if not raw_tags:
        return []
    cleaned = []
    for tag in raw_tags:
        if not tag:
            continue
        cleaned.append(tag.lstrip("#").strip().lower())
    return [t for t in cleaned if t]


def parse_int(value):
    """Strip 'K', 'M', commas, and spaces; convert to int when possible."""
    if value is None:
        return None
    digits = re.sub(r"[^\d]", "", str(value))
    return int(digits) if digits else None


def normalize_record(raw):
    creator = raw.get("creator") or {}
    music = raw.get("music") or {}
    return {
        "video_url": raw.get("url") or "",
        "video_id": raw.get("video_id") or "",
        "creator_handle": creator.get("handle") or "",
        "creator_display_name": creator.get("display_name") or "",
        "caption": raw.get("caption") or "",
        "hashtags": normalize_tags(raw.get("hashtags")),
        "music_title": music.get("title") or "",
        "music_author": music.get("author") or "",
        "view_count": parse_int(raw.get("view_count")),
        "like_count": parse_int(raw.get("like_count")),
        "comment_count": parse_int(raw.get("comment_count")),
        "share_count": parse_int(raw.get("share_count")),
        "published_at": raw.get("published_at") or "",
        "scraped_at": dt.datetime.now(dt.timezone.utc).isoformat(timespec="seconds"),
        "source_repo": SOURCE_REPO,
    }


def main():
    raw = json.loads(RAW_PATH.read_text(encoding="utf-8"))
    record = normalize_record(raw)
    with JSONL_PATH.open("a", encoding="utf-8") as fh:
        fh.write(json.dumps(record, ensure_ascii=False) + "\n")
    print(f"Appended {record['video_url']} to {JSONL_PATH}")


if __name__ == "__main__":
    main()

Three details matter here. First, normalize_tags defends against hashtag drift by lowercasing and stripping the leading hash, so joins against a hashtag list stay stable across runs. Second, parse_int strips locale-formatted suffixes such as K and M, returning None for unknown values so downstream analytics can identify missing data instead of silently coercing it to 0. Third, scraped_at is recorded in UTC so joins across runs and time zones are reproducible.

Representative output

After running the smoke test and the normalization script, videos.jsonl contains one line that looks like the example below. Treat this as a representative shape, not real-time data; verify fields against the current scraper response before wiring the schema into production.

{
  "video_url": "https://www.tiktok.com/@examplecreator/video/7280000000000000001",
  "video_id": "7280000000000000001",
  "creator_handle": "examplecreator",
  "creator_display_name": "Example Creator",
  "caption": "A short demo video #python #data",
  "hashtags": ["python", "data"],
  "music_title": "Demo Track",
  "music_author": "Demo Artist",
  "view_count": 15400,
  "like_count": 920,
  "comment_count": 45,
  "share_count": 12,
  "published_at": "2026-09-10T08:30:00+00:00",
  "scraped_at": "2026-09-10T10:00:19+00:00",
  "source_repo": "tiktok-video-scraper"
}

A stable per-video JSONL schema means downstream tools such as pandas, DuckDB, or a CSV-to-Parquet job can read the file without per-record discovery logic, and any field-level diff between two captures of the same URL falls out naturally.

Use cases

A per-video metadata primitive supports several workflows:

Hashtag velocity research. Walk a hashtag's recent posts through the per-video scraper and compute weekly deltas in view_count to rank hashtags by acceleration rather than absolute volume.
Sound trending. Capture music_title and music_author per video and count co-occurrences across your sample to detect emerging sounds before they appear in official charts.
Creator vetting. Sample a creator's recent uploads and compute per-video engagement rates, useful when comparing reach claims against the actual public record.
Content moderation archive. Snapshot known unsafe URLs with their public fields preserved so an evidence pack remains reproducible if the platform takes action later.
AI-agent context packs. Bundle a curated set of public video records as a versioned text asset that an agent can quote and re-read without refetching the live site each time.

In every case, the analysis stays inside the boundaries of public data: no private accounts, no direct messages, no credentialed views of follower-only live replays.

Build, buy, and tool comparison

Most teams pick one of three paths for TikTok video metadata:

Self-hosted per-video scraper. Best for engineers who want full control over request shape, schema, and cadence. Maintenance burden falls on the team when the source page changes.
Self-hosted hashtag or creator pipeline. Best when the analysis requires thousands of videos per run and the team is willing to maintain pagination and concurrency code. The data-scrape/tiktok-data-scraper-api repository sits in this lane.
Managed scraper API. Best for teams that want a hosted endpoint, a normalized response, and a quota model. Cost and quota must be verified against the provider's current pricing page; do not assume numbers from older blog posts.

A pragmatic sequence is to use data-scrape/tiktok-video-scraper for the first dozen records, validate the per-video schema against your analytical assumptions, and only then evaluate a managed endpoint or a larger pipeline for ongoing refresh.

Checklist before production

Schema is documented, including the units for the count fields.
published_at is recorded in UTC with an explicit offset; scraped_at uses the same convention.
Numeric fields are coerced — no embedded K, M, commas, or ranges remain in the JSONL.
Raw JSON is archived alongside the normalized record so future schema changes can be replayed.
videos.jsonl deduplication is handled by video_id; appending an already-captured record is a deliberate action.
Refresh cadence matches the analytical question: weekly for trend snapshots, daily for active campaigns, hourly only when justified.
Access to the JSONL is scoped to the team that needs it, and sensitive columns are stripped if the file is shared externally.

Compliance and maintenance

Extracting metadata from public TikTok video URLs is acceptable for personal research, internal trend analysis, and academic study, but you must respect the relevant terms of service, applicable privacy law, and any contractual limits in your jurisdiction. Do not collect follower-only posts, direct messages, or any field that is not visible on the public video page. Do not bypass authentication, CAPTCHA, rate limits, or any other access control. If a target page asks you to stop automated access, stop and remove the captured records.

For long-running use, schedule a monthly review of the scraper response, re-validate a sample of records against the live public page, and pin the scraper to a known-good commit. The schema-drift monitoring job is what tells you that a field disappeared or changed type before an analyst does.

FAQ

Is there an official TikTok video metadata API?
TikTok offers official APIs for authorized partners and certain research programs, but access is restricted and requires approval. The repository-based workflow described here is for public web data that does not rely on official API access.

What fields does per-video metadata include?
The exact fields depend on the current scraper response. The schema in this article covers the high-value public fields: video URL and ID, creator handle and display name, caption, hashtags, music title and author, view/like/comment/share counts, publish timestamp, and the scraper repository that produced the record.

Can this access private accounts or direct messages?
No. This workflow is for public videos and public creator profiles only. Private data, direct messages, and account-cracking techniques are out of scope and should not be attempted.

How often should I refresh per-video snapshots?
Match the cadence to the question. Weekly is reasonable for trend snapshots. Daily is reasonable for active campaigns. Hourly refresh is rarely justified and increases the risk of throttling.

What happens when TikTok changes its page layout?
The scraper may stop returning the expected fields. Your normalization layer will log missing fields, giving you a clear signal of what broke. Update the mapper or switch to a newer repository version when available.

Can I feed JSONL into a CRM, a notebook, or an AI agent?
Yes. JSONL is one of the cleanest formats for pandas, DuckDB, or any LLM tool that accepts file-based context. Add a short header comment that documents the capture time and source repository.

Next Steps

If you want a transparent, maintainable way to capture public TikTok video metadata, start with the verified repositories:

data-scrape/tiktok-video-scraper for per-video metadata extraction.
data-scrape/tiktok-data-scraper-api for hashtag and creator discovery pipelines.
data-scrape for the full list of open-source public-web-data repositories.

For an adjacent public-web-data pattern that uses the same per-record → JSONL approach for a different domain — including a similar bulk-export pipeline for real-estate records — see data-scrape/zillow-data-scraper.

Build the normalization layer first, run it on a small sample, and only scale once the schema and compliance checks are solid.

CoreClaw n8n Integration: How to Connect Web Data API to No-Code Automation Workflows

coreclaw — Thu, 10 Sep 2026 01:06:54 +0000

CoreClaw n8n Integration: How to Connect Web Data API to No-Code Automation Workflows

You can connect the CoreClaw web data API to n8n through the built-in HTTP Request node and turn public web data into the starting point of any no-code workflow. The practical path is to copy your current CoreClaw endpoint from the console, configure a single HTTP Request node with authentication and JSON inputs, and pipe the response into Sheets, Slack, Airtable, a CRM, or an AI agent context store. This article shows a complete workflow you can adapt without writing any custom code.

TL;DR

Problem: n8n has hundreds of nodes for SaaS tools, but reliable public web data still has to come from somewhere.
Solution: Trigger CoreClaw through n8n's HTTP Request node, parse the JSON, and forward the result to the next node.
Outcome: A scheduled or webhook-driven workflow that runs a managed scraper and delivers structured records to your stack.
Next step: Open the CoreClaw console, copy the current endpoint, and wire it into the HTTP Request node described below.

Why n8n still needs an external data source

n8n is excellent at moving data between apps, but the public web itself is not a built-in node. Most teams hit the same wall:

They need a list of public business records, product prices, search results, or social posts.
They do not want to write and host a scraper themselves.
They want the result to land in Google Sheets, Airtable, Slack, HubSpot, Notion, or an AI agent.

If you connect n8n to a managed web data API, you skip the hosting and anti-detection work and focus on the orchestration. CoreClaw exposes a callable endpoint for each data source, so n8n treats it like any other HTTP service.

What the CoreClaw endpoint returns

CoreClaw's API is a JSON HTTP endpoint. You send a request that names the data source, the target parameters, and your authentication, and you get a structured response back.

The conceptual shape is:

{
  "job_id": "job_4f7c9b21",
  "status": "completed",
  "data_source": "google_maps",
  "output": {
    "schema": "business_listing",
    "record_count": 18,
    "records": [
      {
        "name": "Example Coffee Shop",
        "address": "123 Main St, Anytown",
        "rating": 4.6,
        "reviews": 213,
        "category": "Cafe"
      }
    ]
  }
}

The exact field names and available data sources depend on your account and the endpoint you call. Always confirm the current schema on the CoreClaw console before you wire it into production.

Build the n8n workflow

This walkthrough uses n8n's built-in nodes only. No custom n8n module or unofficial community node is required.

Step 1: Collect credentials and the current endpoint

In n8n, open Credentials and create a new HTTP Header Auth credential. Store the value of your CoreClaw API key there. Never paste the key directly into a node.

Then open your CoreClaw dashboard and copy the endpoint URL for the data source you want to call. Endpoint paths and versions change, so always copy the current value rather than guessing.

Step 2: Add a Schedule or Webhook trigger

Most production workflows fall into one of two patterns:

Scheduled run. Drop in a Schedule Trigger node and pick a cron expression. Daily at 09:00 works for most refresh-once-a-day use cases.
Event-driven run. Drop in a Webhook node and let an external system start the workflow. Useful for on-demand lead pulls or reactive monitoring.

For testing, use n8n's Manual Trigger node so you can click "Execute Workflow" without waiting for a schedule.

Step 3: Configure the HTTP Request node

Add an HTTP Request node and wire it to the trigger. Use these settings:

Field	Value
Method	POST
URL	The endpoint you copied from the console
Authentication	Generic Credential Type → HTTP Header Auth
Header name	`Authorization`
Header value	`Bearer <your CoreClaw API key>`
Send Body	true
Body Content Type	JSON
Specify Body	Using JSON

The body should describe the scrape you want to run. Keep it small and stable:

{
  "data_source": "google_maps",
  "query": "coffee shops in Anytown",
  "max_results": 25
}

Use data_source, query, and max_results as the conceptual inputs and substitute whatever fields your chosen CoreClaw endpoint actually accepts. Always read the current parameter list from the dashboard.

Step 4: Parse the response with a Code node

Most CoreClaw endpoints return JSON that is already shaped for downstream tools, but n8n often wants a flatter row-per-record shape for Sheets or Airtable. Add a Code node and convert the response into a list of items.

const items = [];

const response = $input.first().json;
const records = response?.output?.records ?? [];

for (const record of records) {
  items.push({
    json: {
      job_id: response.job_id,
      status: response.status,
      ...record,
    },
  });
}

return items;

This keeps each record as its own item, which is how Sheets, Airtable, and most CRM nodes expect to receive rows.

Step 5: Deliver the data

Add the destination node that matches your stack:

Google Sheets: use the Append Row operation. Map each field from the previous node to a column.
Airtable: use Create Record and pick the table and base.
Slack: use the Send Message operation and include the most important fields in the text block.
HubSpot or Salesforce: use the corresponding CRM node and map name, address, rating, and similar fields.
AI agent: use the HTTP Request node again to forward the cleaned records to your agent's retrieval endpoint, or write them to a vector store via the agent's API.

Add an IF node if you want to skip empty results, and an Error Trigger workflow so failed runs page you instead of disappearing.

Step 6: Test end to end

Click Execute Workflow and walk through each node. Confirm:

The HTTP Request node returns HTTP 200 and a JSON body.
The Code node produces one item per record.
The destination node accepts the mapped fields.

Once the manual run works, activate the workflow and let the trigger run it on schedule.

A local Python check before you wire n8n

It is worth validating the endpoint with a one-line Python script before you commit the workflow to a cron schedule. That lets you iterate on parameters without the n8n UI in the way.

import os
import json
import requests

ENDPOINT = os.environ["CORECLAW_ENDPOINT"]
API_KEY = os.environ["CORECLAW_API_KEY"]

payload = {
    "data_source": "google_maps",
    "query": "coffee shops in Anytown",
    "max_results": 10,
}

response = requests.post(
    ENDPOINT,
    headers={
        "Authorization": f"Bearer {API_KEY}",
        "Content-Type": "application/json",
    },
    json=payload,
    timeout=60,
)

response.raise_for_status()
data = response.json()
print(json.dumps(data.get("output", {}).get("records", [])[:3], indent=2))

Run it once, confirm the response shape, and only then paste the same parameters into the n8n HTTP Request node. This saves you from debugging n8n expressions when the real problem is a parameter the endpoint does not recognize.

Deployment checklist

Before you turn the workflow on in production, confirm:

[ ] The endpoint URL was copied from the CoreClaw console and not guessed.
[ ] The API key is stored in an n8n credential, not in the workflow JSON.
[ ] The target data source is allowed by your account and plan.
[ ] The Code node produces one item per record, not one item per workflow run.
[ ] The destination node is mapped to the actual JSON fields, not placeholder names.
[ ] You have an Error Trigger workflow that notifies you when a run fails.
[ ] You respect the target site's terms of service and applicable privacy law.

Business use cases

A CoreClaw + n8n workflow fits several recurring operational needs:

Daily lead refresh. Run a Google Maps scrape every morning and append new businesses to a HubSpot list.
Price monitoring. Poll product pages every few hours and post a Slack alert when a competitor drops below threshold.
SEO rank tracking. Pull SERP results for a keyword list once a day and write ranks to a Google Sheet.
Social listening. Pull public posts or profiles from a social data source and pipe mentions into a Notion database.
Agent context refresh. Re-run a public source on a schedule and push the cleaned records into an AI agent's retrieval store so its answers stay current.

How it compares to other approaches

Dimension	DIY scraper + n8n HTTP node	n8n + CoreClaw	Hosted n8n alternative with built-in data nodes
Best for	Full control, custom code, very large custom datasets	Teams that want managed collection and no-code orchestration	Teams locked into one platform's node ecosystem
Setup model	You build and host the scraper	You configure an HTTP node and supply an API key	You enable the platform's pre-built node
Data coverage	Whatever you code	Depends on the CoreClaw product store endpoints	Limited to the platform's catalog
Output format	You design and maintain	JSON by default; n8n transforms it for the destination node	Usually the platform's native shape
Maintenance burden	High: servers, proxies, anti-detection	Low: CoreClaw handles runtime and IP rotation	Medium: depends on the platform's update cadence
Integration path	Custom webhook + n8n	HTTP node + Sheets / Airtable / Slack / CRM / agent	Built-in node only
Quota / freshness	Set by your own infrastructure	Confirm current limits on the CoreClaw pricing page	Confirm current limits on the platform's docs
Pricing verification	Your own hosting bill	Check official pricing before scaling	Check official pricing before scaling

For most teams, the managed collection + n8n orchestration combination is the shortest path from "I need this public data" to "it is in my Sheet, CRM, or agent."

Limitations and compliance

A managed API does not remove your responsibility to collect data lawfully. Keep these limits in mind:

Public data only. Use the API to access public business listings, public product pages, and public SERP or social content. Do not use it for authenticated or private data.
Terms of service. Respect the target site's terms, robots directives where applicable, and any rate guidance in your CoreClaw plan.
Rate limits. Even with managed infrastructure, aggressive pacing can still trigger blocks or violate terms.
Schema drift. When a data source changes its public layout, the response shape can shift. Validate record_count and required fields in your Code node.
Data retention. Store only what you need and delete records when they are no longer useful.
Regional rules. Confirm that your use case complies with local privacy and data-protection law.

The API removes infrastructure work, not your compliance obligations.

FAQ

Does CoreClaw publish an official n8n node?

The most reliable approach is the built-in HTTP Request node, which calls the same CoreClaw endpoint used by the Python and JavaScript clients. If a community or first-party n8n node appears later, treat it as a thin wrapper around the same HTTP API and validate it against the current console documentation before you rely on it in production.

Which CoreClaw data sources can n8n call?

Any data source available in the product store is callable, including Google Maps, Google Search, Amazon products, Instagram, and YouTube channel data. The exact list and parameters change over time, so check the store and the console for the current options.

How often should the workflow run?

It depends on how fast the source data changes. Daily is enough for most business listings and product prices. Hourly or sub-hourly makes sense for time-sensitive SERP or pricing use cases. Start with a conservative schedule and tighten only if the data justifies it.

What happens when the response shape changes?

Add a validation step in the Code node. Check that output.records exists and that each record has the fields your destination node expects. When validation fails, route the workflow to an error channel and notify yourself instead of silently writing bad rows to your Sheet or CRM.

Can I chain CoreClaw calls inside one n8n workflow?

Yes. Add a second HTTP Request node, pipe the records from the first call into it as parameters, and you get a fan-out workflow. This is useful for "for each business, fetch details" patterns where one endpoint returns IDs and another enriches them.

Will this work in n8n Cloud and self-hosted n8n?

Yes. The HTTP Request node, the Code node, the Schedule Trigger, and the Webhook node are all built-in nodes that work the same way in both editions. The destination nodes (Sheets, Airtable, Slack, HubSpot, etc.) also behave identically.

What should I verify before production use?

Confirm the target data source is public, the endpoint and parameters match the current console documentation, the API key is stored in an n8n credential, and your use case complies with the target site's terms and applicable law. Review the current CoreClaw pricing to confirm the planned run frequency fits your plan.

Summary and next steps

You can wire CoreClaw into n8n with no custom nodes and no custom code. Add an HTTP Request node, point it at the endpoint you copy from the console, parse the JSON in a Code node, and forward the records to Sheets, Slack, Airtable, a CRM, or an AI agent.

Start with a single data source and a manual trigger. Validate the response shape, add the destination node, then switch to a Schedule Trigger once the end-to-end run succeeds. Add error handling and quota checks only after the happy path works.

Zillow Data Scraper GitHub: How to Export Bulk Property Records to CSV with Python

coreclaw — Wed, 09 Sep 2026 02:20:07 +0000

Zillow Data Scraper GitHub: How to Export Bulk Property Records to CSV with Python

The fastest way to build a local real estate dataset from public Zillow listing pages without committing to a paid brokerage feed is to combine a runnable open-source scraper repository with a small Python normalization layer that exports clean CSV files. This article is for real estate analysts, PropTech engineers, independent investors, market researchers, and AI-agent builders who need a repeatable bulk property export they can refresh, version, and feed into downstream analysis tools.

The data-scrape/zillow-data-scraper repository provides a runnable Python reference for collecting public property records, and the same query and output pattern can be paired with data-scrape/zillow-scraper-api when you also want a long-running monitor that watches the same listings for changes.

TL;DR

Use data-scrape/zillow-data-scraper to collect public Zillow property records for a bounded geography and listing type.
Define a narrow target list up front: city, ZIP, neighborhood, or property type. Broad "all listings" queries produce noisy, hard-to-maintain CSVs.
Normalize each raw record into a stable CSV schema with captured_at, property_url, address, price, beds, baths, sqft, lot_size, year_built, property_type, and status.
Aggregate to one row per listing so joins with public county assessor data or in-house CRM records are easy.
Refresh daily or weekly for market analysis; treat the output as a snapshot, not a live stream.
Combine with data-scrape/zillow-scraper-api when you also want change-detection over time on the same address set.

Why Bulk Property Export Is Harder Than It Looks

Public real estate listings look simple until you try to turn them into a reliable CSV. A handful of problems recur:

Selector drift. Listing cards, price badges, status labels, and address blocks change frequently, and an outdated selector silently drops columns or rows from your export.
Mixed property types. Single-family rentals, condos, new construction, auctions, and "off market" pre-listings share one search URL but use different card structures.
Address and unit formatting. Unit numbers, fractional addresses, and city/zip combinations are inconsistent across listings, so a naive join by address fails on roughly five to fifteen percent of rows in real datasets.
Price and status freshness. A price change on the live page may not propagate to the cached version served to scrapers, so two captures of the same URL can disagree on price within the same hour.
Rate limits and bot detection. Aggressive polling hits throttling quickly, and aggressive retries can get an IP blocked long enough to lose a daily refresh window.
Legal and policy scope. Bulk export of public listing pages is a different posture from harvesting private agent notes, owner portals, or authenticated user data; respect the relevant terms of service and applicable law.

This article takes the middle path: start with an open-source reference repository, then add a small, auditable Python layer for normalization and CSV export.

What the Verified Repositories Provide

The data-scrape/zillow-data-scraper repository is a Python reference for collecting public Zillow property records. It ships a scraper.py entry point, a requirements.txt, an examples/ folder, and a README. Read the README for the current setup steps, supported query parameters, and environment variables before running it.

The data-scrape profile hosts related repositories that share a similar pattern. data-scrape/zillow-scraper-api targets the same listings but is shaped for ongoing monitoring use cases, where you want to detect new listings, price drops, or status changes on an address set you already track. The two repositories overlap in raw field names, so the normalization layer you write for bulk export can be reused for monitoring with only the source adapter swapped.

Pipeline Design

A maintainable bulk export pipeline has five stages: capture raw JSON for a bounded geography and listing type, normalize fields into a stable schema, deduplicate by property_url and address, coerce numeric and date fields into typed columns, and write a single CSV with one row per listing.

Setup and Configuration

Clone the repository and install dependencies inside a virtual environment so packages do not collide with system Python:

git clone https://github.com/data-scrape/zillow-data-scraper.git
cd zillow-data-scraper
python -m venv .venv
source .venv/bin/activate  # Windows: .venv\Scripts\activate
pip install -r requirements.txt

Set configuration through environment variables instead of hard-coding geographies or pacing limits. The example below forwards a target list, output path, and request budget to the scraper's CLI. Adjust variable names to match the current version of scraper.py; never commit a real target list or proxy configuration to source control.

export ZILLOW_TARGETS_FILE="targets.txt"
export ZILLOW_OUTPUT_FILE="raw_listings.json"
export ZILLOW_MAX_REQUESTS=400
export ZILLOW_REQUEST_DELAY=2.5
python scraper.py \
  --input "$ZILLOW_TARGETS_FILE" \
  --output "$ZILLOW_OUTPUT_FILE" \
  --max-requests "$ZILLOW_MAX_REQUESTS" \
  --delay "$ZILLOW_REQUEST_DELAY"

Keep targets.txt narrow: one city or ZIP per line, optionally annotated with property type or listing status. A first useful set is three to five geographies you already understand.

Normalization: From Raw JSON to a Stable CSV

The script below reads raw_listings.json, normalizes each record, deduplicates by property_url, coerces numeric fields, and writes a CSV. The output schema is stable across runs, so downstream tools can rely on the column names without per-run discovery.

import csv
import datetime as dt
import json
import os
import pathlib
import re

RAW_PATH = pathlib.Path(os.environ.get("ZILLOW_OUTPUT_FILE", "raw_listings.json"))
CSV_PATH = pathlib.Path(os.environ.get("ZILLOW_CSV_FILE", "listings.csv"))

NUMERIC_FIELDS = ("price", "beds", "baths", "sqft", "lot_size", "year_built")
COLUMNS = [
    "captured_at", "property_url", "address", "city", "state", "zip",
    "price", "beds", "baths", "sqft", "lot_size", "year_built",
    "property_type", "status",
]


def parse_int(value):
    if value is None:
        return ""
    digits = re.sub(r"[^\d]", "", str(value))
    return int(digits) if digits else ""


def normalize_record(raw, captured_at):
    address = raw.get("address") or {}
    return {
        "captured_at": captured_at,
        "property_url": raw.get("url") or "",
        "address": " ".join(
            part for part in [
                address.get("street"),
                address.get("unit"),
            ] if part
        ).strip(),
        "city": address.get("city") or "",
        "state": address.get("state") or "",
        "zip": address.get("zip") or "",
        "price": parse_int(raw.get("price")),
        "beds": parse_int(raw.get("beds")),
        "baths": parse_int(raw.get("baths")),
        "sqft": parse_int(raw.get("sqft")),
        "lot_size": parse_int(raw.get("lot_size")),
        "year_built": parse_int(raw.get("year_built")),
        "property_type": raw.get("property_type") or "",
        "status": raw.get("status") or "",
    }


def main():
    captured_at = dt.datetime.now(dt.timezone.utc).isoformat(timespec="seconds")
    raw_records = json.loads(RAW_PATH.read_text(encoding="utf-8"))

    seen_urls = set()
    rows = []
    for raw in raw_records:
        url = (raw.get("url") or "").strip()
        if not url or url in seen_urls:
            continue
        seen_urls.add(url)
        rows.append(normalize_record(raw, captured_at))

    with CSV_PATH.open("w", encoding="utf-8", newline="") as fh:
        writer = csv.DictWriter(fh, fieldnames=COLUMNS)
        writer.writeheader()
        for row in rows:
            writer.writerow(row)

    print(f"Wrote {len(rows)} rows to {CSV_PATH}")


if __name__ == "__main__":
    main()

The script keeps the schema stable across runs, drops duplicates by property_url, and uses parse_int to strip currency symbols, commas, and ranges from price and square footage. Adjust the column list if you want to add days_on_zillow, hoa_fee, or price_per_sqft.

Representative Output

After running the script above against a small target set, listings.csv looks like the example below. Treat this as a representative shape, not real-time data; verify fields against the current page structure of the source pages.

captured_at,property_url,address,city,state,zip,price,beds,baths,sqft,lot_size,year_built,property_type,status
2026-09-09T10:00:19+00:00,https://www.zillow.com/homedetails/123-Main-St/0000_rid/,123 Main St,Asheville,NC,28801,425000,3,2,1480,7200,1998,SingleFamily,ForSale
2026-09-09T10:00:19+00:00,https://www.zillow.com/homedetails/456-Oak-Ave/0001_rid/,456 Oak Ave,Asheville,NC,28801,389000,2,1,1080,5400,1962,SingleFamily,ForSale
2026-09-09T10:00:19+00:00,https://www.zillow.com/homedetails/789-Pine-Rd/0002_rid/,789 Pine Rd,Asheville,NC,28803,512000,4,3,2120,9100,2005,SingleFamily,Pending

A stable schema means downstream tools such as pandas, DuckDB, or a CSV-to-Parquet job can read the file without per-run discovery logic.

Use Cases

A bulk property export supports several analyst workflows:

Market snapshots for a neighborhood. Capture every active listing in three to five ZIPs once per week and compute median list price, price per square foot, and days-on-market distribution.
Comparable sales research. Pair each in-contract or sold record with public county assessor data using address plus zip as the join key.
Investor outreach lists. Filter for property_type = SingleFamily and price < 400000, then enrich with public parcel data before mailing.
Rental yield modeling. Combine the price column with publicly listed rent ranges and standard expense assumptions to model gross yield by ZIP.
AI-agent context packs. Wrap the CSV as a versioned context file that an agent can quote, cite, and re-read on demand instead of refetching the live site each time.

Build, Buy, and Tool Comparison

Most teams pick one of three paths for bulk property data. Use the same evaluation criteria for each:

Self-hosted open-source scraper. Best for engineers who want full control over query shape, refresh cadence, and CSV schema. Maintenance burden falls on the team when the source page changes.
Managed scraper API. Best for teams that want a hosted endpoint, a normalized response, and a quota model. Cost and quota must be verified against the provider's current pricing page; do not assume numbers from older blog posts.
Manual CSV download. Best for one-off research where a single CSV from the source site is enough. Does not scale to recurring analysis.

A reasonable first step is to use data-scrape/zillow-data-scraper for the first export, validate the CSV against your internal assumptions, and only then evaluate a managed endpoint for ongoing refresh.

Checklist Before Production

Run through these checks before you publish or share a CSV:

Schema is stable and documented, including units for sqft and lot_size.
captured_at is recorded in UTC so joins across time zones are reproducible.
Numeric columns are coerced; no embedded currency symbols or ranges remain.
Duplicates are removed by property_url, not just by address.
Refresh cadence matches the analytical question: daily for active campaigns, weekly for market snapshots.
Output file size and row count are recorded in a small manifest so a broken run is visible.
Refresh runs are idempotent and can be re-run from scratch.
Access to the CSV is scoped to the team that needs it, and sensitive columns are removed if the file is shared externally.

Compliance and Maintenance

Bulk export of public listing pages is acceptable for personal research, internal market analysis, and academic study, but you must respect the relevant terms of service, applicable privacy law, and any contractual limits in your jurisdiction. Do not collect owner names, agent contact details, or any field that is not visible on the public listing page. Do not bypass authentication, CAPTCHA, rate limits, or any other access control. If a target page asks you to stop automated access, stop and remove the captured records.

For long-running use, schedule a monthly review of the source page structure, re-validate a sample of rows against the live page, and pin the scraper to a known-good commit so a bad update does not silently corrupt your downstream dataset.

FAQ

Is there an official Zillow bulk export API?
Zillow does not publish a general-purpose bulk listing export API for third-party developers. The public search pages remain the most common source, which is why open-source scrapers are widely used.

What columns should a Zillow export contain?
At minimum: captured_at, property_url, address, city, state, zip, price, beds, baths, sqft, property_type, and status. Add lot_size, year_built, hoa_fee, and days_on_zillow when you need them.

How often should I refresh the export?
Weekly is a good default for market snapshots. Daily is reasonable for active campaigns. Hourly refresh is usually wasteful and increases the risk of being throttled.

What happens when the source page layout changes?
The scraper may silently drop columns or rows. Compare a sample of listings.csv against the live page on every release of the scraper, and add a small validation job that flags missing fields.

Can I feed the CSV to an AI agent or a notebook?
Yes. CSV is one of the cleanest formats for pandas, DuckDB, and CSV-aware LLM tools. Add a short header comment that documents the capture time and the source repository.

What is the difference between zillow-data-scraper and zillow-scraper-api?
The data scraper is shaped for bulk snapshot exports to CSV or JSON. The scraper API is shaped for ongoing monitoring of a known address set. They share enough of the field schema that the same normalization layer can serve both.

How do I avoid being blocked?
Keep request rates modest, set a real ZILLOW_REQUEST_DELAY, respect any robots directives, and stop immediately if you see a CAPTCHA or a 429 response. Treat the public pages as a shared resource.

Next Steps

If you want to extend the same pipeline to ongoing monitoring, pair this export with data-scrape/zillow-scraper-api and reuse the normalization layer to detect new listings, price drops, and status changes on the same address set. For a broader cross-vertical pattern that also covers hotels, marketplaces, and local businesses, browse the data-scrape profile for the full list of public repositories and their READMEs.

CoreClaw Workers: How to Deploy a Custom Web Scraper Without DevOps

coreclaw — Wed, 09 Sep 2026 01:05:09 +0000

CoreClaw Workers: How to Deploy a Custom Web Scraper Without DevOps

You can deploy a custom web scraper to managed cloud workers and skip the server, proxy pool, and job-queue setup. The practical path is to write a small scraping function, wrap it as a worker job, and let a managed platform handle scheduling, retries, and delivery. This article shows a complete Python workflow you can adapt to any public data source.

TL;DR

Problem: Self-hosting a scraper means managing servers, proxies, retries, storage, and observability.
Solution: Deploy the scraper as a managed worker through CoreClaw Workers and focus on the extraction logic.
Outcome: You get a callable endpoint that runs your scraper in the cloud, handles execution, and returns structured JSON.
Verified links: CoreClaw product store, CoreClaw pricing.

Why self-hosting a scraper becomes a DevOps project

A scraper that works on your laptop rarely works in production without extra work. You need:

A server or container that stays running.
A proxy or IP rotation strategy to avoid blocks.
Retry logic with exponential backoff and jitter.
A queue or scheduler for recurring jobs.
Storage and a way to normalize output.
Logs and alerts so you know when a target changes.

Each of these is solvable, but together they turn a small data task into infrastructure work. If the goal is to collect public web data for a product, a report, or an AI agent, most teams would rather spend time on the data than on the plumbing.

What CoreClaw Workers does

CoreClaw Workers gives you a managed environment to deploy scraper scripts. You provide the extraction logic or choose from the CoreClaw product store, and the platform handles execution, scaling, retries, and result delivery. You call the worker through an HTTP endpoint or schedule it to run on an interval.

The key idea is separation of concerns:

Concern	You handle	Platform handles
Extraction logic	Yes	No
Target URLs and parameters	Yes	No
Servers and runtime	No	Yes
Proxies and IP rotation	No	Yes
Retries and queueing	No	Yes
Structured result delivery	No	Yes

This lets you ship a scraper the same way you ship a serverless function: write code, configure inputs, and get a callable URL.

Build and deploy a custom scraper

This workflow uses Python and environment variables for anything account-specific. You can run the same script locally for testing and then point it at your CoreClaw worker endpoint for production runs.

Step 1: Write a small scraping function

Keep the scraper focused on one public data source. The example below extracts article titles and URLs from a public index page. It uses standard Python libraries plus requests, which most environments already have.

import os
import re
import json
import time
from urllib.parse import urljoin

import requests
from bs4 import BeautifulSoup


def scrape_index_page(url):
    """
    Extract article links from a public index page.
    Returns a list of dicts with title, url, and fetched_at.
    """
    response = requests.get(url, timeout=30)
    response.raise_for_status()

    soup = BeautifulSoup(response.text, "html.parser")
    records = []

    for link in soup.find_all("a", href=True):
        href = link["href"]
        title = link.get_text(strip=True)
        if title and "/article/" in href:
            records.append({
                "title": title,
                "url": urljoin(url, href),
                "source_url": url,
                "fetched_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
            })

    return records


if __name__ == "__main__":
    target = os.environ.get("TARGET_URL", "https://example.com/blog")
    results = scrape_index_page(target)
    print(json.dumps(results[:5], indent=2))

Install dependencies with:

pip install requests beautifulsoup4

Step 2: Create a worker payload

A worker payload tells the platform which function to run and what inputs to pass. Keep the payload simple: a target URL, a few parameters, and a stable job name.

import os
import json
import time
import requests

# Copy these values from your CoreClaw Workers dashboard.
WORKER_ENDPOINT = os.environ["WORKER_ENDPOINT"]
WORKER_API_KEY = os.environ["WORKER_API_KEY"]

job_payload = {
    "job_name": "blog-index-scraper",
    "target_url": "https://example.com/blog",
    "output_schema": "article_links",
    "requested_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
}

headers = {
    "Authorization": f"Bearer {WORKER_API_KEY}",
    "Content-Type": "application/json",
}

response = requests.post(
    WORKER_ENDPOINT,
    headers=headers,
    json=job_payload,
    timeout=60,
)
print(response.status_code)
print(json.dumps(response.json(), indent=2))

The WORKER_ENDPOINT is the full URL shown in the CoreClaw Workers console. Do not hard-code it. Endpoint paths and versions change, so always copy the current value from the dashboard.

Step 3: Poll for results

Managed workers are asynchronous. After you submit a job, poll the status endpoint until the run completes or fails.

import os
import json
import time
import requests

WORKER_API_KEY = os.environ["WORKER_API_KEY"]
status_url = os.environ["WORKER_STATUS_URL"]

def poll_for_result(status_url, api_key, max_attempts=30, sleep_seconds=10):
    headers = {"Authorization": f"Bearer {api_key}"}

    for attempt in range(max_attempts):
        response = requests.get(status_url, headers=headers, timeout=30)
        response.raise_for_status()
        data = response.json()

        if data.get("status") in ("completed", "failed"):
            return data

        print(f"Attempt {attempt + 1}: status={data.get('status')}")
        time.sleep(sleep_seconds)

    raise TimeoutError("Worker did not finish within the polling window")


result = poll_for_result(status_url, WORKER_API_KEY)
print(json.dumps(result, indent=2))

Step 4: Normalize and store the output

Once the worker returns structured JSON, normalize it into the shape your pipeline expects. The example below writes results to a local JSONL file, but you can just as easily send them to a database, CRM, or AI agent context store.

import json
from pathlib import Path

records = result.get("output", {}).get("records", [])
output_path = Path("data") / "articles.jsonl"
output_path.parent.mkdir(parents=True, exist_ok=True)

with output_path.open("a", encoding="utf-8") as f:
    for record in records:
        f.write(json.dumps(record, ensure_ascii=False) + "\n")

print(f"Appended {len(records)} records to {output_path}")

Expected output shape

A successful run returns metadata plus the extracted records. The exact field names depend on your worker configuration, but the structure usually looks like this:

{
  "job_id": "job_9f8a7b6c",
  "status": "completed",
  "started_at": "2026-09-09T08:15:00Z",
  "finished_at": "2026-09-09T08:15:12Z",
  "output": {
    "schema": "article_links",
    "record_count": 24,
    "records": [
      {
        "title": "Deploying Scrapers Without Servers",
        "url": "https://example.com/blog/deploying-scrapers",
        "source_url": "https://example.com/blog",
        "fetched_at": "2026-09-09T08:15:05Z"
      }
    ]
  }
}

Use this structure to build checks in your pipeline:

status must be completed before you ingest records.
record_count should match the length of records.
source_url and fetched_at give you traceability for compliance and freshness.

Deployment checklist

Before you move the script from your laptop to a managed worker, confirm these items:

[ ] The target site exposes the data publicly without authentication.
[ ] You have copied the current WORKER_ENDPOINT from the CoreClaw Workers console.
[ ] Your scraper function handles HTTP errors and returns empty results gracefully.
[ ] You store the API key in an environment variable, not in the source code.
[ ] You respect the target site's robots.txt and terms of service.
[ ] You have a retention plan for the collected data.
[ ] You know how to check worker logs when a run fails.

Business use cases

A managed worker approach fits several recurring data needs:

Lead list refresh. Run a worker every morning to collect public business listings, then load the structured JSON into a CRM or outreach tool.
Price monitoring. Schedule product pages to be checked on an interval and compare current prices against a baseline.
Content research. Collect article titles, publish dates, and authors from public blogs or news sites for trend analysis.
AI agent context. Feed public web data into an agent's retrieval pipeline so answers are grounded in current sources.
Compliance monitoring. Track public regulator pages, terms of service, or announcement boards for changes.

Build it yourself vs. CoreClaw Workers

Dimension	Self-hosted scraper	CoreClaw Workers
Best for	Full control, custom infrastructure, large existing ops team	Teams that want to focus on data, not servers
Setup model	Provision servers, proxies, queues, storage	Deploy function or select from product store
Data coverage	Whatever you code	Depends on worker logic or ready-made scraper
Output format	You design and maintain	JSON by default; configurable
Maintenance burden	High: OS, proxies, anti-detection, alerting	Low: platform handles runtime and retries
Integration path	Custom webhooks, queues, databases	HTTP endpoint, scheduling, direct delivery
Quota/freshness	Set by your own infrastructure	Confirm current limits on the pricing page
Pricing verification	Your own cloud and proxy bills	Check official pricing before scaling

The trade-off is control versus operational speed. If your core competency is using the data, not running scraping infrastructure, a managed worker is usually the faster path to production.

Limitations and compliance

Managed workers are not a bypass for access controls. Keep these limits in mind:

Public data only. Do not use a worker to extract private, authenticated, or restricted content.
Terms of service. Follow the target site's terms and robots directives.
Rate limits. Even with managed proxies, aggressive pacing can still trigger blocks or violate terms.
Layout changes. A scraper breaks when the target site changes markup. Plan for periodic maintenance.
Data retention. Store only what you need and delete data when it is no longer useful.
Regional rules. Confirm that your use case complies with local privacy and data-protection laws.

A worker removes infrastructure pain, but it does not remove your responsibility to collect data lawfully.

FAQ

Is there an official API to deploy a worker?

Yes. The CoreClaw Workers console provides the endpoint and credentials you need to submit jobs programmatically. Copy the endpoint from the dashboard instead of guessing a URL.

What data fields does a worker return?

The worker returns whatever your scraping function produces, wrapped in a standard envelope with job_id, status, timestamps, and an output.records array. You control the record schema.

How often should the workflow run?

It depends on freshness needs. Daily is common for price and lead monitoring. Weekly or monthly works for slower-moving sources. Start conservative and increase frequency only if the data changes often enough to justify it.

What happens when a page layout changes?

Your scraper function will return fewer or different fields. Monitor record_count and run a validation step that alerts you when expected fields are missing. That is the signal to update your selectors.

Can this connect to n8n, a CRM, or an AI agent?

Yes. The worker output is JSON, so you can send it to any webhook, no-code tool, database, or agent context store. Keep the API key in the receiving system's secret manager.

What should I verify before production use?

Confirm the target site allows public access, test your scraper on a small URL set, validate the output schema, and review the current pricing and quota details on the CoreClaw pricing page.

Do I need to manage proxies?

No. The platform handles proxy rotation and runtime infrastructure. You only write and maintain the extraction logic.

Summary and next steps

Deploying a custom scraper does not have to mean becoming a DevOps team. With CoreClaw Workers, you write the extraction logic, submit a job to a managed endpoint, and get structured JSON back without managing servers or proxies.

Start with a single public data source, test the scraper locally, then move the same payload to a worker. Add validation, scheduling, and storage only after the first end-to-end run succeeds.

Twitter Scraper GitHub: How to Build a Public Mention and Sentiment Dataset with Python

coreclaw — Tue, 08 Sep 2026 02:19:01 +0000

Twitter Scraper GitHub: How to Build a Public Mention and Sentiment Dataset with Python

The fastest way to track what people are saying about a brand or topic on X/Twitter without committing to a paid API tier is to collect public posts with an open-source scraper repository and add a small Python analysis layer. This article is for growth marketers, brand analysts, social-listening engineers, and AI-agent builders who need a reproducible mention-and-sentiment dataset built from public posts.

The data-scrape/twitter-scraper repository provides a runnable Python reference for public X/Twitter posts, and the same normalization pattern works for data-scrape/x-scraper and data-scrape/instagram-post-scraper, so the sentiment pipeline you build here can be extended across public social platforms.

TL;DR

Use data-scrape/twitter-scraper to collect public X/Twitter post records.
Define a narrow watchlist up front: target handles, hashtags, or keyword phrases. Broad queries produce noisy, hard-to-maintain datasets.
Normalize each raw post into a stable schema with captured_at, source URL, author, text, engagement metrics, and extracted entities.
Score sentiment with a local, rule-based lexicon so you do not need an external NLP API or fine-tuned model to start.
Aggregate results by hour, day, and target entity so trend direction is easy to read.
Refresh daily for brand monitoring; refresh hourly only during active campaigns or incidents, and collect only public, non-authenticated posts.

Why Public Mention and Sentiment Is Harder Than It Looks

Public social feeds look simple until you try to turn them into a reliable dataset. A handful of problems recur:

Official API access is gated and expensive. The X API requires paid tiers for most search use cases, so many teams prefer to own a lightweight mention pipeline.
HTML is JavaScript-heavy. A basic requests.get() call usually returns a hydration shell, not the visible post text, so you need a browser-automation or embedded-data approach.
Posts are noisy. Hashtags, cashtags, retweets, quote tweets, replies, and threads have different structures, and a brand-name query returns irrelevant matches unless you filter.
Sentiment is context-dependent. A simple word list misclassifies sarcasm, slang, and emoji, so the goal is a directional signal, not ground truth.
Rate limits and layout drift. Aggressive polling hits throttling fast, and selectors break when the platform updates its front end.

This article takes the middle path: start with an open-source reference repository, then add a small, auditable Python layer for normalization and sentiment.

What the Verified Repositories Provide

The data-scrape/twitter-scraper repository is a Python reference for collecting public X/Twitter post records. It ships a scraper.py entry point, a requirements.txt, an examples/ folder, and a README. Read the README for the current setup steps and environment variables.

The same pattern appears across the data-scrape profile: data-scrape/x-scraper returns public post and profile records from the same platform, and data-scrape/instagram-post-scraper returns public post records whose captions and comments can be fed into the same normalization layer. The repositories differ in raw field names but share enough shape that one script can be reused across all three platforms.

Pipeline Design

A maintainable pipeline has five stages: capture raw JSON against a bounded watchlist, normalize fields into a stable schema, extract mentions and hashtags, score sentiment with a local lexicon, and aggregate per-post scores into time-window summaries by entity.

Setup and Configuration

Clone the repository and install dependencies inside a virtual environment so packages do not collide with system Python:

git clone https://github.com/data-scrape/twitter-scraper.git
cd twitter-scraper
python -m venv .venv
source .venv/bin/activate  # Windows: .venv\Scripts\activate
pip install -r requirements.txt

Set configuration through environment variables instead of hard-coding queries or pacing limits. The example below forwards a target list, output path, and request budget to the scraper's CLI. Adjust variable names to match the current version of scraper.py; never commit a real query list or proxy configuration to source control.

export TWITTER_QUERIES_FILE="watchlist.txt"
export TWITTER_OUTPUT_FILE="raw_posts.json"
export TWITTER_MAX_REQUESTS=300
python scraper.py \
  --input "$TWITTER_QUERIES_FILE" \
  --output "$TWITTER_OUTPUT_FILE" \
  --max-requests "$TWITTER_MAX_REQUESTS"

Keep watchlist.txt narrow: one brand handle, two competitor handles, and a couple of campaign hashtags is usually enough to start.

Normalization: From Raw JSON to a Stable Schema

The script below reads raw_posts.json, normalizes each record, extracts mentions and hashtags, scores sentiment with a tiny local lexicon, and writes the result to JSONL. Treat the score as a directional signal, not a definitive emotion label.

import json
import pathlib
import os
import re
import sys
from collections import Counter
from datetime import datetime, timezone


RAW_PATH = pathlib.Path(os.environ.get("TWITTER_RAW_PATH", "raw_posts.json"))
OUT_PATH = pathlib.Path(os.environ.get("TWITTER_OUT_PATH", "normalized_posts.jsonl"))
ROLLUP_PATH = pathlib.Path(os.environ.get("TWITTER_ROLLUP_PATH", "sentiment_rollup.json"))
CAPTURED_AT = datetime.now(timezone.utc).isoformat()

# Minimal lexicon for illustration. Expand it with domain-specific words.
POSITIVE_WORDS = {
    "love", "great", "good", "awesome", "excellent", "happy", "best",
    "amazing", "fantastic", "recommend", "thanks", "thank", "helpful",
    "solid", "smooth", "clean", "fast", "easy", "useful",
}
NEGATIVE_WORDS = {
    "hate", "bad", "terrible", "worst", "broken", "slow", "awful",
    "disappointed", "frustrating", "useless", "bug", "crash", "poor",
    "expensive", "fail", "failed", "annoying", "waste",
}


def to_int(value):
    """Coerce engagement fields into integers."""
    if value is None:
        return None
    try:
        return int(str(value).replace(",", "").split(".")[0])
    except ValueError:
        return None


def extract_entities(text: str) -> dict:
    """Extract mentions, hashtags, and cashtags from a post body."""
    return {
        "mentions": re.findall(r"@(\w+)", text),
        "hashtags": re.findall(r"#(\w+)", text),
        "cashtags": re.findall(r"\$(\w+)", text),
    }


def sentiment_scores(text: str) -> dict:
    """Return rule-based sentiment proportions and a compound score."""
    tokens = re.findall(r"\b\w+\b", text.lower())
    pos = sum(1 for t in tokens if t in POSITIVE_WORDS)
    neg = sum(1 for t in tokens if t in NEGATIVE_WORDS)
    total = len(tokens) or 1
    neu = total - pos - neg
    compound = round((pos - neg) / total, 4)
    return {
        "positive": pos,
        "negative": neg,
        "neutral": max(neu, 0),
        "compound": compound,
    }


def normalize(raw: dict) -> dict:
    text = raw.get("text") or raw.get("content") or raw.get("body") or ""
    metrics = raw.get("metrics") or raw.get("engagement") or {}
    entities = extract_entities(text)

    return {
        "platform": "x-twitter",
        "post_id": raw.get("id") or raw.get("tweet_id") or raw.get("status_id"),
        "author": raw.get("author") or raw.get("username") or raw.get("user_handle"),
        "posted_at": raw.get("posted_at") or raw.get("timestamp") or raw.get("created_at"),
        "captured_at": raw.get("captured_at") or CAPTURED_AT,
        "source_url": raw.get("source_url") or raw.get("url"),
        "text": text,
        "metrics": {
            "likes": to_int(metrics.get("likes") or metrics.get("like_count")),
            "replies": to_int(metrics.get("replies") or metrics.get("reply_count")),
            "retweets": to_int(metrics.get("retweets") or metrics.get("retweet_count")),
            "quotes": to_int(metrics.get("quotes") or metrics.get("quote_count")),
        },
        "entities": entities,
        "sentiment": sentiment_scores(text),
        "metadata": {
            "category": "Social Scrapers",
            "query": raw.get("metadata", {}).get("query"),
        },
    }


def rollup(records: list) -> dict:
    """Aggregate sentiment and volume by entity and day."""
    by_entity = {}
    for record in records:
        day = record.get("posted_at", record.get("captured_at", ""))[:10] or "unknown"
        for entity_type in ("mentions", "hashtags"):
            for entity in record["entities"].get(entity_type, []):
                key = (entity_type, entity.lower(), day)
                bucket = by_entity.setdefault(key, {
                    "entity_type": entity_type,
                    "entity": entity.lower(),
                    "day": day,
                    "posts": 0,
                    "compound_sum": 0.0,
                    "likes": 0,
                    "positive": 0,
                    "negative": 0,
                })
                bucket["posts"] += 1
                bucket["compound_sum"] += record["sentiment"]["compound"]
                bucket["likes"] += record["metrics"]["likes"] or 0
                bucket["positive"] += record["sentiment"]["positive"]
                bucket["negative"] += record["sentiment"]["negative"]

    result = []
    for bucket in by_entity.values():
        bucket["avg_compound"] = round(bucket["compound_sum"] / bucket["posts"], 4) if bucket["posts"] else 0
        result.append(bucket)
    return result


def main() -> int:
    if not RAW_PATH.is_file():
        print(f"Raw file not found: {RAW_PATH}", file=sys.stderr)
        return 1

    raw_data = json.loads(RAW_PATH.read_text(encoding="utf-8"))
    records = raw_data if isinstance(raw_data, list) else [raw_data]

    normalized = [normalize(r) for r in records]

    OUT_PATH.parent.mkdir(parents=True, exist_ok=True)
    with OUT_PATH.open("w", encoding="utf-8") as fp:
        for record in normalized:
            fp.write(json.dumps(record, ensure_ascii=False) + "\n")

    summary = rollup(normalized)
    ROLLUP_PATH.write_text(json.dumps(summary, indent=2, ensure_ascii=False), encoding="utf-8")

    print(f"Wrote {len(normalized)} normalized records to {OUT_PATH}")
    print(f"Wrote {len(summary)} entity-day rollups to {ROLLUP_PATH}")
    return 0


if __name__ == "__main__":
    sys.exit(main())

The script is deliberately dependency-light. If you later need higher accuracy, replace the sentiment_scores function with a library such as VADER or a hosted model, but keep the same output shape so downstream aggregation code does not change.

Representative Output

Each line of the JSONL file looks like this:

{
  "platform": "x-twitter",
  "post_id": "1234567890123456789",
  "author": "example_user",
  "posted_at": "2026-09-08T08:30:00+00:00",
  "captured_at": "2026-09-08T10:00:00+00:00",
  "source_url": "https://x.com/example_user/status/1234567890123456789",
  "text": "Just tried the new update — smooth, fast, and exactly what I needed. Great work @examplebrand! #opensource",
  "metrics": {
    "likes": 42,
    "replies": 3,
    "retweets": 8,
    "quotes": 1
  },
  "entities": {
    "mentions": ["examplebrand"],
    "hashtags": ["opensource"],
    "cashtags": []
  },
  "sentiment": {
    "positive": 2,
    "negative": 0,
    "neutral": 16,
    "compound": 0.1111
  },
  "metadata": {
    "category": "Social Scrapers",
    "query": "examplebrand"
  }
}

Treat this as a representative shape, not a guaranteed output. Read the actual raw_posts.json before assuming field names; the X/Twitter schema changes when the platform updates its front end.

Use Cases

A normalized dataset supports several workflows:

Brand health tracking. Plot average compound sentiment per day for your handle and selected competitors.
Campaign measurement. Compare volume and sentiment before, during, and after a product launch.
Incident detection. Watch for sudden spikes in negative mentions and replies with threshold alerts.
AI-agent context. Feed the JSONL into a retriever so an agent can answer questions like "What did people say about our pricing yesterday?" without scraping live pages.
Cross-platform rollups. Reuse the script with data-scrape/x-scraper and data-scrape/instagram-post-scraper to compare X/Twitter and Instagram conversations side by side.

Comparison: Official API, Open-Source Scraper, and Hosted Listening Platform

Three common sourcing options:

Dimension	Official X API	Open-source repo (e.g., `data-scrape/twitter-scraper`)	Hosted social-listening platform
Best for	Teams with budget and compliance requirements that need high-volume, real-time streams	Teams that want a runnable reference and accept responsibility for hosting, scheduling, and schema maintenance	Teams that want dashboards, alerts, and reports without building infrastructure
Setup model	Apply for access, choose a paid tier, authenticate, and call the documented endpoints	Clone the repo, install requirements, configure the CLI, schedule refreshes	Sign up, connect accounts or search terms, and configure dashboards
Output format	JSON via API; check current docs for field names and rate limits	JSON or CSV from the scraper, plus your own normalized JSONL	Dashboard widgets, CSV export, or API; varies by vendor
Maintenance burden	Low once integrated; policy and pricing changes are vendor-managed	Medium: scheduling, retries, schema drift, proxy policy, and layout changes	Low: vendor maintains the integration, but export formats may change
Sentiment analysis	Not provided; you supply your own NLP layer	You supply and control the lexicon or model	Usually included; verify accuracy and customization limits
Pricing verification	Official X developer pricing page	Free open source; you pay for your own hosting and proxies	The vendor's current pricing page; do not trust third-party summaries

Operational Checklist

Before you run a mention pipeline in any non-trivial environment, walk through this checklist:

[ ] Read X/Twitter's current terms of service, developer policy, and any jurisdiction-specific platform guidelines.
[ ] Your watchlist is narrow enough to return useful results without hammering the platform.
[ ] You set request pacing (delays, jitter) to avoid rate limits or CAPTCHA walls.
[ ] You capture captured_at, the source URL, and the raw text block on every record.
[ ] Your sentiment lexicon is documented and versioned, and you understand its known biases.
[ ] You have a dedupe strategy based on post_id or source_url, not on text content.
[ ] You have a backoff and retry policy and you log the failures.
[ ] You have a refresh cadence that matches your decision speed; daily is enough for most brand monitoring.
[ ] You have reviewed applicable privacy and data-protection laws for the records you collect.

Limits, Maintenance, and Compliance

X/Twitter's HTML, JavaScript, and anti-automation measures evolve, and the repository's README reflects a snapshot of that evolution. The repositories do not ship a hosted proxy, a CAPTCHA solver, or an unlimited request budget; you provide those, or you accept a smaller, slower dataset.

Public X/Twitter data still touches trademark, database, and consumer-protection law in many jurisdictions, and the platform's terms restrict certain automated access. Restrict your collection to public, non-authenticated posts, respect robots directives and the platform's terms, do not attempt to bypass access controls, and do not republish records in misleading ways. Sentiment scores from a small lexicon are a directional signal, not a psychological measurement; always label the method when you share a chart or a number.

FAQ

Is there an official X/Twitter API I can use instead?
Yes. The X API offers search and streaming endpoints, but most useful search access requires a paid tier. Verify current tiers, pricing, and terms on the official X developer documentation page. This article covers the open-source, self-managed path for teams that do not have or do not want API access.

What data fields does the repository return?
Field names depend on the current version of scraper.py and the query. Treat the example record as a representative shape, not a guarantee. Read raw_posts.json and the README before integrating.

How often should the workflow run?
Daily refreshes are sufficient for most brand-health monitoring. Hourly refreshes are reasonable during active campaigns or incidents. Avoid sub-hourly refreshes without a documented reason and a sustainable request budget.

What happens when the page layout changes?
Layout changes break CSS selectors and JSON keys. Keep a small set of "smoke test" posts that fail loudly if a core field disappears; treat a missing text or author field as a hard error.

How do I make the sentiment scoring more accurate?
Replace the small lexicon in sentiment_scores with a domain-specific lexicon, VADER, TextBlob, or a fine-tuned model. Keep the same output keys (positive, negative, neutral, compound) so the rollup logic stays unchanged.

Can this connect to a dashboard, CRM, or AI agent?
Yes. The JSONL output can be loaded by pandas, DuckDB, or a vector store. Emit the rollup file to a scheduled job that writes to Google Sheets, a Slack channel, or a CRM activity feed. For AI agents, expose the JSONL records through a retriever so the agent can answer questions about mention trends without browsing live pages.

What should be verified before production use?
Confirm that your watchlist is narrow, your request pacing respects platform limits, your proxy or IP rotation policy is documented, your sentiment method is calibrated against a sample of manually labeled posts, and your data retention policy complies with applicable law.

Next Steps

Clone the data-scrape/twitter-scraper repository, follow the README, and start with the normalization script. Add data-scrape/x-scraper and data-scrape/instagram-post-scraper when you want to compare X/Twitter and Instagram conversations on the same schema. The rest of the data-scrape profile holds adjacent open-source repositories you can adopt as your pipeline grows.

AI Agent Data Governance: How to Keep Web Data Sources Traceable and Auditable

coreclaw — Tue, 08 Sep 2026 01:05:40 +0000

AI Agent Data Governance: How to Keep Web Data Sources Traceable and Auditable

If your AI agent answers questions with public web data, governance is the layer that decides whether the answer is defensible. Governance is not access control and it is not prompt hardening. For an external-data agent, governance means every retrieved record carries the source URL it came from, the timestamp it was fetched, the consent signal of the target site, and the version of the schema the record conforms to — so that a human auditor can replay any answer the agent ever gave. This article walks through a Python framework that adds that governance layer to a retrieval pipeline, with environment variables for the scraper endpoint and credentials so nothing is hard-coded.

TL;DR

AI agent data governance = provenance + freshness + consent + schema version + audit replay for every record the agent retrieves.
A practical governance record is a small JSON envelope (source_url, fetched_at, schema_version, consent_state, license_class, record_hash) wrapped around every normalized record before it hits your vector store.
The pipeline has five stages: classify (decide what the record is), wrap (attach the governance envelope), verify (re-fetch or hash-check on demand), expire (drop records past their freshness window), and replay (regenerate the agent's answer from the stored records).
You can deploy the scraper side as a managed worker on the CoreClaw Workers deploy console, pull ready-made scrapers with built-in provenance from the CoreClaw Workers store, and confirm current rates on the CoreClaw pricing page before scaling.

Why AI Agent Outputs Break Without Governance

Most retrieval-augmented agents look correct in demos and fall apart in production for the same reason: the records they retrieved have no defensible lineage. Three failure modes show up quickly.

Stale answers dressed as current. An agent says "the top-rated coffee shop near downtown Austin is X" based on a Google Maps record fetched six months ago. The shop closed last week. The user has no way to know the answer was old, and the developer cannot prove it was fresh at the time of the call.
Provenance loss. The agent cites a fact, the user asks where it came from, and the system can only point to "the index." There is no URL, no fetch timestamp, no schema version, no operator identity. Compliance teams reject this on first review.
Consent drift. A site that was public in January starts requiring login or returns a 403 to automated clients in June. The agent keeps emitting answers sourced from a record that no longer reflects the site's current terms. Without a consent signal that is refreshed on every retrieval, the agent is in a quiet violation.

The fix is a thin governance layer around the retrieval pipeline. It does not change the model, the vector store, or the prompt. It changes the records.

What "AI Agent Data Governance" Actually Means

Governance for external-data agents is the discipline of attaching five fields to every record before it enters retrieval:

Field	Purpose	Failure if missing
`source_url`	The exact URL the record came from	Cannot defend a citation
`fetched_at`	ISO 8601 UTC timestamp of retrieval	Cannot prove freshness at time-of-answer
`schema_version`	Version of the normalized record schema	Cannot replay through older schema
`consent_state`	Whether the source was reachable and on-terms at fetch	Quiet ToS violation
`record_hash`	Stable hash of normalized payload	Cannot detect downstream tampering

These five fields are the minimum. Mature governance pipelines also attach operator_id (who or what service fetched the record), fetch_method (scraper_api, browser_worker, manual), and license_class (public, registration_required, partner_only). None of these fields is hard to produce; the hard part is making the agent emit them on every retrieval.

Step-by-Step Governance Pipeline

┌────────────┐    ┌──────────┐    ┌─────────┐    ┌──────────┐    ┌──────────┐
│ Classifier │───▶│ Wrapper  │───▶│ Verifier│───▶│ Indexer  │───▶│ Replayer │
│ (decide)   │    │ (attach) │    │ (check) │    │ (store)  │    │ (audit)  │
└────────────┘    └──────────┘    └─────────┘    └──────────┘    └──────────┘

Step 1 — Classify the Record

Before you wrap anything, classify what kind of record it is. A local-business record, a product record, and a SERP record have different freshness windows and different consent assumptions. Classification is a small mapping function, not a model call.

Step 2 — Wrap with the Governance Envelope

After normalization, wrap the record in a small dict that adds the five governance fields. The wrapper is pure code; no network calls.

Step 3 — Verify on Demand

For high-stakes answers, the verifier re-fetches the source URL on a separate worker and confirms the hash still matches. This is the replay guarantee.

Step 4 — Expire Records Past Their Freshness Window

A 90-day-old Google Maps record is usually stale. A 1-day-old SERP record is fresh. The expirer drops records past their freshness class.

Step 5 — Replay

The replayer can regenerate any past agent answer from the stored governance records. This is the audit guarantee.

Runnable Python: A Minimal Governance Wrapper

The wrapper below takes any normalized record dict, attaches the governance envelope, and returns a versioned governance record. All scraper endpoints and credentials are environment variables.

import hashlib
import json
import os
import uuid
from datetime import datetime, timezone
from typing import Any


# --- Configuration pulled from environment, never hard-coded ---
SCRAPER_ENDPOINT = os.environ["SCRAPER_ENDPOINT"]
SCRAPER_API_KEY = os.environ["SCRAPER_API_KEY"]
OPERATOR_ID = os.environ.get("OPERATOR_ID", "agent-pipeline-v1")

# Per-class freshness windows in seconds; tune per source.
FRESHNESS_WINDOWS = {
    "local_business": int(os.environ.get("LOCAL_FRESHNESS_DAYS", "30")) * 86400,
    "product":        int(os.environ.get("PRODUCT_FRESHNESS_DAYS", "1"))  * 86400,
    "serp":           int(os.environ.get("SERP_FRESHNESS_DAYS", "1"))    * 86400,
    "social_post":    int(os.environ.get("SOCIAL_FRESHNESS_DAYS", "7"))  * 86400,
}

SCHEMA_VERSION = "1.0.0"


def classify(record: dict[str, Any]) -> str:
    """Decide which freshness class the record belongs to."""
    if "rating" in record and "address" in record:
        return "local_business"
    if "price" in record and "title" in record:
        return "product"
    if "rank" in record and "query" in record:
        return "serp"
    if "post_text" in record or "comment_text" in record:
        return "social_post"
    return "local_business"  # default fallback


def stable_hash(payload: dict[str, Any]) -> str:
    """Stable hash of the normalized payload, independent of fetch metadata."""
    canonical = json.dumps(payload, sort_keys=True, separators=(",", ":"))
    return "sha256:" + hashlib.sha256(canonical.encode("utf-8")).hexdigest()


def wrap_with_governance(
    raw_record: dict[str, Any],
    source_url: str,
    fetch_method: str = "scraper_api",
    license_class: str = "public",
) -> dict[str, Any]:
    """Attach provenance, freshness, schema, consent, and hash fields."""
    record_class = classify(raw_record)
    fetched_at = datetime.now(timezone.utc).isoformat()

    # consent_state: "ok" by default; the verifier can downgrade it later.
    return {
        "record_id": str(uuid.uuid4()),
        "operator_id": OPERATOR_ID,
        "schema_version": SCHEMA_VERSION,
        "fetched_at": fetched_at,
        "source_url": source_url,
        "fetch_method": fetch_method,
        "license_class": license_class,
        "consent_state": "ok",
        "record_class": record_class,
        "freshness_window_seconds": FRESHNESS_WINDOWS[record_class],
        "record_hash": stable_hash(raw_record),
        "payload": raw_record,
    }


def is_fresh(governed: dict[str, Any], now: datetime | None = None) -> bool:
    """Return True if the record is still inside its freshness window."""
    now = now or datetime.now(timezone.utc)
    fetched = datetime.fromisoformat(governed["fetched_at"])
    age = (now - fetched).total_seconds()
    return age <= governed["freshness_window_seconds"]


def filter_fresh(records: list[dict[str, Any]]) -> list[dict[str, Any]]:
    return [r for r in records if is_fresh(r)]


# --- Example call: a normalized local-business record from a scraper ---
if __name__ == "__main__":
    example_raw = {
        "name": "Example Coffee",
        "address": "500 Congress Ave, Austin, TX",
        "rating": 4.6,
        "reviews_count": 1240,
        "phone": "+1-512-555-0188",
    }
    example_source = "https://www.google.com/maps/search/example+coffee+austin"

    governed = wrap_with_governance(example_raw, example_source)

    print(json.dumps(governed, indent=2))
    print("Fresh right now?", is_fresh(governed))

This wrapper is intentionally small. Its job is to make provenance, freshness, consent, schema, and hashing non-optional before any record enters the agent's retrieval index. Run the example with:

export SCRAPER_ENDPOINT="https://your-scraper.example.com/run"
export SCRAPER_API_KEY="replace-with-real-key"
python governance_wrapper.py

You should see a governed record with all five fields populated and Fresh right now? True.

Governance Record Example

{
  "record_id": "8f4c1f1c-2a17-4f8c-9b87-1f9c1a1b3a4e",
  "operator_id": "agent-pipeline-v1",
  "schema_version": "1.0.0",
  "fetched_at": "2026-09-08T01:04:12+00:00",
  "source_url": "https://www.google.com/maps/search/example+coffee+austin",
  "fetch_method": "scraper_api",
  "license_class": "public",
  "consent_state": "ok",
  "record_class": "local_business",
  "freshness_window_seconds": 2592000,
  "record_hash": "sha256:9b1f02a4...",
  "payload": {
    "name": "Example Coffee",
    "address": "500 Congress Ave, Austin, TX",
    "rating": 4.6,
    "reviews_count": 1240,
    "phone": "+1-512-555-0188"
  }
}

This is the smallest record an AI agent should be allowed to retrieve. Anything missing the envelope is a record the agent cannot defend.

Business Use Cases

Use case	Why governance matters	Typical freshness window
Local-business Q&A agent	Ratings and hours change weekly; outdated answers lose user trust	7-30 days
E-commerce price comparison agent	Prices shift daily; stale answers misinform buying decisions	1-7 days
SERP rank tracking agent	SERPs are volatile; stale ranks mislead SEO decisions	1 day
Influencer / creator analytics agent	Follower counts and engagement shift over weeks	7-30 days
News / trend summarization agent	Stale trends undermine the whole product	1-3 days

For each use case the wrapper class and freshness window are different, but the envelope is identical. That is the design intent.

DIY Governance vs Scraper Platform vs AI-Native Data API

Dimension	DIY governance on top of raw scraping	Managed scraper with provenance (e.g. CoreClaw Workers)	AI-native data API (purpose-built for agents)
Setup effort	High — you write and maintain the wrapper, verifier, expirer	Low — provenance attached per worker run	Low — API returns governed records
Source coverage	Whatever your scraper reaches	100+ ready-made scrapers in the Workers store	Varies by provider; usually narrow
Freshness control	You set windows per class	Platform records `fetched_at` and `source_url` per result	Provider-controlled
Audit replay	Your responsibility	Replay through the same worker	Depends on retention policy
Cost model	Infrastructure + your time	Pay-per-result, proxy included	Subscription or per-call
Best for	Teams with strong compliance engineering	Teams that need fast coverage and audit-ready records	Single-purpose agents
Verification source	Internal docs	CoreClaw pricing	Provider pricing page

If you only need one source and you have engineering capacity, DIY is fine. If you need breadth and audit-ready records out of the box, a managed scraper with provenance is usually the better trade.

Limits, Compliance, and Freshness

Public data only. This pipeline is for public web data and public business information. It is not a substitute for consent, contracts, or platform licensing.
Respect target-site terms. consent_state should be refreshed on every retrieval. If a site returns 403 to automated clients, the verifier should downgrade the record and the indexer should drop it.
Regional and freshness behavior change. Search engines and directories update layouts; freshness windows that worked last quarter may be too loose or too tight next quarter. Audit the windows quarterly.
Schema drift is real. A scraper that worked yesterday may return a different field set today. The schema_version field is how you replay answers through the schema they were originally produced under; do not skip it.
Privacy and lawful basis. Public does not mean unrestricted. Personal data, even if public, has different rules in different jurisdictions. Build your retrieval around this set, and add a privacy review before any user-facing deployment.

FAQ

1. Does governance slow down the agent?
No. The wrapper is a single dictionary construction per record and a hash on a small payload. It adds microseconds, not milliseconds, and is invisible to the user.

2. What if my scraper does not return a stable URL?
That is a scraper-quality problem, not a governance problem. If a source has no stable URL, it has no provenance, and you should not include it in a governed pipeline. Pick a different data source.

3. How often should the verifier re-check a record?
For local-business and product records, a 24-hour re-check is a reasonable default. For SERP records, re-check on every agent query that touches the record. Tune based on how badly staleness would damage the answer.

4. Can I add governance to a pipeline that already exists?
Yes. Wrap your existing normalized records at the boundary between your normalizer and your indexer. The wrapper is a 20-line function and does not require reworking the rest of the pipeline.

5. What is the minimum viable schema_version?
Any semver string. The point is to be able to identify which schema was active at retrieval time, not to follow semver religiously. 1.0.0 is fine.

6. Do I need a vector database for this?
No. Governance is orthogonal to retrieval. The envelope attaches to the record regardless of whether the index is a vector store, SQLite FTS, or a JSON file.

7. What is the difference between governance and access control?
Governance is what the agent retrieved and from where. Access control is who can talk to the agent. They are different layers and both are needed; this article is about the first.

Summary

AI agent data governance is a thin, opinionated envelope around every record the agent retrieves: provenance, freshness, consent, schema version, and hash. It is small enough to fit in one Python module, fast enough to add microseconds to a query, and necessary enough that any production agent handling external data should treat it as a hard requirement.

If you want governance attached automatically per result rather than built from scratch, deploy a worker on the CoreClaw Workers deploy console, pick a ready-made scraper from the CoreClaw Workers store, and review current rates on CoreClaw pricing.

Web Data for AI Agents: How to Build a RAG Pipeline with Structured Web Data

coreclaw — Thu, 03 Sep 2026 01:05:42 +0000

Web Data for AI Agents: How to Build a RAG Pipeline with Structured Web Data

AI agents that answer questions about the real world need one thing above all: structured, citable external data they can retrieve on demand. The most reliable way to deliver that data is a Retrieval-Augmented Generation (RAG) pipeline that fetches public web data, normalizes it into consistent JSON records, stores them in a searchable index, and injects only the relevant slices into the LLM context window at query time. This article shows you how to build that pipeline in Python, using environment variables for any scraper endpoint so nothing is hard-coded.

TL;DR

A production-grade web-data RAG pipeline has four stages: collect (fetch public data through a scraper API or worker), normalize (map heterogeneous HTML or JSON into a flat schema), index (store records so they are searchable by keyword or semantic similarity), and retrieve (select the top-k records relevant to a user query and pass them to the LLM as context). You can deploy the collection stage as a managed worker on the CoreClaw Workers platform, pick from ready-to-use scrapers in the CoreClaw store, and only pay for results returned — check CoreClaw pricing for current rates.

Why This Is Hard

LLMs are trained on static snapshots. When an agent is asked "What are the top-rated coffee shops near downtown Austin?" or "What is the current price of this Amazon product?", the model either guesses from stale training data or hallucinates. Neither is acceptable in production.

Three challenges make web-data RAG harder than textbook RAG:

Heterogeneous formats. Every website returns HTML with different structure. Even API responses vary in field names, nesting depth, and data types.
Freshness drift. Pages change layouts, products go out of stock, businesses close. A pipeline that worked yesterday may return empty fields today.
Context budget. An LLM context window is finite. Dumping 10,000 raw records into the prompt is both expensive and counterproductive — the model needs the right five records, not every record.

The solution is a normalization layer that converts messy web data into a predictable JSON schema before it ever touches the retrieval index.

What a Web-Data RAG Pipeline Looks Like

┌─────────┐    ┌────────────┐    ┌─────────┐    ┌──────────┐    ┌─────────┐
│ Scraper │───▶│ Normalizer │───▶│ Indexer │───▶│ Retriever │───▶│   LLM   │
│ (collect)│    │ (normalize)│    │ (store) │    │ (top-k)   │    │ (answer)│
└─────────┘    └────────────┘    └─────────┘    └──────────┘    └─────────┘
     ▲                                                        │
     │                  User query ────────────────────────────┘

Scraper: Fetches public web data. Can be a managed API, a deployed worker, or a local script. Must respect rate limits and target-site terms.
Normalizer: Transforms raw HTML/JSON into a flat, typed schema (e.g., {name, address, rating, url, source_domain, fetched_at}).
Indexer: Stores normalized records. For small datasets, a JSON file with keyword search works. For larger ones, use SQLite or a vector database.
Retriever: Given a user query, selects the top-k most relevant records.
LLM: Receives the user query plus the retrieved records as context, and generates a grounded answer with citations.

Step-by-Step Implementation

Step 1 — Define the Normalized Schema

Start by deciding what your pipeline will produce. A flat schema is easier to index and debug than deeply nested JSON.

from dataclasses import dataclass, asdict
from datetime import datetime
from typing import Optional

@dataclass
class WebDataRecord:
    """Flat, typed record for RAG indexing."""
    title: str
    content: str            # description, review text, or summary
    url: str                # source URL for citation
    source_domain: str
    category: str           # e.g., "local_business", "product", "social_post"
    rating: Optional[float] = None
    price: Optional[str] = None      # keep as string — formats vary
    location: Optional[str] = None
    fetched_at: str = datetime.utcnow().isoformat()

    def to_dict(self) -> dict:
        return asdict(self)

Step 2 — Collect Web Data with Environment-Configured Endpoint

Use environment variables for every endpoint, API key, and configuration value. Never hard-code a URL you have not verified.

import os
import requests

SCRAPER_ENDPOINT = os.environ.get("SCRAPER_ENDPOINT", "")
SCRAPER_API_KEY = os.environ.get("SCRAPER_API_KEY", "")
TARGET_URL = os.environ.get("TARGET_URL", "https://example.com")

def fetch_raw_data(query: str, limit: int = 20) -> list[dict]:
    """Call a scraper endpoint and return raw JSON records.

    Set SCRAPER_ENDPOINT to the current API URL from your provider's
    console or documentation. Do not guess endpoints.
    """
    if not SCRAPER_ENDPOINT:
        raise ValueError(
            "Set SCRAPER_ENDPOINT env var to your scraper API URL. "
            "Copy the current endpoint from your provider's console."
        )

    headers = {"Authorization": f"Bearer {SCRAPER_API_KEY}"}
    params = {"query": query, "limit": limit}

    resp = requests.get(
        SCRAPER_ENDPOINT,
        headers=headers,
        params=params,
        timeout=30,
    )
    resp.raise_for_status()
    return resp.json().get("results", [])

Step 3 — Normalize Raw Records into the Flat Schema

This is the layer that makes your pipeline robust. Every data source — whether it returns Google Maps results, Amazon products, or social media posts — gets mapped into the same WebDataRecord shape.

from urllib.parse import urlparse

def normalize_record(raw: dict, category: str = "general") -> WebDataRecord:
    """Map a raw scraper response into a flat WebDataRecord.

    Adapt field names to match what your scraper actually returns.
    The goal is consistency: every record has the same keys.
    """
    source_url = raw.get("url", raw.get("link", ""))
    domain = urlparse(source_url).netloc if source_url else "unknown"

    return WebDataRecord(
        title=raw.get("title", raw.get("name", "Untitled")),
        content=raw.get("description", raw.get("snippet", raw.get("text", ""))),
        url=source_url,
        source_domain=domain,
        category=category,
        rating=raw.get("rating"),
        price=str(raw.get("price", "")) if raw.get("price") else None,
        location=raw.get("address", raw.get("location")),
    )

def normalize_batch(raw_records: list[dict], category: str = "general") -> list[WebDataRecord]:
    return [normalize_record(r, category) for r in raw_records]

Step 4 — Index Records for Retrieval

For a lightweight pipeline, use SQLite with full-text search. This avoids the overhead of a vector database while still supporting keyword queries.

import sqlite3
import json

def create_index(db_path: str = "web_data.db") -> sqlite3.Connection:
    conn = sqlite3.connect(db_path)
    conn.execute("""
        CREATE TABLE IF NOT EXISTS records (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            title TEXT,
            content TEXT,
            url TEXT,
            source_domain TEXT,
            category TEXT,
            rating REAL,
            price TEXT,
            location TEXT,
            fetched_at TEXT
        )
    """)
    conn.execute("""
        CREATE VIRTUAL TABLE IF NOT EXISTS records_fts
        USING fts5(title, content, content='records')
    """)
    conn.commit()
    return conn

def index_records(conn: sqlite3.Connection, records: list[WebDataRecord]) -> None:
    for r in records:
        d = r.to_dict()
        conn.execute(
            """INSERT INTO records
               (title, content, url, source_domain, category, rating, price, location, fetched_at)
               VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)""",
            (d["title"], d["content"], d["url"], d["source_domain"],
             d["category"], d["rating"], d["price"], d["location"], d["fetched_at"]),
        )
        conn.execute(
            "INSERT INTO records_fts (title, content) VALUES (?, ?)",
            (d["title"], d["content"]),
        )
    conn.commit()

Step 5 — Retrieve Top-K Records for the LLM Context

def retrieve_records(
    conn: sqlite3.Connection,
    query: str,
    top_k: int = 5,
) -> list[dict]:
    """Full-text search retrieval. Returns the top-k matching records."""
    cursor = conn.execute(
        """SELECT r.title, r.content, r.url, r.source_domain,
                  r.category, r.rating, r.price, r.location
           FROM records_fts f
           JOIN records r ON r.id = f.rowid
           WHERE records_fts MATCH ?
           ORDER BY rank
           LIMIT ?""",
        (query, top_k),
    )
    columns = [desc[0] for desc in cursor.description]
    return [dict(zip(columns, row)) for row in cursor.fetchall()]

Step 6 — Build the RAG Context Block

def build_context(records: list[dict]) -> str:
    """Format retrieved records as a context block for the LLM."""
    if not records:
        return "No relevant records found."

    blocks = []
    for i, r in enumerate(records, 1):
        block = f"[{i}] {r['title']}\n"
        if r.get("content"):
            block += f"    {r['content']}\n"
        if r.get("rating"):
            block += f"    Rating: {r['rating']}\n"
        if r.get("price"):
            block += f"    Price: {r['price']}\n"
        if r.get("location"):
            block += f"    Location: {r['location']}\n"
        block += f"    Source: {r['url']}\n"
        blocks.append(block)

    return "\n".join(blocks)

Putting It All Together

def rag_pipeline(query: str, scrape_query: str, category: str = "general") -> str:
    # 1. Collect
    raw = fetch_raw_data(scrape_query)

    # 2. Normalize
    records = normalize_batch(raw, category)

    # 3. Index
    conn = create_index()
    index_records(conn, records)

    # 4. Retrieve
    top_records = retrieve_records(conn, query, top_k=5)

    # 5. Build context
    context = build_context(top_records)

    conn.close()
    return context

# Example: run with environment variables set
#   export SCRAPER_ENDPOINT="https://your-verified-endpoint"
#   export SCRAPER_API_KEY="your-key"
#   export TARGET_URL="https://example.com"
#
#   context = rag_pipeline(
#       query="best coffee shops in Austin",
#       scrape_query="coffee shops downtown Austin",
#       category="local_business",
#   )
#   print(context)

Representative Output

When the pipeline runs successfully, the context block looks like:

[1] Blue Bottle Coffee
    Specialty coffee roaster with pour-over and espresso bar.
    Rating: 4.6
    Location: 2501 E 6th St, Austin, TX
    Source: https://maps.example.com/blue-bottle

[2] Houndstooth Coffee
    Espresso and pour-over focus with knowledgeable baristas.
    Rating: 4.5
    Location: 401 Congress Ave, Austin, TX
    Source: https://maps.example.com/houndstooth

[3] Flat Track Coffee
    Neighborhood espresso bar with pastries and bagels.
    Rating: 4.7
    Location: 1119 E 11th St, Austin, TX
    Source: https://maps.example.com/flat-track

The LLM receives this block alongside the user's question and can answer with citations: "Based on the top-rated results, Flat Track Coffee has the highest rating at 4.7, followed by Blue Bottle Coffee at 4.6."

Business Use Cases

Use Case	Data Source	RAG Value
Customer support agent	Product pages, FAQ pages	Agent answers with current specs, not stale training data
Sales intelligence bot	Google Maps, LinkedIn	Agent surfaces local business leads with addresses and ratings
Price comparison assistant	Amazon, Walmart listings	Agent compares current prices across sellers
Content research tool	YouTube, TikTok metadata	Agent summarizes trending topics with creator attribution
SEO monitoring agent	Google SERP results	Agent tracks ranking changes and reports competitor moves

Build vs Buy: Collection Layer

Dimension	Self-Hosted Scraper	Managed Worker (CoreClaw)
Setup model	Provision servers, write scraper code, manage proxies	Deploy a pre-built worker from the store
Data coverage	Only what you build scrapers for	Choose from 100+ ready-made workers
Maintenance burden	Fix broken scrapers when layouts change	Provider handles maintenance
Proxy management	You rotate, validate, and pay for proxies	Included in the platform
Output format	Whatever your code produces	Standardized JSON
Pricing model	Infrastructure + development time	Pay per result — verify current rates on the pricing page

The normalization, indexing, and retrieval layers are worth building yourself — they are tightly coupled to your application logic. The collection layer is where a managed platform saves the most time, because it eliminates proxy management, layout-fix cycles, and infrastructure scaling. You can deploy a new worker or browse the workers store to see what is available.

Limitations and Compliance

Freshness. Web data changes continuously. Schedule re-collection at intervals appropriate to your use case — hourly for prices, daily for reviews, weekly for business directories.
Coverage. No scraper covers every site. Verify that your target domains are supported before relying on the pipeline in production.
Context window limits. Even with retrieval, large records can exceed the LLM's context budget. Trim content fields and cap top_k at 5-10.
Citation accuracy. Always pass the source URL in the context block so the LLM can cite it. Never let the model answer without a source.
Terms and law. Respect each target site's terms of service, robots directives, and applicable data-protection law. Only collect public web data. Do not access private content, bypass authentication, or evade CAPTCHAs.
Schema drift. When a scraper's output fields change, the normalizer must be updated. Build a validator that checks every record against the expected schema and logs mismatches.

FAQ

1. Do I need a vector database for web-data RAG?
No. For keyword-driven queries (product names, business categories, addresses), SQLite full-text search is sufficient and far simpler. Use vector embeddings when queries are semantic ("find me cozy coffee shops" where "cozy" is not a literal keyword match).

2. How often should I re-collect web data for my agent?
It depends on volatility. Price data may need hourly updates. Business directory data changes weekly or monthly. Build a scheduler (cron, Celery, or a managed worker) that runs collection at the interval your use case requires.

3. Can I connect this pipeline to an MCP client like Claude or Cursor?
Yes. The retriever function can be exposed as an MCP tool. An AI agent calls the tool with a query, receives the top-k records as structured context, and generates a grounded answer. See Day 2 of this series for an MCP server setup walkthrough.

4. What happens when a scraper's output fields change?
The normalizer maps field names dynamically (e.g., it checks both title and name), but if a field is removed entirely, the record will have empty values. Build a schema validator that logs records with missing critical fields so you can fix the normalizer.

5. How do I prevent the LLM from hallucinating beyond the retrieved records?
Include an instruction in the system prompt: "Answer only using the provided context records. If the context does not contain the answer, say you do not have current data." Always pass source URLs so the model can cite them.

6. Can I use this for multiple data sources at once?
Yes. Run separate collection jobs for each source (Google Maps, Amazon, YouTube), normalize them all into WebDataRecord with different category values, and index them in the same SQLite database. The retriever will surface mixed results ranked by relevance.

7. What is the simplest way to start without managing scraper infrastructure?
Browse the CoreClaw workers store for a pre-built scraper that matches your data source, deploy it from the console, and pipe its JSON output directly into the normalizer. This lets you focus on the retrieval and LLM layers while the collection layer runs as a managed worker.

Summary

A web-data RAG pipeline turns static LLMs into agents that can answer questions about the real world with citations. The pipeline has four stages — collect, normalize, index, retrieve — and the collection stage is where a managed platform like CoreClaw saves the most engineering time. Start with SQLite FTS for the index, build a flat schema for the normalizer, and always pass source URLs in the context block so the LLM can cite them.

Instagram Post Scraper GitHub: How to Calculate Public Engagement Rate with Python

coreclaw — Tue, 01 Sep 2026 02:12:41 +0000

Instagram Post Scraper GitHub: How to Calculate Public Engagement Rate with Python

The fastest way to turn public Instagram post records into a defensible engagement-rate dataset is to combine an open-source scraper repository with a small Python normalization layer you can read and audit. This article is for growth analysts, social-media engineers, marketing operations teams, and AI-agent builders who need to score posts by likes, comments, and reach without paying for a hosted analytics subscription.

The data-scrape/instagram-post-scraper repository is a runnable Python reference that returns public post records in JSON, and the same schema pattern reappears in data-scrape/twitter-scraper and data-scrape/tiktok-video-scraper, so the engagement pipeline you build here is portable across public social platforms.

TL;DR

Use data-scrape/instagram-post-scraper to collect public post records.
Pick the engagement formula up front and document it in code: (likes + comments) / followers or (likes + comments) / reach.
Normalize each record into a stable schema with captured_at, the source URL, the raw metric block, and a derived block for the rate.
Write normalized rows as JSONL so pandas, DuckDB, or an AI-agent retriever can stream them.
Schedule nightly refreshes for account-level benchmarking; refresh hourly only for narrow campaign windows.
Re-read Instagram's current terms and applicable privacy law before scaling up; do not collect private or gated content.

Why Public Engagement Rate Is Harder Than It Looks

Most public Instagram post pages render key metrics through JavaScript and place them in different DOM regions for photos, carousels, reels, and sponsored posts. A simple requests.get() often returns a hydration shell rather than the likes and comments a human sees. The metric definitions are also inconsistent:

Three competing denominators. Engagement-rate formulas differ by industry: per-post likes+comments divided by followers, per-post likes+comments divided by reach, or per-impression engagement. Pick one and document it.
View vs reach vs plays. Reels expose a play_count; photo posts have neither. Treat them as different content classes.
Locale and number formatting. Numbers appear as 1,240, 1.2K, or 120 万; parsing requires normalization before arithmetic.
Rate-limit and bot detection. Aggressive polling triggers throttling or empty responses; field names drift when the platform rolls out a new layout.

Most teams either pay for a hosted analytics subscription, or accept the maintenance burden of an open-source scraper repository. This article focuses on the second path because the data flow stays transparent and the cost stays predictable.

What the Verified Repositories Provide

The data-scrape/instagram-post-scraper repository is a Python reference for collecting public Instagram post records. It ships a scraper.py entry point, a requirements.txt, an examples/ folder, and a README. Read the README for the current setup steps and any environment variables the current version exposes.

The same schema pattern is used across the data-scrape profile: data-scrape/twitter-scraper returns public post and reply records you can score with the same (likes + comments) / followers formula, and data-scrape/tiktok-video-scraper returns public video records with play_count, like_count, comment_count, and share_count you can feed into (likes + comments + shares) / plays. The three repositories differ in raw field names but share enough of an input/output shape that one normalization layer can be reused across all three verticals.

Pipeline Design

A maintainable engagement-rate pipeline has four stages:

Ingest. Run the scraper against a bounded set of public post URLs or profile URLs; capture raw JSON to disk and tag every record with captured_at and the source URL.
Resolve authors. Map each post to the author handle and fetch the author's follower count via a sidecar lookup; cache the result.
Normalize. Map raw fields into a stable schema with metrics (likes, comments, view count) and a derived block holding the rate.
Aggregate. Roll per-post rates into per-author and per-time-window summaries; persist both granularities.

Setup and Configuration

Clone the repository and install dependencies inside a virtual environment so packages do not collide with system Python:

git clone https://github.com/data-scrape/instagram-post-scraper.git
cd instagram-post-scraper
python -m venv .venv
source .venv/bin/activate  # Windows: .venv\Scripts\activate
pip install -r requirements.txt

Set configuration through environment variables instead of hard-coding queries or pacing limits. The example below forwards a target list, output path, and request budget to the scraper's CLI. Adjust the variable names to match the current version of scraper.py; never commit a real post list, a proxy list, or a session cookie to source control.

export IG_POST_URLS_FILE="posts.txt"
export IG_OUTPUT_FILE="raw_posts.json"
export IG_MAX_REQUESTS=200
python scraper.py \
  --input "$IG_POST_URLS_FILE" \
  --output "$IG_OUTPUT_FILE" \
  --max-requests "$IG_MAX_REQUESTS"

Normalization: From Raw JSON to a Stable Schema

Define the engagement formula up front and store it in code, not in a spreadsheet. The script below reads raw_posts.json, maps each record to a stable dict, computes two engagement-rate variants (per-post against follower count and per-post against view count for reels), and writes the result to a JSONL file.

import json
import pathlib
import os
import sys
from datetime import datetime, timezone


RAW_PATH = pathlib.Path(os.environ.get("IG_RAW_PATH", "raw_posts.json"))
OUT_PATH = pathlib.Path(os.environ.get("IG_OUT_PATH", "normalized_posts.jsonl"))
AUTHOR_FOLLOWERS_PATH = pathlib.Path(
    os.environ.get("IG_AUTHOR_FOLLOWERS_PATH", "author_followers.json")
)
CAPTURED_AT = datetime.now(timezone.utc).isoformat()


def to_int(value):
    """Coerce raw like/comment/view fields into integers.

    Accepts strings like '1,240', '1.2K', '12 万', or None.
    Returns None when the value cannot be parsed.
    """
    if value is None:
        return None
    if isinstance(value, (int, float)):
        return int(value)
    text = str(value).strip().replace(",", "").replace(" ", "")
    if text.endswith("K") or text.endswith("k"):
        try:
            return int(float(text[:-1]) * 1_000)
        except ValueError:
            return None
    if text.endswith("M") or text.endswith("m"):
        try:
            return int(float(text[:-1]) * 1_000_000)
        except ValueError:
            return None
    try:
        return int(float(text))
    except ValueError:
        return None


def load_followers_cache(path: pathlib.Path) -> dict:
    """Load {author_handle: follower_count} from a sidecar JSON file."""
    if not path.is_file():
        return {}
    data = json.loads(path.read_text(encoding="utf-8"))
    if isinstance(data, list):
        return {row.get("author"): to_int(row.get("followers")) for row in data}
    if isinstance(data, dict):
        return {k: to_int(v) for k, v in data.items()}
    return {}


def safe_div(numerator, denominator):
    if numerator is None or denominator in (None, 0):
        return None
    return round((numerator / denominator) * 100.0, 4)


def normalize(raw: dict, followers_cache: dict) -> dict:
    metrics = raw.get("metrics") or raw.get("engagement") or {}
    likes = to_int(metrics.get("likes") or metrics.get("like_count"))
    comments = to_int(metrics.get("comments") or metrics.get("comment_count"))
    views = to_int(metrics.get("views") or metrics.get("view_count") or metrics.get("play_count"))
    author = raw.get("author") or raw.get("owner_username") or "unknown"
    follower_count = followers_cache.get(author)
    interactions = sum(v for v in (likes, comments) if v is not None)

    return {
        "platform": "Instagram",
        "post_id": raw.get("id") or raw.get("shortcode"),
        "author": author,
        "captured_at": raw.get("captured_at") or CAPTURED_AT,
        "source_url": raw.get("source_url") or raw.get("url"),
        "post_type": raw.get("type") or raw.get("media_type"),
        "metrics": {
            "likes": likes,
            "comments": comments,
            "views": views,
        },
        "derived": {
            "interactions": interactions,
            "followers": follower_count,
            "engagement_rate_pct_vs_followers": safe_div(interactions, follower_count),
            "engagement_rate_pct_vs_views": safe_div(interactions, views),
        },
        "metadata": {
            "category": "Social Scrapers",
            "query": raw.get("metadata", {}).get("query"),
            "raw_metrics": metrics,
        },
    }


def main() -> int:
    if not RAW_PATH.is_file():
        print(f"Raw file not found: {RAW_PATH}", file=sys.stderr)
        return 1

    raw_data = json.loads(RAW_PATH.read_text(encoding="utf-8"))
    records = raw_data if isinstance(raw_data, list) else [raw_data]
    followers_cache = load_followers_cache(AUTHOR_FOLLOWERS_PATH)

    OUT_PATH.parent.mkdir(parents=True, exist_ok=True)
    written = 0
    with OUT_PATH.open("w", encoding="utf-8") as fp:
        for raw in records:
            fp.write(json.dumps(normalize(raw, followers_cache), ensure_ascii=False) + "\n")
            written += 1

    print(f"Wrote {written} normalized records to {OUT_PATH}")
    return 0


if __name__ == "__main__":
    sys.exit(main())

The to_int helper is deliberately defensive so a single locale change does not break the whole pipeline. The safe_div helper guards against divide-by-zero when an author lookup fails or when a reel post has no public view count. Treat the engagement-rate values as calculated outputs, not as ground-truth platform metrics; the formula you choose should match the question you are trying to answer.

Representative Output

Each line of the JSONL file looks like this:

{
  "platform": "Instagram",
  "post_id": "CExamplePost01",
  "author": "example_handle",
  "captured_at": "2026-09-01T10:00:00+00:00",
  "source_url": "https://www.instagram.com/p/CExamplePost01/",
  "post_type": "reel",
  "metrics": {"likes": 1240, "comments": 86, "views": 18450},
  "derived": {
    "interactions": 1326,
    "followers": 28400,
    "engagement_rate_pct_vs_followers": 4.669,
    "engagement_rate_pct_vs_views": 7.1879
  },
  "metadata": {
    "category": "Social Scrapers",
    "query": "example_handle",
    "raw_metrics": {"like_count": "1,240", "comment_count": 86, "play_count": 18450}
  }
}

Treat this as a representative shape, not a guaranteed output. Always read the actual raw_posts.json before assuming a field name; the Instagram schema changes when the platform rolls out new layouts.

Use Cases

A maintainable engagement-rate pipeline supports several research and operations use cases:

Account benchmarking. Roll per-post rates up to per-author summaries; compare author A vs author B on the same time window.
Content-format analysis. Compare photo, carousel, and reel posts separately; reels expose a different denominator (views) than photo posts.
Campaign post-mortems. Snapshot engagement around a campaign window and compare it to a control window.
Influencer shortlisting. Score candidates on the same formula, then filter by a minimum interaction count rather than raw follower count.
AI-agent retrieval. Wrap the JSONL records in a small retriever so an agent can answer questions like "Which watchlist authors had engagement above 5% last week?" without scraping the web itself.
Cross-platform rollups. Reuse the JSONL loader with data-scrape/twitter-scraper and data-scrape/tiktok-video-scraper on the same formula structure.

Comparison: Build, Open-Source Scrapers, and Hosted Analytics

Three common ways to source public engagement data, each with different trade-offs.

Dimension	DIY with Playwright/Scrapy	Open-source repo (e.g., `data-scrape/instagram-post-scraper`)	Hosted analytics subscription
Best for	Teams with strong scraping engineering capacity and time to invest in proxy, anti-bot, and schema work	Teams that want a runnable reference and accept responsibility for hosting, scheduling, and schema maintenance	Teams that want a hosted dashboard and accept vendor lock-in
Setup model	Build and host everything yourself	Clone the repo, install requirements, configure the CLI, schedule refreshes	Sign up, integrate the widget or export, configure dashboards
Output format	Whatever you build (JSONL, CSV, database)	JSON or CSV, plus your own normalization	Dashboard widgets, sometimes CSV export; check vendor docs
Maintenance burden	High: proxies, retries, schema changes, anti-bot work	Medium: scheduling, retries, schema drift, proxy policy	Low: vendor maintains the integration
Engagement formula	Yours to choose and document	Yours to choose and document	Vendor's formula; usually fixed, sometimes configurable
Pricing verification	N/A	Free (open source)	The vendor's current pricing page; do not trust third-party summaries

Operational Checklist

Before you run an engagement pipeline in any non-trivial environment, walk through this checklist:

[ ] Read Instagram's current terms of service and any jurisdiction-specific platform guidelines.
[ ] Your post list is narrow enough to return useful results without hammering the platform.
[ ] You set request pacing (delays, jitter) to avoid rate limits or CAPTCHA walls.
[ ] You capture captured_at, the source URL, and the raw metric block on every record.
[ ] Your engagement formula is documented in code with the denominator spelled out.
[ ] You have a dedupe strategy based on post_id or source_url, not caption text.
[ ] You have a backoff and retry policy and you log the failures.
[ ] You have a refresh cadence for the author follower cache so per-author rates stay accurate.
[ ] You have reviewed applicable privacy and data-protection laws for the records you collect.

Limits, Maintenance, and Compliance

Instagram's HTML and JavaScript evolve, and the repository's README reflects a snapshot of that evolution. The repositories do not ship a hosted proxy, a CAPTCHA solver, or an unlimited request budget; you provide those, or you accept a smaller, slower dataset.

Public Instagram data still touches trademark, database, and consumer-protection law in many jurisdictions, and Instagram's terms restrict certain automated access. Restrict your collection to public, non-authenticated posts, respect the platform's terms, do not attempt to bypass access controls, and do not republish records in misleading ways. Engagement-rate formulas are an analyst convention, not a platform metric, so always label the formula you used when you share a chart or a number.

FAQ

Is there an official Instagram API I can use instead?
The Instagram Graph API requires a Business or Creator account and an app review for most use cases. The path in this article is the open-source, self-managed path for teams without API access. Verify current Instagram Graph API availability on the official Meta developer documentation page.

How often should the workflow run?
Daily is enough for most account-level benchmarking. Hourly refreshes during a narrow campaign window are reasonable; avoid sub-hourly refreshes without a documented reason and a sustainable request budget.

What happens when Instagram changes the page layout?
Layout changes break CSS selectors and JSON keys. Keep a small set of "smoke test" posts that fail loudly if a core field disappears; treat a missing metric as a hard error.

How do I extend this to Twitter and TikTok?
Clone data-scrape/twitter-scraper and data-scrape/tiktok-video-scraper next to the Instagram project. Map each platform's metric names into the same metrics block and reuse derived to score them on the same formula.

Next Steps

Clone the data-scrape/instagram-post-scraper repository, follow the README, and start with the normalization script. Layer data-scrape/twitter-scraper and data-scrape/tiktok-video-scraper into the same JSONL loader when you want Instagram, Twitter, and TikTok scored on one formula. The rest of the data-scrape profile holds adjacent open-source repositories you can adopt as your pipeline grows.

Web Scraping Job Scheduling: How to Orchestrate Recurring Scrapes with Python

coreclaw — Tue, 01 Sep 2026 01:07:10 +0000

Web Scraping Job Scheduling: How to Orchestrate Recurring Scrapes with Python

The shortest reliable way to schedule recurring web scraping jobs in Python is to put the orchestration in a single small entrypoint script that reads its target list and configuration from environment variables, persists state between runs, and uses exponential backoff with bounded retries for transient failures. That pattern gives you predictable schedules, recoverable runs, and a single seam where you can later swap a self-hosted scheduler for a managed workers platform without rewriting your scrapers.

This guide is for backend developers, data engineers, and small automation teams who are tired of cron lines they cannot see, ad-hoc loops in notebooks, and silent failures that nobody notices until a downstream dashboard goes blank.

TL;DR

Treat each scrape as a discrete job with a stable name, a target, an output destination, and a schedule, rather than one long-lived crawler script.
Keep the scheduler separate from the scraping logic so you can move from cron → Python scheduler → managed platform without rewriting extraction code.
Use environment variables for everything that varies between environments: endpoints, API keys, rate limits, target URLs, output paths.
Persist run state (last cursor, last successful timestamp, attempts) so retries and restarts do not duplicate work or skip pages.
Apply bounded retries with exponential backoff for network errors; treat permanent failures (4xx, schema change, missing auth) as terminal and log them loudly.
When the orchestration footprint becomes bigger than the scraping logic itself, move to a managed platform that already ships retries, scheduling, and a pay-per-result model — for example the CoreClaw Workers Store, where ready-made scrapers can be deployed in minutes without you holding any infrastructure. Confirm the current unit economics on the published CoreClaw pricing page before committing.

Why "Just Add a Cron Job" Stops Working

The first scrape you run on a schedule is easy: a cron line, a requests.get(), a CSV append. The fifth scrape is fine. The twentieth scrape is where the cracks appear:

You cannot see what ran. Cron's only logging is mail, and most servers stopped mailing a long time ago. A job silently fails at 03:00 and you only notice when the morning report is empty.
Retries are ad hoc. Every scraper reinvents its own try / except, often without backoff, often without idempotency, and almost always without upper bounds.
Schedules collide. Three jobs that each loop through their own paginated target accidentally all hit the same upstream at the same minute, and the rate-limiting reply cascades into your other jobs.
State lives nowhere. A job crashes after updating some records; the next run starts from page 1 and you either duplicate rows or write a fragile offset marker into a local file.
Infrastructure is invisible. Who watches the scheduler itself? Who applies security patches to the cron host? Who kills the zombie processes when one job hangs forever?

A real orchestration layer fixes all of these — but it does not have to be a full Airflow cluster on day one. The minimum viable version lives in one Python entrypoint, one schedule declaration, one place to persist state.

The Pattern: One Script Per Job, One Scheduler For All Jobs

The architectural shift is the smallest one that solves the most pain: separate the scheduler from the scraper.

┌─────────────────────────┐    triggers     ┌──────────────────────────┐
│  Scheduler / cron /     │ ───────────────▶│  scrape_<job>.py         │
│  managed workers        │                 │  (single-purpose entry)  │
└─────────────────────────┘                 └────────────┬─────────────┘
                                                       │
                                       reads env vars  │  writes state
                                                       ▼
                                              ┌─────────────────┐
                                              │  state.json /   │
                                              │  SQLite / queue │
                                              └─────────────────┘

Two rules follow from this split:

Each scrape_*.py script knows nothing about schedules. It knows one job: which target, which output, which state file.
The scheduler knows nothing about extraction. It knows when to run, how many retries to allow, and where to send notifications.

When you adopt this split, switching from a self-hosted scheduler to a managed platform becomes a deployment decision, not a rewrite decision.

Step 1: Define the Job Contract

Before writing any scraping code, write a small dataclass that names every job field. This is the contract your scheduler, your retry policy, and your monitoring will all read from.

# job_contract.py
from dataclasses import dataclass, field
from typing import Callable, Awaitable

@dataclass
class ScrapeJob:
    name: str                       # stable, e.g. "google_maps_leads_uk"
    target_url: str                 # the page or API endpoint to call
    output_path: str                # where to write the result
    schedule_cron: str              # "*/15 * * * *" for every 15 minutes
    max_attempts: int = 3           # bounded retries
    backoff_seconds: float = 2.0    # exponential base
    state_path: str = ""            # where to persist cursors / last success

The contract does not say how to parse the response or what the scraper does. That stays in the entrypoint. The contract only says what the scheduler needs to know to call it.

Step 2: Environment-Driven Endpoints and Credentials

Every value that depends on environment — endpoint URL, API key, target list, output bucket — is read from environment variables. This is the line that turns "a script on one machine" into "a script that can move" when you graduate to a managed platform.

# config.py
import os
from dataclasses import dataclass

@dataclass
class JobConfig:
    scraper_endpoint: str
    scraper_api_key: str
    target_urls: list[str]
    output_dir: str
    rate_limit_per_minute: int

    @classmethod
    def from_env(cls) -> "JobConfig":
        raw_targets = os.environ.get("TARGET_URLS", "").split(",")
        targets = [t.strip() for t in raw_targets if t.strip()]
        if not targets:
            raise RuntimeError("TARGET_URLS env var must list at least one URL")
        return cls(
            # The endpoint is environment-supplied so the same script
            # can be pointed at a self-hosted scraper, a staging proxy,
            # or a managed platform like the CoreClaw console.
            scraper_endpoint=os.environ["SCRAPER_ENDPOINT"],
            scraper_api_key=os.environ["SCRAPER_API_KEY"],
            target_urls=targets,
            output_dir=os.environ.get("OUTPUT_DIR", "./out"),
            rate_limit_per_minute=int(os.environ.get("RATE_LIMIT", "30")),
        )

When you migrate the same job to a managed platform later, you change SCRAPER_ENDPOINT and SCRAPER_API_KEY — nothing else.

Step 3: Runnable Orchestrator With Bounded Retries

The script below is the smallest self-contained orchestrator that genuinely earns the title "scheduled." It loads a job contract from a JSON file, fetches each target with exponential backoff, persists the last successful timestamp, and exits with a non-zero status so the scheduler can react.

# run_job.py
"""Orchestrate a single scraping job. Safe to call from cron or any scheduler."""
import json
import os
import sys
import time
import pathlib
from datetime import datetime, timezone
from typing import Any

import requests

from config import JobConfig


def load_job(path: str) -> dict[str, Any]:
    with open(path, "r", encoding="utf-8") as f:
        return json.load(f)


def fetch_with_retries(
    session: requests.Session,
    endpoint: str,
    api_key: str,
    target_url: str,
    max_attempts: int,
    base_backoff: float,
) -> dict[str, Any]:
    last_error: Exception | None = None
    for attempt in range(1, max_attempts + 1):
        try:
            resp = session.post(
                endpoint,
                headers={"Authorization": f"Bearer {api_key}"},
                json={"url": target_url},
                timeout=60,
            )
            if resp.status_code == 429 or 500 <= resp.status_code < 600:
                raise requests.HTTPError(f"transient {resp.status_code}")
            resp.raise_for_status()
            return resp.json()
        except (requests.RequestException, requests.HTTPError) as exc:
            last_error = exc
            sleep_for = base_backoff * (2 ** (attempt - 1))
            print(f"attempt {attempt} failed: {exc}; sleeping {sleep_for:.1f}s")
            time.sleep(sleep_for)
    raise RuntimeError(f"exhausted retries for {target_url}") from last_error


def update_state(state_path: str, target_url: str, ok: bool) -> None:
    state = {}
    if pathlib.Path(state_path).exists():
        state = json.loads(pathlib.Path(state_path).read_text(encoding="utf-8"))
    state.setdefault(target_url, {})
    state[target_url]["last_run_at"] = datetime.now(timezone.utc).isoformat()
    state[target_url]["last_status"] = "ok" if ok else "failed"
    pathlib.Path(state_path).write_text(json.dumps(state, indent=2), encoding="utf-8")


def run_job(job_path: str) -> int:
    cfg = JobConfig.from_env()
    job = load_job(job_path)
    state_path = job.get("state_path", f"./state/{job['name']}.json")
    pathlib.Path(state_path).parent.mkdir(parents=True, exist_ok=True)

    failures = 0
    with requests.Session() as session:
        for target in job["targets"]:
            try:
                payload = fetch_with_retries(
                    session=session,
                    endpoint=cfg.scraper_endpoint,
                    api_key=cfg.scraper_api_key,
                    target_url=target,
                    max_attempts=job.get("max_attempts", 3),
                    base_backoff=job.get("backoff_seconds", 2.0),
                )
                out_file = pathlib.Path(cfg.output_dir) / f"{job['name']}.jsonl"
                out_file.parent.mkdir(parents=True, exist_ok=True)
                with out_file.open("a", encoding="utf-8") as f:
                    f.write(json.dumps(payload, ensure_ascii=False) + "\n")
                update_state(state_path, target, ok=True)
            except Exception as exc:
                print(f"job {job['name']} target {target} failed: {exc}", file=sys.stderr)
                update_state(state_path, target, ok=False)
                failures += 1

    return 0 if failures == 0 else 1


if __name__ == "__main__":
    if len(sys.argv) != 2:
        print("usage: run_job.py <job.json>", file=sys.stderr)
        sys.exit(2)
    sys.exit(run_job(sys.argv[1]))

What this entrypoint gives you that a cron + ad-hoc script never did:

Bounded retries with exponential backoff and a clear terminal-failure exit code.
State persistence so a partially-completed run leaves a recoverable record, not silent gaps.
Single purpose — the orchestrator never grows parsing logic, so it never becomes the fragile part.

Step 4: Schedule It With Cron, Airflow, or a Managed Worker

Once run_job.py exists, the scheduler is interchangeable:

Scheduler	What you write	When it makes sense
Cron	`/15 * * * cd /srv/jobs && /usr/bin/python run_job.py jobs/google_maps_leads_uk.json >> logs/uk.log 2>&1`	One or two jobs, single host, low maintenance appetite
APScheduler / Prefect / Airflow	A Python schedule declaration bound to the same `run_job.py` entrypoint	Many jobs, need dashboards, ability to host your own scheduler
Managed Workers platform	A UI schedule + the same job script uploaded as a Worker	Many jobs across teams, you do not want to own the scheduler host

The same script runs in all three. The only thing that changes is who triggers it and where the logs land.

What the Output Looks Like

The orchestrator appends one JSON object per target per run. Representative example for a job tracking public product listings:

{
  "request": {"url": "https://example.com/products/widget-123"},
  "fetched_at": "2026-09-01T08:00:12Z",
  "status": 200,
  "data": {
    "title": "Widget 123",
    "price_usd": 19.99,
    "availability": "in_stock",
    "sku": "WDG-123"
  }
}

And the state.json file that survives restarts:

{
  "https://example.com/products/widget-123": {
    "last_run_at": "2026-09-01T08:00:12Z",
    "last_status": "ok"
  }
}

If a job run is interrupted, the next run sees the previous timestamp and can decide whether to re-fetch the same target or skip ahead, depending on your idempotency policy.

Where This Pattern Pays Off

Sales ops. Recurring local-business data refresh, with state showing which ZIP codes were scraped in the last 24 hours.
Pricing intelligence. Hourly pulls on a fixed product list, with the script deciding whether to re-fetch or skip based on the last_status field.
SEO rank tracking. Daily SERP snapshots for the same keyword set, persisted as JSONL per run so trendlines are auditable.
AI agent ingestion. A scheduled pull that pushes fresh public records into a vector store or RAG index, instead of an agent scraping live on every query.

Build vs Buy: Where the Line Sits

Dimension	Self-hosted scheduler + script	Managed Workers platform
Setup time	Hours to days	Minutes
Retries and backoff	You write it	Included
Schedule UI	You script it	Click in console
Logs and observability	You wire it	Included
Cost model	Your server + your time	Pay per successful result (see pricing)
Lock-in	None	Low if `run_job.py` stays the contract
When it makes sense	One or two jobs you fully understand	Many jobs across teams or domains

The crossover point is usually around the third concurrent scheduled job, or the first failed-batch incident that you only noticed three days later.

Limits, Compliance, and Honest Caveats

This pattern works for publicly accessible data. Respect each target's robots.txt, terms of service, and applicable privacy law (GDPR, CCPA, sector-specific rules).
Do not use it to scrape behind logins, evade CAPTCHAs, harvest private profiles, or republish copyrighted records at scale.
Exponential backoff reduces pressure on shared targets but does not make aggressive scraping polite. Always pair retries with a real rate_limit_per_minute.
If the upstream renames a field, the next run will write an unexpected key. Treat a sudden drift in the JSON schema as a P1 and pin a contract test.
Self-hosted schedulers still need patching, log rotation, and disk-space monitoring. Do not assume "it runs because the cron line ran" is the same thing as "it is healthy."

FAQ

Is there an official Python scheduler for scraping?
There is no single official scheduler. Teams commonly use cron for one or two jobs, then move to APScheduler, Prefect, or Airflow as the job count grows, or to a managed Workers platform when the orchestration overhead becomes its own project.

What is the safest schedule interval?
Start at 15 or 30 minutes for high-traffic public targets. Daily is fine for SERP tracking or product-price snapshots. Anything below 1 minute is almost never appropriate for a public site you do not own.

How do I avoid duplicate rows when a job retries?
Write append-only JSONL, then deduplicate downstream by a stable key (url + fetched_at, or an upstream-provided record id). Idempotency in the scraper itself is harder to get right than idempotency at the destination.

What should change first when I move to a managed platform?
Only the deployment details — the entrypoint script, the job contract, and the output format stay the same. The platform replaces your scheduler, retry logic, and observability glue.

Can this run inside a serverless function?
Yes, as long as the function is allowed enough time and memory to finish a single job. For long-running scrapers, a worker with a real timeout is a better fit than a short-lived function.

Summary

Recurring scraping becomes reliable the moment you treat each scrape as a discrete job with its own contract, its own state file, and its own retry policy — and stop trying to make cron do the work of an orchestrator. Keep your endpoint, API key, and target list in environment variables so the same script runs anywhere, then move the scheduler to a managed Workers platform the moment the operational overhead starts competing with the actual data work. If you want a starting point, browse the CoreClaw Workers Store for ready-made scrapers you can deploy on a schedule today.

DEV Community: coreclaw

Web Data Normalization Pipeline: How to Turn Raw Scraped JSON into Consistent Records with Python

TL;DR

Why Raw Scraped Data Is Inconsistent

The Normalization Pipeline Architecture

Runnable Python Pipeline

Expected Output

Adapting the Pipeline for Multiple Sources

Business Use Cases

DIY Normalization vs CoreClaw Structured Data vs Custom Pipeline

Limitations and Compliance

FAQ

Summary

Related Reading

YouTube Video Scraper API: How to Extract Public Video Metadata and Stats with Python

YouTube Video Scraper API: How to Extract Public Video Metadata and Stats with Python

TL;DR

Why per-video metadata is the right primitive

What the verified repositories provide

Pipeline design

Setup and configuration

Runnable Python: normalize and append

Representative output

Use cases

Comparison: per-video vs channel-level scraping

Limitations and compliance

Maintenance checklist

FAQ

Next steps

Zapier Web Scraping: How to Connect CoreClaw Data to Your CRM and Sheets

Zapier Web Scraping: How to Connect CoreClaw Data to Your CRM and Sheets

TL;DR

Why Web Data + Zapier Breaks Without a Structured Source

What CoreClaw + Zapier Looks Like

Step-by-Step: Build a Google Maps Leads → Zapier → Sheets Workflow

Step 1 — Create the Zapier Webhook Trigger

Step 2 — Test the Webhook Locally

Step 3 — Send CoreClaw Data to the Webhook

Step 4 — Format and Map in Zapier

Step 5 — Add Error Handling

Representative Output Shape

Business Use Cases

Manual Scraping vs CoreClaw + Zapier vs Enterprise Integration

Limitations and Compliance

FAQ

Summary

Related Reading

TikTok Video Scraper GitHub: How to Extract Public Video Metadata with Python

TikTok Video Scraper GitHub: How to Extract Public Video Metadata with Python

TL;DR

Why per-video metadata is the right primitive

What the verified repositories provide

Pipeline design

Setup and configuration

Normalization: From raw JSON to a stable JSONL record

Representative output

Use cases

Build, buy, and tool comparison

Checklist before production

Compliance and maintenance

FAQ

Next Steps

CoreClaw n8n Integration: How to Connect Web Data API to No-Code Automation Workflows

CoreClaw n8n Integration: How to Connect Web Data API to No-Code Automation Workflows

TL;DR

Why n8n still needs an external data source

What the CoreClaw endpoint returns

Build the n8n workflow

Step 1: Collect credentials and the current endpoint

Step 2: Add a Schedule or Webhook trigger

Step 3: Configure the HTTP Request node

Step 4: Parse the response with a Code node

Step 5: Deliver the data

Step 6: Test end to end

A local Python check before you wire n8n

Deployment checklist

Business use cases

How it compares to other approaches

Limitations and compliance

FAQ