How to build a supplier intelligence dataset with Python

Sourcing software is only as good as its supplier data, and supplier data goes stale. Companies move, change what they make and close. A list somebody built by hand once is a snapshot with no way back to where each fact came from. If your product sells supplier search or discovery, the refresh matters as much as the first build.

This post builds a supplier intelligence dataset with Python: twelve suppliers of compostable and recyclable food packaging, found by queries aimed at the UK and Europe, each with a name, website, offering and location, and the source page for every field. Spidra Search finds the supplier websites, Map locates the pages worth reading, and a batch scrape with one schema extracts a record from each page.

Then we check every value against its page, store the result in SQLite so you can search it, and show how to refresh it.

It's a dated snapshot of one query set, not a ranking, and not an endorsement of anyone in it.

The code below comes from small modules: config.py, discover.py, select_suppliers.py, pages.py, extract.py, merge.py, ground.py, store.py and pipeline.py. Put it in one file, in the order it appears, with these imports at the top, and every example runs. The calls to Spidra need your API key.

import re
import sqlite3
import time
import unicodedata
from collections import Counter
from dataclasses import dataclass, field
from urllib.parse import urlsplit

import pycountry
import tldextract

Decide what a supplier record is before you search

The deliverable is a table you can search, so fix its columns first. Each supplier gets a name, a website, an offering, an address as written, a city, a region and a country, plus product categories. Every value also carries the page it came from.

Some columns are missing on purpose. Certifications, capacity, prices and minimum orders change often, matter a great deal to a buyer, and are easy to get wrong. A named person's email address is personal data you don't need. If your product needs a certification, store what the page claims, with the quote, and confirm it with the body that issues it before anyone relies on it. That check belongs to a person, not a scraper.

The schema asks for each field as text, and as null when the page doesn't say:

FIELDS = {
    "company_name": "The company that owns this website",
    "offering": "What the company makes or sells, in one sentence, in the page's own words",
    "address_text": "The postal address exactly as written on the page",
    "city": "City, as written",
    "region": "Region, state or county, as written",
    "country": "Country, as written",
}

EXTRA = {
    "page_type": {
        "type": "string",
        "description": "supplier_own_site, directory_or_marketplace, article_or_listicle or other",
    },
    "product_categories": {
        "type": "array",
        "items": {"type": "string"},
        "description": "Up to 6 product types the page names",
    },
}

The page type and the categories sit beside the text fields, and the schema is assembled from the three:

SCHEMA = {
    "type": "object",
    "properties": {
        **{n: {"type": ["string", "null"], "description": d} for n, d in FIELDS.items()},
        **EXTRA,
    },
}

A field the page doesn't show comes back as null, never a guess. The prompt says the same in words, and tells the extractor not to work out a country from a language or a phone number:

PROMPT = (
    "Describe the company that owns this website, using only what this page says. "
    "If the page does not say, use null. Do not work out a country from a language, "
    "a domain name or a phone number."
)

Find candidates with a query set, not a query

One query returns ten or twenty results, and they come from the same few sites. A query set varies the product term and the country, so different suppliers surface:

QUERIES = [
    ("q1", "compostable food packaging manufacturer", "gb"),
    ("q2", "paper takeaway cups and lids wholesale supplier", "gb"),
    ("q3", "recyclable food trays and containers supplier", "gb"),
    ("q4", "compostable food packaging manufacturer", "de"),
    ("q5", "compostable food packaging manufacturer", "nl"),
    ("q6", "compostable food packaging manufacturer", "fr"),
    ("q7", "compostable food packaging supplier", "ie"),
    ("q8", "moulded fibre food packaging manufacturer", "eu"),
]

All eight queries are in English, so suppliers whose sites exist only in another language are underrepresented. That's a limit of this query set, and adding queries in other languages widens it.

The country option leans results toward a country. It's a hint, not a filter, so expect some results from elsewhere. exclude_domains is enforced, and we use it to drop marketplaces and directories, which aren't suppliers' own sites:

