DEV Community: coreclaw

Google Maps Lead Scraper: How to Build a Local B2B Outreach List with CoreClaw

coreclaw — Sat, 19 Sep 2026 01:14:50 +0000

Google Maps Lead Scraper: How to Build a Local B2B Outreach List with CoreClaw

The fastest way to build a local B2B outreach list is to turn Google Maps business listings into structured, filterable records. A Google Maps lead scraper extracts public business names, addresses, phone numbers, websites, categories, ratings, and review counts, then lets you score and load the results into a CRM or outreach tool. This guide is for sales teams, local SEO agencies, and SaaS founders who need a repeatable Python workflow for collecting and qualifying local leads without maintaining their own scraping infrastructure.

TL;DR

A Google Maps lead scraper reads public business listings and returns structured fields you can filter by category, rating, and location.
Use a managed scraper like the CoreClaw Google Maps Scraper to skip proxy rotation, browser management, and anti-bot maintenance.
A short Python script normalizes the output, deduplicates by phone or website, scores leads by review signals, and exports a CSV ready for a CRM or cold-outreach tool.
Always collect only public data, respect Google Maps terms of service, and comply with local privacy and telemarketing rules.

Why Manual Local Lead Research Fails

Building a local lead list by hand is straightforward for five businesses and painful for five hundred. You open Google Maps, search for a category and city, click each listing, copy the name, guess the email, paste the phone number into a spreadsheet, and repeat. Within an hour the spreadsheet is full of inconsistent fields, missing websites, and duplicate chain locations.

The problems multiply quickly:

Scale. A single city can contain thousands of businesses in one category. Manual collection becomes a multi-day task.
Freshness. Phone numbers, hours, and websites change. A spreadsheet copied last quarter already contains stale records.
Deduplication. The same business can appear under slightly different names or addresses. Without a canonical identifier, your list fills with duplicates.
Qualification. A raw list tells you a business exists; it does not tell you which ones are worth calling first. You need signals like rating, review count, category, and web presence to prioritize.
Provenance. When a record is wrong, you need to know where it came from and when it was collected. Manual copy-paste loses that trail.

A structured scraping workflow solves these problems by producing timestamped, normalized records you can filter, score, and reload on a schedule.

What a Google Maps Lead Scraper Returns

A lead scraper does not invent contact information. It collects the fields that are publicly visible on a Google Maps listing and organizes them into a consistent schema. Typical output fields include:

Field	Why it matters for outreach
`business_name`	The public trading name of the business.
`address`	Full street address for geo-targeting and routing.
`phone`	Direct public phone number; the best deduplication key.
`website`	Domain for email discovery and enrichment.
`category`	Primary business category for segmentation.
`rating`	Average public review score; a quality signal.
`review_count`	Volume of public reviews; indicates visibility and engagement.
`hours`	Operating hours for call-timing decisions.
`plus_code` / `place_id`	Stable identifiers for deduplication and re-scraping.
`collected_at`	Timestamp so you know how fresh the record is.

The exact field names depend on the scraper you use. The important part is that the output is machine-readable, not a copy-paste mess.

A Runnable Python Lead Qualification Workflow

The script below assumes you have already run a Google Maps scraper and have a JSONL file of raw records, or you are calling a managed worker endpoint that returns structured JSON. It reads configuration from environment variables, normalizes the records, scores each lead, deduplicates by phone or website, and writes a CSV ready for import into a CRM.

import csv
import json
import os
from datetime import datetime, timezone
from urllib.parse import urlparse

# Read endpoint and credentials from environment variables.
# Never hard-code API keys or worker URLs in source control.
CORECLAW_API_KEY = os.environ.get("CORECLAW_API_KEY")
CORECLAW_WORKER_ENDPOINT = os.environ.get("CORECLAW_WORKER_ENDPOINT", "")
INPUT_FILE = os.environ.get("LEAD_INPUT_FILE", "google_maps_raw.jsonl")
OUTPUT_FILE = os.environ.get("LEAD_OUTPUT_FILE", "qualified_leads.csv")

# Outreach targets: edit these to match your ideal customer profile.
TARGET_CATEGORIES = {"marketing agency", "software company", "it services", "consultant"}
MIN_RATING = 3.5
MIN_REVIEW_COUNT = 5


def normalize_record(raw: dict) -> dict:
    """Flatten a raw Google Maps record into a consistent lead schema."""
    website = raw.get("website") or raw.get("site") or ""
    domain = ""
    if website:
        parsed = urlparse(website if website.startswith("http") else f"https://{website}")
        domain = parsed.netloc.replace("www.", "")

    phone = raw.get("phone") or raw.get("phone_number") or ""
    if phone:
        phone = "".join(c for c in phone if c.isdigit() or c == "+")

    return {
        "business_name": (raw.get("name") or raw.get("business_name") or "").strip(),
        "address": (raw.get("address") or "").strip(),
        "phone": phone,
        "website": website,
        "domain": domain,
        "category": (raw.get("category") or "").lower().strip(),
        "rating": _to_float(raw.get("rating")),
        "review_count": _to_int(raw.get("review_count") or raw.get("reviews")),
        "place_id": raw.get("place_id") or raw.get("plus_code") or "",
        "collected_at": datetime.now(timezone.utc).isoformat(),
    }


def _to_float(value):
    try:
        return float(value)
    except (TypeError, ValueError):
        return None


def _to_int(value):
    try:
        return int(str(value).replace(",", "").replace("+", ""))
    except (TypeError, ValueError):
        return 0


