Python と Scrapeless を使って複数のウェブページをスクレイピングする方法

要約

• 複数ページのスクレイピングには入力元台帳と上限付きキューが必要です。 正規化した各 URL の結果を必ず保持し、「ループが最後まで回った＝すべて取得できた」とみなさないでください。
• URL の重複排除とレコードの重複排除は別問題です。 同じ入力 URL は 2 回リクエストすべきではありませんが、同じ内容のページから得られたレコードにはレコードの同一性ルールが必要です。
• 空ページには肯定的な証拠が必要です。 セレクタが見つからないというだけでは、そのソースにレコードが存在しない証明にはなりません。
• エクスポートの検査ではファイルを開き直します。 CSV と JSONL のレコード識別子を比較し、完了・空・失敗の各ページをキューと突き合わせて整合させます。

複数ページをスクレイピングする場合、作業は 1 回の抽出ではなく収集プロセスになります。このプロセスは、何を訪問するよう指示され、どのページを受け入れ、どのページから書き出されたレコードなのかを把握している必要があります。

したがって、複数の Web ページからデータをスクレイピングする方法を学ぶには、まずスコープと記録管理から始めます。Python のループはあくまで取得メカニズムにすぎません。このチュートリアルでは、上限付き URL キューを構築し、各ソースを検証し、ページの結果を保持し、レコードを重複排除し、エクスポートしたファイルを検査します。

公開デモには、直接の HTTP 経路を通じて Quotes to Scrape を使います。同じプログラムには、アカウントキーを必要とし前提条件としてラベル付けされた、Scrapeless Web Unlocker による取得ブランチも含まれています。インタラクティブなページネーションは、ソースがそれを必要とする場合に用いる別のブラウザーワークフローです。

パイプラインの全体像

このパイプラインは、承認済み URL リストを、整合性の取れたページ結果と受理済みデータファイルへと変換します。失敗が空のエクスポートの中に隠れてしまわないよう、これらの成果物は別々に保存してください。

ソースのオペレーションとアプリケーションのスキーマは区別してください。同じソースから複数のレコードが生成されることも、レコードが 0 件であることも、レスポンスが利用不能であることもあります。そのため、ページ数とレコード数には別々の名称が必要です。

URL リストかソースのページネーションを選ぶ

保存済みの URL リストは、ソースのスコープがすでに分かっているタスクに適しています。ページネーションは、ソース自身のナビゲーションから、次に許可されたページを発見していくタスクに適しています。

リスト方式では、元の入力と正規化されたキューを保持します。ページネーション方式では、現在のページ、観測された次リンクまたはインタラクション、訪問済み集合、ページ予算を保持します。任意のページ番号パラメータがサイトの実際のナビゲーションと一致していると決めつけてはいけません。

ページネーションワークフローの比較では、異なる前進条件と停止条件を説明しています。このチュートリアルでは有限のリストを用いるため、試行したすべてのページを入力と直接照合できます。

意味を失わずに URL を正規化する

URL の正規化では、取得に無関係な差異を取り除きつつ、ソースに影響するパラメータは保持すべきです。この例では、フラグメントだけが異なる重複を削除しつつ、パスとクエリはそのまま残します。

URI の構成要素と参照解決では、フラグメントがパスやクエリと区別されています。2 つの URL が似て見えるからといって、安易にクエリパラメータを削除しないでください。異なるレコードやフィルターを指定している可能性があります。

プログラムはまた、入力を承認済みのデモホストに制限します。実際のコレクターは、任意に新しく発見された宛先を受け入れるのではなく、ソースマニフェストを用いるべきです。

前提条件とインストール

この例では、プロジェクト環境に requests と beautifulsoup4 がインストールされた Python が必要です。検証した環境では、Requests 2.34.2 と Beautiful Soup 4.15.0 を使用しました。再現する際は、自分のプロジェクトのバージョンを記録してください。