EXCLUDE = [
    "alibaba.com", "made-in-china.com", "amazon.com", "amazon.co.uk", "ebay.com", "ebay.co.uk",
    "linkedin.com", "facebook.com", "instagram.com", "pinterest.com", "youtube.com",
    "wikipedia.org", "reddit.com", "thomasnet.com", "europages.com",
]

def as_row(query_id, country, result):
    return {"query_id": query_id, "country": country, "url": result.url,
            "title": result.title, "position": result.position}


def search_rows(client, queries=QUERIES, limit=20):
    """One row per web result, tagged with the query that found it."""
    rows = []
    for query_id, text, country in queries:
        found = client.search(
            text, sources=["web"], limit=limit, country=country, exclude_domains=EXCLUDE
        )
        rows += [as_row(query_id, country, r) for r in found.data.web]
    return rows

Search costs 1 credit per 10 results returned. Our eight queries returned 119 results and cost 12 credits. Four of them returned only 10 results although we asked for 20, so what you pay depends on what comes back, not on limit. None of the results came from an excluded domain, so that filter did its job.

The results overlap, which is the point of a query set: 119 results came from 80 domains, and one of them, enviropack.org.uk, turned up in six of the eight queries.

One supplier, one domain

Search returns pages, not suppliers. A supplier can appear three times, once with a blog post, and a directory can look like a supplier. So we reduce the results to one candidate per domain.

The obvious tool is a public suffix list, and it has a trap. Two unrelated shops on the same platform share a suffix, so a plain lookup merges them into one supplier. Telling it to treat the platform's domains as suffixes keeps them apart:

_EXTRACT = tldextract.TLDExtract(
    suffix_list_urls=(), cache_dir=None, include_psl_private_domains=True
)

def registrable_domain(url):
    """The domain a supplier owns. Subdomains are dropped, but shared platforms stay apart."""
    host = urlsplit(url).hostname or ""
    return _EXTRACT(host).top_domain_under_public_suffix or host.removeprefix("www.")

plain = tldextract.TLDExtract(suffix_list_urls=(), cache_dir=None)
for n in ("one", "two"):
    url = f"https://example-store-{n}.myshopify.com/"
    print(plain(url).top_domain_under_public_suffix, "|", registrable_domain(url))

Output:

myshopify.com | example-store-one.myshopify.com
myshopify.com | example-store-two.myshopify.com

Next we flag results that are not a supplier's own page: titles with words like "best", "guide" or "report", or that start with a number, paths such as /blog/ or /factsheet, and government, university and EU domains. A flag doesn't remove a result.

LISTICLE = re.compile(
    r"\b(best|top \d+|top ten|\d+ (best|top)|list of|directory|compared?|guide|factsheet|report|"
    r"trends|archives|buyer's)\b|^\d+ |^\[pdf\]",
    re.I,
)
ARTICLE_PATH = re.compile(
    r"/(blogs?|news|articles?|guides?|insights|press|magazine|stories|resources|hub|files|"
    r"document|industry-analysis|tag|media-news|partnering-opportunities|end-of-life|members)"
    r"(/|$)|/(factsheet|what-is|top-\d+|how-it-works)|aboutevent",
    re.I,
)
INSTITUTION = re.compile(r"(\.(gov|edu|ac)(\.[a-z]{2})?|^europa\.eu)$")

def flags_for(title, url):
    """Why a result is probably not a supplier's own page: article, listicle, institution."""
    flags = ["listicle"] if LISTICLE.search(title or "") else []
    flags += ["article_path"] if ARTICLE_PATH.search(urlsplit(url).path) else []
    return flags + (["institution"] if INSTITUTION.search(registrable_domain(url)) else [])

checks = [
    ("10 best compostable packaging suppliers", "https://a.test/"),
    ("Compostable Food Packaging: UK Business Guide", "https://a.test/guide"),
    ("Plastics pots, tubs and trays | County Council", "https://county.gov.uk/waste"),
    ("Northfold Pack | Compostable boxes", "https://a.test/"),
]
for title, url in checks:
    print(f"{flags_for(title, url)!s:30s} {title}")

Output:

