Bingスクレイピング入門｜PythonとRola IPで検索結果を取得

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

mkdir bing-scraper
cd bing-scraper
py -m venv .venv
.\.venv\Scripts\python.exe -m pip install requests beautifulsoup4

python -m pip install jupyterlab
python -m jupyterlab

python --version
python -m pip show requests beautifulsoup4

import argparse
import base64
import csv
import getpass
import json
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


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):
    # Validate every redirect before sending the next request.
    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; Retry-After: {retry_after}; stopped.")
        if r.status_code == 403:
            raise RuntimeError(f"HTTP {r.status_code}; 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"}:
        if 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):
    soup = BeautifulSoup(html, "html.parser")
    if soup.select_one('#b_captcha, iframe[src*="captcha"]'):
        raise RuntimeError("Verification page; stopped.")
    rows = []
    stamp = datetime.now(timezone.utc).isoformat(timespec="seconds")
    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")
        if snippet is None:
            snippet = 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": stamp,
        })
    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/https/socks5h): ").strip()
    host = input("Proxy host only: ").strip()
    port = int(input("Proxy port: "))
    if scheme not in {"http", "https", "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): ")
    auth = quote(user, safe="") + ":" + quote(password, safe="")
    proxy = f"{scheme}://{auth}@{host}:{port}"
    session.proxies.update({"http": proxy, "https": proxy})

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"] = "ja-JP,ja;q=0.9"
    url = "https://www.bing.com/search?" + urlencode(
        {"q": query, "setlang": "ja-JP"})
    all_rows, seen_pages, seen_signatures = [], set(), set()
    try:
        if proxy:
            configure_proxy(session)
        for page in range(1, max_pages + 1):
            if url in seen_pages:
                break
            seen_pages.add(url)
            r = fetch(session, url)
            (folder / f"page-{page}.html").write_text(r.text, encoding="utf-8")
            rows, next_url = parse_page(r.text, r.url, query, page)
            signature = tuple(row["url"] for row in rows)
            if signature in seen_signatures:
                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")
            print("Response host:", urlsplit(r.url).hostname)
            if not next_url or page == max_pages:
                break
            url = next_url
            time.sleep(5)
    finally:
        session.close()
        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()
    try:
        scrape(args.query, args.pages, args.proxy)
    except (requests.RequestException, RuntimeError, ValueError) as error:
        # Never print proxy exception details, which can contain credentials.
        message = str(error) if not isinstance(error, requests.RequestException) else type(error).__name__
        raise SystemExit("Stopped: " + message)

python bing_scraper.py "Python ドキュメント" --pages 1

.\.venv\Scripts\python.exe bing_scraper.py "Python ドキュメント" --pages 2

python bing_scraper.py "Python ドキュメント" --pages 1 --proxy

python -m pip install "requests[socks]"