如何用 Python 抓取 Yelp 数据商业listings 与评分

python --version

python -m venv yelp_env
source yelp_env/bin/activate

pip install crawlbase beautifulsoup4 pandas

from urllib.parse import urlencode
from crawlbase import CrawlingAPI

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

def build_url(query, location, start=0):
    base = "https://www.yelp.com/search?"
    params = {"find_desc": query, "find_loc": location, "start": start}
    return base + urlencode(params)

def crawl(page_url):
    response = api.get(page_url, {"ajax_wait": "true", "page_wait": "3000"})
    if response["headers"]["cb_status"] == "200":
        return response["body"].decode("latin1")
    print(f"Request failed: {response['headers']['cb_status']}")
    return None

if __name__ == "__main__":
    url = build_url("Italian Restaurants", "San Francisco, CA")
    html = crawl(url)
    print(html[:500] if html else "No HTML returned")

from bs4 import BeautifulSoup

BASE = "https://www.yelp.com"

def text_of(node):
    return node.get_text(strip=True) if node else None

def extract_business(card):
    name = card.select_one('div[class*="businessName"] h3 > span > a')
    rating = card.select_one('div.css-volmcs + div.css-1jq1ouh > span:first-child')
    reviews = card.select_one('div.css-volmcs + div.css-1jq1ouh > span:last-child')
    category = card.select('div[class*="priceCategory"] div > p > span:first-child a')
    address = card.select_one('div[class*="priceCategory"] div > p > span:last-child')

    return {
        "name": text_of(name),
        "rating": text_of(rating),
        "review_count": text_of(reviews),
        "category": ", ".join(c.get_text(strip=True) for c in category) if category else None,
        "address": text_of(address),
        "link": BASE + name["href"] if name and name.get("href") else None,
    }

def extract_businesses(html):
    soup = BeautifulSoup(html, "html.parser")
    cards = soup.select('div[data-testid="serp-ia-card"]:not(.ABP)')
    return [extract_business(card) for card in cards]

import json
from urllib.parse import urlencode
from crawlbase import CrawlingAPI
from bs4 import BeautifulSoup

api = CrawlingAPI({"token": "YOUR_CRAWLBASE_TOKEN"})
BASE = "https://www.yelp.com"

def build_url(query, location, start=0):
    params = {"find_desc": query, "find_loc": location, "start": start}
    return BASE + "/search?" + urlencode(params)

def crawl(page_url):
    response = api.get(page_url, {"ajax_wait": "true", "page_wait": "3000"})
    if response["headers"]["cb_status"] == "200":
        return response["body"].decode("latin1")
    print(f"Request failed: {response['headers']['cb_status']}")
    return None

def text_of(node):
    return node.get_text(strip=True) if node else None

def extract_business(card):
    name = card.select_one('div[class*="businessName"] h3 > span > a')
    rating = card.select_one('div.css-volmcs + div.css-1jq1ouh > span:first-child')
    reviews = card.select_one('div.css-volmcs + div.css-1jq1ouh > span:last-child')
    category = card.select('div[class*="priceCategory"] div > p > span:first-child a')
    address = card.select_one('div[class*="priceCategory"] div > p > span:last-child')
    return {
        "name": text_of(name),
        "rating": text_of(rating),
        "review_count": text_of(reviews),
        "category": ", ".join(c.get_text(strip=True) for c in category) if category else None,
        "address": text_of(address),
        "link": BASE + name["href"] if name and name.get("href") else None,
    }

def extract_businesses(html):
    soup = BeautifulSoup(html, "html.parser")
    cards = soup.select('div[data-testid="serp-ia-card"]:not(.ABP)')
    return [extract_business(card) for card in cards]

def main():
    url = build_url("Italian Restaurants", "San Francisco, CA")
    html = crawl(url)
    if not html:
        return
    data = extract_businesses(html)
    print(json.dumps(data, indent=2))

if __name__ == "__main__":
    main()

[
  {
    "name": "Bella Trattoria",
    "rating": "4.3",
    "review_count": "(1.9k reviews)",
    "category": "Italian, Bars, Pasta Shops",
    "address": "Inner Richmond",
    "link": "https://www.yelp.com/biz/bella-trattoria-san-francisco"
  },
  {
    "name": "Sotto Mare",
    "rating": "4.3",
    "review_count": "(5.2k reviews)",
    "category": "Seafood, Italian, Bars",
    "address": "North Beach/Telegraph Hill",
    "link": "https://www.yelp.com/biz/sotto-mare-san-francisco"
  }
]

import json
import time
import pandas as pd

def scrape_all_pages(query, location, max_pages=5):
    all_rows = []
    for page in range(max_pages):
        start = page * 10
        url = build_url(query, location, start)
        html = crawl(url)
        if not html:
            print(f"Stopping at offset {start}: no HTML")
            break
        rows = extract_businesses(html)
        if not rows:
            print(f"No results at offset {start}; reached the end")
            break
        all_rows.extend(rows)
        print(f"Offset {start}: {len(rows)} businesses")
        time.sleep(2)
    return all_rows

if __name__ == "__main__":
    rows = scrape_all_pages("Italian Restaurants", "San Francisco, CA", max_pages=5)

    with open("yelp_businesses.json", "w") as f:
        json.dump(rows, f, indent=2)

    pd.DataFrame(rows).to_csv("yelp_businesses.csv", index=False)
    print(f"Saved {len(rows)} businesses to JSON and CSV")