['listicle']                   10 best compostable packaging suppliers
['listicle', 'article_path']   Compostable Food Packaging: UK Business Guide
['institution']                Plastics pots, tubs and trays | County Council
[]                             Northfold Pack | Compostable boxes

This is the second version of the rule. The first caught only titles like "best" and "top 10" and paths like /blog/. On the real results it flagged 15 of the 80 domains, and let through a market research report, a trade association's member list, a county council's recycling page and a trade-fair article. Reading the unflagged results is how we found what to add. The widened rule flags 36.

collect keeps the best position and the queries that found each domain, and one rule that matters: a domain is flagged only if every result for it is. A supplier with a blog post is still a supplier.

@dataclass
class Candidate:
    domain: str
    url: str
    title: str
    queries: list = field(default_factory=list)
    country_hint: str = ""
    best_position: int = 99
    flags: list = field(default_factory=list)

def collect(rows):
    """rows: dicts with query_id, country, url, title, position. -> one Candidate per domain."""
    found = {}
    for r in rows:
        domain, flags = registrable_domain(r["url"]), flags_for(r["title"], r["url"])
        c = found.setdefault(domain, Candidate(domain, r["url"], r["title"], flags=flags))
        c.flags = sorted(set(c.flags) & set(flags))  # flagged only if every result for it is
        c.queries = sorted({*c.queries, r["query_id"]})
        if r["position"] < c.best_position:
            c.url, c.title = r["url"], r["title"]
            c.best_position, c.country_hint = r["position"], r["country"]
    return sorted(found.values(), key=lambda c: (c.best_position, c.domain))

These rows are made up, on reserved .test domains:

def hit(query, url, title, position):
    return {"query_id": query, "country": "gb", "url": url, "title": title, "position": position}


rows = [
    hit("q1", "https://www.northfold-pack.test/", "Northfold Pack | Compostable boxes", 1),
    hit("q1", "https://directory.example.test/best", "10 best compostable suppliers", 2),
    hit("q2", "https://northfold-pack.test/cups", "Paper cups | Northfold Pack", 3),
    hit("q2", "https://lindau.test/", "Verpackung Lindau GmbH", 2),
    hit("q2", "https://lindau.test/blog/trends", "Packaging trends", 5),
]
for c in collect(rows):
    print(f"{c.domain:24s} {','.join(c.queries):6s} best {c.best_position} {c.flags}")

Output:

northfold-pack.test      q1,q2  best 1 []
directory.example.test   q1     best 2 ['listicle']
lindau.test              q2     best 2 []

Run over the real results, collect gave 80 domains: 36 flagged and 44 not.

Select suppliers with a rule you can re-run

Twelve suppliers is a sample you can verify by hand. The rule: never a flagged domain, domains found by more queries first, and no more than four from one country hint, so the sample isn't all one market.

SELECTION_SIZE = 12
MAX_PER_COUNTRY_HINT = 4

def rank(c):
    return (-len(c.queries), c.best_position, c.domain)


def propose_selection(candidates, size=SELECTION_SIZE, per_country=MAX_PER_COUNTRY_HINT):
    """-> [(candidate, reason)]. Flagged candidates (listicles, articles) are never proposed."""
    taken, chosen = Counter(), []
    for c in sorted((c for c in candidates if not c.flags), key=rank):
        if len(chosen) < size and taken[c.country_hint] < per_country:
            taken[c.country_hint] += 1
            chosen.append((c, f"{len(c.queries)} queries, best position {c.best_position}"))
    return chosen

for candidate, reason in propose_selection(collect(rows), size=2):
    print(candidate.domain, "|", reason)

Output:

northfold-pack.test | 2 queries, best position 1
lindau.test | 1 queries, best position 2

A rule is a starting point, not a verdict. The first version of it, with the first version of the flags, did not propose twelve suppliers. It proposed a market research publisher, a training company's guide and a magazine article among them, because ranking by the number of queries that found a domain rewards content that ranks for everything. That's why the flags above are wider.

With the wider flags, the rule proposed twelve domains, and we changed two by hand, for reasons visible in the search results. Country hints are soft: none of our queries asked for India or Melbourne, and a result naming each still came back.

