AI Data Collection Workflow: From Pages to a Usable Dataset

1. Define what one record represents

An AI data collection workflow should produce records you can trace back to their source and check before use. Start by deciding what one row means: a help article, a product specification or a reviewed question and answer. Mixing those units in one text column makes later checks harder. A successful download only tells you that content arrived; you still need to establish what it contains.

This example uses invented help pages. Each page becomes one record with a heading and body text. The return policy below is fictional, not an IPMusk policy. The goal is a small, inspectable dataset, not a claim that it is ready for training or retrieval. If you have not chosen the downstream task, read the training-data versus RAG-data guide linked below first.

The proposed fields are a starting point for this example. Keep a source update date separately if you have one: the time you observed a page does not tell you when its author last corrected it. For a real collection run, capture the actual observation time with a time zone rather than copying the fixed demonstration timestamp.

2. Choose the source and reading method

Use an export or documented API when it already supplies the fields you need. For pages you own or are allowed to read, inspect a few examples before writing the extraction rule. Check whether the useful text is present in the downloaded HTML. A page that fills its body with JavaScript needs a different reading method from the static HTML used here.

Keep connection setup separate from extraction. IPMusk provides proxy connections; its documentation covers HTTP and SOCKS5 configuration. Your collection program still chooses the source, reads the response and prepares the records. A proxy connection does not extract text or establish whether a dataset is useful. The Python Requests guide covers that networking step if your project needs it.

The demonstration below makes no network requests. Its example.test URLs are source labels in synthetic input, not destinations to fetch. This lets you inspect every input and repeat the processing without an account, proxy credentials or a live website. For a real source, keep the original response and retrieval metadata so extraction mistakes can be investigated without immediately downloading it again.

3. Start with a sample that can expose mistakes

Save the following invented input as fixture.json in an empty working folder. It contains four pages: a valid returns page, a whitespace variation of that page, a page missing its heading and a different delivery page. The expected result is two retained records, one duplicate linked to its retained record and one rejection with a reason.

A sample made only of clean pages will not show whether your workflow notices missing fields. Include the page shapes you expect to encounter, then inspect the extraction against the original content. These four records test the narrow example below; they do not establish coverage of a real website or provide a recommended sample size.

[
  {"record_id":"a","source_url":"https://example.test/help/returns","html":"<h1>Returns</h1><p>Return unused items within 30 days.</p>"},
  {"record_id":"b","source_url":"https://example.test/help/returns-copy","html":"<h1> Returns </h1><p>Return unused items   within 30 days.</p>"},
  {"record_id":"c","source_url":"https://example.test/help/incomplete","html":"<p>This example has no heading.</p>"},
  {"record_id":"d","source_url":"https://example.test/help/delivery","html":"<h1>Delivery</h1><p>Delivery times depend on the destination.</p>"}
]

4. Extract fields using an explicit page rule

The script uses Python's standard-library HTMLParser callbacks to collect text inside h1 and p elements. It requires no extra packages. Each synthetic page has at most one h1 and one p. The parser is deliberately small: it does not execute JavaScript, choose the main article region or remove navigation from arbitrary websites. Adapt and check the page rule before using it on another structure.

Extraction errors should stay visible. The script rejects a record when its extracted title or text is empty, and records its input identifier. It stops on invalid JSON or missing input keys rather than silently substituting invented values. A nonempty title and paragraph are only minimum checks: a login screen or an error message could also pass them.

When adapting the workflow, compare retained text with the page you intended to collect. Look for dropped qualifications, extra navigation and missing paragraphs. Do this before collecting more pages. Preserve the input and the extraction-rule version so a future correction can be applied to the same source material.

5. Normalize text and record duplicate decisions

Save this script as prepare.py beside fixture.json. It collapses whitespace, then computes SHA-256 over a JSON pair containing the normalized title and text. Source URLs and observation times are excluded from that comparison so identical content from two source labels can be detected. The first accepted input wins; the duplicate report records which identifier it matched.

That rule is a choice for this help-page sample. Two product variants might share their wording while still needing separate records. If your task depends on product identifiers, language or region, include those distinctions in the duplicate rule or preserve them through a separate source mapping. Do not delete records merely because the same rule worked on this fixture.

