迷惑をかけないPythonスクレイピング: robots.txtとアクセス間隔の基本

# polite_sleep_get.py
# 目的: 最小構成でUser-Agentを名乗り、ランダム待機を挟んでGETする
import time
import random
import requests

UA = "MyPoliteScraper/1.0 (+https://example.com/contact)"  # 自分の連絡先を含めるのが望ましい
HEADERS = {"User-Agent": UA}
TIMEOUT = 10  # 秒。応答がないときは諦める

def polite_get(url: str, min_wait: float = 3.0, max_wait: float = 7.0) -> str:
    # 次のアクセス前にランダムな待機を入れる(3〜7秒)
    wait = random.uniform(min_wait, max_wait)
    print(f"[INFO] Waiting {wait:.2f}s before requesting {url}")
    time.sleep(wait)

    # 実際のHTTP GET。タイムアウトとUser-Agentを設定する
    resp = requests.get(url, headers=HEADERS, timeout=TIMEOUT)
    print(f"[INFO] GET {url} -> {resp.status_code}")
    resp.raise_for_status()  # 4xx/5xxなら例外を発生させる
    return resp.text  # HTML本文など

if __name__ == "__main__":
    # 学習用にはURL数を最小限にし、結果の利用は自己責任で
    urls = [
        "https://www.python.org/",
        "https://www.python.org/downloads/",
    ]
    for u in urls:
        try:
            html = polite_get(u)
            print(f"[INFO] Received {len(html)} bytes")
        except requests.HTTPError as e:
            print(f"[ERROR] HTTPError: {e}")
        except requests.RequestException as e:
            print(f"[ERROR] Request failed: {e}")

# robots_check.py
# 目的: robots.txtを読み、can_fetch(許可/不許可)とcrawl_delay(推奨待機)を確認する
from urllib import robotparser
from urllib.parse import urlparse, urljoin

def robots_for(url: str, ua: str = "MyPoliteScraper/1.0") -> tuple[bool, float | None]:
    parsed = urlparse(url)
    robots_url = f"{parsed.scheme}://{parsed.netloc}/robots.txt"

    rp = robotparser.RobotFileParser()
    rp.set_url(robots_url)
    try:
        rp.read()  # robots.txtを取得・解析
    except Exception:
        # 取得失敗時は不明扱い。安全側に倒すならFalseを返す
        return (False, None)

    allowed = rp.can_fetch(ua, url)
    # UAに固有のcrawl-delayがなければ'*'を参照する
    delay = rp.crawl_delay(ua)
    if delay is None:
        delay = rp.crawl_delay("*")
    return (allowed, delay)

if __name__ == "__main__":
    url = "https://www.python.org/downloads/"
    ok, delay = robots_for(url)
    print(f"Allowed: {ok}, Crawl-delay: {delay}")

# polite_scraper.py
# 目的:
# - robots.txtを確認してアクセス可否とCrawl-delayを尊重
# - ランダム待機 + 指数バックオフ
# - 設定値で挙動を管理
from dataclasses import dataclass
from typing import Optional, Dict
from urllib import robotparser
from urllib.parse import urlparse
import time
import random
import requests

@dataclass
class ScrapeConfig:
    user_agent: str = "MyPoliteScraper/1.0 (+https://example.com/contact)"
    timeout: float = 15.0  # 秒
    base_wait_min: float = 3.0
    base_wait_max: float = 7.0
    respect_crawl_delay: bool = True
    max_retries: int = 3
    backoff_factor: float = 2.0  # 待機を倍々に増やす
    max_redirects: int = 5

class RobotsCache:
    """ドメインごとにrobots.txtの結果をキャッシュする簡易クラス"""
    def __init__(self, ua: str) -> None:
        self.ua = ua
        self.cache: Dict[str, robotparser.RobotFileParser] = {}

    def get_rp(self, netloc: str) -> robotparser.RobotFileParser:
        if netloc in self.cache:
            return self.cache[netloc]
        rp = robotparser.RobotFileParser()
        rp.set_url(f"https://{netloc}/robots.txt")
        try:
            rp.read()
        except Exception:
            # 取得できない場合は空ルールとみなし、安全側に倒す判断もありうる
            pass
        self.cache[netloc] = rp
        return rp

    def can_fetch(self, url: str) -> bool:
        netloc = urlparse(url).netloc
        rp = self.get_rp(netloc)
        try:
            return rp.can_fetch(self.ua, url)
        except Exception:
            # パース不能時は安全側に倒す
            return False

    def crawl_delay(self, url: str) -> Optional[float]:
        netloc = urlparse(url).netloc
        rp = self.get_rp(netloc)
        # UA固有 -> '*' の順に確認
        delay = rp.crawl_delay(self.ua)
        if delay is None:
            delay = rp.crawl_delay("*")
        return delay