The final twelve, each with the search result that found it:

One row needs a word. Search found bioleaderpack.com through a "Top 10" article on its own site. We don't extract from that article: Map finds the site's pages, and we read its home page and a contact page.

This is the output of one query set on one day. It's not a ranking, and a place in it says nothing about quality. Whether each one really supplies food packaging, and where it is, is what the next steps find out.

Map each site and pick two pages

Map lists a site's pages without reading them. It costs 1 credit, and a site where it finds nothing costs nothing. With search set, the best matches come first, and respect_robots=True leaves out URLs the site's robots.txt disallows. We want two pages per supplier: the home page, and the page most likely to hold an address.

Map also returns translated copies of pages (/de/, /fr/), which would double the pages we pay for. drop_translations removes a copy when the original exists:

MAP_SEARCH = "contact about us impressum"

LANG = re.compile(r"^/([a-z]{2})(/|$)")
CONTACT = re.compile(
    r"contact|kontakt|contacto|contatt|impressum|imprint|mentions-l[eé]gales|colophon", re.I
)
ABOUT = re.compile(
    r"about|over-ons|ueber|uber-uns|über|a-propos|qui-sommes|company|who-we-are|our-story", re.I
)

def drop_translations(urls):
    """Drop /de/... copies when the same page exists without the language prefix."""
    paths = {urlsplit(u).path or "/" for u in urls}

    def is_copy(url):
        path = urlsplit(url).path or "/"
        return bool(LANG.match(path)) and (path[3:] or "/") in paths

    return [u for u in urls if not is_copy(u)]

def pick_pages(home_url, links, max_pages=2):
    """-> [(url, role)]: the home page, then the best-ranked contact page, or else about page."""
    domain = registrable_domain(home_url)
    origin = "{0.scheme}://{0.netloc}/".format(urlsplit(home_url))
    own = [u for u in links if registrable_domain(u) == domain and u.rstrip("/") != origin[:-1]]
    urls = drop_translations(own)
    for role, pattern in (("contact", CONTACT), ("about", ABOUT)):
        match = next((u for u in urls if pattern.search(urlsplit(u).path)), None)
        if match:
            return [(origin, "home"), (match, role)][:max_pages]
    return [(origin, "home")]

links = [
    "https://acme-fibre.test/about",
    "https://acme-fibre.test/de/about",
    "https://acme-fibre.test/de/kontakt",
    "https://acme-fibre.test/products",
    "https://other-site.test/contact",
]
for url, role in pick_pages("https://acme-fibre.test/", links):
    print(f"{role:8s} {url}")

Output:

home     https://acme-fibre.test/
contact  https://acme-fibre.test/de/kontakt

An Impressum page counts as a contact page, because German sites usually publish their address there.

Extract with one schema, one page at a time

A scrape merges up to three URLs into one answer, which loses which page said what. A batch gives every URL its own result and job ID, up to 50 per request. That's the property provenance needs:

def run_batch(client, urls, timeout=600, **options):
    """-> (response, schema warnings). Starts the batch, then polls until every item is done."""
    queued = client.start_batch_scrape(urls, **options)
    deadline = time.monotonic() + timeout
    while time.monotonic() < deadline:
        response = client.get_batch_scrape(queued.batch_id)
        if response.completed_count + response.failed_count >= response.total_urls:
            return response, queued.schema_warnings
        time.sleep(5)
    raise TimeoutError(f"batch {queued.batch_id} did not finish in {timeout} s")

@dataclass
class PageResult:
    url: str
    status: str
    content: "dict | str | None"
    error: "str | None"
    credits: int
    job_id: "str | None"

def page_results(response):
    """A batch is 'completed' even when items failed, so every item is read on its own."""
    return [
        PageResult(i.url, i.status, i.result, i.error, i.credits_used or 0, i.job_id)
        for i in response.items
    ]

def extract_pages(client, urls):
    """Up to 50 URLs per batch. -> (page results, schema warnings)."""
    results, warnings = [], []
    for start in range(0, len(urls), 50):
        response, found = run_batch(client, urls[start : start + 50], prompt=PROMPT, schema=SCHEMA)
        results += page_results(response)
        warnings += found or []
    return results, warnings

