用Python抓取Bing搜索结果：分页、CSV导出与代理配置

mkdir bing-scraper
cd bing-scraper
python3 -m venv .venv
source .venv/bin/activate
python -m pip install requests beautifulsoup4

python -m pip install jupyterlab
python -m jupyterlab

import sys
import requests
import bs4
print("Executable:", sys.executable)
print("Python:", sys.version.split()[0])
print("Requests:", requests.__version__)
print("Beautiful Soup:", bs4.__version__)

import argparse
import base64
import csv
import getpass
import json
import platform
import sys
import time
from datetime import datetime, timezone
from pathlib import Path
from urllib.parse import parse_qs, quote, urlencode, urljoin, urlsplit

import requests
from bs4 import BeautifulSoup
from importlib.metadata import version

def bing_url(url):
    p = urlsplit(url)
    return (p.scheme == "https" and p.hostname in
            {"www.bing.com", "cn.bing.com", "bing.com"}
            and p.path == "/search" and not p.username
            and p.port in (None, 443))

def fetch(session, url):
    for _ in range(5):
        if not bing_url(url):
            raise RuntimeError("Unexpected search URL; stopped.")
        r = session.get(url, timeout=(10, 30), allow_redirects=False)
        if r.is_redirect:
            url = urljoin(r.url, r.headers["Location"])
            continue
        if r.status_code == 429:
            retry_after = r.headers.get("Retry-After", "not provided")
            raise RuntimeError(f"HTTP 429; stopped. Retry-After: {retry_after}")
        if r.status_code == 403:
            raise RuntimeError("HTTP 403; stopped.")
        r.raise_for_status()
        if "text/html" not in r.headers.get("Content-Type", ""):
            raise RuntimeError("Expected HTML; stopped.")
        r.encoding = "utf-8"
        return r
    raise RuntimeError("Too many redirects; stopped.")

def destination(href, base_url):
    url = urljoin(base_url, href)
    p = urlsplit(url)
    if p.hostname in {"www.bing.com", "cn.bing.com", "bing.com"} and p.path == "/ck/a":
        value = parse_qs(p.query).get("u", [""])[0]
        if value.startswith("a1"):
            try:
                payload = value[2:]
                decoded = base64.urlsafe_b64decode(
                    payload + "=" * (-len(payload) % 4)
                ).decode("utf-8")
                if urlsplit(decoded).scheme in {"http", "https"}:
                    return decoded
            except (ValueError, UnicodeError):
                pass
    return url if p.scheme in {"http", "https"} else ""

def parse_page(html, page_url, query, page_number, fetched_at_utc):
    soup = BeautifulSoup(html, "html.parser")
    if soup.select_one('#b_captcha, iframe[src*="captcha"]'):
        raise RuntimeError("Verification page; stopped.")
    rows = []

    for item in soup.select("li.b_algo"):
        a = item.select_one("h2 a[href]")
        if a is None or not a.get_text(strip=True):
            continue
        url = destination(a["href"], page_url)
        if not url:
            continue
        snippet = item.select_one(".b_caption p") or item.select_one(
            "p.b_lineclamp2, p.b_lineclamp3")
        rows.append({
            "query": query, "page": page_number,
            "position_on_page": len(rows) + 1,
            "title": a.get_text(" ", strip=True), "url": url,
            "snippet": snippet.get_text(" ", strip=True) if snippet else "",
            "raw_href": a["href"],
            "source_url": page_url, "fetched_at_utc": fetched_at_utc,
        })
    if not rows:
        raise RuntimeError("No organic rows; inspect HTML, do not assume end.")
    link = soup.select_one("a.sb_pagN[href]")
    next_url = urljoin(page_url, link["href"]) if link else None
    if next_url and not bing_url(next_url):
        raise RuntimeError("Unexpected next-page URL; stopped.")
    return rows, next_url

def excel_safe(value):
    if isinstance(value, str):
        if value.lstrip().startswith(("=", "+", "-", "@")):
            return "'" + value
        if value.startswith(("\t", "\r", "\n")):
            return "'" + value
    return value

def save_rows(rows, folder):
    if not rows:
        raise RuntimeError("No data to export.")
    folder = Path(folder)
    folder.mkdir(parents=True, exist_ok=True)
    (folder / "results.json").write_text(
        json.dumps(rows, ensure_ascii=False, indent=2), encoding="utf-8")
    with (folder / "results.csv").open("w", encoding="utf-8-sig", newline="") as f:
        writer = csv.DictWriter(f, fieldnames=list(rows[0]))
        writer.writeheader()
        writer.writerows({k: excel_safe(v) for k, v in r.items()} for r in rows)

