Building a Local Business Data Pipeline with Python
Local business data looks simple until you try to use it in software.
A person can open a website and understand that it belongs to a business, identify its name, recognize its location, find its services, and decide whether the information appears complete.
Software does not get that context for free.
A page may contain structured JSON LD, visible text, navigation labels, footer information, duplicate addresses, social links, service pages, and marketing language at the same time.
If we want to build a directory, search tool, research dashboard, entity graph, or internal data platform, simply scraping the visible text is not enough.
We need a pipeline that can answer several questions consistently.
What entity does this page represent?
Which attributes are factual?
Where did each value come from?
How confident are we in the extracted value?
Can two pages be recognized as belonging to the same entity?
How should multilingual content be normalized?
This article builds a practical Python architecture for solving that problem.
The goal is not to create a general purpose crawler.
The goal is to design a small but reliable local business data pipeline with explicit provenance, validation, normalization, and confidence scoring.
Start with the data model
Before writing a scraper, it helps to define what the application actually needs.
A local business record might look like this:
from dataclasses import dataclass, field
from typing import Optional
@dataclass
class BusinessRecord:
name: Optional[str] = None
city: Optional[str] = None
district: Optional[str] = None
address: Optional[str] = None
phone: Optional[str] = None
website: Optional[str] = None
services: list[str] = field(default_factory=list)
This looks reasonable, but one important thing is missing.
We do not know where each value came from.
If the address was extracted from JSON LD, that may deserve more confidence than an address guessed from a paragraph.
If the business name came from the page title but a different name appeared in structured data, we need to preserve that disagreement.
So the raw business model should not be the first object in the pipeline.
We need an intermediate representation.
Store facts before building entities
Instead of immediately assigning values to a business, I prefer storing observations.
from dataclasses import dataclass
@dataclass
class Observation:
field: str
value: str
source_url: str
source_type: str
confidence: float
Now the system can store several observations for the same field.
For example:
observations = [
Observation(
field="city",
value="Tehran",
source_url="https://example.com",
source_type="json_ld",
confidence=0.95,
),
Observation(
field="city",
value="Tehran",
source_url="https://example.com/contact",
source_type="visible_text",
confidence=0.80,
),
]
Both observations agree.
That agreement can increase confidence when the final entity is assembled.
More importantly, if they disagree, the system does not silently overwrite one value with another.
Why provenance matters
Consider a directory that says a business is located in a particular district.
Six months later someone asks:
Where did that information come from?
If the system stored only the final value, the answer may be impossible to recover.
Provenance fixes that.
Every extracted fact should be connected to its source.
For a production system, I would store at least:
The source URL
The extraction date
The extraction method
The original raw value
The normalized value
The confidence score
The parser version
A slightly richer object might look like this:
from datetime import datetime
@dataclass
class Fact:
field: str
raw_value: str
normalized_value: str
source_url: str
source_type: str
confidence: float
extracted_at: datetime
parser_version: str
This makes data changes explainable.
If the address changes later, we can compare the old and new observations.
Fetching a page responsibly
For a small research pipeline, the fetching layer can remain simple.
import requests
def fetch_page(url):
headers = {
"User-Agent": "LocalDataResearch/1.0"
}
response = requests.get(
url,
headers=headers,
timeout=15,
)
response.raise_for_status()
return response.text
In a larger crawler, I would also add caching, retry logic, request scheduling, domain level limits, robots handling, and proper error classification.
The important architectural point is that fetching should be separate from parsing.
A parser should receive HTML.
It should not care how that HTML was downloaded.
That separation makes testing much easier.
Parse structured data first
Many business websites expose structured information using JSON LD.
That is often the best place to start because the page author has already attempted to represent business information in a machine readable format.
We can extract JSON LD blocks with Beautiful Soup.
import json
from bs4 import BeautifulSoup
def extract_json_ld(html):
soup = BeautifulSoup(html, "html.parser")
blocks = []
for script in soup.select(
'script[type="application/ld+json"]'
):
content = script.string
if not content:
continue
try:
blocks.append(json.loads(content))
except json.JSONDecodeError:
continue
return blocks
Real websites can contain several JSON LD objects.
One may describe the organization.
Another may describe breadcrumbs.
Another may describe an article.
Another may contain a graph with multiple connected entities.
So extraction is only the first step.
Flatten JSON LD graphs
Schema data frequently appears inside an @graph array.
It is useful to flatten those nodes before classification.
def flatten_json_ld(blocks):
nodes = []
for block in blocks:
if isinstance(block, list):
nodes.extend(block)
continue
if not isinstance(block, dict):
continue
graph = block.get("@graph")
if isinstance(graph, list):
nodes.extend(graph)
else:
nodes.append(block)
return nodes
Now every structured object can be inspected individually.
Find candidate business entities
A page may use several Schema types for a provider.
We can define a small set of types that our application understands.
BUSINESS_TYPES = {
"LocalBusiness",
"Organization",
"MedicalBusiness",
"HealthAndBeautyBusiness",
"ProfessionalService",
}
Then search the structured nodes.
def get_schema_types(node):
schema_type = node.get("@type")
if isinstance(schema_type, str):
return {schema_type}
if isinstance(schema_type, list):
return set(schema_type)
return set()
def find_business_nodes(nodes):
candidates = []
for node in nodes:
if not isinstance(node, dict):
continue
types = get_schema_types(node)
if types.intersection(BUSINESS_TYPES):
candidates.append(node)
return candidates
This gives us candidate entities.
It does not yet prove that every candidate refers to the primary business represented by the page.
That distinction matters.
Do not trust structured data blindly
JSON LD is useful, but it is not automatically correct.
A website may contain outdated structured data.
A template may accidentally reuse information from another page.
Multiple organization entities may appear.
Fields may also be incomplete.
So instead of directly accepting a schema value, convert it into an observation.
def schema_name_observation(node, url):
name = node.get("name")
if not isinstance(name, str):
return None
return Observation(
field="name",
value=name.strip(),
source_url=url,
source_type="json_ld",
confidence=0.95,
)
The same approach can be used for address, phone number, or service information.
The parser extracts evidence.
The resolver decides what becomes the final entity.
Using a real provider page as input data
A useful pipeline should work against real websites rather than only synthetic HTML.
For example, when testing a multilingual local business parser, a public Persian provider page such as کلینیک زیبایی در تهران زعفرانیه can be treated as an input document from which the system attempts to extract observable attributes.
The parser should not infer that the provider is better than another provider.
It should only attempt to answer questions such as:
What organization name is present?
What city or district is stated?
Which services are listed?
Does the page expose structured data?
Are contact details available?
Do multiple parts of the site agree about the same attributes?
This distinction is fundamental.
Extraction collects evidence.
Ranking creates a judgment.
Those should be different layers.
Extracting visible page text
Structured data may be incomplete.
We therefore also need a controlled way to inspect visible text.
def visible_text(html):
soup = BeautifulSoup(html, "html.parser")
for tag in soup(
["script", "style", "noscript"]
):
tag.decompose()
text = soup.get_text(
separator=" ",
strip=True,
)
return text
This produces a simplified text representation.
For a production parser, I would avoid searching the entire document indiscriminately.
Navigation menus, cookie banners, related posts, and footer content can introduce noise.
Section aware extraction is usually better.
Extracting headings as semantic hints
Headings provide useful context.
def extract_headings(html):
soup = BeautifulSoup(html, "html.parser")
headings = []
for tag in soup.find_all(
["h1", "h2", "h3"]
):
text = tag.get_text(
" ",
strip=True,
)
if text:
headings.append({
"level": tag.name,
"text": text,
})
return headings
This can help identify service sections, location sections, team information, and contact information.
It also gives the pipeline more structure than a single giant block of text.
Persian text requires normalization
Multilingual extraction introduces another problem.
Two strings can look almost identical to a user but contain different Unicode characters.
Persian text commonly requires normalization of Arabic and Persian forms of certain letters.
For example, Arabic ي and Persian ی are different Unicode code points.
The same applies to Arabic ك and Persian ک.
We can normalize them.
PERSIAN_REPLACEMENTS = {
"ي": "ی",
"ى": "ی",
"ك": "ک",
"ة": "ه",
}
def normalize_persian(text):
result = text
for source, target in PERSIAN_REPLACEMENTS.items():
result = result.replace(
source,
target,
)
return result
This simple step improves matching significantly.
Normalize invisible spacing
Persian content may also contain different whitespace forms.
A normalization function can collapse standard whitespace while preserving meaningful text.
import re
def normalize_spaces(text):
text = text.replace(
"\u200c",
" ",
)
text = re.sub(
r"\s+",
" ",
text,
)
return text.strip()
Depending on the application, replacing the Persian half space may or may not be desirable.
For entity matching, aggressive normalization can be useful.
For displaying text back to users, preserving typography is usually better.
This is why I prefer storing both raw and normalized values.
Normalize Persian digits
Phone numbers and addresses may contain Persian numerals.
A matching system should understand them.
PERSIAN_DIGITS = "۰۱۲۳۴۵۶۷۸۹"
ENGLISH_DIGITS = "0123456789"
DIGIT_MAP = str.maketrans(
PERSIAN_DIGITS,
ENGLISH_DIGITS,
)
def normalize_digits(text):
return text.translate(DIGIT_MAP)
Now:
value = "۰۲۱۱۲۳۴۵۶۷۸"
print(normalize_digits(value))
produces:
02112345678
This makes phone normalization easier.
Create a general normalization pipeline
Instead of applying every transformation manually, combine them.
def normalize_text(text):
text = normalize_persian(text)
text = normalize_digits(text)
text = normalize_spaces(text)
return text
Every extracted value can now keep both forms.
raw_name = "کلینیک نمونه"
fact = Fact(
field="name",
raw_value=raw_name,
normalized_value=normalize_text(
raw_name
),
source_url="https://example.com",
source_type="visible_text",
confidence=0.75,
extracted_at=datetime.utcnow(),
parser_version="1.0",
)
This gives the resolver a stable representation for comparisons.
Extract address data from JSON LD
Addresses are often represented as nested objects.
def extract_schema_address(node):
address = node.get("address")
if not isinstance(address, dict):
return {}
return {
"street": address.get(
"streetAddress"
),
"city": address.get(
"addressLocality"
),
"region": address.get(
"addressRegion"
),
"country": address.get(
"addressCountry"
),
}
The returned fields can become separate observations.
This is preferable to storing the entire address as one string.
Cities, districts, countries, and street addresses have different semantics.
Service extraction is harder
Service names may appear in JSON LD, navigation elements, cards, headings, or internal links.
They are therefore more difficult to extract reliably.
One basic strategy is to collect internal links and classify their anchor text.
from urllib.parse import urljoin
from urllib.parse import urlparse
def internal_links(html, base_url):
soup = BeautifulSoup(html, "html.parser")
base_host = urlparse(base_url).netloc
links = []
for tag in soup.find_all(
"a",
href=True,
):
href = urljoin(
base_url,
tag["href"],
)
host = urlparse(href).netloc
if host != base_host:
continue
label = tag.get_text(
" ",
strip=True,
)
if not label:
continue
links.append({
"label": label,
"url": href,
})
return links
This does not automatically tell us which links are services.
But it gives us another set of structured observations.
Classify links instead of assuming intent
A rule based classifier can identify likely service links.
SERVICE_TERMS = {
"service",
"services",
"treatment",
"درمان",
"خدمات",
}
def looks_like_service_link(label):
normalized = normalize_text(
label
).lower()
for term in SERVICE_TERMS:
if term in normalized:
return True
return False
This rule is deliberately conservative.
A production system could use a classifier or language model, but even then I would preserve the original evidence.
Models should propose classifications.
They should not erase provenance.
Building confidence scores
Now we can assign confidence based on extraction method.
SOURCE_CONFIDENCE = {
"json_ld": 0.95,
"page_heading": 0.85,
"contact_section": 0.85,
"visible_text": 0.70,
"navigation": 0.60,
"inferred": 0.40,
}
This is an engineering heuristic.
It is not a universal truth.
The main advantage is consistency.
A system can later tune the values using evaluation data.
Agreement should increase confidence
Suppose the city appears in both JSON LD and the contact section.
That agreement should influence the final result.
def agreement_score(observations):
grouped = {}
for item in observations:
key = normalize_text(
item.value
).lower()
grouped.setdefault(
key,
[],
).append(item)
best_value = None
best_score = 0
for value, items in grouped.items():
score = sum(
item.confidence
for item in items
)
if score > best_score:
best_value = value
best_score = score
return best_value, best_score
This is still a simple resolver.
But it already performs better than taking the first value found.
Preserve disagreement
Do not discard conflicting values.
Imagine these observations:
city_observations = [
Observation(
field="city",
value="Tehran",
source_url="https://example.com",
source_type="json_ld",
confidence=0.95,
),
Observation(
field="city",
value="Karaj",
source_url="https://example.com/contact",
source_type="visible_text",
confidence=0.70,
),
]
A weak pipeline chooses one value and forgets the other.
A stronger pipeline stores the conflict.
@dataclass
class Resolution:
field: str
selected_value: str
confidence: float
alternatives: list[str]
Now the final entity can explain uncertainty.
Entity resolution comes next
Once facts are normalized, we need to decide whether several pages represent the same business.
Useful matching signals include:
Normalized business name
Domain
Phone number
Street address
City
Structured identifier
Social profile references
Phone numbers are particularly useful because they are less ambiguous than names.
Normalize phone numbers
A basic phone normalizer can remove formatting.
def normalize_phone(phone):
phone = normalize_digits(phone)
allowed = []
for char in phone:
if char.isdigit():
allowed.append(char)
return "".join(allowed)
Now visually different phone formats can be compared more reliably.
Build a simple entity similarity score
We can assign weights to matching fields.
def entity_similarity(a, b):
score = 0
if a.name and b.name:
if normalize_text(
a.name
).lower() == normalize_text(
b.name
).lower():
score += 3
if a.phone and b.phone:
if normalize_phone(
a.phone
) == normalize_phone(
b.phone
):
score += 4
if a.city and b.city:
if normalize_text(
a.city
).lower() == normalize_text(
b.city
).lower():
score += 1
if a.website and b.website:
if urlparse(
a.website
).netloc == urlparse(
b.website
).netloc:
score += 4
return score
Again, the exact weights are not the main point.
The architecture is.
Entity matching rules should be explicit and testable.
Separate facts from rankings
This is especially important for local search products.
Suppose the system successfully extracts that a business:
Operates in Tehran
Has a stated district
Publishes contact information
Lists several services
Provides physician information
Contains structured data
Those are facts about the page.
They do not prove that the business is the best provider in the city.
A ranking is a different data product.
It requires a methodology.
That methodology may include verified credentials, review quality, accessibility, service availability, data completeness, user preferences, or other factors.
The extraction pipeline should not quietly create ranking claims.
Model ranking separately
If ranking is required, create an explicit object.
@dataclass
class RankingSignal:
name: str
value: float
weight: float
explanation: str
Then the system can explain why an entity received a certain score.
This is much better than adding a field called:
best_provider
That field has no transparent meaning.
Design for multilingual location matching
Local business platforms often need to match location names written in several forms.
For example, a district may appear in Persian script on one page and Latin script in another dataset.
A location table can help.
LOCATION_ALIASES = {
"زعفرانیه": "zafaraniyeh",
"تهران": "tehran",
}
def canonical_location(value):
normalized = normalize_text(
value
).lower()
return LOCATION_ALIASES.get(
normalized,
normalized,
)
In production, this should come from a dedicated location database rather than a hard coded dictionary.
But the principle remains useful.
Normalize before matching.
Keep raw data forever when possible
Normalization is useful.
It can also destroy information.
Imagine converting every Persian location into an English identifier.
That may simplify matching but lose the original representation.
The better pattern is:
@dataclass
class NormalizedValue:
raw: str
normalized: str
The raw value remains available for display and debugging.
The normalized value is used for matching.
This pattern is useful far beyond Persian text.
Add validation rules
The next layer should validate extracted data.
For example:
def valid_name(value):
if not value:
return False
if len(value.strip()) < 2:
return False
return True
def valid_city(value):
if not value:
return False
return len(value.strip()) >= 2
Validation should remain separate from extraction.
A parser can find a string.
A validator decides whether the string is plausible.
A resolver decides whether it becomes the final value.
Keeping those stages separate makes the system easier to maintain.
Track parser versions
Websites change.
Parsers change too.
If a data record suddenly changes after a deployment, you need to know whether the website changed or your parser did.
That is why every observation should contain a parser version.
PARSER_VERSION = "1.2"
Store it with each fact.
This small field becomes extremely valuable when debugging large datasets.
Add page fingerprints
A content fingerprint can help determine whether a page has materially changed.
import hashlib
def page_fingerprint(html):
normalized = normalize_spaces(
html
)
return hashlib.sha256(
normalized.encode("utf8")
).hexdigest()
If the fingerprint is unchanged, you may not need to parse the page again.
This reduces unnecessary work.
Caching improves both speed and politeness
A crawler should not request the same unchanged page repeatedly.
A simple cache object might look like this:
@dataclass
class CachedPage:
url: str
html: str
fingerprint: str
fetched_at: datetime
Before fetching a page again, the pipeline can check whether a recent cached copy is sufficient.
For production systems, HTTP cache headers should also be respected where appropriate.
Observability matters
Once the pipeline processes many pages, failures need to be measurable.
Useful metrics include:
Fetch success rate
JSON LD detection rate
Business entity detection rate
Address extraction rate
Phone extraction rate
Conflict rate
Average confidence
Parsing errors
Validation failures
Entity merge rate
Without metrics, data quality problems can remain invisible.
Create a parser report
One useful pattern is returning a report with every parsing result.
@dataclass
class ParseReport:
url: str
observations_found: int
schema_nodes_found: int
business_nodes_found: int
warnings: list[str]
This makes debugging much easier than simply returning a BusinessRecord.
The pipeline can tell us what happened.
Testing should use stored fixtures
Avoid testing a parser only against live websites.
A page can change between test runs.
Instead, save representative HTML fixtures and run deterministic tests.
def test_extract_json_ld():
html = """
<html>
<head>
<script type="application/ld+json">
{
"@type": "LocalBusiness",
"name": "Example Business"
}
</script>
</head>
</html>
"""
blocks = extract_json_ld(
html
)
assert len(blocks) == 1
Tests like this make refactoring safer.
Test multilingual normalization too
Persian normalization deserves its own tests.
def test_persian_normalization():
original = "كلينيك زيبايي"
normalized = normalize_persian(
original
)
assert "ک" in normalized
assert "ی" in normalized
A multilingual parser should test the languages it claims to support.
Avoid turning extraction into recommendation
There is a broader lesson here.
Software often begins with a narrow task:
Extract information from a page.
Then someone asks:
Can we tell users which provider is best?
That second question is much harder.
Extraction is primarily an information retrieval problem.
Recommendation introduces judgment.
Ranking introduces methodology.
Medical or financial ranking can introduce additional responsibility.
Do not let these layers blur together.
A cleaner architecture
The final pipeline can be separated into several components.
Fetcher
Parser
Normalizer
Validator
Observation Store
Resolver
Entity Store
Ranking Layer
API
Each component has one job.
The fetcher retrieves documents.
The parser extracts candidate values.
The normalizer creates comparable representations.
The validator rejects obviously invalid values.
The observation store preserves provenance.
The resolver chooses canonical values.
The entity store represents businesses.
The ranking layer is optional and operates only on resolved data.
The API exposes the result.
This separation makes the system easier to test and easier to trust.
Why this architecture scales better
A quick scraper often starts like this:
business_name = soup.find("h1").text
That may work for one website.
It does not scale to hundreds of domains.
Real websites contain conflicting information, several languages, multiple entity types, incomplete structured data, old templates, and duplicated content.
A robust system therefore needs to preserve uncertainty instead of hiding it.
That is the main architectural idea in this article.
Do not force every page into a perfect record immediately.
Collect observations first.
Resolve them later.
Final thoughts
Building a local business data pipeline is not mainly a scraping problem.
It is an entity modeling problem.
The HTML is only the raw input.
A useful system needs to distinguish between what was observed, what was normalized, what was validated, what was inferred, and what was eventually selected as the canonical value.
That distinction becomes even more important when working with multilingual websites or sensitive industries.
The core principles are simple.
Preserve provenance.
Prefer structured data, but verify it.
Normalize text before matching.
Keep raw values.
Represent disagreement explicitly.
Assign confidence to observations.
Separate extraction from entity resolution.
Separate entity resolution from ranking.
Test with multilingual fixtures.
Track parser versions.
Measure data quality.
Once those foundations are in place, the same architecture can support directories, search engines, knowledge graphs, research tools, recommendation systems, and internal data platforms.
The difficult part is not finding text on a web page.
The difficult part is deciding what that text actually means as data.