Three behaviours matter. A batch is completed even when items failed, so every item is read on its own. Failed pages are free, but a page that loads a cookie wall can come back nearly empty and still be charged. The API can retry a failed batch, but the Python client doesn't expose that, so you re-submit the failed URLs yourself.

We set aside any page whose result isn't an object or whose page_type isn't one we asked for, and keep the reason:

PAGE_TYPES = {"supplier_own_site", "directory_or_marketplace", "article_or_listicle", "other"}


def check_content(content):
    """-> (content, None) for a usable result, or (None, the reason the page is set aside)."""
    if not isinstance(content, dict):
        return None, f"content is a {type(content).__name__}, not an object"
    if content.get("page_type") not in PAGE_TYPES:
        return None, f"page_type is {content.get('page_type')!r}"
    return content, None

def page_entry(url, role, r):
    """One page: what was asked, what came back, and why it was set aside, if it was."""
    if r is None or r.status != "completed":
        content, problem = None, (r.error if r else "no result")
    else:
        content, problem = check_content(r.content)
    sent = {"url": url, "role": role, "content": content, "problem": problem}
    got = {"job_id": r.job_id, "status": r.status, "credits": r.credits} if r else {}
    return {"job_id": None, "status": "missing", "credits": 0} | sent | got

def build_pages(plan, results):
    by_url = {r.url.rstrip("/"): r for r in results}
    return {
        domain: [page_entry(url, role, by_url.get(url.rstrip("/"))) for url, role in picks]
        for domain, picks in plan.items()
    }

Merge pages into one record, with a source for every field

A supplier has two pages, and they won't always agree. Each field is taken from the page most likely to have it: the contact page for an address, the home page for a name and offering. The page it came from stays with the value, and a disagreement is recorded, not hidden:

FIELDS = ("company_name", "offering", "address_text", "city", "region", "country")
ROLE_ORDER = {
    "company_name": ("home", "about", "contact"),
    "offering": ("home", "about", "contact"),
}
DEFAULT_ORDER = ("contact", "about", "home")  # an address is most likely on the contact page

def squash(value):
    return " ".join(str(value).split()).casefold()


def candidates_for(name, pages):
    """Every non-empty value of one field, best page first: [(value, page)]."""
    order = ROLE_ORDER.get(name, DEFAULT_ORDER)
    ranked = sorted(pages, key=lambda p: order.index(p["role"]) if p["role"] in order else 9)
    return [(p["content"][name], p) for p in ranked if p["content"].get(name)]

def fact_of(seen):
    """The best value of a field and the page it came from, or None when no page has it."""
    if not seen:
        return None
    value, page = seen[0]
    return {"value": value, "source_url": page["url"], "job_id": page["job_id"]}


def categories_of(pages):
    """Product categories from every page, one spelling each, the first seen."""
    unique = {}
    for page in pages:
        for category in page["content"].get("product_categories") or []:
            unique.setdefault(squash(category), category)
    return list(unique.values())

def merge_supplier(domain, pages):
    """pages: dicts with url, role, job_id and content. -> one record with a source per field."""
    record = {"domain": domain, "fields": {}, "conflicts": [], "categories": categories_of(pages)}
    for name in FIELDS:
        seen = candidates_for(name, pages)
        record["fields"][name] = fact_of(seen)
        if len({squash(v) for v, _ in seen}) > 1:
            record["conflicts"].append({"field": name, "values": [(v, p["url"]) for v, p in seen]})
    return record

pages = [
    {"url": "https://northfold-pack.test/", "role": "home", "job_id": "job-1",
     "content": {"company_name": "Northfold Pack Ltd", "offering": "Compostable boxes and cups"}},
    {"url": "https://northfold-pack.test/contact", "role": "contact", "job_id": "job-2",
     "content": {"company_name": "Northfold Pack Limited", "city": "Leeds", "country": "England"}},
]
record = merge_supplier("northfold-pack.test", pages)
for name, fact in record["fields"].items():
    print(f"{name:13s}", fact and (fact["value"], fact["source_url"].rsplit("/", 1)[1] or "home"))
