GoogleニュースのスクレイピングをPythonで行う方法｜RSS取得とSocks5.IOの設定

python -m venv .venv
.\.venv\Scripts\python.exe -m pip install "requests[socks]>=2.32,<3"

https://news.google.com/rss/search?q=artificial+intelligence&hl=en-US&gl=US&ceid=US%3Aen

import argparse
import csv
import ipaddress
import json
import os
import re
import sys
import xml.etree.ElementTree as ET
from datetime import datetime, timezone
from email.utils import parsedate_to_datetime
from pathlib import Path
from urllib.parse import parse_qs, quote, urlencode, urlsplit
from uuid import uuid4
from xml.sax.saxutils import escape

import requests

EDITIONS = {"US": {"hl": "en-US", "gl": "US", "ceid": "US:en"},
            "GB": {"hl": "en-GB", "gl": "GB", "ceid": "GB:en"}}
MAX_BYTES = 2_000_000
COLUMNS = ["run_id", "retrieved_at_utc", "source_kind", "query", "requested_edition",
           "observed_language", "item_id", "title", "publisher", "publisher_homepage",
           "google_news_url", "published_raw", "published_at_utc", "review_reason"]


def search_url(query, edition, rss=True):
    path = "/rss/search" if rss else "/search"
    return "https://news.google.com" + path + "?" + urlencode({"q": query, **EDITIONS[edition]})


def demo_feed(query, edition):
    item = '''<item><title>Fictional research update - Demo Publisher</title>
<link>https://news.google.com/rss/articles/DEMO-A</link><guid>demo-a</guid>
<pubDate>Tue, 20 Oct 2026 10:00:00 GMT</pubDate>
<source url="https://publisher.example.invalid/">Demo Publisher</source></item>'''
    second = '''<item><title>Another fictional headline</title>
<link>https://news.google.com/rss/articles/DEMO-B</link><guid>demo-b</guid>
<pubDate>Tue, 20 Oct 2026 11:00:00 GMT</pubDate>
<source url="https://second.example.invalid/">Second Demo</source></item>'''
    return (f'<?xml version="1.0" encoding="UTF-8"?><rss version="2.0"><channel>'
            f'<title>Local fictional feed</title><link>{escape(search_url(query, edition, False))}</link>'
            f'<language>{EDITIONS[edition]["hl"]}</language>'
            + item + item + second + '</channel></rss>').encode("utf-8")


def parse_feed(data, query, edition, run_id, retrieved_at, source_kind):
    if len(data) > MAX_BYTES:
        raise ValueError("Feed exceeds tutorial size limit")
    text = data.decode("utf-8-sig")
    if "<!DOCTYPE" in text.upper() or "<!ENTITY" in text.upper():
        raise ValueError("DTD and entity declarations are not supported")
    root = ET.fromstring(text)
    channel = root.find("channel")
    if root.tag != "rss" or channel is None:
        raise ValueError("Expected RSS channel, not HTML or another feed format")
    channel_link = channel.findtext("link", "").strip()
    p = urlsplit(channel_link)
    observed = parse_qs(p.query)
    expected = {"q": query, **EDITIONS[edition]}
    if (p.scheme != "https" or p.hostname != "news.google.com" or p.path != "/search"
            or any(observed.get(k) != [v] for k, v in expected.items())):
        raise ValueError("Channel search context is missing or differs from the request")
    language = channel.findtext("language", "").strip()
    if language != EDITIONS[edition]["hl"]:
        raise ValueError("Channel language differs from the requested edition")
    rows, seen, duplicates = [], set(), 0
    items = channel.findall("item")
    for item in items:
        title = item.findtext("title", "").strip()
        link = item.findtext("link", "").strip()
        link_parts = urlsplit(link)
        if not title or link_parts.scheme != "https" or link_parts.hostname != "news.google.com":
            raise ValueError("Missing title or expected Google News item link")
        identity = item.findtext("guid", "").strip() or link
        if identity in seen:
            duplicates += 1
            continue
        seen.add(identity)
        source = item.find("source")
        publisher = "" if source is None else (source.text or "").strip()
        homepage = "" if source is None else source.get("url", "").strip()
        raw_date = item.findtext("pubDate", "").strip()
        flags, published = [], ""
        try:
            dt = parsedate_to_datetime(raw_date)
            if dt.tzinfo is None:
                raise ValueError("Publication timezone missing")
            published = dt.astimezone(timezone.utc).isoformat()
        except (ValueError, TypeError, OverflowError):
            flags.append("publication_time_unknown")
        if not publisher:
            flags.append("publisher_unknown")
        rows.append(dict(zip(COLUMNS, [run_id, retrieved_at, source_kind, query, edition,
                    language, identity, title, publisher, homepage, link, raw_date,
                    published, ";".join(flags)])))
    return rows, {"items_seen": len(items), "duplicates_removed": duplicates,
                  "records_exported": len(rows), "observed_channel_link": channel_link,
                  "observed_language": language}