ダイレクトモードでは公開デモページを使用し、Scrapeless API のクレデンシャルは不要です。unlocker モードには SCRAPELESS_API_KEY、Web Unlocker へのアカウントアクセス、および選択したソースを収集する許可が必要です。
Web Unlocker は、レスポンス指向の取得プロダクトです。その現在のレンダリング構成は、v2 リクエストエンドポイントとネストされた jsRender フィールドを使用します。API ブランチは HTML を要求し、ダイレクトモードと同じソース固有パーサーにフィードします。

実行前にソーススコープを確認してください。例では短い固定リストを逐次的に実行します。これは無制限クローラーでも、許可なくリクエスト量を増やすためのテンプレートでもありません。

完全な境界付きコレクターを実行する

コレクターは入力 URL を重複排除し、キューに入れられた各ページの結果を記録し、受け入れられたレコードのみをエクスポートします。そのソース契約は、検査した Quotes to Scrape レイアウトに一致します。

このデモでは、著者名、タグ、レコードハッシュ、およびソース URL をエクスポートします。引用文テキストは内部でコンテンツアイデンティティを構成するために使用しますが、エクスポートされたデータセットには完全な引用文を再現しません。

注: ダイレクトモードは公開デモページに対して実行されました。unlocker ブランチには実際の Scrapeless API キーとアカウントアクセスが必要であり、実行されていません。そのリクエストフィールドは最新ドキュメントと照合されていますが、ダイレクトモードの結果は Web Unlocker のパフォーマンスとして報告されていません。

import csv
import hashlib
import json
import os
from datetime import datetime, timezone
from pathlib import Path
from urllib.parse import urlsplit, urlunsplit
import requests
from bs4 import BeautifulSoup

INPUT = [
    "https://quotes.toscrape.com/page/1/",
    "https://quotes.toscrape.com/page/2/",
    "https://quotes.toscrape.com/page/2/#top",
    "https://quotes.toscrape.com/page/99/",
    "https://quotes.toscrape.com/missing/"
]
MODE = os.environ.get("FETCH_MODE", "direct")
if MODE not in {"direct", "unlocker"}:
    raise ValueError("Unsupported acquisition mode")

def normalize(url):
    parts = urlsplit(url)
    if parts.scheme != "https" or parts.hostname != "quotes.toscrape.com":
        raise ValueError("URL is outside the permitted demo scope")
    return urlunsplit((parts.scheme, parts.netloc.lower(),
                      parts.path or "/", parts.query, ""))

queue = list(dict.fromkeys(normalize(url) for url in INPUT))
pages = []
records = {}
duplicate_records = 0
with requests.Session() as client:
    for url in queue:
        page = {"requested_url": url}
        try:
            if MODE == "unlocker":
                response = client.post(
                    "https://api.scrapeless.com/api/v2/unlocker/request",
                    headers={"x-api-token": os.environ["SCRAPELESS_API_KEY"]},
                    json={"actor": "unlocker.webunlocker",
                          "proxy": {"country": "ANY"},
                          "input": {"url": url, "jsRender": {
                              "enabled": True, "response": {"type": "html"}}}},
                    timeout=60
                )
                response.raise_for_status()
                envelope = response.json()
                html = envelope.get("data")
                if envelope.get("code") != 200 or not isinstance(html, str):
                    raise ValueError("Expected a successful HTML envelope")
                page["final_url"] = None
            else:
                response = client.get(url, timeout=30)
                response.raise_for_status()
                html = response.text
                page["final_url"] = response.url
            soup = BeautifulSoup(html, "html.parser")
            heading = soup.select_one("h1 a")
            grids = soup.select("div.container > div.row > div.col-md-8")
            if not heading or heading.get_text(strip=True) != "Quotes to Scrape" or not grids:
                raise ValueError("Source identity does not match the demo contract")
            grid = grids[-1]
            cards = grid.select("div.quote")
            if not cards and (grid.select_one("ul.pager") is None or
                              "No quotes found!" not in grid.get_text(" ", strip=True)):
                raise ValueError("Empty state is not confirmed")
            extracted = []
            for card in cards:
                text_node = card.select_one("span.text")
                author_node = card.select_one("small.author")
                if not text_node or not author_node:
                    raise ValueError("Required source field missing")
                text = text_node.get_text(strip=True)
                author = author_node.get_text(strip=True)
                if not text or not author:
                    raise ValueError("Required source field empty")
                digest = hashlib.sha256((text + "\n" + author).encode()).hexdigest()
                extracted.append({"record_id": digest, "author": author,
                                  "tags": [x.get_text(strip=True) for x in card.select("a.tag")],
                                  "source_urls": [url]})
            for record in extracted:
                digest = record["record_id"]
                if digest in records:
                    duplicate_records += 1
                    records[digest]["source_urls"].append(url)
                else:
                    records[digest] = record
            page.update(status="completed" if cards else "empty", records=len(cards),
                        captured_at=datetime.now(timezone.utc).isoformat())
        except (requests.RequestException, ValueError, KeyError) as error:
            page.update(status="failed", reason=type(error).__name__)
        pages.append(page)