print("conflicts:", [c["field"] for c in record["conflicts"]])

Output:

company_name  ('Northfold Pack Ltd', 'home')
offering      ('Compostable boxes and cups', 'home')
address_text  None
city          ('Leeds', 'contact')
region        None
country       ('England', 'contact')
conflicts: ['company_name']

Countries need care. Sites write UK, England and Deutschland, and the ISO lookup knows none of them. We resolve a small table of aliases and everything the lookup knows, and leave the rest unresolved, with the text kept as written:

COUNTRY_ALIASES = {
    "uk": "GB", "united kingdom": "GB", "great britain": "GB", "england": "GB", "scotland": "GB",
    "wales": "GB", "northern ireland": "GB", "deutschland": "DE", "the netherlands": "NL",
    "holland": "NL", "nederland": "NL", "republic of ireland": "IE", "éire": "IE",
    "österreich": "AT", "schweiz": "CH", "españa": "ES", "italia": "IT", "usa": "US",
}

def resolve_country(text):
    """-> an ISO code, or None when the text is not a country we can name for sure."""
    key = " ".join((text or "").split()).casefold()
    if key in COUNTRY_ALIASES:
        return COUNTRY_ALIASES[key]
    try:
        return pycountry.countries.lookup(key).alpha_2
    except LookupError:
        return None

for text in ("UK", "England", "Deutschland", "Germany", "Narnia", "Europe"):
    print(f"{text!r:14s} -> {resolve_country(text)}")

Output:

'UK'           -> GB
'England'      -> GB
'Deutschland'  -> DE
'Germany'      -> DE
'Narnia'       -> None
'Europe'       -> None

Check every value against its page

An extractor can paraphrase, misread or fill a gap. So for every extracted value we ask whether it's on the page it came from. A second batch fetches the same pages as markdown, at 1 credit each, and check_value compares:

def page_texts(client, urls):
    """The same pages as markdown (1 credit each), to check every value against its page."""
    response, _ = run_batch(client, urls, output="markdown", extract_content_only=True)
    return {i.url: i.result for i in response.items if isinstance(i.result, str)}

def norm(text):
    """Case, spacing and no-break spaces should not decide whether a value is on the page."""
    text = unicodedata.normalize("NFKC", str(text)).casefold().replace("\u2019", "'")
    return " ".join(text.split())

def check_value(value, page_text):
    """'found' (the value is on the page), 'partial' (most of its words are), or 'not_found'."""
    if value is None:
        return "empty"
    if not page_text:
        return "no_text"
    wanted, text = norm(value), norm(page_text)
    if wanted in text:
        return "found"
    words = [w for w in re.findall(r"\w+", wanted) if len(w) >= 4]
    return "partial" if words and sum(w in text for w in words) / len(words) >= 0.8 else "not_found"

def ground_record(record, texts):
    """Adds a grounding verdict to every field that has a value. texts: page url -> page text."""
    index = {url.rstrip("/"): text for url, text in texts.items()}
    for name, fact in record["fields"].items():
        if fact:
            fact["grounding"] = check_value(
                fact["value"], index.get(fact["source_url"].rstrip("/"))
            )
    return record

texts = {
    "https://northfold-pack.test/": "Northfold Pack Ltd. Compostable boxes and cups.",
    "https://northfold-pack.test/contact": "Contact. Northfold Pack Limited, Leeds, UK.",
}
for name, fact in ground_record(record, texts)["fields"].items():
    print(f"{name:13s}", fact and fact["grounding"])

Output:

company_name  found
offering      found
address_text  None
city          found
region        None
country       not_found

A value is found, partial (most of its words are on the page) or not_found. In the example, the extractor said England and the page says UK, so the check flags it. What it can't catch is a wrong value that really is on the page, such as a different office's address. That needs a person.

So we split the checking. Every not_found and partial value goes to a person, plus a random sample of the rest. worksheet_rows builds that list, with the doubtful values first, the source link for each value, and blank columns for the verdict and notes:

VERDICT_ORDER = {"not_found": 0, "partial": 1, "no_text": 2, "found": 3}


