はじめに
仕事でスクレイピングをする機会がまたやってきそうだ。
しかし、その後の処理のスピード面から、Pythonでできた方がよさそうだと思ったので、今回はPythonでスクレイピングに挑戦し、その記録をここに書いていこうと思う。
学部生の頃に一度習ったことがあり、BeautifulSoupを使うということは覚えているのだが、当時はPyhonはおろかRの知識も乏しかったので、もったいないことをしたなと思う。改めて勉強していこう。
方針
時間も限られているので、今回は以下の記事で実行したRのコードをChat-GPTにPython用に改変してもらい、それをなぞりながら練習するというやり方で進めていきたい。
一応Rでのコードも貼り付けておく
library(rvest)
# データを保存したいフォルダを指定
save_dir <- here::here("data/pop_by_mesh")
# 保存したいフォルダがない場合に作成
if (!dir.exists(save_dir)) {
dir.create(save_dir)
}
base_url <- "https://www.e-stat.go.jp"
for (i in 1:8) {
url <- paste0(
"https://www.e-stat.go.jp/gis/statmap-search?page=",
i,
"&type=1&toukeiCode=00200521&toukeiYear=2020&aggregateUnit=H",
"&serveyId=H002005112020&statsId=T001141&datum=2011"
)
html <- read_html_live(url)
Sys.sleep(1)
links <- html |>
html_elements("a") |>
html_attr("href")
csv_links <- links[grepl("data\\?statsId=T001141", links)]
full_urls <- paste0(base_url, csv_links)
for (j in seq_along(full_urls)) {
code <- sub(".*code=([0-9]+).*", "\\1", csv_links[j])
zip_path <- file.path(save_dir, paste0("pop_mesh", code, ".zip"))
txt_path <- file.path(save_dir, paste0("pop_mesh", code, ".txt"))
# 既にファイルがある場合は次のループへ
if (file.exists(txt_path)) next
tryCatch({
download.file(full_urls[j], destfile = zip_path, mode = "wb")
Sys.sleep(1)
unzip_files <- unzip(zip_path, list = TRUE)$Name
unzip(zip_path, exdir = save_dir)
for (original_name in unzip_files) {
if (grepl("\\.txt$", original_name, ignore.case = TRUE)) {
old_path <- file.path(save_dir, original_name)
new_path <- txt_path
file.rename(old_path, new_path)
}
}
file.remove(zip_path)
}, error = function(e) {
warning(paste("エラー：", full_urls[j]))
})
Sys.sleep(2)
}
}出力してもらったPythonコード
Chat-GPTに頼んでできたのがこれだー。
import re
import shutil
import zipfile
from pathlib import Path
from typing import Optional, Iterable, Tuple
import requests
from bs4 import BeautifulSoup
from tqdm import tqdm
from concurrent.futures import ThreadPoolExecutor, as_completed
import threading
from selenium import webdriver
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
# ==============================
# 設定
# ==============================
SAVE_DIR = Path("data/pop_by_mesh")
SAVE_DIR.mkdir(parents=True, exist_ok=True)
BASE_URL = "https://www.e-stat.go.jp"
MAX_WORKERS = 6 # 並列ダウンロード数（回線や先方負荷を見て調整）
REQUEST_TIMEOUT = 60 # 秒
# ==============================
# Selenium セットアップ
# ==============================
def get_selenium_driver() -> webdriver.Chrome:
"""Chromeをヘッドレスで安定起動するモダン設定。"""
options = Options()
options.add_argument("--headless=new")
options.add_argument("--no-sandbox")
options.add_argument("--disable-gpu")
options.add_argument("--window-size=1280,1800")
options.add_argument("--disable-dev-shm-usage")
options.add_argument("--disable-blink-features=AutomationControlled")
options.add_argument(
"user-agent=Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
"AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126 Safari/537.36"
)
driver = webdriver.Chrome(options=options)
driver.set_page_load_timeout(60)
driver.set_script_timeout(60)
return driver
# ==============================
# 動的ページ読み込み
# ==============================
def read_html_live(
url: str,
driver: webdriver.Chrome,
wait_for_css: Optional[str] = None,
timeout: int = 20
) -> str:
"""JS実行後のHTMLを取得。"""
driver.get(url)
# 1) ドキュメントの読み込み完了を待つ
WebDriverWait(driver, timeout).until(
lambda d: d.execute_script("return document.readyState") == "complete"
)
# 2) 目的の要素が現れるまで待つ（指定がある場合）
if wait_for_css:
WebDriverWait(driver, timeout).until(
EC.presence_of_all_elements_located((By.CSS_SELECTOR, wait_for_css))
)
return driver.page_source
# ==============================
# CSVリンク抽出
# ==============================
def extract_csv_links(html: str) -> list[str]:
soup = BeautifulSoup(html, "html.parser")
links = [a.get("href") for a in soup.find_all("a", href=True)]
csv_links = [link for link in links if link and "data?statsId=T001141" in link]
return csv_links
# ==============================
# requests Session（スレッドローカル）
# ==============================
_thread_local = threading.local()
def get_session() -> requests.Session:
"""各スレッド専用のSessionを用意（接続プール＆自動リトライ付き）。"""
if getattr(_thread_local, "session", None) is None:
s = requests.Session()
retries = Retry(
total=3,
backoff_factor=0.8, # 0.8, 1.6, 3.2秒…指数バックオフ
status_forcelist=(429, 500, 502, 503, 504),
allowed_methods=("GET",)
)
adapter = HTTPAdapter(pool_connections=MAX_WORKERS, pool_maxsize=MAX_WORKERS, max_retries=retries)
s.mount("http://", adapter)
s.mount("https://", adapter)
_thread_local.session = s
return _thread_local.session
# ==============================
# ZIPダウンロード＆解凍（並列ワーカー用）
# ==============================
def sanitize_member_name(name: str) -> str:
"""Zipエントリ名の安全化（ディレクトリトラバーサル対策の簡易版）。"""
name = name.replace("\\", "/")
name = name.lstrip("/.")
# サブディレクトリを無視して最後のパス要素だけ使う
return name.split("/")[-1]
def download_and_extract_zip_worker(task: Tuple[str, str]) -> tuple[str, bool, str]:
"""
並列ワーカー：ZIPを取得して.txtを保存。
Returns: (url, success, message)
"""
full_url, code = task
zip_path = SAVE_DIR / f"pop_mesh{code}.zip"
txt_path = SAVE_DIR / f"pop_mesh{code}.txt"
if txt_path.exists():
return (full_url, True, "skipped (exists)")
try:
session = get_session()
with session.get(full_url, stream=True, timeout=REQUEST_TIMEOUT) as resp:
resp.raise_for_status()
with open(zip_path, "wb") as f:
for chunk in resp.iter_content(chunk_size=8192):
if chunk:
f.write(chunk)
found_txt = False
with zipfile.ZipFile(zip_path, "r") as zf:
for member in zf.namelist():
if re.search(r"\.txt$", member, re.IGNORECASE):
safe_name = sanitize_member_name(member)
extract_tmp = SAVE_DIR / safe_name
zf.extract(member, SAVE_DIR)
# 抽出元がサブディレクトリでも safe_name で移動
src = SAVE_DIR / member
# srcが存在しない（ディレクトリ含む）場合に備えて
if not src.exists() and extract_tmp.exists():
src = extract_tmp
shutil.move(str(src), txt_path)
found_txt = True
break
zip_path.unlink(missing_ok=True)
if not found_txt:
return (full_url, False, "no .txt in zip")
return (full_url, True, "ok")
except Exception as e:
# エラー時はZIPを片付ける
try:
zip_path.unlink(missing_ok=True)
except Exception:
pass
return (full_url, False, f"error: {e}")
# ==============================
# メイン処理（並列ダウンロード）
# ==============================
def main():
driver = get_selenium_driver()
try:
# 1) 各ページからCSVリンクを収集
all_tasks: list[tuple[str, str]] = []
for i in tqdm(range(1, 9), desc="ページ巡回"):
url = (
f"https://www.e-stat.go.jp/gis/statmap-search?page={i}"
"&type=1&toukeiCode=00200521&toukeiYear=2020&aggregateUnit=H"
"&serveyId=H002005112020&statsId=T001141&datum=2011"
)
html = read_html_live(
url,
driver,
wait_for_css="a[href*='data?statsId=T001141']",
timeout=25,
)
csv_links = extract_csv_links(html)
full_urls = [BASE_URL + link for link in csv_links]
# ダウンロードタスク（(url, code)）を積む
for full_url, csv_link in zip(full_urls, csv_links):
m = re.search(r"code=([0-9]+)", csv_link)
code = m.group(1) if m else "unknown"
all_tasks.append((full_url, code))
# 2) 並列ダウンロード実行
successes = 0
failures = 0
skipped = 0
with ThreadPoolExecutor(max_workers=MAX_WORKERS) as ex:
futures = [ex.submit(download_and_extract_zip_worker, task) for task in all_tasks]
for fut in tqdm(as_completed(futures), total=len(futures), desc="ダウンロード"):
url, ok, msg = fut.result()
if ok and msg.startswith("ok"):
successes += 1
elif ok and "skipped" in msg:
skipped += 1
else:
failures += 1
# 進捗の詳細を見たい場合は次の行のコメントアウトを外す
# print(f"[{ 'OK' if ok else 'NG' }] {msg} - {url}")
print(f"\nDone. ok={successes}, skipped={skipped}, failed={failures}")
finally:
driver.quit()
if __name__ == "__main__":
main()Pythonにしてもらうついでに並列処理を可能にし、高速でダウンロードできるようにしてもらったため、結構複雑な処理になっている。
ポイントとしてはJavaScriptに対応したスクレイピングをするためにSeleniumを使用しているところだと思う。Rで言えばread_html()ではなくread_html_live()を使用しているのと同じような違いだ。
また、僕がread_html_live()をもとに作るよう依頼したせいか、read_html_live()関数を定義して使用しているようだ。なんだかR訛りのあるPythonコードになっているような気がする。
まとめ
実行してみたのだが、並列処理が可能になっているおかげでだいぶ高速でダウンロードできるようになった。
これまではR信者としてなんとかRで実行しようとしていたが、スピードも求められる仕事ではPythonの恩恵にあずかることになる気がしてきた。
詳細をまとめようと思ったのだが、いかんせん見慣れないコードが並んでいるので、もう少し噛み砕いてからtipsの方でまとめようと思う。
こんな感じでできるのか、というのがわかっただけでも収穫としよう。