如何从网站抓取数据实用 Python 演练

python --version

python -m venv crawler_env
source crawler_env/bin/activate

pip install crawlbase beautifulsoup4

import time
from crawlbase import CrawlingAPI

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

def fetch_html(page_url, max_retries=2):
    for attempt in range(max_retries + 1):
        response = api.get(page_url)
        if response["headers"]["pc_status"] == "200":
            return response["body"].decode("utf-8")
        if attempt < max_retries:
            print(f"Retrying ({attempt + 1}/{max_retries})...")
            time.sleep(1)
    print(f"Failed: {page_url} ({response['headers']['pc_status']})")
    return None

if __name__ == "__main__":
    html = fetch_html("https://example.com")
    print(html[:500] if html else "No HTML returned")

from urllib.parse import urljoin, urldefrag
from bs4 import BeautifulSoup

def extract_links(html, base_url):
    soup = BeautifulSoup(html, "html.parser")
    links = set()
    for a in soup.select("a[href]"):
        href = a["href"].strip()
        if not href or href.startswith(("mailto:", "tel:", "javascript:")):
            continue
        absolute = urljoin(base_url, href)
        absolute, _ = urldefrag(absolute)
        links.add(absolute)
    return links

from urllib.parse import urlparse

def in_scope(url, root):
    root_parts = urlparse(root)
    url_parts = urlparse(url)
    if url_parts.scheme not in ("http", "https"):
        return False
    if url_parts.netloc != root_parts.netloc:
        return False
    return url_parts.path.startswith(root_parts.path)

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

def attr_of(soup, selector, attr):
    el = soup.select_one(selector)
    return el.get(attr) if el else None

def parse_page(html, url):
    soup = BeautifulSoup(html, "html.parser")
    summary = (
        attr_of(soup, 'meta[name="description"]', "content")
        or text_of(soup, "article p")
    )
    return {
        "url": url,
        "title": text_of(soup, "h1") or text_of(soup, "title"),
        "summary": summary,
        "date": attr_of(soup, "time[datetime]", "datetime"),
    }

import csv
import json
import time
from collections import deque
from urllib.parse import urljoin, urldefrag, urlparse
from crawlbase import CrawlingAPI
from bs4 import BeautifulSoup

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

def fetch_html(page_url, max_retries=2):
    for attempt in range(max_retries + 1):
        response = api.get(page_url)
        if response["headers"]["pc_status"] == "200":
            return response["body"].decode("utf-8")
        if attempt < max_retries:
            time.sleep(1)
    return None

def extract_links(html, base_url):
    soup = BeautifulSoup(html, "html.parser")
    links = set()
    for a in soup.select("a[href]"):
        href = a["href"].strip()
        if not href or href.startswith(("mailto:", "tel:", "javascript:")):
            continue
        absolute, _ = urldefrag(urljoin(base_url, href))
        links.add(absolute)
    return links

def in_scope(url, root):
    r, u = urlparse(root), urlparse(url)
    return (
        u.scheme in ("http", "https")
        and u.netloc == r.netloc
        and u.path.startswith(r.path)
    )

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

def attr_of(soup, selector, attr):
    el = soup.select_one(selector)
    return el.get(attr) if el else None

def parse_page(html, url, link_count):
    soup = BeautifulSoup(html, "html.parser")
    summary = (
        attr_of(soup, 'meta[name="description"]', "content")
        or text_of(soup, "article p")
    )
    return {
        "url": url,
        "title": text_of(soup, "h1") or text_of(soup, "title"),
        "summary": summary,
        "date": attr_of(soup, "time[datetime]", "datetime"),
        "links": link_count,
    }

def crawl(start_url, max_pages=25):
    queue = deque([start_url])
    visited = set()
    records = []
    while queue and len(visited) < max_pages:
        url = queue.popleft()
        if url in visited:
            continue
        visited.add(url)
        html = fetch_html(url)
        if not html:
            continue
        found = {l for l in extract_links(html, url) if in_scope(l, start_url)}
        records.append(parse_page(html, url, len(found)))
        for link in found:
            if link not in visited:
                queue.append(link)
        print(f"[{len(visited)}/{max_pages}] {url}")
        time.sleep(2)
    return records

def dedupe(records):
    seen = {}
    for record in records:
        seen[record["url"].rstrip("/")] = record
    return list(seen.values())

def save_outputs(records):
    with open("crawl_results.json", "w") as f:
        json.dump(records, f, indent=2)
    if not records:
        return
    with open("crawl_results.csv", "w", newline="") as f:
        writer = csv.DictWriter(f, fieldnames=records[0].keys())
        writer.writeheader()
        writer.writerows(records)

def main():
    records = crawl("https://example.com", max_pages=25)
    records = dedupe(records)
    save_outputs(records)
    print(f"Saved {len(records)} pages")

if __name__ == "__main__":
    main()

[
  {
    "url": "https://example.com/articles/web-data",
    "title": "A Practical Guide to Web Data",
    "summary": "How teams turn public pages into clean, structured records.",
    "date": "2024-09-18",
    "links": 12
  },
  {
    "url": "https://example.com/articles/crawling-basics",
    "title": "Crawling Basics",
    "summary": "Discovery, scope, and dedupe explained from first principles.",
    "date": "2024-08-02",
    "links": 9
  }
]