rows = list(records.values())
with Path("records.jsonl").open("w", encoding="utf-8") as output:
    for row in rows:
        output.write(json.dumps(row, ensure_ascii=False) + "\n")
with Path("records.csv").open("w", encoding="utf-8", newline="") as output:
    writer = csv.DictWriter(output, fieldnames=["record_id", "author", "tags", "source_urls"])
    writer.writeheader()
    for row in rows:
        writer.writerow({**row, "tags": json.dumps(row["tags"], ensure_ascii=False),
                         "source_urls": json.dumps(row["source_urls"])})
with Path("records.csv").open(encoding="utf-8", newline="") as source:
    csv_rows = list(csv.DictReader(source))
jsonl_rows = [json.loads(line) for line in Path("records.jsonl").read_text().splitlines()]
assert [r["record_id"] for r in csv_rows] == [r["record_id"] for r in jsonl_rows]
summary = {"mode": MODE, "input_urls": len(INPUT), "duplicate_inputs": len(INPUT)-len(queue),
           "queued_pages": len(queue), "completed": sum(p["status"]=="completed" for p in pages),
           "empty": sum(p["status"]=="empty" for p in pages),
           "failed": sum(p["status"]=="failed" for p in pages),
           "unique_records": len(rows), "duplicate_records": duplicate_records,
           "csv_rows": len(csv_rows), "jsonl_rows": len(jsonl_rows)}
assert summary["completed"] + summary["empty"] + summary["failed"] == len(queue)
Path("pages.json").write_text(json.dumps(pages, indent=2), encoding="utf-8")
Path("summary.json").write_text(json.dumps(summary, indent=2), encoding="utf-8")
print(json.dumps(summary))

プログラムを multi_page.py として保存し、プロジェクトディレクトリで実行します。モードを上書きしない場合はダイレクト取得を使用します。そのブランチの認証情報とソース許可の前提条件を満たした後でのみ FETCH_MODE=unlocker を選択してください。

プログラムは、キューに入れられた一意の URL ごとに 1 つのページ結果を保持します。カードをパースする前に想定されるソース見出しを確認し、空ページを受け入れる前に、想定される「結果なし」のメッセージとナビゲーションコンテナの両方を要求します。

Scrapeless でスクレイピングを始めよう

Scrapeless で Web スクレイピングと自動化ワークフローを強化しましょう。
今日登録すると 5 ドル分の無料クレジット を獲得できます — クレジットカードは不要です。

今すぐ Scrapeless ダッシュボード で無料クレジットを受け取りましょう。

各レコードを抽出して検証する

レコード検証では、有意味なソースデータを特定するフィールドを必須とするべきです。本例では、ハッシュに使用する引用文テキストと、観測された著者名の両方を、ページのレコードを確定する前の必須項目としています。

セレクタは検査したデモソースに属します: div.quote、span.text、small.author、および a.tag。メインデータコンテナは、ヘッダー列とは異なる、最終的に一致するコンテンツ列です。ソースが変更されたら、これらの前提を再確認してください。