def worksheet_rows(records):
    """One row per value, with the page to open and blank columns for a person to fill in."""
    rows = [
        {"domain": r["domain"], "field": name, "value": f["value"], "source_url": f["source_url"],
         "job_id": f["job_id"], "grounding": f["grounding"], "verified": "", "notes": ""}
        for r in records
        for name, f in r["fields"].items()
        if f
    ]
    return sorted(
        rows, key=lambda x: (VERDICT_ORDER.get(x["grounding"], 9), x["domain"], x["field"])
    )

for row in worksheet_rows([ground_record(record, texts)])[:3]:
    print(row["grounding"], row["field"], repr(row["value"]), row["source_url"])

Output:

not_found country 'England' https://northfold-pack.test/contact
found city 'Leeds' https://northfold-pack.test/contact
found company_name 'Northfold Pack Ltd' https://northfold-pack.test/

Store it so you can search it

SQLite is enough for a dataset this size, and its built-in full-text index (FTS5) gives ranked search with no extra service. Three tables hold the facts and their sources, and one index holds the searchable text:

SCHEMA_SQL = """
CREATE TABLE IF NOT EXISTS suppliers (
    domain TEXT PRIMARY KEY, name TEXT, offering TEXT, categories TEXT,
    address_text TEXT, city TEXT, region TEXT, country_text TEXT, country_code TEXT,
    first_seen TEXT, last_seen TEXT);
CREATE TABLE IF NOT EXISTS source_pages (
    url TEXT PRIMARY KEY, domain TEXT, role TEXT, job_id TEXT, fetched_at TEXT, status TEXT);
CREATE TABLE IF NOT EXISTS facts (
    domain TEXT, field TEXT, value TEXT, source_url TEXT, grounding TEXT,
    PRIMARY KEY (domain, field));
CREATE VIRTUAL TABLE IF NOT EXISTS supplier_search USING fts5(
    domain UNINDEXED, name, offering, categories, city, region,
    tokenize='porter unicode61 remove_diacritics 2');
"""

COLUMNS = (
    "domain",
    "name",
    "offering",
    "categories",
    "address_text",
    "city",
    "region",
    "country_text",
    "country_code",
)
PAGE_COLUMNS = ("url", "role", "job_id", "status")
FACT_COLUMNS = ("value", "source_url", "grounding")
INDEXED = ("name", "offering", "categories", "city", "region")

def open_db(path=":memory:"):
    db = sqlite3.connect(path)
    db.row_factory = sqlite3.Row
    db.executescript(SCHEMA_SQL)
    return db


def pick(source, keys):
    return {key: source.get(key) for key in keys}

def put(db, table, values):
    """Insert or replace one row, whatever columns it has."""
    marks = ", ".join(":" + name for name in values)
    db.execute(f"INSERT OR REPLACE INTO {table} ({', '.join(values)}) VALUES ({marks})", values)

def save_supplier(db, row, record, pages, now):
    """Upsert one supplier, its pages and one fact per field. first_seen survives a refresh."""
    domain = row["domain"]
    old = db.execute("SELECT first_seen FROM suppliers WHERE domain = ?", (domain,)).fetchone()
    seen = {"first_seen": old[0] if old else now, "last_seen": now}
    put(db, "suppliers", dict.fromkeys(COLUMNS) | row | seen)
    for page in pages:
        put(db, "source_pages", pick(page, PAGE_COLUMNS) | {"domain": domain, "fetched_at": now})
    for name, fact in record["fields"].items():
        fact and put(db, "facts", pick(fact, FACT_COLUMNS) | {"domain": domain, "field": name})
    db.execute("DELETE FROM supplier_search WHERE domain = ?", (domain,))
    put(db, "supplier_search", pick(row, INDEXED) | {"domain": domain})
    db.commit()

first_seen survives a refresh and last_seen moves with it. The search function turns each word into a prefix match for words of three letters or more, and ranks by relevance:

