Web Scraping Yellow Pages with Python: Workflow and HTML Demo

# Windows PowerShell
python -m venv .venv
.\.venv\Scripts\python.exe -m pip install requests==2.33.1 beautifulsoup4==4.14.3

project/
  serve_pages.py
  fetch_pages.py
  yellowpages_pipeline.py
  urls.txt
  rejected_urls.txt
  html/
    approved_directory_page_1.html
    approved_directory_page_2.html
    approved_directory_page_3.html

# Terminal 1
python serve_pages.py --port 8765

# Terminal 2
python fetch_pages.py --urls urls.txt --origin http://127.0.0.1:8765 --out-dir downloaded

for card in soup.select("article.result"):
    name = field(card, ".business-name")
    category = field(card, ".categories")
    location = field(card, ".locality")
    phone_raw = field(card, ".phones")
    phone = normalize_phone(phone_raw)

python yellowpages_pipeline.py --input-dir downloaded --out-dir results

HTML files: 3 | Cards: 12
Unique businesses saved: 10
Duplicates removed: 2
Missing phone: 1

python fetch_pages.py --urls rejected_urls.txt --origin http://127.0.0.1:8765 --out-dir rejected-run

$env:CONFIRMED_DATA_PERMISSION = 'yes'
python fetch_pages.py --urls approved_urls.txt --origin https://approved.example --out-dir real-pilot --delay 2

python yellowpages_pipeline.py --input-dir real-pilot --out-dir real-pilot-results

# macOS / Linux: replace placeholders with your connection details
export HTTPS_PROXY="http://USERNAME:PASSWORD@HOST:PORT"
export CONFIRMED_DATA_PERMISSION=yes
python fetch_pages.py --urls approved_urls.txt --origin https://approved.example --out-dir external-run --use-env-proxy --delay 2

"""Start a local HTTP server for reproducible, permission-controlled pages."""
from __future__ import annotations
import argparse
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path

BASE = Path(__file__).resolve().parent / 'html'

class DirectoryHandler(BaseHTTPRequestHandler):
    def do_GET(self):
        route = self.path.split('?', 1)[0]
        if route.startswith('/catalog/page-') and route.endswith('.html'):
            number = route.removeprefix('/catalog/page-').removesuffix('.html')
            if number in {'1', '2', '3'}:
                path = BASE / f'approved_directory_page_{number}.html'
                content = path.read_bytes()
                self.send_response(200)
                self.send_header('Content-Type', 'text/html; charset=utf-8')
                self.send_header('Content-Length', str(len(content)))
                self.end_headers()
                self.wfile.write(content)
                return
        if route in {'/status/403', '/status/429', '/status/503'}:
            status = int(route.rsplit('/', 1)[-1])
            self.send_response(status)
            self.send_header('Content-Type', 'text/plain; charset=utf-8')
            self.end_headers()
            self.wfile.write(f'HTTP {status}: request not available'.encode())
            return
        self.send_error(404, 'Route not found')

    def log_message(self, fmt, *args):
        print('HTTP access: ' + fmt % args, flush=True)


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument('--port', type=int, default=8765)
    args = ap.parse_args()
    print(f'Local permitted directory: http://127.0.0.1:{args.port}/catalog/page-1.html', flush=True)
    ThreadingHTTPServer(('127.0.0.1', args.port), DirectoryHandler).serve_forever()

if __name__ == '__main__':
    main()

"""Retrieve one permitted origin; stop the batch on HTTP 403 or 429."""
from __future__ import annotations
import argparse
import csv
import hashlib
import os
import time
from datetime import datetime, timezone
from pathlib import Path
from urllib.parse import urlparse
import requests

COLUMNS = ('requested_url', 'final_url', 'status', 'content_type',
           'saved_file', 'sha256', 'retrieved_utc', 'outcome')
MAX_BYTES = 2_000_000