HTTP レスポンスセマンティクスは、プロトコルの結果を説明するものであり、これらソースフィールドの正しさを説明するものではありません。成功したレスポンスであっても、アプリケーション契約を満たす必要があります。

パーサーは、ページの候補レコードを構築してから、それらを出力セットに追加します。必須フィールドが欠けている場合、そのページは、部分的な取得を完全なものとして黙ってエクスポートするのではなく、拒否されます。

完了・失敗・空ページを追跡する

ページ結果は、キューに入れられた各ソースで何が起きたかを説明する必要があります。このデモでは、有効なレコードページ、確認済みの空ページ、および HTTP またはソース契約の失敗を区別します。

境界付き実行では 5 つの URL 入力が使用されました。フラグメントのみ重複を削除すると、4 つのキュー済みページが残りました。ダイレクト取得の実行では、2 つの完了ページ、1 つの確認済み空ページ、1 つの失敗ページが生成され、20 個の一意レコードが両方のファイルに書き込まれました。

これらのカウントは、境界付きの公開デモを説明するものであり、別サイトでの保証された結果や、マネージド取得のベンチマークではありません。利用不可パスは結果の突き合わせをテストするために含められており、失敗リストから消えるまで繰り返し送信されるわけではありません。

レコードの重複排除とソースの系譜の保持

レコードの重複排除は、抽出後に定義されたコンテンツアイデンティティを比較します。本例では、引用文テキストと著者名を組み合わせてハッシュ化し、同じアイデンティティが再度現れたときにはソース URL を保持します。
これは、この実演で示されたレコード型に対する厳密な同一性ルールです。製品識別子、改訂版、ロケールには、別のキーが必要になる場合があります。差異がタスクにとって重要であるレコードを、1 つの汎用テキストハッシュで結合しないでください。

受け入れたデータには、ソースドキュメントを紐付けて保持します。重複排除では、重複して観測された箇所を消去するのではなく、ソースキャプチャとの関係が維持されるようにします。

CSV と JSONL をエクスポートし、その後に照合する

エクスポートの検証では、シリアライズ後の成果物を読み取る必要があります。プログラムは CSV と JSONL を再オープンし、結果のサマリーを報告する前に、順序付きレコード識別子を比較します。

このデモでは、識別子が一致する 20 行の CSV と 20 件の JSONL レコードが生成されました。summary.json はそれらの件数を保持し、pages.json は個々のページ結果を保持します。

CSV フィールドのクォートがシリアライズの詳細を扱います。フォーマットの妥当性はソースの受理とは別物なので、両方のチェックが必須です。例では、CSV ではリスト値フィールドを JSON 文字列としてセルに格納し、JSONL では配列として格納しています。

保存性チェックは単純です。完了 + 空 + 失敗ページがユニークなキューサイズに等しく、エクスポートされた識別子がファイル間で一致していることを確認します。これらのチェックではソースのすべての事実を証明することはできませんが、落とされたページや不整合な成果物を検出することはできます。

ページネーションにブラウザ操作が必要な場合

インタラクティブなページネーションは、ソースの次の状態にクリックや別のページ操作が必要なときにブラウザワークフローを必要とします。1 つの HTML ドキュメントをレンダリングしても、それだけで永続的なマルチステップセッションになるわけではありません。

そのケースには専用の Agent Browser ワークフローを選択し、実際の次のコントロールを調べ、必要な新規レコードを待機します。同じページ予算、訪問済み状態の追跡、レコード受理ポリシーをブラウザ取得時にも維持します。

パスを見えない形で混在させないでください。ソースが直接 HTTP、Web Unlocker のレスポンス、インタラクティブなブラウザセッションのどれで取得されたかを記録します。取得方法の変更により、ページ状態やコンテンツ範囲が変化しうるからです。

ソースのスコープを責任を持って扱う

複数ページのタスクは、承認されたソースリストの範囲内に留まり、その目的に必要なデータのみを使用すべきです。範囲を広げたコレクターを有効化する前に、利用規約と robots 排除ルールを確認してください。

