如何使用 Python 抓取本地商家信息名称、地址、评分等字段

python --version

python -m venv listings_env
source listings_env/bin/activate

pip install crawlbase beautifulsoup4

from urllib.parse import quote_plus
from crawlbase import CrawlingAPI

api = CrawlingAPI({"token": "YOUR_CRAWLBASE_TOKEN"})

def build_url(category, city):
    terms = quote_plus(category)
    geo = quote_plus(city)
    return f"https://www.yellowpages.com/search?search_terms={terms}&geo_location_terms={geo}"

def crawl(page_url):
    options = {"country": "US"}
    response = api.get(page_url, options)
    if response["status_code"] == 200:
        return response["body"].decode("latin1")
    print(f"Request failed: {response['status_code']}")
    return None

if __name__ == "__main__":
    url = build_url("plumbers", "Austin, TX")
    html = crawl(url)
    print(html[:500] if html else "No HTML returned")

from bs4 import BeautifulSoup

def text_of(card, selector):
    el = card.select_one(selector)
    return el.get_text(strip=True) if el else None

def parse_rating(card):
    el = card.select_one("div.result-rating")
    if not el:
        return None
    words = {"one": 1, "two": 2, "three": 3, "four": 4, "five": 5}
    rating = None
    for cls in el.get("class", []):
        base = cls.replace("-half", "")
        if base in words:
            rating = words[base] + (0.5 if "-half" in cls else 0)
    return rating

def scrape_results(html):
    soup = BeautifulSoup(html, "html.parser")
    cards = soup.select("div.result")
    results = []
    for card in cards:
        try:
            website = card.select_one("a.track-visit-website")
            results.append({
                "name": text_of(card, "a.business-name"),
                "address": text_of(card, "div.street-address"),
                "locality": text_of(card, "div.locality"),
                "phone": text_of(card, "div.phones"),
                "category": text_of(card, "div.categories"),
                "rating": parse_rating(card),
                "reviews": text_of(card, "span.count"),
                "website": website["href"] if website else None,
            })
        except Exception as e:
            print(f"Skipped a card: {e}")
    return results

import csv
import json
from urllib.parse import quote_plus
from crawlbase import CrawlingAPI
from bs4 import BeautifulSoup

api = CrawlingAPI({"token": "YOUR_CRAWLBASE_TOKEN"})

def build_url(category, city):
    terms = quote_plus(category)
    geo = quote_plus(city)
    return f"https://www.yellowpages.com/search?search_terms={terms}&geo_location_terms={geo}"

def crawl(page_url):
    response = api.get(page_url, {"country": "US"})
    if response["status_code"] == 200:
        return response["body"].decode("latin1")
    print(f"Request failed: {response['status_code']}")
    return None

def text_of(card, selector):
    el = card.select_one(selector)
    return el.get_text(strip=True) if el else None

def parse_rating(card):
    el = card.select_one("div.result-rating")
    if not el:
        return None
    words = {"one": 1, "two": 2, "three": 3, "four": 4, "five": 5}
    rating = None
    for cls in el.get("class", []):
        base = cls.replace("-half", "")
        if base in words:
            rating = words[base] + (0.5 if "-half" in cls else 0)
    return rating

def scrape_results(html):
    soup = BeautifulSoup(html, "html.parser")
    results = []
    for card in soup.select("div.result"):
        try:
            website = card.select_one("a.track-visit-website")
            results.append({
                "name": text_of(card, "a.business-name"),
                "address": text_of(card, "div.street-address"),
                "locality": text_of(card, "div.locality"),
                "phone": text_of(card, "div.phones"),
                "category": text_of(card, "div.categories"),
                "rating": parse_rating(card),
                "reviews": text_of(card, "span.count"),
                "website": website["href"] if website else None,
            })
        except Exception as e:
            print(f"Skipped a card: {e}")
    return results

def save(rows, name):
    with open(f"{name}.json", "w") as f:
        json.dump(rows, f, indent=2)
    if rows:
        with open(f"{name}.csv", "w", newline="") as f:
            writer = csv.DictWriter(f, fieldnames=rows[0].keys())
            writer.writeheader()
            writer.writerows(rows)

def main():
    url = build_url("plumbers", "Austin, TX")
    html = crawl(url)
    if not html:
        return
    data = scrape_results(html)
    save(data, "listings")
    print(json.dumps(data, indent=2))

if __name__ == "__main__":
    main()

[
  {
    "name": "Austin Plumbing Co",
    "address": "1200 W 5th St",
    "locality": "Austin, TX 78703",
    "phone": "(512) 555-0142",
    "category": "Plumbers, Water Heaters",
    "rating": 4.5,
    "reviews": "(38)",
    "website": "https://www.austinplumbingco.example"
  },
  {
    "name": "Lone Star Drain & Sewer",
    "address": "904 E Cesar Chavez St",
    "locality": "Austin, TX 78702",
    "phone": "(512) 555-0188",
    "category": "Plumbers",
    "rating": null,
    "reviews": null,
    "website": null
  }
]

import time

def scrape_city(category, city, max_pages=5):
    base = build_url(category, city)
    collected = []
    for page in range(1, max_pages + 1):
        html = crawl(f"{base}&page={page}")
        if not html:
            break
        rows = scrape_results(html)
        if not rows:
            break
        for row in rows:
            row["city"] = city
        collected.extend(rows)
        print(f"{city} page {page}: {len(rows)} listings")
        time.sleep(2)
    return collected

def scrape_cities(category, cities):
    all_rows = []
    for city in cities:
        all_rows.extend(scrape_city(category, city))
    return all_rows

data = scrape_cities("restaurants", ["Austin, TX", "Denver, CO", "Phoenix, AZ"])
save(data, "multi_city")