def search(db, text, country=None, limit=5):
    """Ranked full-text search over name, offering, categories and place. Optional country code."""
    words = re.findall(r"\w+", text)
    if not words:
        return []
    query = " ".join(f'"{w}"*' if len(w) >= 3 else f'"{w}"' for w in words)
    sql = (
        "SELECT s.domain, s.name, s.city, s.country_code, bm25(supplier_search) AS score "
        "FROM supplier_search JOIN suppliers s ON s.domain = supplier_search.domain "
        "WHERE supplier_search MATCH ? AND (? IS NULL OR s.country_code = ?) ORDER BY score LIMIT ?"
    )
    return [dict(r) for r in db.execute(sql, (query, country, country, limit))]

db = open_db()
for domain, name, offering, city, code in (
    ("northfold-pack.test", "Northfold Pack Ltd", "Compostable boxes and cups", "Leeds", "GB"),
    ("vezelwerk.test", "Vezelwerk B.V.", "Moulded fibre trays", "Utrecht", "NL"),
    ("koeln-karton.test", "Köln Karton AG", "Recyclable food boxes", "Köln", "DE"),
):
    row = {"domain": domain, "name": name, "offering": offering, "city": city, "country_code": code}
    save_supplier(db, row, {"fields": {}}, [], "2026-11-05")
for text, country in (("compost", None), ("fibre", "NL"), ("koln", None)):
    print(repr(text), country, [r["name"] for r in search(db, text, country)])

Output:

'compost' None ['Northfold Pack Ltd']
'fibre' NL ['Vezelwerk B.V.']
'koln' None ['Köln Karton AG']

If creating the index fails with a "no such module" error, your Python's SQLite was built without FTS5, and you need a build that includes it.

Keep it fresh, because the second run is the product

The reason to build this as code is the second run. Re-run the same query set, extract again, and compare by domain:

TRACKED = ("name", "offering", "address_text", "city", "region", "country_code")

def diff_runs(old, new):
    """old, new: {domain: row}. -> what was added, what disappeared, and which fields changed."""
    changed = [
        {"domain": d, "field": f, "old": old[d].get(f), "new": new[d].get(f)}
        for d in sorted(old.keys() & new.keys())
        for f in TRACKED
        if old[d].get(f) != new[d].get(f)
    ]
    return {
        "added": sorted(new.keys() - old.keys()),
        "removed": sorted(old.keys() - new.keys()),
        "changed": changed,
    }

old = {"vezelwerk.test": {"city": "Utrecht"}, "gone.test": {"city": "Leiden"}}
new = {"vezelwerk.test": {"city": "Amersfoort"}, "new.test": {"city": "Delft"}}
print(diff_runs(old, new))

{'added': ['new.test'], 'removed': ['gone.test'], 'changed': [{'domain': 'vezelwerk.test', 'field': 'city', 'old': 'Utrecht', 'new': 'Amersfoort'}]}

A refresh costs about what the first run did: 1 credit per 10 search results, 1 per Map, 2 per extracted page and 1 per page text. A supplier that stops appearing in results keeps an old last_seen, which is your cue to look at it. Monitors can watch a few important pages between runs and alert when one changes. The Free plan includes 2, so use them on the pages you'd least like to miss.

What to check before you rely on it

Check each site's terms before you collect from it, and read our guide on whether web scraping is legal. Map's respect_robots option leaves out URLs a site's robots.txt disallows, and we turn it on, but a site's terms are a separate question from its robots.txt.

Keep the dataset to company information. We collect no names, emails or phone numbers of people, and neither should you without a clear reason and a basis for holding them.

Treat everything about certifications, capacity and prices as a claim until the issuer or the supplier confirms it. And before you publish a company's name next to extracted details, have a person check them against the page, as we did. The result is a snapshot of one query set on one day.

Try it on a vertical of your own

You now have a pipeline that turns a query set into a searchable dataset where every value points to the page it came from. Create a free account at app.spidra.io, put your own vertical in QUERIES, and run search_rows to see which suppliers Search finds.

The Search, Map and batch scrape docs cover the options we skipped, and how credits work has the pricing rule. If you want the next step after this one, our post on validating scraped JSON with Pydantic shows how to hold each page result to a stricter model.