ホストごとの予算は保守的に設定してください。この例は逐次処理ですが、並行性を用いるパイロットでは、想定されるサービス最大値ではなく、ホストごと 3 ワーカーまでといった許可された収集ポリシーの範囲内に留まる必要があります。

未解決の結果は、レビューのために保持します。失敗リストは完了したジョブの証拠の一部であり、最終レポートから省くべき「不便なもの」ではありません。

結論

Python を使って複数のウェブページをスクレイピングするには、境界付きキュー、ソース固有の受理ルール、ファイルの照合作業が必要です。コレクションプロセスが完了したと見なせるのは、キューに入れたすべての入力に結果があり、書き出した成果物が受理されたレコードと一致するときだけです。

許可された静的ソースには直接取得を、レスポンス指向のパスが必要なタスクにはドキュメント化された Web Unlocker ブランチを、インタラクティブなページネーションには別個のブラウザワークフローを使用します。これらの選択をまたいでも、台帳と出力契約は安定して保ってください。

照合済みコレクションジョブを構築する準備はできましたか？

Scrapeless で境界付きワークフローを計画し、取得要件を 最新の料金 と照らして評価してください。Telegram でキューおよびエクスポート契約について議論できます。

FAQ

Q: 公開されていることは、複数ページの収集許可を意味しますか？

公開されていることは、あらゆる収集や利用の許可を意味するわけではありません。ソースの利用規約や適用される要件を確認し、意図する活動が不確かな場合は、資格のある法律専門家に相談してください。

Q: いつコレクターはページネーションではなく URL リストを使うべきですか？

許可されたソース集合が既知である場合は URL リストを使用してください。タスクが、調査したリンクやインタラクションから次のページを特定する場合は、ソースのページネーションを利用します。

Q: 空のセレクタ結果は、有効な空ページと見なせますか？

空のセレクタ結果だけでは未解決です。ソース側の明示的な空状態の証拠を要求し、ページの同一性を確認してから、有効な空結果として記録してください。

Q: すべての入力 URL を処理したことは、どのように示せますか？

生の入力、重複排除されたキュー、およびキューに入れた各ソースにつき 1 つの結果を保存します。そのうえで、完了・空・失敗の各結果をキューと照合します。

Q: Web Unlocker ブランチには、別途設定したプロキシが必要ですか？

表示されているマネージドリクエストは、ドキュメント化された country フィールドを使用します。スタンドアロンのプロキシ製品をクライアント側で使用する場合には、別途割り当てられたプロキシ認証情報が必要です。

Q: チャレンジページやアクセス拒否ドキュメントはどうすべきですか？

コレクターは不適切なソースコンテンツを拒否し、取得理由を保持する必要があります。HTTP の成功ステータスが返ってきても、それだけではチャレンジドキュメントを有効なデータページとして受け入れる理由にはなりません。

Q: ソースのマークアップがローテーションした場合、何を変更すべきですか？

ページの同一性と必要なフィールドを再確認し、ソース固有のセレクターと空状態チェックを更新してください。マークアップが変化したことを、ソースにレコードが存在しない証拠として扱ってはいけません。

Q: チュートリアルではどの程度の並行性を使うべきですか？

デモされているコレクターは逐次実行されます。許可されたパイロットでは、1 ホストあたり 3 ワーカーのような控えめな上限を設定し、その運用を評価してから拡大してください。

Q: このワークフローには AI エージェントが必要ですか？

Python のキュー、抽出処理、およびエクスポートチェックは、AI エージェントなしで動作します。エージェントは、受理されたレコードを消費したり、それらの決定論的コンポーネントを取り巻くタスクを監督したりすることができます。

Q: リクエストされた URL は、最終 URL として保存すべきですか、それとも正規 URL として保存すべきですか？

リクエストされた URL は、観測された最終 URL または正規の識別子とは別に保存してください。ダイレクトモードではレスポンスの最終 URL を記録しますが、提示された API ブランチでは、リダイレクト先をでっち上げるのではなく、そのフィールドを「不明」のままにしておきます。