0
1

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

【第2回】CSVダウンロードを自動化し、URL列を抽出する

0
Posted at

はじめに

第1回では、Playwright(Firefox)でログインしてCSVページへ遷移する最小実行を作りました。
第2回では、実際に次の2点を動かします。

  • CSVダウンロードボタンを押してファイルを保存する
  • 保存したCSVからURL列を取り出して list[str] にする

ここまでできると、次回から「URL巡回して情報抽出」の本処理に入れます。


この回のゴール

最終的に、以下の関数が動作する状態を目指します。

  • download_csv(page, save_dir, selector) -> Path
  • extract_urls_from_csv(csv_path, column_name) -> list[str]

プロジェクト構成(第2回時点)

ticket-workflow/
  ├─ .env
  ├─ .env.example
  ├─ requirements.txt
  ├─ data/
  │  └─ downloads/
  └─ src/
     ├─ config.py
     ├─ browser_client.py
     ├─ csv_pipeline.py
     └─ main.py

1. 依存パッケージを追加する

requirements.txt を次のようにします。

playwright
python-dotenv
pydantic
pandas

反映コマンド:

pip install -r requirements.txt

2. config.py を少し強化する

第1回の src/config.py はそのまま使えます。
第2回では、ダウンロード先フォルダも環境変数で管理するように1項目だけ増やします。

.env.example に追加:

DOWNLOAD_DIR=./data/downloads

src/config.pySettings に追加:

# ダウンロード先ディレクトリ
download_dir: str

load_settings() の返却にも追加:

download_dir=os.getenv("DOWNLOAD_DIR", "./data/downloads"),

これで、コード側に固定パスをベタ書きせずに済みます。


3. CSVダウンロード処理を作る(src/browser_client.py

以下は、実装例です。

from pathlib import Path
from playwright.sync_api import Page


def download_csv(page: Page, save_dir: str, download_button_selector: str) -> Path:
    """
    CSVダウンロードボタンを押して、ファイルを指定フォルダへ保存する。

    Parameters
    ----------
    page : Page
        Playwrightのページオブジェクト。
    save_dir : str
        CSVファイルを保存したいディレクトリパス。
    download_button_selector : str
        CSVダウンロードボタンのセレクタ(CSSやXPath)。

    Returns
    -------
    Path
        保存したCSVファイルのフルパス。
    """
    # 保存先ディレクトリがなければ作成します。
    # parents=True で親フォルダがなくてもまとめて作れます。
    save_path = Path(save_dir)
    save_path.mkdir(parents=True, exist_ok=True)

    # expect_download() を使う理由:
    # 「これから発生するダウンロード」をPlaywrightに待ってもらい、
    # ダウンロードオブジェクトを安全に受け取るためです。
    with page.expect_download() as download_info:
        # ダウンロードボタンをクリックします。
        page.click(download_button_selector)

    # 実際のダウンロード結果を取得します。
    download = download_info.value

    # ダウンロードされた元のファイル名を取得します。
    # 取得できないケースに備えて、デフォルト名を用意しておきます。
    filename = download.suggested_filename or "downloaded.csv"

    # 保存先の最終パスを作ります。
    final_path = save_path / filename

    # ダウンロードファイルを指定パスへ保存します。
    download.save_as(str(final_path))

    return final_path

4. CSVからURL列を抽出する(src/csv_pipeline.py

次に、保存したCSVから対象列だけを取り出します。

from pathlib import Path
import pandas as pd


def extract_urls_from_csv(csv_path: Path, column_name: str) -> list[str]:
    """
    CSVファイルから指定列を読み取り、URLリストを返す。

    Parameters
    ----------
    csv_path : Path
        CSVファイルのパス。
    column_name : str
        URLが入っている列名(例: target_url)。

    Returns
    -------
    list[str]
        URL文字列のリスト(空欄や重複は除去)。
    """
    # CSVの文字コードは環境によって異なるため、
    # まずUTF-8で試し、失敗したらcp932(Shift_JIS系)で再試行します。
    try:
        df = pd.read_csv(csv_path, encoding="utf-8")
    except UnicodeDecodeError:
        df = pd.read_csv(csv_path, encoding="cp932")

    # 指定列がない場合は、どの列があるかを表示して原因を特定しやすくします。
    if column_name not in df.columns:
        raise ValueError(
            f"指定列 '{column_name}' が見つかりません。利用可能列: {list(df.columns)}"
        )

    # 欠損値を除去し、文字列化して前後空白を取り除きます。
    # さらに空文字を除去して、重複URLも取り除きます。
    series = (
        df[column_name]
        .dropna()
        .astype(str)
        .map(lambda x: x.strip())
    )

    urls = [u for u in series if u]

    # dict.fromkeys() を使うと順序を保ったまま重複除去できます。
    unique_urls = list(dict.fromkeys(urls))

    return unique_urls

5. main.py でつなぐ

ここでは「ログイン→CSVページ→ダウンロード→URL列抽出」までを1回で実行します。

from playwright.sync_api import sync_playwright

from config import load_settings
from browser_client import download_csv
from csv_pipeline import extract_urls_from_csv


def run() -> None:
    settings = load_settings()

    with sync_playwright() as p:
        # 第2回もFirefoxを使用します。
        browser = p.firefox.launch(headless=False)
        context = browser.new_context()
        page = context.new_page()

        # ログインページへ移動
        page.goto(settings.login_url)

        # ログインフォーム入力
        page.fill(settings.sel_login_username, settings.username)
        page.fill(settings.sel_login_password, settings.password)
        page.click(settings.sel_login_submit)

        # CSVページへ移動
        page.goto(settings.csv_page_url)

        # CSVをダウンロード
        csv_path = download_csv(
            page=page,
            save_dir=settings.download_dir,
            download_button_selector=settings.sel_csv_download_button,
        )

        # CSVからURL列を抽出
        urls = extract_urls_from_csv(
            csv_path=csv_path,
            column_name=settings.csv_url_column_name,
        )

        # 結果確認(第2回時点ではprintでOK)
        print(f"CSVファイル: {csv_path}")
        print(f"抽出URL件数: {len(urls)}")

        # 先頭5件だけ表示(大量出力を避ける)
        for i, url in enumerate(urls[:5], start=1):
            print(f"{i}: {url}")

        browser.close()


if __name__ == "__main__":
    run()

6. 実行して確認する

python src/main.py

確認ポイント:

  • data/downloads にCSVが保存される
  • コンソールにURL件数が表示される
  • 先頭URLが期待どおりの値になっている

ハマりどころと対策

1) ダウンロードが始まらない

  • SEL_CSV_DOWNLOAD_BUTTON がずれている可能性があります。
  • まず手動でそのボタンが本当に押せるか確認します。

2) CSVは保存されるがURL件数が0

  • CSV_URL_COLUMN_NAME が違う可能性があります。
  • 列名は大文字小文字・全角半角・余計な空白まで一致させます。

3) 文字化けで読み込めない

  • 本文コードは UTF-8 → cp932 の順でフォールバックしています。
  • それでも失敗する場合は、実CSVをExcelで「UTF-8 CSV」として再保存して確認します。

まとめ

第2回では、次を完成させました。

  • expect_download() を使ったCSV保存
  • CSVからURL列を安全に抽出する関数
  • main.py で一連の流れを接続

次回(第3回)は、抽出したURLを巡回して
商品ID1 / 商品ID2 / 商品名 / issue1 / issue2 を構造化する処理を作ります。

0
1
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
0
1

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?