The hash identifies matches under the chosen normalization rule. It does not detect paraphrases, assess factual accuracy or establish semantic similarity. A changed sentence will normally produce a different hash. The raw fixture and duplicate mapping are retained so you can review a decision or change the rule without losing the second source.

import hashlib
import json
from html.parser import HTMLParser
from pathlib import Path


class Fields(HTMLParser):
    def __init__(self):
        super().__init__()
        self.parts = {"h1": [], "p": []}
        self.field = None

    def handle_starttag(self, tag, attrs):
        if tag in self.parts:
            self.field = tag

    def handle_endtag(self, tag):
        if tag == self.field:
            self.field = None

    def handle_data(self, data):
        if self.field:
            self.parts[self.field].append(data)


raw = json.loads(Path("fixture.json").read_text(encoding="utf-8"))
# Fixed synthetic observation time; a live fetch must record its actual time.
collected_at = "2026-11-05T10:00:00+08:00"
records, rejected, duplicates, seen = [], [], [], {}
for item in raw:
    parser = Fields()
    parser.feed(item["html"])
    parser.close()
    title, text = [
        " ".join(" ".join(parser.parts[tag]).split()) for tag in ("h1", "p")
    ]
    if not title or not text:
        rejected.append({"record_id": item["record_id"], "reason": "missing title or text"})
        continue
    payload = json.dumps([title, text], ensure_ascii=False).encode("utf-8")
    digest = hashlib.sha256(payload).hexdigest()
    if digest in seen:
        duplicates.append({"record_id": item["record_id"], "kept_id": seen[digest]})
        continue
    seen[digest] = item["record_id"]
    records.append({
        "record_id": item["record_id"], "source_url": item["source_url"],
        "collected_at": collected_at, "title": title, "text": text,
        "content_hash": digest,
    })

report = {"input": len(raw), "kept": len(records),
          "rejected": rejected, "duplicates": duplicates}
Path("records.json").write_text(
    json.dumps(records, ensure_ascii=False, indent=2) + "\n", encoding="utf-8"
)
Path("report.json").write_text(json.dumps(report, indent=2) + "\n", encoding="utf-8")
print(json.dumps(report))

6. Save the records and reconcile the report

Run the commands below from that folder with Python 3.12. The first checks Python syntax; the second creates records.json and report.json, replacing files with those names in the working folder. These commands were run locally on the synthetic fixture. They do not test fetching, a real proxy connection or a live data source.

python3 -I -m py_compile prepare.py
python3 -I prepare.py

The observed report is input=4 and kept=2, with c rejected for missing title or text and b linked to retained record a. Retained records are a (Returns) and d (Delivery). Both retain their source_url and the fixed synthetic collected_at value. Four inputs reconcile as two kept plus one rejected plus one duplicate.

Review the sample before expanding collection

Open records.json and compare both entries with the fixture. Confirm the returns text still includes its 30-day condition and the delivery text still qualifies timing by destination. Check the rejected and duplicate identifiers against the original input. Counts help reveal lost records, but correct counts alone cannot establish correct extraction.

Keep the output, raw sample, processing report and script together as one reviewable run. JSON here stores each complete array or report as a single document. For a larger job, choose storage and batching around your actual volume and update process; do not add infrastructure before the small workflow produces records you can explain.

Before expansion, write down what remains unchecked: other page layouts, source freshness, coverage and suitability for the AI task. Fix the first extraction problem you find, rerun the sample and inspect the changed output. This is the handoff point to downstream data review and evaluation, not evidence of improved model answers.

Technical references

• Python 3.12 documentation: HTMLParser callbacks and parsing limits, docs.python.org/3.12/library/html.parser.html
• Python 3.12 documentation: SHA-256 and hexadecimal digests, docs.python.org/3.12/library/hashlib.html
• Python 3.12 documentation: JSON serialization and ensure_ascii, docs.python.org/3.12/library/json.html

Official references checked November 5, 2026. Input pages, policies, source labels and observation time are synthetic. The local result verifies this fixture and processing rule only.