Add Data Quality Gates to a Web Crawler
Check page evidence and record fields before publishing crawler output, isolate invalid rows, and distinguish duplicates, conflicts, and incomplete batches.
An HTTP 200 response and an existing export do not establish useful data. The response may be a login page, a price selector may collect shipping, or a category may unexpectedly yield no products.
This tutorial adds a quality gate between an existing collector and its downstream storage. You will produce accepted records, quarantined records, duplicate evidence, and a batch report. The example covers page failures as well as missing fields and incorrect types.
Define the collector's input contract
Each page package contains a source URL, HTTP status, page classification, and extracted records. The collector must classify pages from actual target-specific evidence: a product list, an explicit empty result, login, a challenge, or an unknown page.
Failure to find a product selector must not automatically mean a genuine empty list. Require an explicit empty-state marker or an equivalent, interpretable API response.
The page_kind field below is an input contract. The quality gate does not magically recognize every website. Put that classification in the collector's parsing stage, and use unknown when evidence is insufficient rather than defaulting to products.
Run a complete teaching example
Use Python 3.12 or later with no additional packages. Create quality_gate.py and add the following code. The page packages are supplied teaching inputs; no external website is contacted.
import json
from urllib.parse import urlsplit
base = {"sku": "sku-1", "name": "红茶 500g", "price_minor": 1290,
"currency": "CNY", "source_url": "https://example.16yun.cn/p/1"}
pages = [
{"url": "https://example.16yun.cn/list/1", "status": 200,
"page_kind": "products", "records": [base, {**base, "sku": "sku-2", "name": ""}]},
{"url": "https://example.16yun.cn/list/2", "status": 200,
"page_kind": "products", "records": [
base, {**base, "sku": "sku-3", "price_minor": -1},
{**base, "sku": "sku-4", "price_minor": "12.90"}]},
{"url": "https://example.16yun.cn/list/3", "status": 200,
"page_kind": "challenge", "records": []},
{"url": "https://example.16yun.cn/list/4", "status": 200,
"page_kind": "products", "records": []},
]
def reasons_for(row):
reasons = []
for field in ("sku", "name"):
if not isinstance(row.get(field), str) or not row[field].strip():
reasons.append(f"missing_or_invalid_{field}")
amount = row.get("price_minor")
if type(amount) is not int or amount < 0:
reasons.append("invalid_price_minor")
if row.get("currency") not in {"CNY", "USD", "GBP"}:
reasons.append("unsupported_currency")
source = row.get("source_url")
if not isinstance(source, str):
reasons.append("missing_source")
elif urlsplit(source).scheme not in {"http", "https"} or not urlsplit(source).hostname:
reasons.append("invalid_source_url")
return reasons
accepted, rejected, duplicates, page_errors = [], [], [], []
seen = {}
for page in pages:
if page["status"] != 200 or page["page_kind"] not in {"products", "empty"}:
page_errors.append({"url": page["url"], "reason": "unexpected_page"})
continue
if page["page_kind"] == "empty" and page["records"]:
page_errors.append({"url": page["url"], "reason": "inconsistent_empty_page"})
continue
if page["page_kind"] == "products" and not page["records"]:
page_errors.append({"url": page["url"], "reason": "unexplained_empty_result"})
continue
for row in page["records"]:
reasons = reasons_for(row)
if reasons:
rejected.append({"page": page["url"], "record": row, "reasons": reasons})
continue
key = (row["sku"], row["currency"])
if key in seen:
duplicates.append({
"key": list(key), "record": row,
"conflict": row != seen[key],
})
continue
seen[key] = row
accepted.append(row)
report = {
"accepted": len(accepted), "rejected": len(rejected),
"duplicates": len(duplicates), "page_errors": page_errors,
"publishable": not rejected and not page_errors
and not any(item["conflict"] for item in duplicates),
}
for filename, data in [
("accepted.json", accepted), ("rejected.json", rejected),
("duplicates.json", duplicates), ("quality-report.json", report),
]:
with open(filename, "w", encoding="utf-8") as stream:
json.dump(data, stream, ensure_ascii=False, indent=2)
print(json.dumps(report, ensure_ascii=False, indent=2))Run:
python quality_gate.pyFrom these inputs, expect one accepted product, three rejected records, and one duplicate. The third page is a challenge; the fourth has an unexplained empty product result. Both appear in page errors, so publishable is false. One good record does not justify replacing a complete historical snapshot.
Separate structural, business, and batch checks
Structural checks detect changes such as an integer minor-unit amount becoming the decimal string "12.90". The exact integer type check also avoids accepting a Python boolean as money.
Business checks describe the task: this one accepts zero prices, rejects negative prices, and supports a specified currency set. These are not universal financial rules. Establish the business requirement before changing them.
Batch checks detect conditions that individual rows cannot express: missing page processing, unexplained empty lists, and conflicting records for one key.
The example uses conservative full-snapshot publication. A streaming system may publish accepted partitions, but must disclose partial completion instead of presenting an incomplete directory as complete.
Preserve duplicate evidence
Two lists may legitimately advertise the same SKU. Identical records can be deduplicated; different prices require investigation into market, variant, source, or observation time.
The code saves duplicates and flags conflicts instead of applying last-write-wins. The last network request to finish does not necessarily contain the newest price. Selecting a preferred offer needs an explicit timestamp or source policy.
Likewise, the demonstrated key is task-specific. Regional work needs a market dimension; variant work usually needs a SKU rather than a parent product ID.
Integrate the gate into an existing collector
Build a page package after downloading and parsing, run the gate, then write accepted data to staging. Page failure, record quarantine, and publication are separate states.
In real collection, keep a reference to the original response or an appropriately redacted page snapshot. Attach it to page and record errors. This teaching input is embedded, but a production error message alone rarely explains a later layout change.
Repair according to the reported stage. A challenge points to access conditions. An unexplained empty result points to list selectors or loading. Invalid minor units point to price parsing. Recollect the affected pages, then reevaluate the combined batch.
Understand what the report cannot prove
A semantically wrong field can still have the correct type and range. A shipping amount may pass as a plausible nonnegative product price. Compare representative records with their source fields, or retain additional field-location evidence.
The gate also cannot know about a second page that discovery never found. Complete coverage needs expected-page information, pagination termination evidence, or other upstream checks.
Historical volume comparisons need explicit denominators. A reduction from one thousand records to eight hundred could reflect removals or missing pages; a percentage threshold alone cannot distinguish them or justify deletion.
Troubleshoot the gate and its integration
Invalid data still reaches downstream storage: check whether publication actually reads publishable and whether any writer bypasses staging. Rules that do not participate in publication cannot protect the dataset.
Valid empty lists are rejected: add a source-specific, explicit empty classification. The example allows empty pages with no records. Do not reclassify every zero-row page automatically.
One product has conflicting values: compare variant, market, currency, and observation time. Add missing identity dimensions before choosing a record.
A corrected field is still reported: establish that the corrected data belongs to the current batch and that old failure files were not mixed in. Bind real reports to a batch ID.
Only some pages failed: retain a targeted recollection list. After filling gaps, assess the combined batch, not just the latest successful subset.
A useful quality gate explains where every page and record went, what needs repair, and whether the data can replace an existing result.
Need an enterprise proxy plan?
We can tailor architecture to your target domains, concurrency, and reliability goals.