def score_lead(lead: dict) -> int:
    """Score a lead based on public signals. Higher is better."""
    score = 0
    if lead["category"] in TARGET_CATEGORIES:
        score += 30
    if lead["rating"] and lead["rating"] >= MIN_RATING:
        score += 20
    if lead["review_count"] and lead["review_count"] >= MIN_REVIEW_COUNT:
        score += min(lead["review_count"] // 10, 20)
    if lead["website"]:
        score += 15
    if lead["phone"]:
        score += 10
    return score


def load_records(path: str) -> list:
    records = []
    with open(path, "r", encoding="utf-8") as f:
        for line in f:
            line = line.strip()
            if not line:
                continue
            records.append(json.loads(line))
    return records


def main() -> None:
    raw_records = load_records(INPUT_FILE)
    leads = [normalize_record(r) for r in raw_records]

    # Score and filter.
    qualified = []
    for lead in leads:
        lead["lead_score"] = score_lead(lead)
        if lead["lead_score"] >= 30:
            qualified.append(lead)

    # Deduplicate by phone, then by domain, then by place_id.
    seen: set = set()
    unique = []
    for lead in qualified:
        key = lead["phone"] or lead["domain"] or lead["place_id"]
        if not key or key in seen:
            continue
        seen.add(key)
        unique.append(lead)

    # Sort by score descending.
    unique.sort(key=lambda x: x["lead_score"], reverse=True)

    fieldnames = [
        "business_name", "category", "rating", "review_count",
        "phone", "website", "domain", "address", "lead_score", "collected_at",
    ]
    with open(OUTPUT_FILE, "w", newline="", encoding="utf-8") as f:
        writer = csv.DictWriter(f, fieldnames=fieldnames)
        writer.writeheader()
        for lead in unique:
            writer.writerow({k: lead.get(k, "") for k in fieldnames})

    print(f"Raw records: {len(raw_records)}")
    print(f"Qualified leads: {len(unique)}")
    print(f"Output written to: {OUTPUT_FILE}")


if __name__ == "__main__":
    main()

If you are calling a live worker instead of reading a local file, replace load_records with an authenticated request to your verified endpoint. Keep the endpoint, API key, and output schema in environment variables so the script stays portable.

Representative Output and Checklist

A qualified lead row looks like this:

business_name,category,rating,review_count,phone,website,domain,address,lead_score,collected_at
Acme Digital,marketing agency,4.7,42,+15551234567,https://acmedigital.com,acmedigital.com,"123 Main St, Austin, TX",95,2026-09-19T09:00:00+00:00

Before importing into your CRM or outreach tool, run this checklist:

[ ] Remove records without a phone number or website unless you have another enrichment path.
[ ] Verify that the category matches your ideal customer profile.
[ ] Exclude businesses below your minimum rating threshold.
[ ] Deduplicate against your existing CRM by phone or domain.
[ ] Add a source column that references Google Maps and the collection date.
[ ] Review the list for chain locations if you are targeting independent businesses.
[ ] Confirm compliance with local telemarketing and privacy regulations before calling.

Build vs Buy: Where CoreClaw Fits

Approach	Best for	Maintenance burden	Lead quality	Time to first list
Manual research	A handful of high-value accounts	Low	High	Hours
Open-source scraper + custom code	Engineering teams with scraping experience	High	Medium	Days to weeks
Managed scraper API	Teams that need reliable, scalable lists	Low	High	Minutes

Building your own Google Maps scraper in 2026 means fighting TLS fingerprinting, proxy rotation, CAPTCHA handling, and selector drift every time Google updates its front end. For most sales and marketing teams, that engineering time is better spent on messaging and follow-up.

The CoreClaw Google Maps Scraper returns structured business records in bulk, including reviews, reviewer details, contact information, hours, and photos, without requiring you to manage proxies or browser farms. If your use case is not covered by the ready-made worker, you can build a custom worker in Python, Node.js, or Go and deploy it through the CoreClaw worker console.

For a broader view of ready-made scrapers, the CoreClaw Store lists workers for Google Maps, Google Search, Amazon, Instagram, YouTube, TikTok, and other common targets.

Business Use Cases

Local SEO agencies use lead lists to prospect businesses with weak review profiles. A restaurant with a 3.2 rating and no website is an obvious candidate for reputation management or web design services.

SaaS outbound sales target specific categories in a city or metro area. A fleet-management tool might search for "delivery service," "courier," or "logistics company" within a 50-mile radius.

Franchise and multi-location brands monitor competitor density when choosing new locations. A structured Maps dataset shows which neighborhoods are saturated and which are underserved.

Event and conference organizers build sponsor or attendee prospect lists around venues and nearby businesses. A tech conference in Chicago can target "software company" and "IT consultant" listings within walking distance of the venue.

Market research teams compare category counts, ratings, and review velocity across cities to measure local market health.

Limitations and Compliance

A lead scraper is a research accelerator, not a license to ignore rules. Keep the following limits in mind:

Public data only. Only collect information that is publicly visible on the business listing. Do not attempt to access private accounts, restricted pages, or non-public contact details.
Respect platform terms. Google Maps has terms of service and technical protections. Run requests at a reasonable pace and review the latest terms before deploying.
No credential abuse. Do not use stolen credentials, fake accounts, or session hijacking to access data.
No CAPTCHA evasion. If the site presents a challenge, stop and reduce frequency or switch to an officially supported data source.
Privacy and telemarketing laws. Phone numbers collected from public listings may still be regulated by local laws. In many jurisdictions you need a lawful basis to call or message a business. Review GDPR, TCPA, or equivalent rules that apply to your market.
Data freshness. A scraped record is a snapshot, not a live feed. Re-run your workflow on a schedule and treat old records as suspects.

For more on collecting public web data responsibly, see the public web data compliance guide on the CoreClaw Web Data Guide.

FAQ

Do I need a Google Maps API key?
No. A Google Maps scraper reads public business listings directly. You do not need a Google Cloud project, API key, or OAuth flow.

What output format does the CoreClaw Google Maps Scraper use?
CoreClaw workers typically return JSON, CSV, or Excel. You can choose the format that fits your downstream workflow.

Can I get email addresses from Google Maps?
No. Google Maps listings generally do not expose email addresses. Use the website or phone number to initiate contact, or run a separate enrichment step on the domain using a tool you trust.

How do I avoid duplicate leads?
Deduplicate by phone number first, then by domain, then by place ID or plus code. Chain locations often share a domain but have different addresses, so decide whether you want one entry per brand or one per location.

How often should I refresh the lead list?
For most outreach campaigns, weekly or monthly refreshes are enough. Daily refreshes are useful only when you are monitoring fast-moving markets or competitor openings.

Can I schedule the workflow to run automatically?
Yes. You can run the script from cron, GitHub Actions, or a cloud scheduler. Pair it with a managed scraper that exposes an API endpoint so you do not have to maintain the browser layer.

Is this legal?
Laws vary by jurisdiction. Collecting publicly visible business information is generally permissible in many regions, but you must still comply with platform terms, privacy laws, and telemarketing regulations. This guide does not constitute legal advice; consult a lawyer for your specific use case.

Next Steps

Start by defining your ideal customer profile: category, location, minimum rating, and minimum review count. Then run a small test with the CoreClaw Google Maps Scraper, normalize the output with the Python script above, and review a dozen records manually before scaling.

If Google Maps is only one of several data sources you need, browse the CoreClaw Store for ready-made workers covering search, e-commerce, and social platforms. For a deeper look at Google Maps scraping workflows, the CoreClaw Google Maps scraper guide walks through public-data collection patterns and compliance considerations.

Etsy Price Tracker: How to Monitor Listing Price Changes with Python

coreclaw — Fri, 18 Sep 2026 02:07:43 +0000

Etsy Price Tracker: How to Monitor Listing Price Changes with Python

The reliable way to track price changes on Etsy is to snapshot public listing data on a schedule and diff each snapshot against the previous one. You can build that workflow in a couple hundred lines of Python around the open-source Etsy scraper maintained by the data-scrape profile, which handles the collection side while your code handles storage, comparison, and alerting. This article shows the full loop: what to store, how to diff, and how to turn raw changes into a report you can actually act on.

TL;DR

Use the data-scrape/etsy-scraper repository to collect public listing records (title, price, availability, shop, URL) for the listings and shops you care about.
Append each run's output to a dated JSONL snapshot file — never overwrite, because the history is the product.
Diff consecutive snapshots in Python to detect price drops, price increases, and sold-out or relisted items.
Apply the same snapshot-diff pattern to other marketplaces; the AliExpress scraper repository follows the same repository-led approach for cross-marketplace monitoring.
Respect Etsy's terms, robots directives, and applicable privacy law. This workflow is for public marketplace research, not for bulk harvesting personal data.

Why manual Etsy price tracking breaks down

If you watch five handmade listings, a spreadsheet works. If you watch five hundred — competitor shops, a product niche, or your own catalog across seasons — manual tracking fails for predictable reasons:

Prices move quietly. Etsy sellers reprice for sales, holidays, and material costs. A weekly manual check misses most intraweek changes.
Listings disappear and return. A sold-out listing that relists at a higher price looks like two unrelated records unless you track listing IDs over time.
Copy-paste loses provenance. Without a timestamp and source URL on every record, you cannot reconstruct when a change happened or trust the history.
One-off exports go stale immediately. A CSV saved today is a fossil next week.

The fix is not a bigger spreadsheet. It is a small, boring pipeline: collect → snapshot → diff → report. The analysis is easy; the collection and consistency are the hard parts, which is exactly what a maintained scraper repository removes.

What the repository-led workflow looks like

The data-scrape/etsy-scraper repository provides runnable code for collecting public Etsy listing and shop data. Treat it as the collection layer of your pipeline: you run it with your own documented inputs, it produces structured records, and everything downstream — storage, diffing, reporting — is plain Python you own.

The pipeline has four stages:

Collect. Run the repository's documented workflow against your target listings or shop pages. Output: one structured record per listing.
Snapshot. Normalize each record to a stable schema and append it to snapshots/YYYY-MM-DD.jsonl. One file per run, append-only.
Diff. Compare the newest snapshot against the previous one on listing_id. Produce a change list: price up, price down, delisted, relisted, availability changed.
Report. Write the changes to a CSV and, optionally, a markdown summary you can paste into Slack or email.

Because every record carries a source URL and a run date, you can always answer "what did this listing cost on which day" — the property that manual tracking never gives you.

Step 1 — Set up the environment

Clone the repository and create a working directory for snapshots:

git clone https://github.com/data-scrape/etsy-scraper.git
cd etsy-scraper
mkdir -p ../etsy-price-tracker/snapshots

Configure any account-specific values through environment variables rather than hard-coded strings. The exact variable names depend on the repository's current README — read it before running, because documented inputs change over time:

export ETSY_TARGETS_FILE="../etsy-price-tracker/targets.txt"   # listing or shop URLs you follow
export SNAPSHOT_DIR="../etsy-price-tracker/snapshots"

Keep a plain targets.txt with one public listing or shop URL per line. Starting from a file instead of hard-coded URLs means adding a competitor shop is a one-line change, and the file doubles as documentation of what you monitor.

Step 2 — Run a collection pass and snapshot the output

Run the repository's documented collection workflow once. Its exact invocation is in the README; the important discipline is what you do with the output: normalize it into a stable schema before storing it, so that diffing code never has to know about raw-format quirks.

A normalization wrapper looks like this:

import json
import pathlib
from datetime import datetime, timezone

def normalize_record(raw: dict) -> dict:
    """Map a raw collected record to the tracker's stable schema.

    Field names in `raw` depend on the repository's current output format.
    Adjust the keys after inspecting one real record — do not guess.
    """
    return {
        "listing_id": str(raw["listing_id"]),
        "title": raw.get("title", ""),
        "price": raw.get("price"),              # keep as-is; parsed below
        "currency": raw.get("currency", "USD"),
        "availability": raw.get("availability", "unknown"),
        "shop": raw.get("shop", ""),
        "url": raw.get("url", ""),
        "collected_at": datetime.now(timezone.utc).isoformat(timespec="seconds"),
    }

def append_snapshot(records: list[dict], snapshot_dir: str) -> pathlib.Path:
    snap_dir = pathlib.Path(snapshot_dir)
    snap_dir.mkdir(parents=True, exist_ok=True)
    today = datetime.now(timezone.utc).date().isoformat()
    path = snap_dir / f"{today}.jsonl"
    with path.open("a", encoding="utf-8") as f:
        for rec in records:
            f.write(json.dumps(rec, ensure_ascii=False) + "\n")
    return path

Two design decisions matter here:

Append, never overwrite. If you run the collection twice in one day, both passes land in the same dated file. The diff step handles duplicates by keeping the last record per listing per day.
Normalize at ingestion. A price stored as "$24.00" on Monday and 24.0 on Tuesday will silently break your diff. Parse and canonicalize once, at the boundary.

Step 3 — Diff consecutive snapshots

The core of the tracker is a pure function: given yesterday's records and today's records, produce a change list. No network, no parsing — which makes it trivially testable.

import csv
import json
import pathlib
from decimal import Decimal, InvalidOperation

def load_latest_records(snapshot_dir: str, date: str) -> dict[str, dict]:
    """Load one day's JSONL file, keeping the LAST record per listing."""
    path = pathlib.Path(snapshot_dir) / f"{date}.jsonl"
    records: dict[str, dict] = {}
    if not path.exists():
        return records
    with path.open(encoding="utf-8") as f:
        for line in f:
            rec = json.loads(line)
            records[rec["listing_id"]] = rec
    return records

def parse_price(value) -> Decimal | None:
    if value is None:
        return None
    try:
        return Decimal(str(value).replace("$", "").replace(",", "").strip())
    except InvalidOperation:
        return None

def diff_snapshots(old: dict[str, dict], new: dict[str, dict]) -> list[dict]:
    changes = []
    for lid, new_rec in new.items():
        old_rec = old.get(lid)
        if old_rec is None:
            changes.append({"listing_id": lid, "change": "new_listing",
                            "url": new_rec["url"], "detail": new_rec["title"]})
            continue
        old_price, new_price = parse_price(old_rec["price"]), parse_price(new_rec["price"])
        if old_price is not None and new_price is not None and old_price != new_price:
            direction = "price_increase" if new_price > old_price else "price_decrease"
            changes.append({"listing_id": lid, "change": direction,
                            "url": new_rec["url"],
                            "detail": f"{old_price} -> {new_price} {new_rec['currency']}"})
        if old_rec.get("availability") != new_rec.get("availability"):
            changes.append({"listing_id": lid, "change": "availability_changed",
                            "url": new_rec["url"],
                            "detail": f"{old_rec.get('availability')} -> {new_rec.get('availability')}"})
    for lid, old_rec in old.items():
        if lid not in new:
            changes.append({"listing_id": lid, "change": "delisted",
                            "url": old_rec["url"], "detail": old_rec["title"]})
    return changes

def write_change_report(changes: list[dict], out_path: str) -> None:
    with open(out_path, "w", newline="", encoding="utf-8") as f:
        writer = csv.DictWriter(f, fieldnames=["listing_id", "change", "detail", "url"])
        writer.writeheader()
        writer.writerows(changes)

Glue it together with a small daily runner:

import sys
from datetime import date, timedelta

def main(snapshot_dir: str) -> None:
    today = date.today().isoformat()
    yesterday = (date.today() - timedelta(days=1)).isoformat()
    new = load_latest_records(snapshot_dir, today)
    old = load_latest_records(snapshot_dir, yesterday)
    if not new:
        sys.exit(f"No snapshot found for {today} — run the collection step first.")
    changes = diff_snapshots(old, new)
    out = f"changes-{today}.csv"
    write_change_report(changes, out)
    print(f"{len(changes)} changes written to {out}")

if __name__ == "__main__":
    main(sys.argv[1] if len(sys.argv) > 1 else "snapshots")

Representative output

The change report is deliberately boring and machine-friendly. Example rows (illustrative, not from a live run):

listing_id,change,detail,url
1284402917,price_decrease,32.00 -> 24.00 USD,https://www.etsy.com/listing/1284402917
1290773451,price_increase,18.50 -> 21.00 USD,https://www.etsy.com/listing/1290773451
1281023764,availability_changed,in_stock -> sold_out,https://www.etsy.com/listing/1281023764
1275539980,delisted,Handmade ceramic mug set,https://www.etsy.com/listing/1275539980
1300182734,new_listing,Personalized dog bandana,https://www.etsy.com/listing/1300182734

From this one file you can answer the questions manual tracking never could: which competitor started a sale, which niche is seeing price creep, and which listings rotate in and out of stock.

Comparison: build vs. spreadsheet vs. managed service

Dimension	This repository workflow	Spreadsheet + manual checks	Managed scraping platform
Best for	Developers tracking a defined list of listings	Fewer than ~10 listings, one-off checks	Teams needing scale, uptime, and support
Setup model	Clone repo, run documented workflow, own the code	None	Account, plan selection, endpoint config
Maintenance burden	You adapt when page layouts change	None, but coverage decays	Provider maintains parsers
Cost	Code is open source; your time	Free, but hours of labor	Verify current pricing on the provider's official page
Data provenance	Full: URL + timestamp on every record	Poor	Usually good; depends on provider
Change detection	Built in (this article's diff step)	Manual	Varies by provider

The honest trade-off: a repository-led workflow costs you maintenance attention in exchange for full control and provenance. If you only need a handful of price checks per month, a spreadsheet wins. If you need auditable history across hundreds of listings, the pipeline pays for itself.

Use cases this supports

Competitive repricing. Handmade and craft sellers watch adjacent shops; a price-decrease alert is often the first visible signal of a promotion.
Resale and arbitrage research. Track listing prices over weeks to distinguish stable prices from temporary discounts before making inventory decisions.
Niche market research. Aggregate price distributions per category over time to see whether a niche is moving up- or down-market. The same snapshot-diff pattern extends to other marketplaces via the AliExpress scraper for cross-marketplace product data.
Catalog QA for your own shop. Run the tracker against your own public listings to catch accidental price edits or unintended sold-out states.

Limits, compliance, and maintenance

Be realistic about all three:

Layout changes are inevitable. When a diff suddenly reports everything as "delisted," suspect a parsing break upstream, not a mass exodus of sellers. Keep the collection layer isolated so fixes stay contained.
Pace your collection. Run daily or a few times per day, not per minute. Low-frequency snapshots capture nearly all economically meaningful price changes and keep load on target servers minimal.
Public data only. Track public listing pages. Do not attempt to access private data, bypass access controls, or evade anti-bot protections — those approaches create legal exposure and produce unreliable data anyway.
Respect the platform. Review Etsy's terms and robots directives, comply with applicable privacy and consumer-protection law, and use the data for research and competitive analysis rather than republishing sellers' content wholesale.
Verify before production. The repository's documented inputs and output fields evolve. Inspect a real record before wiring assumptions into your schema, and pin the commit you deploy so upgrades are deliberate.

FAQ

Is there an official Etsy API for prices?
Etsy offers an official API for sellers and approved applications with its own scope and authentication requirements. The repository workflow in this article is for public-page research where the official API does not fit your use case; check Etsy's developer documentation for current API access rules before choosing.

What fields does each snapshot contain?
The normalized schema used here: listing_id, title, price, currency, availability, shop, url, and collected_at. Adjust after inspecting the repository's real output — do not assume field names.

How often should the workflow run?
Once daily is enough for most price-monitoring goals. Increase frequency only if you have evidence that intraday changes matter for your niche, and keep frequency proportionate to avoid unnecessary load.

What happens when Etsy changes its page layout?
The collection step (repository) may need updating; your diff and reporting code keeps working because it only sees normalized records. This separation is the main reason to keep raw collection isolated from analysis.

Can this send alerts to Slack or email?
Yes. The change report is a plain CSV and list of dicts — add a webhook call or SMTP send in the runner's main() after write_change_report(). Keep alert logic separate from diff logic so you can test each independently.

Can I track other marketplaces with the same pattern?
Yes. The snapshot-diff-report skeleton is marketplace-agnostic; only the collection layer changes. The AliExpress repository linked above uses the same repository-led approach for a different marketplace.

Do I need a paid service to run this?
No managed service is required — the code runs locally on a schedule you control (cron, Task Scheduler, or a CI cron job). Costs are your own compute time. Commercial alternatives exist and can be worth it at scale; verify current pricing on any provider's official page before committing.

Next step

Clone the data-scrape/etsy-scraper repository, run one documented collection pass against a small targets.txt, and get a two-day snapshot loop working before scaling up. Browse the rest of the data-scrape repository profile for sibling scrapers covering other marketplaces, platforms, and data sources you may want to fold into the same pipeline.

Google Maps Scraper GitHub: How to Monitor Local Business Review Trends with Python

coreclaw — Thu, 17 Sep 2026 02:03:53 +0000

Google Maps Scraper GitHub: How to Monitor Local Business Review Trends with Python

The fastest way to track how local businesses are perceived on Google Maps is to snapshot public review data on a schedule and diff the results. You can do this with a small Python workflow built around the open-source Google Maps scraper for Python maintained by the data-scrape profile. This article shows you how to collect rating, review count, and review text fields, store snapshots, and surface trends without writing a full scraping engine from scratch.

TL;DR

Clone the data-scrape/scrape-google-maps repository and run it with documented inputs.
Schedule a Python script to fetch public business records, normalize rating and review fields, and append dated snapshots to JSONL files.
Diff consecutive snapshots to detect rating drops, review-count spikes, and sentiment shifts.
Pair it with the Yellow Pages scraper when you need additional local business context such as category or contact signals.
Respect Google Maps terms, robots directives, and privacy rules; this workflow is for public data research only.

Why review trends matter for local market research

A single star-rating snapshot tells you almost nothing. A time series of rating, review count, and review text tells you:

Whether a competitor is improving or deteriorating in customer perception.
When a new business starts accumulating reviews rapidly.
Which service issues recur across multiple listings in the same neighborhood.
How your own locations trend after operational changes.

The hard part is not the analysis. It is getting consistent, timestamped records without maintaining a browser farm, parsing layout changes, or managing proxies. That is where a repository-led workflow helps.

The limits of manual tracking and one-off exports

Manual checks fail for anything beyond a handful of listings because:

Ratings are live and change daily.
Review counts move as new reviews are added or filtered.
Copy-pasting from a browser loses provenance and timestamps.
One-off CSV exports become stale the moment they are saved.

A script that runs on a schedule solves the freshness problem, but only if the underlying scraper is maintained. Writing your own Google Maps parser is fragile because markup and anti-bot measures change. Using a maintained open-source repository lets you focus on the analysis layer instead of the extraction layer.

What the data-scrape Google Maps scraper provides

The data-scrape/scrape-google-maps repository contains a Python-based Google Maps scraping workflow. It is designed for public business records: names, addresses, ratings, review counts, phone numbers, websites, hours, and review samples where publicly visible.

Key characteristics:

Repository-led, not a managed service. You run it in your own environment.
Input-driven. You supply a search query or list of place identifiers.
Output is structured JSON/JSONL that maps cleanly into Python data frames or databases.
No invented endpoints. The README and example scripts define the inputs the repository actually accepts.

Because it is open source, you can inspect how fields are parsed, adapt it to your own validation rules, and store results however you want.

Workflow design

A review-trend workflow has four stages:

Input preparation: define the businesses or search area you want to track.
Extraction: run the scraper and collect public business records.
Normalization: keep only the fields needed for trend analysis and add timestamps.
Diffing and alerting: compare the latest snapshot to the previous one and flag changes.

The Python script below assumes you have cloned the repository, installed its documented dependencies, and set your inputs through environment variables. It does not contain hardcoded credentials or fake endpoints.

Environment setup

# Clone the repository
git clone https://github.com/data-scrape/scrape-google-maps.git
cd scrape-google-maps

# Create a virtual environment and install documented dependencies
python -m venv venv
source venv/bin/activate  # Windows: venv\Scripts\activate
pip install -r requirements.txt

Create a project directory for snapshots next to the repository:

mkdir -p review-snapshots/data

Set inputs through environment variables instead of editing the script:

export SEARCH_QUERY="coffee shops in Seattle, WA"
export SNAPSHOT_DIR="./review-snapshots/data"
export RUN_DATE="2026-09-17"

Runnable Python workflow

The script runs the repository's scraper, normalizes each record, writes a dated JSONL snapshot, and prints businesses whose rating or review count changed since the previous run.

import json
import os
import subprocess
from datetime import datetime, timezone
from pathlib import Path

SEARCH_QUERY = os.environ["SEARCH_QUERY"]
SNAPSHOT_DIR = Path(os.environ["SNAPSHOT_DIR"])
RUN_DATE = os.environ.get("RUN_DATE", datetime.now(timezone.utc).strftime("%Y-%m-%d"))

SNAPSHOT_DIR.mkdir(parents=True, exist_ok=True)


def run_scraper(query: str, output_path: Path) -> None:
    """Run the repository scraper with the documented CLI.

    The exact command depends on the repository README. Adjust the script
    name and flags to match the current version of data-scrape/scrape-google-maps.
    """
    cmd = [
        "python",
        "scrape_google_maps.py",
        "--query", query,
        "--output", str(output_path),
    ]
    subprocess.run(cmd, check=True)


def normalize_record(raw: dict, run_date: str) -> dict:
    """Keep only the fields needed for review-trend analysis."""
    return {
        "place_id": raw.get("place_id") or raw.get("id"),
        "name": raw.get("name"),
        "address": raw.get("address"),
        "rating": _to_float(raw.get("rating")),
        "review_count": _to_int(raw.get("review_count") or raw.get("reviews")),
        "category": raw.get("category"),
        "phone": raw.get("phone"),
        "website": raw.get("website"),
        "sample_reviews": raw.get("reviews_list", [])[:5],
        "snapshot_date": run_date,
        "fetched_at": datetime.now(timezone.utc).isoformat(),
    }


def _to_float(value):
    try:
        return float(value)
    except (TypeError, ValueError):
        return None


def _to_int(value):
    try:
        return int(str(value).replace(",", "").replace("+", ""))
    except (TypeError, ValueError):
        return None


def write_snapshot(records: list, run_date: str) -> Path:
    path = SNAPSHOT_DIR / f"google_maps_reviews_{run_date}.jsonl"
    with path.open("w", encoding="utf-8") as f:
        for record in records:
            f.write(json.dumps(record, ensure_ascii=False) + "\n")
    return path


def load_previous_snapshot(current_path: Path) -> dict:
    """Load the most recent snapshot before the current one."""
    files = sorted(SNAPSHOT_DIR.glob("google_maps_reviews_*.jsonl"))
    for file in reversed(files):
        if file.name != current_path.name:
            records = {}
            with file.open(encoding="utf-8") as f:
                for line in f:
                    rec = json.loads(line)
                    records[rec.get("place_id")] = rec
            return records
    return {}


def diff_snapshots(current: list, previous: dict) -> list:
    changes = []
    for rec in current:
        place_id = rec.get("place_id")
        prev = previous.get(place_id)
        if not prev:
            changes.append({"type": "new", "place_id": place_id, "name": rec["name"]})
            continue
        rating_delta = (rec["rating"] or 0) - (prev["rating"] or 0)
        review_delta = (rec["review_count"] or 0) - (prev["review_count"] or 0)
        if rating_delta or review_delta:
            changes.append({
                "type": "changed",
                "place_id": place_id,
                "name": rec["name"],
                "rating_delta": round(rating_delta, 2),
                "review_delta": review_delta,
                "previous_rating": prev["rating"],
                "current_rating": rec["rating"],
                "previous_review_count": prev["review_count"],
                "current_review_count": rec["review_count"],
            })
    return changes


def main() -> None:
    raw_output = SNAPSHOT_DIR / f"raw_{RUN_DATE}.json"
    run_scraper(SEARCH_QUERY, raw_output)

    raw_records = json.loads(raw_output.read_text(encoding="utf-8"))
    normalized = [normalize_record(r, RUN_DATE) for r in raw_records]
    snapshot_path = write_snapshot(normalized, RUN_DATE)

    previous = load_previous_snapshot(snapshot_path)
    changes = diff_snapshots(normalized, previous)

    print(f"Snapshot written: {snapshot_path}")
    print(f"Records collected: {len(normalized)}")
    print(f"Changes since last run: {len(changes)}")
    for change in changes[:10]:
        print(json.dumps(change, ensure_ascii=False))


if __name__ == "__main__":
    main()

Representative output

A single normalized record looks like this:

{
  "place_id": "ChIJN1t_tDeuEmsRUsoyG83frY4",
  "name": "Seattle Coffee Works",
  "address": "2401 Utah Ave S, Seattle, WA 98134",
  "rating": 4.6,
  "review_count": 1247,
  "category": "Coffee shop",
  "phone": "+1 206-624-1975",
  "website": "https://seattlecoffeeworks.com",
  "sample_reviews": [
    {"text": "Great single-origin espresso.", "rating": 5, "relative_time": "2 weeks ago"},
    {"text": "Long line on weekends.", "rating": 4, "relative_time": "1 month ago"}
  ],
  "snapshot_date": "2026-09-17",
  "fetched_at": "2026-09-17T10:00:00+00:00"
}

A diff entry looks like this:

{
  "type": "changed",
  "place_id": "ChIJN1t_tDeuEmsRUsoyG83frY4",
  "name": "Seattle Coffee Works",
  "rating_delta": -0.1,
  "review_delta": 12,
  "previous_rating": 4.7,
  "current_rating": 4.6,
  "previous_review_count": 1235,
  "current_review_count": 1247
}

Trend analysis checklist

When you review the diff, ask these questions:

Did the rating move by more than 0.1 stars? Small moves are often noise; larger moves indicate a real perception shift.
Did the review count jump faster than usual? A sudden spike may follow a marketing campaign, a PR issue, or a policy change.
Are negative keywords clustering in recent sample reviews? Look for repeated terms like "wait," "cold," "rude," or "closed."
Are new competitors appearing in the search results? A "new" record in the diff may signal market entry.
Are review counts flat for a long period? Stagnation can mean a listing is losing visibility or the business is inactive.

Comparison: build, repository, or managed service

Approach	Best for	Maintenance burden	Data ownership	Cost model
Build from scratch	Teams with dedicated scraping engineers	High: parsers, proxies, anti-bot	Full	Infrastructure + labor
Repository workflow (data-scrape/scrape-google-maps)	Developers who want inspectable code and local storage	Medium: updates come from the repo	Full	Your own infrastructure
Managed scraper API	Teams that need scale without operations	Low: provider handles extraction	Provider-dependent	Provider pricing (verify on official site)

The repository workflow is the middle path. You keep ownership of the data and the logic, but you do not start from a blank page.

Use cases

Competitor monitoring for multi-location brands
Track rating and review velocity for every competitor location in a city. Identify which neighborhoods have improving or deteriorating service quality.

Francheise quality assurance
Compare review trends across your own franchise locations. A location that falls behind the network average may need operational attention.

Local SEO research
Understand which businesses in a category have the most review momentum. Review count and recency are signals that influence local search visibility.

Market entry analysis
Before opening a new location, collect review trends for incumbent businesses. A saturated market with declining ratings is different from a growing market with rising ratings.

Compliance and maintenance notes

Public data only. This workflow collects information that is publicly visible on Google Maps listings. Do not attempt to access private accounts, restricted pages, or non-public data.
Respect terms and robots directives. Google Maps has terms of service and technical protections. Run requests at a reasonable pace and review the latest terms before deploying.
No CAPTCHA evasion. If the site presents a challenge, stop and reduce frequency or switch to an officially supported data source.
Keep scraper code current. The data-scrape/scrape-google-maps repository may update selectors and parsing logic. Pull updates before production runs.
Store provenance. Always keep the raw output alongside normalized snapshots so you can trace a trend back to its source.
Privacy. Remove or anonymize personal information from review samples before sharing reports internally or externally.

FAQ

Is there an official Google Maps API for this?
Yes. The Google Places API returns business details, ratings, and review counts for supported place IDs. It has quota and pricing limits that vary by project. Use it when you need guaranteed uptime and structured fields. Use the repository workflow when you want full control over storage and scheduling.

What fields are returned by the scraper?
The repository typically returns name, address, rating, review count, category, phone, website, hours, and a sample of review text. Check the current README for the exact field list because it changes as the repository evolves.

How often should I run the workflow?
For most local-market research, daily or weekly snapshots are enough. Running more frequently increases the chance of hitting rate limits without adding much analytical value.

What happens when Google Maps changes its layout?
The repository maintainers update the parsing logic. Pull the latest version, run a test extraction, and compare the output fields to your previous snapshots before scheduling a production run.

Can I combine this with other local data sources?
Yes. The data-scrape/yellow-pages-scraper repository is useful for adding category, contact, or business-age context to your Google Maps records. Match records by name and address rather than assuming a shared identifier.

How do I alert when a competitor's rating drops?
Add a threshold check to the diff_snapshots function. For example, emit a notification when rating_delta <= -0.2 or when review_delta exceeds a baseline. Send the alert to email, Slack, or a queue.

Can this run in the cloud on a schedule?
Yes. You can run the script from a cron job, GitHub Actions, or a small cloud VM. Keep the scraper repository as a dependency and pin to a specific commit for reproducibility.

CTA and related repositories

If you want a maintainable, code-first way to track local business review trends, start with the open-source Google Maps scraper for Python from the data-scrape profile. Combine it with the Yellow Pages scraper when you need broader local business context, and adapt the diffing logic to your own alerting rules.

Make.com Web Scraping: How to Build a No-Code Data Pipeline with CoreClaw

coreclaw — Thu, 17 Sep 2026 01:04:37 +0000

You can build a reliable web scraping pipeline in Make.com by calling a managed web data API from the HTTP module and mapping the response into the apps you already use. This keeps the no-code workflow free from proxy management, browser automation, and anti-bot maintenance, because the data collection layer lives in a service that handles those concerns for you.

In this guide, I will show a practical Make.com scenario that fetches structured public web data from CoreClaw and routes it into Google Sheets, Airtable, HubSpot, or Slack. You will see the exact HTTP module configuration, a local Python script you can run to preview the payload shape, output fields, a production checklist, and a short FAQ for the most common setup questions.

TL;DR

Make.com does not include a native "scrape any URL" module, but its HTTP > Make a request module can call any REST API.
Point the HTTP module at the CoreClaw endpoint you want to use, pass the public URL or keyword as a query parameter, and map the JSON response into downstream modules.
This works for Google SERP, Google Maps, Amazon products, Instagram posts, YouTube channels, and other CoreClaw ready-made scrapers.
A local Python preview script lets you test the payload structure and field mapping before you wire it into Make.com.
Use Make.com's error handling, retries, and scheduling to turn a single API call into a recurring data pipeline.

Why Make.com needs an external web data API

Make.com is excellent at connecting applications and running scheduled logic, but it is not a scraping platform. If you try to extract structured data directly with the HTTP module from a site that uses JavaScript rendering, bot detection, or regional blocking, you will spend most of your time fighting:

Dynamic content: Static HTTP responses miss data rendered by the browser.
Blocking: Repeated requests from the same IP trigger rate limits or captchas.
Schema drift: When a site changes its HTML, your regex or CSS selector stops working.
Maintenance overhead: You end up managing proxies, headers, retries, and parsers instead of building the workflow.

A managed web data API solves these problems by returning pre-parsed JSON. Your Make.com scenario only needs to know how to authenticate, pass the right input, and route the structured output.

What the CoreClaw + Make.com workflow looks like

The architecture is simple:

Trigger: Make.com schedule, webhook, or manual run.
HTTP request: Make.com calls the CoreClaw API with the target URL or search parameters.
Parse response: Make.com converts the JSON response into bundles.
Route data: Map fields into Google Sheets, Airtable, HubSpot, Pipedrive, Slack, email, or any other Make.com module.

You do not host a scraper, manage proxies, or parse HTML. The CoreClaw API returns structured records you can treat like any other SaaS integration.

Required inputs and where to get them

Before building the scenario, collect three values:

Value	Where to find it	Make.com field
API endpoint	Copy the endpoint for the scraper you want from the CoreClaw console or product page	URL in HTTP module
API key	Generated inside your CoreClaw account	Header `x-api-key` or query parameter
Input parameter	The public URL, keyword, place ID, ASIN, or handle you want to process	Query parameter or JSON body

Store the API key in Make.com's Connections or use a module-level variable so it is not hard-coded in every scenario.

A local Python preview of the API call

Run this script on your machine to preview the request Make.com will send and the response shape you will map. Replace the endpoint, key, and input with your own values from the CoreClaw console.

import os
import json
import requests

ENDPOINT = os.environ.get("CORECLAW_ENDPOINT", "https://api.coreclaw.com/v1/scrape")
API_KEY = os.environ.get("CORECLAW_API_KEY", "your_api_key_here")
TARGET_URL = os.environ.get("CORECLAW_TARGET_URL", "https://example.com/public-page")

headers = {
    "x-api-key": API_KEY,
    "Content-Type": "application/json",
}

payload = {
    "url": TARGET_URL,
    "render": True,
    "format": "json",
}

resp = requests.post(ENDPOINT, headers=headers, json=payload, timeout=120)
resp.raise_for_status()
data = resp.json()

print(json.dumps(data, indent=2, ensure_ascii=False))

The output will look similar to this example:

{
  "job_id": "job_abc123",
  "status": "completed",
  "url": "https://example.com/public-page",
  "scraped_at": "2026-09-17T09:00:00Z",
  "result": {
    "title": "Example Public Page",
    "headings": ["Feature A", "Feature B"],
    "links": ["https://example.com/a", "https://example.com/b"],
    "paragraphs": ["A short public description."]
  }
}

When you move to Make.com, you will send the same JSON body and map result.title, result.headings, and the other fields into downstream modules. The Python script is only a local preview; the live pipeline runs entirely inside Make.com.

How to configure the HTTP module in Make.com

Add an HTTP > Make a request module.
Set Method to POST.
Set URL to your CoreClaw endpoint, for example the endpoint copied from the CoreClaw product store.
Add a header: x-api-key with your API key.
Set Body type to Raw and Content-Type to application/json.
Enter the JSON body. Use Make.com variables for dynamic inputs:

{
  "url": "{{1.query_url}}",
  "render": true,
  "format": "json"
}

Run the module once. Make.com will show the response bundles.
Add a module for your destination, such as Google Sheets > Add a row, and map fields from the response bundles.

Mapping the response to Google Sheets

After the HTTP module returns a bundle, add Google Sheets > Add a row. Map the columns like this:

Google Sheets column	Make.com field reference	Example value
A - Scraped URL	`result.url`	`https://example.com/public-page`
B - Title	`result.title`	`Example Public Page`
C - Headings	`result.headings` joined by newline	`Feature A\nFeature B`
D - Collected at	`scraped_at`	`2026-09-17T09:00:00Z`

Use Make.com's Iterator if the response contains an array of records, such as multiple search results or product listings.

Output checklist before going live

Use this checklist to confirm the scenario is ready for production:

[ ] API key is stored in a variable or connection, not pasted into every module.
[ ] Endpoint URL matches the CoreClaw scraper you want to use.
[ ] Input parameters are passed as variables, not hard-coded values.
[ ] Response bundles contain the expected fields.
[ ] Iterator is added when the API returns an array.
[ ] Error handling is configured on the HTTP module.
[ ] Retry logic handles transient failures.
[ ] Scheduling interval matches your data freshness needs.
[ ] Compliance notes for the target site are documented.

Build vs. no-code comparison

Approach	Best for	Setup time	Maintenance	Flexibility
Self-hosted Python scraper	Full control, custom parsing	Days to weeks	High	Very high
CoreClaw + Make.com	Teams that need scheduled, routed data	Hours	Low	Medium
CoreClaw + Python/Ruby/Node.js	Engineers building custom pipelines	Hours	Low	High
Browser extension + manual export	One-off research	Minutes	Very low	Low

For ops teams, marketing agencies, and sales teams that want recurring data without a developer, CoreClaw plus Make.com is usually the fastest path to production.

Business use cases

Lead enrichment: Pass a Google Maps place or LinkedIn URL into Make.com, receive structured business data, and append it to HubSpot or Pipedrive.
Price monitoring: Schedule an Amazon ASIN scrape every morning, compare the price against a threshold, and post a Slack alert when it drops.
SERP tracking: Run a keyword scrape weekly, write the top-ranking URLs and titles to a Google Sheet, and build a ranking dashboard.
Creator research: Scrape a YouTube channel or Instagram profile on demand and add metrics to an Airtable base for outreach.
Review monitoring: Extract public reviews from a product page, summarize them with a Make.com OpenAI module, and route the summary to email.

Limitations and compliance

A few practical limits to keep in mind:

Public web data only. Do not try to access login-protected areas, private accounts, or content behind access controls.
Respect robots.txt and the terms of service for every site you scrape.
Frequency matters. Even with a managed API, aggressive polling can strain a target site and violate its policies.
API endpoints and field schemas can evolve. Check the current response shape in the CoreClaw console before relying on specific fields.
Pricing models differ by provider. Review the current CoreClaw pricing page to understand how results are counted.
Regional results may vary. A search or listing viewed from one location can return different fields than the same query from another region.

FAQ

Can I call CoreClaw from the free Make.com plan?
Yes, as long as your Make.com scenario stays within the operation limits of your plan. The HTTP module and webhook triggers consume operations like any other module.

Do I need a CoreClaw account to use this workflow?
Yes. You will need an API key from the CoreClaw console. Keep it in a Make.com variable or connection instead of pasting it into each scenario.

What if the response contains an array of records?
Add an Iterator after the HTTP module. Make.com will create one bundle per array item, which you can then route to Sheets, CRM rows, or Slack messages.

How do I handle errors from the API?
Right-click the HTTP module, choose Add error handler, and route failures to a Slack message, email, or a retry branch. CoreClaw returns standard HTTP status codes and a JSON error body you can inspect.

Can I run this on a schedule?
Yes. Add a Schedule trigger to run hourly, daily, or weekly. For most use cases, daily or weekly is enough and keeps operation usage reasonable.

Which CoreClaw scrapers work with Make.com?
Any scraper that exposes a JSON API endpoint works. Common choices include the Google SERP scraper, Google Maps scraper, Amazon product scraper, Instagram post scraper, and YouTube channel scraper. Pick the one that matches your data source.

Do I still need proxies or a headless browser?
No. The CoreClaw API handles rendering, proxy rotation, and anti-bot logic. Your Make.com scenario only sends the request and routes the response.

Summary

Make.com becomes a much more powerful automation platform when you feed it clean, structured web data. By calling the CoreClaw API from the HTTP module, you avoid the maintenance of self-hosted scrapers and still route public web data into the tools your team already uses.

Start by picking the scraper that matches your data source from the CoreClaw product store, copy the endpoint, and test the response shape locally with the Python script above. Then wire the request into a Make.com scenario and let it run on a schedule.

If you need a pay-per-result model with no proxy management, check the current CoreClaw pricing. And for a broader overview of how CoreClaw fits into no-code and developer stacks, visit coreclaw.com.

Webhook Data Delivery: How to Build a Reliable Webhook Receiver for Web Data with Python

coreclaw — Tue, 15 Sep 2026 01:08:11 +0000

Most web data platforms that run scraping jobs — including CoreClaw Workers — can push results to your HTTP endpoint instead of forcing you to poll for completion. The reliable way to receive that push is to build a small webhook receiver that verifies incoming signatures, deduplicates by idempotency key, acknowledges within the timeout window, and retries failures with exponential backoff. This article shows the full pattern in runnable Python.

TL;DR

A webhook receiver is an HTTP endpoint that accepts POST requests containing structured web data results.
You need four components: signature verification, idempotency, fast acknowledgement, and a background retry queue.
Use environment variables for endpoint secrets and API keys — never hardcode them.
Test locally with a Flask or FastAPI server, then deploy behind a reverse proxy that handles TLS.
The CoreClaw Workers platform lets you deploy scrapers that can deliver results via webhook, and the CoreClaw store offers ready-made Workers you can configure without writing a scraper from scratch.

Why Polling Falls Short

When you run scraping jobs at scale — whether you are monitoring Amazon prices, tracking Google SERP positions, or collecting Google Maps business data — you eventually hit a choice: poll the API for job status, or receive a push notification when results are ready.

Polling works for small volumes. But it creates three problems as you scale:

Wasted requests. If jobs average 45 seconds and you poll every 5 seconds, you issue 9 requests that return "still processing" for every 1 that returns results. At 500 jobs per day, that is 4,500 wasted API calls.
Latency floor. You only learn about results on your next poll cycle. If you poll every 30 seconds, your downstream pipeline has a built-in 15-second average delay — even if the job finished immediately.
Rate-limit pressure. Polling consumes your API quota for status checks, leaving less headroom for actual data requests.

Webhook delivery flips the model: the scraping platform pushes results to your endpoint the moment they are ready. No wasted requests, no latency floor, no quota pressure from status checks.

What a Webhook Receiver Must Do

A production webhook receiver is not just @app.post("/webhook") and a print(request.json). It must handle four concerns:

Concern	Why It Matters	Failure Mode If Ignored
Signature verification	Confirm the payload came from the platform, not an attacker	Spoofed data injected into your pipeline
Idempotency	The platform may retry delivery if your server is slow to respond	Duplicate records in your database
Fast acknowledgement	Most platforms time out within 5–10 seconds	Legitimate deliveries marked as failed
Retry / dead-letter queue	Transient database or downstream failures should not lose data	Data lost on the first error

The Architecture

Scraping Platform
       │
       ▼
  POST /webhook/data  (TLS terminated by reverse proxy)
       │
       ▼
  ┌─────────────────────┐
  │ Signature verifier   │  ← HMAC-SHA256 using shared secret
  └──────────┬──────────┘
             ▼
  ┌─────────────────────┐
  │ Idempotency check    │  ← Redis or DB: have we seen this event_id?
  └──────────┬──────────┘
             ▼
  ┌─────────────────────┐
  │ 202 Accepted         │  ← Respond immediately, before processing
  └──────────┬──────────┘
             ▼
  ┌─────────────────────┐
  │ Background worker    │  ← Process payload, write to DB/storage
  └──────────┬──────────┘
             ▼
         Success? 
         ├─ Yes → mark event processed
         └─ No  → enqueue retry with backoff
                    └─ after N retries → dead-letter queue

The key insight: acknowledge before processing. Your HTTP handler should verify the signature, check idempotency, enqueue the payload, and return 202 — all in under one second. The actual data processing happens in a background worker.

Runnable Python Implementation

This example uses FastAPI for the webhook endpoint and an in-process background task for processing. In production, replace the in-process queue with Redis, Celery, or a message broker.

"""
Webhook receiver for web data delivery.

Environment variables:
  WEBHOOK_SECRET  - shared secret for HMAC signature verification
  WEBHOOK_PORT    - port to listen on (default 8000)

The scraping platform should be configured to send POST requests
to https://your-domain.com/webhook/data with:
  - Header: X-Signature: sha256=<hex-hmac>
  - Body: JSON payload containing job results
  - Header: X-Event-Id: unique event identifier (for idempotency)
"""
import os
import hmac
import hashlib
import json
import asyncio
import logging
from collections import OrderedDict
from typing import Dict, Any

from fastapi import FastAPI, Request, HTTPException, BackgroundTasks
from fastapi.responses import JSONResponse

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
logger = logging.getLogger(__name__)

WEBHOOK_SECRET = os.environ.get("WEBHOOK_SECRET", "dev-secret-change-me")
WEBHOOK_PORT = int(os.environ.get("WEBHOOK_PORT", "8000"))

app = FastAPI(title="Web Data Webhook Receiver")

# --- LRU cache for idempotency (use Redis in production) ---
class LRUCache:
    """Simple ordered dict-based LRU cache for seen event IDs."""
    def __init__(self, capacity: int = 10000):
        self.capacity = capacity
        self._cache: OrderedDict[str, bool] = OrderedDict()

    def seen(self, key: str) -> bool:
        if key in self._cache:
            self._cache.move_to_end(key)
            return True
        return False

    def mark(self, key: str) -> None:
        self._cache[key] = True
        self._cache.move_to_end(key)
        if len(self._cache) > self.capacity:
            self._cache.popitem(last=False)

seen_events = LRUCache(capacity=10000)

# --- Retry queue (use Celery/RabbitMQ in production) ---
retry_queue: list[dict] = []
MAX_RETRIES = 5
BASE_BACKOFF_SECONDS = 2

def verify_signature(payload: bytes, signature_header: str, secret: str) -> bool:
    """
    Verify HMAC-SHA256 signature.

    The platform sends: X-Signature: sha256=<hex>
    This function recomputes the HMAC and compares in constant time.
    """
    if not signature_header.startswith("sha256="):
        return False
    expected = signature_header.removeprefix("sha256=")
    computed = hmac.new(
        secret.encode("utf-8"),
        payload,
        hashlib.sha256,
    ).hexdigest()
    return hmac.compare_digest(expected, computed)


async def process_payload(event_id: str, payload: dict) -> None:
    """
    Background task: process the webhook payload.

    In production, this writes to your database, triggers downstream
    pipelines, or forwards to a message queue. Here we simulate work.
    """
    try:
        logger.info(f"Processing event {event_id}, keys: {list(payload.keys())}")

        # --- Your business logic here ---
        # Example: store results, trigger enrichment, notify Slack
        records = payload.get("results", [])
        logger.info(f"Received {len(records)} records in event {event_id}")

        # Simulate processing time
        await asyncio.sleep(0.5)
        logger.info(f"Successfully processed event {event_id}")

    except Exception as e:
        logger.error(f"Failed to process event {event_id}: {e}")
        retry_queue.append({
            "event_id": event_id,
            "payload": payload,
            "attempt": 1,
        })


async def process_retries() -> None:
    """Background loop: retry failed payloads with exponential backoff."""
    while True:
        if retry_queue:
            item = retry_queue.pop(0)
            attempt = item["attempt"]
            if attempt > MAX_RETRIES:
                logger.error(
                    f"Event {item['event_id']} exceeded max retries — sending to dead-letter"
                )
                # In production: write to a dead-letter table or queue
                continue

            backoff = BASE_BACKOFF_SECONDS * (2 ** (attempt - 1))
            logger.info(f"Retrying event {item['event_id']} in {backoff}s (attempt {attempt})")
            await asyncio.sleep(backoff)

            try:
                await process_payload(item["event_id"], item["payload"])
            except Exception as e:
                logger.error(f"Retry {attempt} failed for {item['event_id']}: {e}")
                item["attempt"] += 1
                retry_queue.append(item)
        else:
            await asyncio.sleep(1)


@app.on_event("startup")
async def startup_event():
    asyncio.create_task(process_retries())


@app.post("/webhook/data")
async def receive_webhook(
    request: Request,
    background_tasks: BackgroundTasks,
):
    """
    Main webhook endpoint.

    Verifies signature, checks idempotency, enqueues background
    processing, and returns 202 within ~100ms.
    """
    raw_body = await request.body()

    # 1. Verify signature
    signature = request.headers.get("x-signature", "")
    if not verify_signature(raw_body, signature, WEBHOOK_SECRET):
        logger.warning("Invalid signature — rejecting")
        raise HTTPException(status_code=401, detail="Invalid signature")

    # 2. Parse JSON
    try:
        payload = json.loads(raw_body)
    except json.JSONDecodeError:
        raise HTTPException(status_code=400, detail="Invalid JSON")

    # 3. Idempotency check
    event_id = request.headers.get("x-event-id", "")
    if not event_id:
        # Fall back to a payload hash if no event ID is provided
        event_id = hashlib.sha256(raw_body).hexdigest()[:16]

    if seen_events.seen(event_id):
        logger.info(f"Duplicate event {event_id} — acknowledging without processing")
        return JSONResponse(status_code=202, content={"status": "duplicate", "event_id": event_id})

    seen_events.mark(event_id)

    # 4. Enqueue background processing
    background_tasks.add_task(process_payload, event_id, payload)

    # 5. Acknowledge immediately
    return JSONResponse(
        status_code=202,
        content={"status": "accepted", "event_id": event_id},
    )


@app.get("/health")
async def health():
    return {"status": "ok", "queue_size": len(retry_queue)}


if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=WEBHOOK_PORT)

Expected Payload Shape

The exact payload depends on the platform you use. A typical structured web data webhook looks like this:

{
  "job_id": "job_a1b2c3d4",
  "source": "google_maps_scraper",
  "status": "completed",
  "completed_at": "2026-09-15T01:05:00Z",
  "results": [
    {
      "name": "Example Business",
      "address": "123 Main St, Anytown, USA",
      "phone": "+1-555-0100",
      "rating": 4.5,
      "reviews_count": 127,
      "category": "Restaurant"
    }
  ],
  "result_count": 1,
  "metadata": {
    "query": "restaurants in Anytown",
    "pages_processed": 1
  }
}

Your receiver should not assume a fixed schema across all data sources. A Google Maps job returns business records; an Amazon job returns product records; a SERP job returns ranking entries. Use the source field to route payloads to the correct handler.

Testing Your Receiver Locally

You can test without a live platform by sending mock requests:

"""
Send a test webhook to your local receiver.
"""
import os
import hmac
import hashlib
import json
import requests

WEBHOOK_SECRET = os.environ.get("WEBHOOK_SECRET", "dev-secret-change-me")
WEBHOOK_URL = os.environ.get("WEBHOOK_URL", "http://localhost:8000/webhook/data")

payload = {
    "job_id": "test_job_001",
    "source": "google_maps_scraper",
    "status": "completed",
    "results": [
        {"name": "Test Business", "rating": 4.0, "reviews_count": 50}
    ],
    "result_count": 1,
}

body = json.dumps(payload).encode("utf-8")
signature = hmac.new(
    WEBHOOK_SECRET.encode("utf-8"),
    body,
    hashlib.sha256,
).hexdigest()

response = requests.post(
    WEBHOOK_URL,
    data=body,
    headers={
        "Content-Type": "application/json",
        "X-Signature": f"sha256={signature}",
        "X-Event-Id": "evt_test_001",
    },
    timeout=10,
)

print(f"Status: {response.status_code}")
print(f"Response: {response.json()}")

Business Use Cases

Use Case	What the Webhook Delivers	Downstream Action
Price monitoring	Amazon product price + stock for tracked ASINs	Trigger repricing engine, alert if drop > 5%
SERP rank tracking	Position + URL for tracked keywords	Update dashboard, notify SEO team on position changes
Lead enrichment	Google Maps business records matching criteria	Push to CRM, trigger sales outreach sequence
Creator analytics	YouTube/TikTok engagement metrics	Update influencer database, flag viral content
Competitor monitoring	Social media post data for competitor accounts	Feed to sentiment analysis pipeline

Limitations and Compliance

Endpoint availability is your responsibility. If your server is down, the platform may retry a limited number of times before dropping the delivery. Use a managed queue or serverless function for high availability.
Signature schemes vary. The HMAC-SHA256 pattern shown here is common, but some platforms use a different header name or algorithm. Always check the platform's documentation for the exact verification method.
Idempotency requires durable storage. The LRU cache in the example loses state on restart. In production, use Redis or a database table to track processed event IDs.
Rate limits apply. A platform that can deliver 1,000 webhooks per second will overwhelm an under-provisioned receiver. Use a queue between the HTTP handler and your database to absorb bursts.
TLS is mandatory. Webhooks carry structured data over the public internet. Your endpoint must use HTTPS. Most platforms refuse to deliver to plain HTTP.
Respect data source terms. Web data delivered via webhook is still subject to the target website's terms of service, robots directives, and applicable privacy law. Receiving data via push does not change your compliance obligations. For a broader discussion of public web data compliance, consult the CoreClaw pricing page to understand pay-per-result models, and review the terms of any data source you monitor.

Self-Hosted vs Managed Webhook Delivery

Dimension	Self-Hosted Polling	Managed Webhook Delivery
Latency	Bounded by poll interval	Near-immediate on job completion
API quota	Consumed by status checks	Used only for data requests
Infrastructure	Client-side scheduler + retry logic	Receiver endpoint + queue
Failure modes	Missed polls, scheduler crashes	Endpoint down, signature mismatch
Best for	Low-volume, infrequent jobs	High-volume, time-sensitive pipelines
Scaling	Linear with poll frequency	Horizontal: add receiver replicas

Checklist Before Production

[ ] Endpoint is behind HTTPS (TLS terminated by reverse proxy or load balancer)
[ ] Signature verification uses hmac.compare_digest (constant-time comparison)
[ ] Idempotency store is durable (Redis or database, not in-memory)
[ ] HTTP handler returns 202 within 2 seconds of receiving the request
[ ] Background processing has a retry mechanism with exponential backoff
[ ] Dead-letter queue or table captures payloads that exceed max retries
[ ] Monitoring alerts on queue depth and retry failure rate
[ ] Health check endpoint exists and is monitored
[ ] Secrets are loaded from environment variables, not hardcoded

FAQ

1. Do I need a webhook receiver if I only run a few jobs per day?

No. If you run fewer than 20–30 jobs per day and latency is not critical, polling is simpler and perfectly adequate. Webhook receivers become worthwhile when volume, speed, or API quota pressure makes polling expensive.

2. What happens if my server is down when the platform sends a webhook?

Most platforms retry delivery after a delay (commonly 1, 5, and 30 minutes). If your server remains down after the final retry, the delivery is dropped. A durable queue (Redis, SQS, or similar) ensures you do not lose data during outages — the platform retries until your receiver is back.

3. How do I verify that the webhook came from the platform and not an attacker?

Use HMAC signature verification. The platform computes HMAC-SHA256(payload, shared_secret) and sends the result in a header. Your receiver recomputes the HMAC and compares it in constant time using hmac.compare_digest. An attacker without the shared secret cannot produce a valid signature.

4. Can I use the same receiver for multiple data sources?

Yes. Route by the source field in the payload or by a custom header. Each source can have its own handler function. Just make sure your idempotency key includes the source identifier if event IDs are only unique within a single source.

5. How do I deploy a scraper that delivers results to my webhook?

On the CoreClaw Workers platform, you can deploy a custom Worker or choose a ready-made one from the store. Configure the webhook URL and shared secret in the Worker settings. The platform handles job execution and pushes results to your endpoint when the job completes.

6. Should I store raw webhook payloads or only processed results?

Store both, at least temporarily. Raw payloads serve as an audit trail and let you reprocess if your handler logic changes. A common pattern: write the raw payload to an append-only log (S3, GCS, or a database JSON column), then process it into structured tables.

7. What is the difference between a webhook and an API callback?

They are often used interchangeably. A webhook is an HTTP callback: the platform makes an HTTP POST to your URL when an event occurs. Some platforms also support long-polling or Server-Sent Events (SSE), but webhooks are the most common push-delivery mechanism because they work across firewalls and do not require a persistent connection.

Summary

Webhook delivery eliminates the latency, waste, and quota pressure of polling — but only if your receiver is built to handle the realities of network-delivered data. The four pillars are signature verification, idempotency, fast acknowledgement, and a durable retry queue. Start with the FastAPI implementation above, test locally with mock payloads, and move the queue and idempotency store to Redis before production traffic.

If you want to deploy a scraper that pushes results to your endpoint, explore the CoreClaw console for custom deployments or browse the CoreClaw store for ready-made Workers. You can check pay-per-result pricing to estimate delivery costs before you scale.

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.

DEV Community: coreclaw

Google Maps Lead Scraper: How to Build a Local B2B Outreach List with CoreClaw

Google Maps Lead Scraper: How to Build a Local B2B Outreach List with CoreClaw

TL;DR

Why Manual Local Lead Research Fails

What a Google Maps Lead Scraper Returns

A Runnable Python Lead Qualification Workflow

Representative Output and Checklist

Build vs Buy: Where CoreClaw Fits

Business Use Cases

Limitations and Compliance

FAQ

Next Steps

Etsy Price Tracker: How to Monitor Listing Price Changes with Python

Etsy Price Tracker: How to Monitor Listing Price Changes with Python

TL;DR

Why manual Etsy price tracking breaks down

What the repository-led workflow looks like

Step 1 — Set up the environment

Step 2 — Run a collection pass and snapshot the output

Step 3 — Diff consecutive snapshots

Representative output

Comparison: build vs. spreadsheet vs. managed service

Use cases this supports

Limits, compliance, and maintenance

FAQ

Next step

Google Maps Scraper GitHub: How to Monitor Local Business Review Trends with Python

Google Maps Scraper GitHub: How to Monitor Local Business Review Trends with Python

TL;DR

Why review trends matter for local market research

The limits of manual tracking and one-off exports

What the data-scrape Google Maps scraper provides

Workflow design

Environment setup

Runnable Python workflow

Representative output

Trend analysis checklist

Comparison: build, repository, or managed service

Use cases

Compliance and maintenance notes

FAQ

CTA and related repositories

Make.com Web Scraping: How to Build a No-Code Data Pipeline with CoreClaw

TL;DR

Why Make.com needs an external web data API

What the CoreClaw + Make.com workflow looks like

Required inputs and where to get them

A local Python preview of the API call

How to configure the HTTP module in Make.com

Mapping the response to Google Sheets

Output checklist before going live

Build vs. no-code comparison

Business use cases

Limitations and compliance

FAQ

Summary

Webhook Data Delivery: How to Build a Reliable Webhook Receiver for Web Data with Python

TL;DR

Why Polling Falls Short

What a Webhook Receiver Must Do

The Architecture

Runnable Python Implementation

Expected Payload Shape

Testing Your Receiver Locally

Business Use Cases

Limitations and Compliance

Self-Hosted vs Managed Webhook Delivery

Checklist Before Production

FAQ

Summary

Related Reading

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