def configure_proxy(session):
    scheme = input("Proxy scheme (http/socks5h): ").strip()
    host = input("Proxy host only: ").strip()
    port = int(input("Proxy port: "))
    if scheme not in {"http", "socks5h"}:
        raise ValueError("Unsupported scheme.")
    if not host or any(c.isspace() or c in "/@?#:[]" for c in host):
        raise ValueError("Enter only the proxy hostname or IPv4 address.")
    if not 1 <= port <= 65535:
        raise ValueError("Invalid port.")
    user = getpass.getpass("Proxy username (hidden): ")
    password = getpass.getpass("Proxy password (hidden): ")
    if not user or not password:
        raise ValueError("This example requires username/password authentication.")
    auth = quote(user, safe="") + ":" + quote(password, safe="")
    proxy = f"{scheme}://{auth}@{host}:{port}"
    session.proxies.update({"http": proxy, "https": proxy})

python bing_scraper.py "python documentation" --pages 1 --proxy

python -m pip install "requests[socks]"

def scrape(query, max_pages=2, proxy=False):
    if not 1 <= max_pages <= 3:
        raise ValueError("This example supports 1 to 3 pages.")
    folder = Path("runs") / datetime.now().strftime("%Y%m%d-%H%M%S-%f")
    folder.mkdir(parents=True)
    session = requests.Session()
    session.trust_env = False
    session.headers["Accept-Language"] = "en-US,en;q=0.9"
    url = "https://www.bing.com/search?" + urlencode({"q": query, "setlang": "en-US"})
    all_rows, seen_pages, seen_signatures = [], set(), set()
    run_meta = {
        "started_at_utc": datetime.now(timezone.utc).isoformat(),
        "os": platform.platform(), "python": sys.version,
        "requests": version("requests"), "beautifulsoup4": version("beautifulsoup4"),
        "query": query, "max_pages": max_pages, "proxy_enabled": proxy,
        "accept_language": session.headers["Accept-Language"],
        "timeout_seconds": [10, 30], "page_interval_seconds": 5,
        "automatic_retries": 0, "pages": [], "stop_reason": "not started",
    }
    try:
        if proxy:
            configure_proxy(session)
        for page in range(1, max_pages + 1):
            if url in seen_pages:
                run_meta["stop_reason"] = "Repeated request URL"
                break
            seen_pages.add(url)
            r = fetch(session, url)
            fetched_at_utc = datetime.now(timezone.utc).isoformat(timespec="seconds")
            page_meta = {"page": page, "source_url": r.url,
                         "fetched_at_utc": fetched_at_utc, "http_status": r.status_code}
            run_meta["pages"].append(page_meta)
            (folder / f"page-{page}.html").write_text(r.text, encoding="utf-8")
            rows, next_url = parse_page(r.text, r.url, query, page, fetched_at_utc)
            page_meta["organic_rows"] = len(rows)
            print(f"Response host: {urlsplit(r.url).hostname}")
            signature = tuple(row["url"] for row in rows)
            if signature in seen_signatures:
                run_meta["stop_reason"] = "Repeated ordered URL signature"
                print("Repeated page; stopped.")
                break
            seen_signatures.add(signature)
            all_rows.extend(rows)
            print(f"Page {page}: HTTP {r.status_code}, {len(rows)} organic rows")
            if not next_url:
                run_meta["stop_reason"] = "No next-page link found"
                break
            if page == max_pages:
                run_meta["stop_reason"] = "Requested page limit reached"
                break
            url = next_url
            time.sleep(5)
    except requests.RequestException as exc:
        # Do not persist proxy URLs or credentials from exception messages.
        run_meta["stop_reason"] = type(exc).__name__
        raise RuntimeError(f"{type(exc).__name__}; check network or proxy settings.") from None
    except (RuntimeError, ValueError, OSError) as exc:
        run_meta["stop_reason"] = str(exc)
        raise
    finally:
        session.close()
        run_meta["finished_at_utc"] = datetime.now(timezone.utc).isoformat()
        run_meta["saved_rows"] = len(all_rows)
        (folder / "run.json").write_text(
            json.dumps(run_meta, ensure_ascii=False, indent=2), encoding="utf-8")
        print("Stopped:", run_meta["stop_reason"])
        if all_rows:
            save_rows(all_rows, folder)
            print(f"Saved {len(all_rows)} rows to {folder}")
    return all_rows, folder

if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("query")
    parser.add_argument("--pages", type=int, default=2)
    parser.add_argument("--proxy", action="store_true")
    args = parser.parse_args()
    scrape(args.query, args.pages, args.proxy)

python bing_scraper.py "python documentation" --pages 2