def fetch(urls_file: Path, out_dir: Path, origin: str,
          proxy_env: bool = False, delay: float = 1.0):
    parsed_origin = urlparse(origin)
    if (parsed_origin.scheme not in ('http', 'https') or not parsed_origin.netloc
            or parsed_origin.username or parsed_origin.path not in ('', '/')
            or parsed_origin.query or parsed_origin.fragment):
        raise ValueError('Origin must contain only scheme, host and optional port')
    if delay < 0:
        raise ValueError('Delay must be nonnegative')
    expected = (parsed_origin.scheme, parsed_origin.netloc)
    urls = [s.strip() for s in urls_file.read_text(encoding='utf-8').splitlines()
            if s.strip() and not s.lstrip().startswith('#')]
    if not urls:
        raise ValueError('No URLs supplied')
    # Validate the whole batch before sending any requests.
    for url in urls:
        parsed = urlparse(url)
        if (parsed.scheme, parsed.netloc) != expected or parsed.username:
            raise ValueError('URL outside approved origin or contains credentials')
    if out_dir.exists() and any(out_dir.iterdir()):
        raise ValueError('Output directory must be new or empty; use a fresh run directory')
    out_dir.mkdir(parents=True, exist_ok=True)
    report = []
    with requests.Session() as session:
        session.trust_env = bool(proxy_env)
        for i, url in enumerate(urls, 1):
            if i > 1:
                time.sleep(delay)
            row = dict.fromkeys(COLUMNS, '')
            row.update(requested_url=url, retrieved_utc=datetime.now(timezone.utc).isoformat(timespec='seconds'))
            try:
                with session.get(url, timeout=(5, 15), allow_redirects=False, stream=True) as response:
                    row.update(final_url=response.url, status=str(response.status_code),
                               content_type=response.headers.get('Content-Type', ''))
                    if response.status_code in (403, 429):
                        row['outcome'] = f'stopped_http_{response.status_code}'
                    elif 300 <= response.status_code < 400:
                        row['outcome'] = 'redirect_not_followed'
                    elif response.status_code == 200 and 'text/html' in row['content_type'].lower():
                        content = bytearray()
                        for chunk in response.iter_content(65536):
                            if len(content) + len(chunk) > MAX_BYTES:
                                row['outcome'] = 'skipped_size_limit'
                                break
                            content.extend(chunk)
                        else:
                            filename = f'page_{i:02d}.html'
                            (out_dir / filename).write_bytes(content)
                            row.update(saved_file=filename, sha256=hashlib.sha256(content).hexdigest(), outcome='saved')
                    else:
                        row['outcome'] = 'skipped_non_html_or_status'
            except requests.RequestException as exc:
                row['outcome'] = 'connection_error_' + type(exc).__name__
            report.append(row)
            print(f"HTTP {row['status'] or '-':>3}  {row['outcome']:<26} {url}")
            if row['status'] in ('403', '429'):
                break
    with (out_dir / 'http_manifest.csv').open('w', encoding='utf-8-sig', newline='') as f:
        writer = csv.DictWriter(f, fieldnames=COLUMNS)
        writer.writeheader()
        writer.writerows(report)
    print('Manifest:', out_dir / 'http_manifest.csv')
    print('Saved pages:', sum(r['outcome'] == 'saved' for r in report))
    return report

def main():
    p = argparse.ArgumentParser(description=__doc__)
    p.add_argument('--urls', type=Path, default=Path('urls.txt'))
    p.add_argument('--out-dir', type=Path, default=Path('downloaded'))
    p.add_argument('--origin', default='http://127.0.0.1:8765')
    p.add_argument('--use-env-proxy', action='store_true')
    p.add_argument('--delay', type=float, default=1.0)
    args = p.parse_args()
    if args.origin not in ('http://127.0.0.1:8765', 'http://localhost:8765'):
        if os.environ.get('CONFIRMED_DATA_PERMISSION') != 'yes':
            p.error('Confirm permission before setting CONFIRMED_DATA_PERMISSION=yes')
    fetch(args.urls, args.out_dir, args.origin, args.use_env_proxy, args.delay)