class PoliteScraper:
    def __init__(self, cfg: ScrapeConfig) -> None:
        self.cfg = cfg
        self.session = requests.Session()
        self.session.headers.update({"User-Agent": cfg.user_agent})
        self.robots = RobotsCache(cfg.user_agent)
        # ドメインごとの直近アクセス時刻
        self.last_access: Dict[str, float] = {}

    def _sleep_before_request(self, url: str) -> None:
        netloc = urlparse(url).netloc
        # 基本のランダム待機
        wait = random.uniform(self.cfg.base_wait_min, self.cfg.base_wait_max)

        # robots.txtのCrawl-delayを尊重
        if self.cfg.respect_crawl_delay:
            cd = self.robots.crawl_delay(url)
            if cd is not None:
                wait = max(wait, float(cd))

        # 直前アクセスとの間隔も確保(短すぎたら延長)
        now = time.time()
        last = self.last_access.get(netloc, 0.0)
        elapsed = now - last
        if elapsed < wait:
            to_sleep = wait - elapsed
            print(f"[WAIT] {netloc}: sleeping {to_sleep:.2f}s (base {wait:.2f}s, elapsed {elapsed:.2f}s)")
            time.sleep(max(0.0, to_sleep))
        else:
            # すでに十分空いている場合でも、微小ジッタを追加
            jitter = random.uniform(0.1, 0.5)
            print(f"[WAIT] {netloc}: sleeping jitter {jitter:.2f}s")
            time.sleep(jitter)

    def get(self, url: str) -> requests.Response:
        parsed = urlparse(url)
        if parsed.scheme not in ("http", "https"):
            raise ValueError("Unsupported scheme")

        # robots.txtの許可確認
        if not self.robots.can_fetch(url):
            raise PermissionError(f"Blocked by robots.txt: {url}")

        self._sleep_before_request(url)

        tries = 0
        backoff = 0.0  # 直前エラーによる延長待機
        redirects = 0

        while True:
            tries += 1
            if backoff > 0:
                print(f"[BACKOFF] sleeping {backoff:.2f}s before retry")
                time.sleep(backoff)

            try:
                resp = self.session.get(url, timeout=self.cfg.timeout, allow_redirects=False)
                status = resp.status_code

                # リダイレクトは回数制限の上で追従
                if 300 <= status < 400 and "Location" in resp.headers:
                    redirects += 1
                    if redirects > self.cfg.max_redirects:
                        raise requests.TooManyRedirects(f"Too many redirects for {url}")
                    url = requests.compat.urljoin(url, resp.headers["Location"])
                    print(f"[REDIR] -> {url}")
                    # リダイレクトも1リクエストなので、次のGET前に待機
                    self._sleep_before_request(url)
                    # バックオフは維持(または軽減)。ここでは維持。
                    continue

                # 成功
                if 200 <= status < 300:
                    self.last_access[parsed.netloc] = time.time()
                    return resp

                # 一時的エラーはバックオフで再試行
                if status in (429, 503) or 500 <= status < 600:
                    # Retry-After優先
                    ra = resp.headers.get("Retry-After")
                    wait_hint = float(ra) if ra and ra.isdigit() else None
                    if wait_hint is None:
                        # 次の待機は指数的に増加
                        base = max(self.cfg.base_wait_min, 1.0)
                        backoff = (backoff or base) * self.cfg.backoff_factor
                    else:
                        backoff = wait_hint
                    if tries <= self.cfg.max_retries:
                        print(f"[RETRY] status={status}, try={tries}/{self.cfg.max_retries}")
                        continue

                # ここまで来たら再試行しない
                resp.raise_for_status()

            except (requests.Timeout, requests.ConnectionError) as e:
                # 接続系エラーも指数バックオフで再試行
                base = max(self.cfg.base_wait_min, 1.0)
                backoff = (backoff or base) * self.cfg.backoff_factor
                if tries <= self.cfg.max_retries:
                    print(f"[RETRY] network error: {e}, try={tries}/{self.cfg.max_retries}")
                    continue
                raise
            finally:
                self.last_access[parsed.netloc] = time.time()

    def close(self) -> None:
        self.session.close()

if __name__ == "__main__":
    cfg = ScrapeConfig()
    scraper = PoliteScraper(cfg)
    targets = [
        "https://www.python.org/",
        "https://www.python.org/downloads/",
    ]
    try:
        for url in targets:
            try:
                resp = scraper.get(url)
                print(f"[OK] {url} -> {resp.status_code}, {len(resp.content)} bytes")
            except PermissionError as e:
                print(f"[BLOCKED] {e}")
            except requests.HTTPError as e:
                print(f"[HTTP ERROR] {e}")
            except Exception as e:
                print(f"[ERROR] {e}")
    finally:
        scraper.close()