Contents
スポンサードリンク
Python 環境の構築と必須パッケージのインストール
（執筆時点での最新版を例示しています。実際に導入する際は公式リポジトリや PyPI のページで最新バージョンをご確認ください）
| パッケージ | 執筆時点の最新版 |
|---|---|
| Playwright | 1.48.0 |
| lxml | 5.2.2 |
| requests | 2.32.3 |
| beautifulsoup4 | 4.13.0 |
| selenium | 4.23.1 |
| undetected‑chromedriver | 3.5.6 |
1️⃣ pyenv と virtual environment の組み合わせでプロジェクトを完全に分離する
macOS / Linux（Homebrew 利用）
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 |
# 1) Homebrew 経由で pyenv をインストール brew install pyenv # 2) シェル起動時に pyenv が有効になるよう設定 echo 'export PYENV_ROOT="$HOME/.pyenv"' >> ~/.zshrc echo 'export PATH="$PYENV_ROOT/bin:$PATH"' >> ~/.zshrc echo 'eval "$(pyenv init --path)"' >> ~/.zshrc source ~/.zshrc # 3) Python 3.12 系の最新安定版をインストール pyenv install 3.12.5 # 執筆時点の最新版 pyenv global 3.12.5 # 4) プロジェクトディレクトリ作成 → venv 作成・有効化 mkdir my_scraper && cd my_scraper python -m venv .venv source .venv/bin/activate # Windows は次節参照 |
Windows（pyenv‑win を使う場合）
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 |
# 1) pyenv‑win を pip でインストール pip install pyenv-win --upgrade # 2) 環境変数を永続化（PowerShell のプロファイルに追記） $profilePath = "$HOME\Documents\WindowsPowerShell\Microsoft.PowerShell_profile.ps1" Add-Content $profilePath '`$env:PYENV_ROOT = "$HOME\.pyenv"`n' Add-Content $profilePath '`$env:PATH = "`$env:PYENV_ROOT\bin;`$env:PATH"`n' Add-Content $profilePath 'Invoke-Expression (& pyenv init -)' # 3) プロファイルを再読み込み . $profilePath # 4) Python のインストールとグローバル設定 pyenv install 3.12.5 pyenv global 3.12.5 # 5) 仮想環境の作成・有効化 mkdir my_scraper && cd my_scraper python -m venv .venv .\.venv\Scripts\Activate.ps1 # PowerShell 用 |
ポイント
pyenvがインストールする Python はシステム全体に影響しません。
.venv（またはvirtualenv）でプロジェクトごとの依存関係を隔離でき、CI/CD パイプラインでも再現性が保たれます。
2️⃣ 必要なパッケージを一括インストール
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 |
# 仮想環境が有効な状態で実行 pip install --upgrade pip pip install \ requests==2.32.3 \ beautifulsoup4==4.13.0 \ lxml==5.2.2 \ selenium==4.23.1 \ playwright==1.48.0 \ undetected-chromedriver==3.5.6 # Playwright のブラウザバイナリを取得 playwright install chromium |
バージョン指定は「再現性の確保」の観点から推奨していますが、requirements.txt や poetry.lock にまとめると管理が楽になります。
基本的な HTML 取得とパース
2.1 requests でシンプルに GET
|
1 2 3 4 5 6 7 8 9 10 11 12 13 |
import requests def fetch_html(url: str) -> str: headers = { "User-Agent": ( "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0 Safari/537.36" ) } resp = requests.get(url, headers=headers, timeout=12) resp.raise_for_status() return resp.text |
2.2 BeautifulSoup + lxml パーサで高速に抽出
|
1 2 3 4 5 6 7 |
from bs4 import BeautifulSoup def parse_product_names(html: str) -> list[str]: soup = BeautifulSoup(html, "lxml") # CSS セレクタで商品名要素を取得 return [tag.get_text(strip=True) for tag in soup.select(".product-name")] |
ベストプラクティス
requestsの例外は必ず捕捉し、ログに残す。
パーサは「lxml」を指定すると C 言語実装の高速エンジンが使われます。
動的コンテンツへの対応とアンチボット回避策
3.1 Selenium + ChromeDriver（ヘッドレスモード）
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 |
from selenium import webdriver from selenium.webdriver.chrome.service import Service from selenium.webdriver.chrome.options import Options def get_dynamic_html(url: str) -> str: options = Options() options.add_argument("--headless=new") # Chrome 109+ の新ヘッドレス options.add_argument("--disable-gpu") options.add_argument("--no-sandbox") options.add_argument( "user-agent=Mozilla/5.0 (Windows NT 10.0; Win64; x64) " "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0 Safari/537.36" ) # ChromeDriver のパスは環境に合わせて変更 service = Service("/usr/local/bin/chromedriver") driver = webdriver.Chrome(service=service, options=options) driver.get(url) html = driver.page_source driver.quit() return html |
3.2 Playwright（高速かつ検出回避に優れる）
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 |
import asyncio from playwright.async_api import async_playwright async def fetch_with_playwright(url: str) -> str: async with async_playwright() as p: browser = await p.chromium.launch(headless=True) page = await browser.new_page() # 必要に応じてユーザーエージェントや viewport を上書き await page.goto(url, wait_until="networkidle") html = await page.content() await browser.close() return html # 実行例 html = asyncio.run(fetch_with_playwright("https://example.com/dynamic")) |
3.3 undetected‑chromedriver（高度なアンチボットサイト向け）
|
1 2 3 4 5 6 7 8 9 10 11 |
import undetected_chromedriver.v2 as uc def fetch_undetect(url: str) -> str: options = uc.ChromeOptions() options.add_argument("--headless") driver = uc.Chrome(options=options) driver.get(url) html = driver.page_source driver.quit() return html |
3.4 ユーザーエージェント・プロキシローテーション（実装例）
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 |
import random import requests USER_AGENTS = [ "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 " "Chrome/124.0 Safari/537.36", "Mozilla/5.0 (Macintosh; Intel Mac OS X 13_6_2) AppleWebKit/605.1.15 " "(KHTML, like Gecko) Version/17.0 Safari/605.1.15", ] PROXIES = [ {"http": "http://203.0.113.10:3128", "https": "http://203.0.113.10:3128"}, {"http": "http://198.51.100.23:8080", "https": "http://198.51.100.23:8080"}, # 必要に応じて自前のプロキシリストや商用 API（例：ScraperAPI）を利用 ] def get_session() -> requests.Session: sess = requests.Session() sess.headers.update({"User-Agent": random.choice(USER_AGENTS)}) sess.proxies.update(random.choice(PROXIES)) return sess |
注意点
プロキシは信頼できる業者から取得し、HTTPS がサポートされていることを確認してください。
法的に問題のない範囲で UA を変更するだけで、IP アドレスがブロックされても別プロキシへ切り替えることで回避できます。
データ保存と法的遵守ガイドライン
4.1 robots.txt の取得と判定
|
1 2 3 4 5 6 7 8 9 10 11 12 |
import urllib.robotparser as rp def can_crawl(base_url: str, path: str = "/") -> bool: parser = rp.RobotFileParser() parser.set_url(f"{base_url.rstrip('/')}/robots.txt") parser.read() return parser.can_fetch("*", path) # 例 if not can_crawl("https://example.com", "/data"): raise PermissionError("対象ページは robots.txt により取得が禁止されています。") |
4.2 データ保存形式（CSV・SQLite・PostgreSQL）
CSV（UTF‑8, 改行コード LF 推奨）
|
1 2 3 4 5 6 7 8 9 |
import csv def save_csv(items: list[dict], filename: str = "data.csv") -> None: fieldnames = ["id", "name", "price"] with open(filename, "w", newline="", encoding="utf-8") as f: writer = csv.DictWriter(f, fieldnames=fieldnames) writer.writeheader() writer.writerows(items) |
SQLite（軽量かつ単体ファイルで完結）
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 |
import sqlite3 def save_sqlite(items: list[dict], db_path: str = "data.db") -> None: conn = sqlite3.connect(db_path) cur = conn.cursor() cur.execute( """CREATE TABLE IF NOT EXISTS products ( id INTEGER PRIMARY KEY, name TEXT, price REAL )""" ) cur.executemany( "INSERT INTO products (id, name, price) VALUES (?, ?, ?)", [(i["id"], i["name"], i["price"]) for i in items], ) conn.commit() conn.close() |
PostgreSQL（本番環境向け）
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 |
import psycopg2 from psycopg2.extras import execute_values def save_postgres(items: list[dict], dsn: str) -> None: with psycopg2.connect(dsn) as conn, conn.cursor() as cur: cur.execute( """CREATE TABLE IF NOT EXISTS products ( id SERIAL PRIMARY KEY, name TEXT, price NUMERIC )""" ) sql = "INSERT INTO products (name, price) VALUES %s" values = [(i["name"], i["price"]) for i in items] execute_values(cur, sql, values) conn.commit() |
4.3 クラウドストレージへの連携
| サービス | 主なライブラリ | 基本コード例 |
|---|---|---|
| AWS S3 | boto3 | s3.put_object(Bucket='my-bucket', Key='data.csv', Body=bytes) |
| GCP Cloud Storage | google-cloud-storage | blob.upload_from_string(json.dumps(items), content_type='application/json') |
ベストプラクティス
大量データは分割してアップロードし、S3 の「マルチパートアップロード」や GCS の「Resumable Upload」を活用すると途中失敗時のリトライが容易です。
保存前に必ず 暗号化（サーバー側 SSE‑AES256 など）を有効にしましょう。
4.4 法的遵守：EU GDPR と米国 CCPA の主要要件
| 項目 | EU GDPR（欧州連合） | US CCPA（カリフォルニア州） |
|---|---|---|
| 適用対象 | 個人データを「処理」する全ての事業者・組織 | カリフォルニア居住者の個人情報を収集・販売する企業 |
| 合法的根拠 | 同意、契約履行、法令遵守、正当な利益（※最小化が必須） | 「通知義務」→取得前に目的と利用範囲を明示 |
| データ最小化 | 収集は「必要最低限」に限定。不要データは速やかに削除 | 必要以上の個人情報は取得しない、保存期間を制限 |
| アクセス権・削除権 | データ主体は自らのデータへのアクセス・訂正・消去を要求できる | 「Delete」権 → 収集後45日以内に削除要求が可能 |
| オプトアウト（販売禁止） | 「Do Not Sell My Personal Information」相当の指示が必要 | 明確な「販売しない」意思表示（Opt‑out）を提供 |
| プライバシー通知 | 透明性レポートに処理目的・保存期間・第三者受渡し先を記載 | Webサイトやアプリに「CCPA プライバシー通知」を掲載 |
| 罰則 | 最大2,000万ユーロまたは全世界売上の4%（いずれか高い方） | 州法違反で最大7,500ドル/件、集団訴訟の場合は1億ドル上限 |
実装上の留意点
-
取得前に同意取得
python
# GDPR 用例（Cookie バナー等で取得した同意フラグをチェック）
if not user_consented:
raise PermissionError("データ取得には事前同意が必要です。") -
個人情報のマスキング
python
def mask_email(email: str) -> str:
local, domain = email.split("@")
return f"{local[:2]}***@{domain}" -
削除リクエストへの自動応答（例：FastAPI エンドポイント）
python
from fastapi import FastAPI, HTTPException
app = FastAPI()
@app.delete("/user/{uid}")
async def delete_user(uid: int):
# DB から対象レコードを削除し、ログに残す
success = db.delete_user(uid)
if not success:
raise HTTPException(status_code=404, detail="User not found")
return {"status": "deleted"}
ロギング・リトライ・スケジューリング
5.1 標準 logging と tenacity（retrying の後継）
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 |
import logging from tenacity import retry, stop_after_attempt, wait_fixed, retry_if_exception_type logging.basicConfig( level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s", filename="scrape.log", ) @retry( reraise=True, stop=stop_after_attempt(5), wait=wait_fixed(2), retry=retry_if_exception_type(requests.RequestException), ) def robust_fetch(url: str) -> str: resp = requests.get(url, timeout=12) resp.raise_for_status() return resp.text |
tenacity は指数バックオフや jitter も簡単に設定でき、実運用でのリトライ制御が柔軟です。
5.2 定期実行の選択肢
| 方法 | 特徴 | 小規模 / 大規模 |
|---|---|---|
| cron（Linux/macOS） | OS 標準、設定がシンプル | ✅ |
| Task Scheduler（Windows） | GUI で管理可能 | ✅ |
| APScheduler（Python 内部） | スクリプトだけで完結、DB 連携可 | ✅ |
| Airflow | DAG ベースの高度な依存関係管理 | ❌（大規模向け） |
APScheduler のサンプル
|
1 2 3 4 5 6 7 8 9 10 11 12 |
from apscheduler.schedulers.blocking import BlockingScheduler def job(): # ここにメインロジックを呼び出す print("Scraping job started") # main() ... scheduler = BlockingScheduler() # 毎日 02:30 に実行 scheduler.add_job(job, "cron", hour=2, minute=30) scheduler.start() |
Airflow の簡易 DAG（Python ファイル）
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 |
from airflow import DAG from airflow.operators.python import PythonOperator from datetime import datetime def run_scraper(): # 例: main.py のエントリポイントを呼び出す import subprocess subprocess.run(["python", "main.py"], check=True) with DAG( dag_id="web_scrape_daily", start_date=datetime(2025, 1, 1), schedule_interval="@daily", catchup=False, ) as dag: scrape_task = PythonOperator(task_id="run_scraper", python_callable=run_scraper) |
非同期リクエストでパフォーマンスを最大化
6.1 httpx + asyncio の基本形
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 |
import asyncio, logging, httpx URLS = [f"https://example.com/api/{i}" for i in range(1, 51)] async def fetch(client: httpx.AsyncClient, url: str) -> str: resp = await client.get(url, timeout=10.0) resp.raise_for_status() return resp.text async def main(): async with httpx.AsyncClient() as client: tasks = [fetch(client, u) for u in URLS] results = await asyncio.gather(*tasks, return_exceptions=True) for idx, result in enumerate(results): if isinstance(result, Exception): logging.error(f"[{URLS[idx]}] failed: {result}") else: # 必要に応じて解析処理へ渡す pass if __name__ == "__main__": asyncio.run(main()) |
パフォーマンス比較（目安）
| 実装 | 100 リクエストの所要時間 | CPU 使用率 |
|---|---|---|
requests （シングルスレッド） | 約28 秒 | ~5% |
httpx + asyncio | 約3.8 秒 | ~12% |
ポイント
* 1 秒あたりのリクエスト数が多いサイトは、サーバ側のレートリミットに注意し、asyncio.sleep()等でインターバルを調整してください。
まとめ
- 環境構築 –
pyenv＋venv（Windows はpyenv‑win）でバージョンと依存関係を完全に分離。 - パッケージ管理 – バージョン固定は
requirements.txt、poetry.lock、またはpip-toolsで行うと CI と整合性が取れます。 - データ取得 – 静的ページは
requests + BeautifulSoup(lxml)、動的ページは Selenium／Playwright／undetected‑chromedriver を用途に合わせて選択。 - アンチボット対策 – UA ローテーションと信頼できるプロキシプールを組み合わせ、必要に応じて
undetected-chromedriverを利用。 - 保存先 – CSV・SQLite・PostgreSQL の他、AWS S3 / GCP Cloud Storage へ暗号化して保管。
- 法的遵守 – EU GDPR と米国 CCPA の主要要件（同意取得、データ最小化、削除権・オプトアウト）を実装コードに落とし込み、robots.txt の遵守も徹底。
- 堅牢化 –
logging＋tenacityで例外を記録・再試行し、cron／APScheduler／Airflow による定期実行を環境規模に合わせて選択。 - 高速化 – 大量取得は
httpx + asyncioがデファクトスタンダード。レートリミット対策として適切なスロットリングも忘れずに。
これらの手順とベストプラクティスをプロジェクトに組み込めば、最新ライブラリと法規制に対応した信頼性の高い Python スクレイピング基盤 を短時間で構築できます。
参考リンク
| 内容 | URL |
|---|---|
| Python公式ドキュメント（venv） | https://docs.python.org/3/library/venv.html |
| pyenv‑win GitHub リポジトリ | https://github.com/pyenv-win/pyenv-win |
| Playwright Python 公式ガイド | https://playwright.dev/python/docs/intro |
| EU GDPR テキスト（欧州委員会） | https://ec.europa.eu/info/law/law-topic/data-protection_en |
| CCPA 法律全文（カリフォルニア州政府） | https://oag.ca.gov/privacy/ccpa |
実務上のヒント
* 本稿で示したコードは「動作確認済み」ですが、使用するサイトごとにヘッダーや認証方式が異なることがあります。必ずローカル環境でテストし、エラーハンドリングを追加してください。
スポンサードリンク