if __name__ == '__main__':
    main()

"""Read local business-directory HTML and create structured CSV reports.

The workflow reads HTML files already available on disk. It does not
contact a website. Only process content you are allowed to use.
"""

from __future__ import annotations

import argparse
import csv
import re
from collections import Counter
from pathlib import Path

from bs4 import BeautifulSoup


FIELDS = [
    "business_id", "name", "category", "city", "state",
    "phone_raw", "phone", "address", "listing_url", "sponsored", "source_file",
]


def field(node, css: str) -> str:
    """Return text from one matching element; otherwise an empty string."""
    element = node.select_one(css)
    if element is None:
        return ""
    return element.get_text(" ", strip=True)


def normalize_phone(text: str) -> str:
    """Format recognizable North American 10-digit phone numbers."""
    digits = re.sub(r"\D", "", text)
    if len(digits) == 11 and digits.startswith("1"):
        digits = digits[1:]
    if len(digits) != 10:
        return ""
    return f"{digits[:3]}-{digits[3:6]}-{digits[6:]}"


def extract_page(path: Path) -> list[dict[str, str]]:
    """Extract business cards from one locally saved HTML file."""
    html = path.read_text(encoding="utf-8")
    soup = BeautifulSoup(html, "html.parser")
    records = []

    for card in soup.select("article.result"):
        name_link = card.select_one("a.business-name")
        location = field(card, ".locality")
        city, separator, state = location.rpartition(",")

        record = {
            "business_id": card.get("data-business-id", "").strip(),
            "name": field(card, ".business-name"),
            "category": field(card, ".categories"),
            "city": city.strip() if separator else location.strip(),
            "state": state.strip() if separator else "",
            "phone_raw": field(card, ".phones"),
            "phone": normalize_phone(field(card, ".phones")),
            "address": field(card, ".street-address"),
            "listing_url": (
                name_link.get("href", "").strip()
                if name_link else ""
            ),
            "sponsored": (
                "Yes" if card.get("data-sponsored") == "Y"
                else "No"
            ),
            "source_file": path.name,
        }
        records.append(record)

    return records


def save_csv(path: Path, columns: list[str], records: list[dict]):
    """Write Excel-compatible UTF-8 CSV with a header row."""
    with path.open("w", encoding="utf-8-sig", newline="") as f:
        writer = csv.DictWriter(f, fieldnames=columns)
        writer.writeheader()
        writer.writerows(records)