def make_session(use_proxy):
    session = requests.Session()
    session.trust_env = False
    session.headers["User-Agent"] = "NewsMetadataTutorial/1.0"
    if use_proxy:
        host = os.environ["SOCKS5IO_HOST"]
        port = int(os.environ["SOCKS5IO_PORT"])
        if not re.fullmatch(r"[A-Za-z0-9.-]+", host) or not 1 <= port <= 65535:
            session.close()
            raise ValueError("Check issued proxy host and port")
        user = quote(os.environ["SOCKS5IO_USER"], safe="")
        password = quote(os.environ["SOCKS5IO_PASSWORD"], safe="")
        proxy = f"socks5h://{user}:{password}@{host}:{port}"
        session.proxies.update(http=proxy, https=proxy)
    return session


def get_body(session, url, accepted_types):
    with session.get(url, timeout=(5, 20), allow_redirects=False, stream=True) as response:
        if response.status_code != 200:
            raise ValueError(f"HTTP {response.status_code}: stop; review access, quota or redirect")
        mime = response.headers.get("Content-Type", "").split(";", 1)[0].strip().lower()
        if mime not in accepted_types:
            raise ValueError("Unexpected content type; do not parse a challenge as news")
        body = bytearray()
        for chunk in response.iter_content(65536):
            body.extend(chunk)
            if len(body) > MAX_BYTES:
                raise ValueError("Response exceeds tutorial size limit")
        return bytes(body)


def check_exit(session):
    body = get_body(session, "https://api.ipify.org?format=json", {"application/json"})
    address = ipaddress.ip_address(json.loads(body)["ip"])
    if not address.is_global:
        raise ValueError("Exit checker returned a non-public address")
    return {"observed_ip": str(address), "checked_at_utc": datetime.now(timezone.utc).isoformat()}


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--query", default="artificial intelligence")
    parser.add_argument("--edition", choices=EDITIONS, default="US")
    parser.add_argument("--live", action="store_true")
    parser.add_argument("--approved", action="store_true")
    parser.add_argument("--proxy", action="store_true")
    parser.add_argument("--route-label", default="private-run")
    parser.add_argument("--out", type=Path, required=True)
    args = parser.parse_args()
    query = args.query.strip()
    if not query:
        parser.error("Query must not be empty")
    if args.proxy and not args.live:
        parser.error("The local fictional feed is not a proxy test")
    if args.live and not args.approved:
        parser.error("Review current source conditions and request budget before --approved")
    log_path = args.out.with_suffix(".run.json")
    if args.out.exists() or log_path.exists():
        parser.error("Use new output filenames; existing files will not be replaced")
    run_id = uuid4().hex
    url = search_url(query, args.edition)
    manifest = {"run_id": run_id, "query": query, "requested_edition": args.edition,
                "requested_feed_url": url, "proxy_requested": args.proxy,
                "route_label": args.route_label}
    if args.live:
        with make_session(args.proxy) as session:
            manifest["before"] = check_exit(session)
            data = get_body(session, url, {"application/rss+xml", "application/xml", "text/xml"})
            stamp = datetime.now(timezone.utc).isoformat()
            manifest["after"] = check_exit(session)
        manifest["same_observed_ip"] = manifest["before"]["observed_ip"] == manifest["after"]["observed_ip"]
        kind = "requested_google_news_rss"
    else:
        data = demo_feed(query, args.edition)
        stamp = datetime.now(timezone.utc).isoformat()
        kind = "synthetic_fixture"
    rows, stats = parse_feed(data, query, args.edition, run_id, stamp, kind)
    manifest.update(stats, source_kind=kind, retrieved_at_utc=stamp)
    with log_path.open("x", encoding="utf-8") as handle:
        json.dump(manifest, handle, indent=2)
    with args.out.open("x", encoding="utf-8", newline="") as handle:
        writer = csv.DictWriter(handle, fieldnames=COLUMNS)
        writer.writeheader()
        writer.writerows(rows)
    print(f"Saved {len(rows)} records; removed {stats['duplicates_removed']} within-feed duplicates.")


if __name__ == "__main__":
    try:
        main()
    except (ValueError, UnicodeError, ET.ParseError) as error:
        print("Stopped:", type(error).__name__, "- review response structure and source context.", file=sys.stderr)
        sys.exit(1)
    except (KeyError, TypeError, OSError, requests.RequestException):
        print("Stopped: inspect connection, issued fields and output paths privately.", file=sys.stderr)
        sys.exit(1)

### 手順4：ローカルの実行結果を確認する

```powershell
python google_news_rss.py --out news-demo.csv
python google_news_rss.py --query "renewable energy" --edition GB --out news-gb-demo.csv

python google_news_rss.py --live --approved --query "renewable energy" --edition US --route-label direct-US --out news-live-us.csv

$env:SOCKS5IO_HOST = Read-Host 'Issued proxy host'
$env:SOCKS5IO_PORT = Read-Host 'Issued proxy port'
$env:SOCKS5IO_USER = Read-Host 'Issued proxy username'
$newsProxySecret = Read-Host 'Issued proxy password' -AsSecureString
$env:SOCKS5IO_PASSWORD = [System.Net.NetworkCredential]::new('', $newsProxySecret).Password

python google_news_rss.py --live --approved --proxy --query "renewable energy" --edition US --route-label residential-US --out news-proxy-us.csv

Remove-Item Env:SOCKS5IO_PASSWORD
Remove-Variable newsProxySecret