def run(input_dir: Path, out_dir: Path) -> dict[str, int]:
    """Parse, deduplicate, summarize, and export local HTML records."""
    manifest = input_dir / "http_manifest.csv"
    if manifest.exists():
        with manifest.open(encoding="utf-8-sig", newline="") as f:
            names = [row["saved_file"] for row in csv.DictReader(f)
                     if row["outcome"] == "saved"]
        if any(Path(name).name != name for name in names):
            raise ValueError("Invalid filename in manifest")
        html_files = [input_dir / name for name in names]
    else:
        html_files = sorted(input_dir.glob("*.html"))
    if not html_files:
        raise ValueError(f"No .html files in {input_dir}")

    candidates = []
    for path in html_files:
        candidates.extend(extract_page(path))

    if not candidates:
        raise ValueError("No business cards found; inspect the HTML and selectors")

    kept = []
    excluded = []
    seen = set()

    for record in candidates:
        # Incomplete fallback identities stay separate for manual review.
        identity = tuple(" ".join(record[k].casefold().split())
                         for k in ("name", "city", "state", "address"))
        key = (("id", record["business_id"]) if record["business_id"] else
               ("address", *identity) if all(identity) else ("unresolved", len(kept)))
        if not record["name"] or not record["city"]:
            excluded.append({**record, "reason": "missing essential field"})
        elif key in seen:
            excluded.append({**record, "reason": "duplicate"})
        else:
            seen.add(key)
            kept.append(record)

    kept.sort(key=lambda r: (r["city"], r["category"], r["name"]))
    category_counts = Counter(r["category"] for r in kept)
    city_counts = Counter((r["city"], r["state"]) for r in kept)
    cross_counts = Counter(
        (r["city"], r["state"], r["category"]) for r in kept
    )

    out_dir.mkdir(parents=True, exist_ok=True)
    save_csv(out_dir / "businesses.csv", FIELDS, kept)

    category_rows = [
        {"category": name, "count": count}
        for name, count in sorted(category_counts.items())
    ]
    save_csv(
        out_dir / "categories.csv", ["category", "count"],
        category_rows,
    )

    city_rows = [
        {"city": city, "state": state, "count": count}
        for (city, state), count in sorted(city_counts.items())
    ]
    save_csv(out_dir / "cities.csv", ["city", "state", "count"], city_rows)

    cross_rows = [
        {"city": city, "state": state, "category": category, "count": count}
        for (city, state, category), count in sorted(cross_counts.items())
    ]
    save_csv(
        out_dir / "city_category.csv",
        ["city", "state", "category", "count"], cross_rows,
    )
    save_csv(out_dir / "excluded.csv", FIELDS + ["reason"], excluded)

    summary = {
        "files": len(html_files),
        "cards": len(candidates),
        "saved": len(kept),
        "duplicates": sum(
            r["reason"] == "duplicate" for r in excluded
        ),
        "missing_phone": sum(not r["phone"] for r in kept),
    }

    print(f"HTML files: {summary['files']} | Cards: {summary['cards']}")
    print(f"Unique businesses saved: {summary['saved']}")
    print(f"Duplicates removed: {summary['duplicates']}")
    print(f"Missing phone: {summary['missing_phone']}")
    print(f"Outputs: {out_dir.resolve()}")
    for category, count in sorted(category_counts.items()):
        print(f"  {category:<16} {count}")

    return summary


def main() -> None:
    parser = argparse.ArgumentParser(
        description="Authorized business directory HTML -> CSV"
    )
    parser.add_argument(
        "--input-dir", type=Path, default=Path("html")
    )
    parser.add_argument(
        "--out-dir", type=Path, default=Path("results")
    )
    args = parser.parse_args()
    run(args.input_dir, args.out_dir)


if __name__ == "__main__":
    main()

http://127.0.0.1:8765/catalog/page-1.html
http://127.0.0.1:8765/catalog/page-2.html
http://127.0.0.1:8765/catalog/page-3.html

http://127.0.0.1:8765/status/429
http://127.0.0.1:8765/catalog/page-1.html

# Terminal 1, from the project folder
python serve_pages.py --port 8765

# Terminal 2, using the project environment
python -m pip install requests==2.33.1 beautifulsoup4==4.14.3
python fetch_pages.py --urls urls.txt --origin http://127.0.0.1:8765 --out-dir downloaded
python yellowpages_pipeline.py --input-dir downloaded --out-dir results
python fetch_pages.py --urls rejected_urls.txt --origin http://127.0.0.1:8765 --out-dir rejected-run

<!doctype html>
<html lang="en">
<meta charset="utf-8">
<title>Synthetic directory sample</title>
<h1>Fictional teaching data</h1>
<article class="result" data-business-id="B01" data-sponsored="N">
<a class="business-name" href="https://example.com/business/B01">Sample Orchard Plumbing</a>
<span class="categories">Plumbers</span>
<span class="locality">San Francisco, CA</span>
<span class="phones">415-555-0101</span>
<span class="street-address">10 Demo Street</span>
</article>
<article class="result" data-business-id="B02" data-sponsored="Y">
<a class="business-name" href="https://example.com/business/B02">Sample Harbor Dental</a>
<span class="categories">Dentists</span>
<span class="locality">San Francisco, CA</span>
<span class="phones">415-555-0102</span>
<span class="street-address">20 Demo Street</span>
</article>
<article class="result" data-business-id="B03" data-sponsored="N">
<a class="business-name" href="https://example.com/business/B03">Sample Northstar Electric</a>
<span class="categories">Electricians</span>
<span class="locality">Oakland, CA</span>
<span class="phones">510-555-0103</span>
<span class="street-address">30 Demo Avenue</span>
</article>
<article class="result" data-business-id="B04" data-sponsored="N">
<a class="business-name" href="https://example.com/business/B04">Sample Bay Auto</a>
<span class="categories">Auto Repair</span>
<span class="locality">San Jose, CA</span>
<span class="phones">408-555-0104</span>
<span class="street-address">40 Demo Road</span>
</article>
</html>

<!doctype html>
<html lang="en">
<meta charset="utf-8">
<title>Synthetic directory sample</title>
<h1>Fictional teaching data</h1>
<article class="result" data-business-id="B05" data-sponsored="N">
<a class="business-name" href="https://example.com/business/B05">Sample City Plumbing</a>
<span class="categories">Plumbers</span>
<span class="locality">Oakland, CA</span>
<span class="phones">510-555-0105</span>
<span class="street-address">50 Demo Avenue</span>
</article>
<article class="result" data-business-id="B06" data-sponsored="N">
<a class="business-name" href="https://example.com/business/B06">Sample Mission Dental</a>
<span class="categories">Dentists</span>
<span class="locality">San Jose, CA</span>
<span class="phones">
</span>
<span class="street-address">60 Demo Road</span>
</article>
<article class="result" data-business-id="B07" data-sponsored="N">
<a class="business-name" href="https://example.com/business/B07">Sample Maple Auto</a>
<span class="categories">Auto Repair</span>
<span class="locality">San Francisco, CA</span>
<span class="phones">415-555-0107</span>
<span class="street-address">70 Demo Street</span>
</article>
<article class="result" data-business-id="B08" data-sponsored="N">
<a class="business-name" href="https://example.com/business/B08">Sample Westline Electric</a>
<span class="categories">Electricians</span>
<span class="locality">Oakland, CA</span>
<span class="phones">510-555-0108</span>
<span class="street-address">80 Demo Avenue</span>
</article>
</html>

<!doctype html>
<html lang="en">
<meta charset="utf-8">
<title>Synthetic directory sample</title>
<h1>Fictional teaching data</h1>
<article class="result" data-business-id="B09" data-sponsored="N">
<a class="business-name" href="https://example.com/business/B09">Sample Valley Plumbing</a>
<span class="categories">Plumbers</span>
<span class="locality">San Jose, CA</span>
<span class="phones">408-555-0109</span>
<span class="street-address">90 Demo Road</span>
</article>
<article class="result" data-business-id="B10" data-sponsored="N">
<a class="business-name" href="https://example.com/business/B10">Sample Summit Dental</a>
<span class="categories">Dentists</span>
<span class="locality">San Francisco, CA</span>
<span class="phones">415-555-0110</span>
<span class="street-address">100 Demo Street</span>
</article>
<article class="result" data-business-id="B01" data-sponsored="N">
<a class="business-name" href="https://example.com/business/B01">Sample Orchard Plumbing</a>
<span class="categories">Plumbers</span>
<span class="locality">San Francisco, CA</span>
<span class="phones">415-555-0101</span>
<span class="street-address">10 Demo Street</span>
</article>
<article class="result" data-business-id="B02" data-sponsored="Y">
<a class="business-name" href="https://example.com/business/B02">Sample Harbor Dental</a>
<span class="categories">Dentists</span>
<span class="locality">San Francisco, CA</span>
<span class="phones">415-555-0102</span>
<span class="street-address">20 Demo Street</span>
</article>
</html>