【第3回】CSVのURLを巡回し、商品IDとissue情報を構造化する

はじめに

第2回では、CSVダウンロードとURL列抽出まで進めました。
第3回では、抽出したURLを実際に巡回して、後続処理で使える構造化データを作成します。

この回のゴール:

• URLごとにページを開いて必要なタグを取得する
• product_id1 / product_id2 / product_name を抽出する
• issue1 / issue2 を暫定ルールで判定する
• 失敗URLがあっても全体処理を継続できる形にする

---

1. データモデルを作る（src/models.py）

まずは、1URL分の情報を保持する器を作ります。
dataclass を使うと、読みやすく扱いやすいです。

from dataclasses import dataclass
from typing import Optional


@dataclass
class ProductRecord:
    """
    1つのURLから抽出した商品情報を保持するモデル。
    """
    # 一次ページで取得する主キーとなるID
    product_id1: str
    # 任意項目（存在しない場合がある）
    product_id2: Optional[str]
    # 商品名
    product_name: str
    # issueは存在しないケースがあるためOptional
    issue1: Optional[str]
    issue2: Optional[str]
    # どのURLから取ったかを追跡するために保持
    source_url: str

---

2. 抽出処理を作る（src/extractor.py）

ここが第3回の中心です。
ポイントは「値取得の共通化」と「例外時に止まらない設計」です。

from __future__ import annotations

from typing import Optional
from playwright.sync_api import Page

from models import ProductRecord


def safe_text_content(page: Page, selector: str) -> Optional[str]:
    """
    指定セレクタのtext_contentを安全に取得する。

    - 要素が見つからない
    - text_contentがNone
    - 前後に不要な空白がある
    といったケースをこの関数で吸収する。
    """
    if not selector:
        # セレクタ未設定は設定ミスなので、ここではNoneとして扱う。
        return None

    try:
        text = page.text_content(selector, timeout=5000)
    except Exception:
        # 一時的なDOM状態などで失敗することがあるため握りつぶしてNone返却
        return None

    if text is None:
        return None

    cleaned = text.strip()
    return cleaned if cleaned else None


def classify_issues(issue_source_text: Optional[str]) -> tuple[Optional[str], Optional[str]]:
    """
    issue判定の暫定ルール。

    本番では業務ルールに合わせて差し替える前提で、
    まずはキーワードベースで判定する。
    """
    if not issue_source_text:
        return None, None

    text = issue_source_text.lower()

    # 例: 注意喚起系のキーワード
    issue1 = "在庫注意" if ("在庫" in text or "欠品" in text) else None
    # 例: 品質・状態系のキーワード
    issue2 = "品質注意" if ("破損" in text or "汚れ" in text) else None

    return issue1, issue2


def extract_product_record_from_current_page(
    page: Page,
    source_url: str,
    sel_product_id1: str,
    sel_product_id2: str,
    sel_product_name: str,
    sel_issue_source: str,
) -> ProductRecord:
    """
    現在開いているページから商品情報を抽出してProductRecordに詰める。
    """
    product_id1 = safe_text_content(page, sel_product_id1) or ""
    product_id2 = safe_text_content(page, sel_product_id2)
    product_name = safe_text_content(page, sel_product_name) or ""

    issue_source_text = safe_text_content(page, sel_issue_source)
    issue1, issue2 = classify_issues(issue_source_text)

    return ProductRecord(
        product_id1=product_id1,
        product_id2=product_id2,
        product_name=product_name,
        issue1=issue1,
        issue2=issue2,
        source_url=source_url,
    )

---

3. URL群を巡回する処理を main.py に追加する

第2回で作った urls を使って、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
from extractor import extract_product_record_from_current_page


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

    with sync_playwright() as p:
        browser = p.firefox.launch(headless=False)
        context = browser.new_context()
        page = context.new_page()

        # ログインとCSV取得（第2回までの処理）
        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)
        page.goto(settings.csv_page_url)

        csv_path = download_csv(
            page=page,
            save_dir=settings.download_dir,
            download_button_selector=settings.sel_csv_download_button,
        )
        urls = extract_urls_from_csv(csv_path, settings.csv_url_column_name)

        # 第3回のメイン処理: URL巡回して構造化する
        records = []
        failed_urls = []

        for i, url in enumerate(urls, start=1):
            try:
                # 各URLに移動
                page.goto(url, timeout=15000)

                # 必要に応じて描画完了待ち
                page.wait_for_load_state("domcontentloaded")

                # 現在ページから商品情報を抽出
                record = extract_product_record_from_current_page(
                    page=page,
                    source_url=url,
                    sel_product_id1=settings.sel_product_id1,
                    sel_product_id2=settings.sel_product_id2,
                    sel_product_name=settings.sel_product_name,
                    sel_issue_source=settings.sel_issue_source,
                )
                records.append(record)

                # 進捗ログ（最初はprintで十分）
                print(f"[OK] {i}/{len(urls)}: {url}")

            except Exception as e:
                # 1件失敗しても全体を止めずに次へ進む
                print(f"[NG] {i}/{len(urls)}: {url} ({e})")
                failed_urls.append(url)

        print("-----")
        print(f"成功件数: {len(records)}")
        print(f"失敗件数: {len(failed_urls)}")

        # 先頭3件だけ表示して中身を確認
        for rec in records[:3]:
            print(rec)

        browser.close()


if __name__ == "__main__":
    run()

---

4. 期待される実行結果

[OK] 1/120: https://example.local/item/001
[OK] 2/120: https://example.local/item/002
[NG] 3/120: https://example.local/item/003 (Timeout 15000ms exceeded)
...
-----
成功件数: 118
失敗件数: 2
ProductRecord(product_id1='A1001', product_id2=None, product_name='...')

---

ハマりどころと対策

text_content() が None になる

• 要素は存在しても中身が空のケースがあります。
• 本文の safe_text_content() で None を吸収する設計にしています。

ページ遷移は成功しているのに値が取れない

• JS描画が遅いケースです。
• wait_for_load_state("domcontentloaded") の後に、必要なら wait_for_timeout() を短く入れて確認しましょう。

issue1/issue2 の判定が期待と違う

• 今回はあくまで暫定ルールです。
• 判定条件を関数分離しているので、次回以降に差し替えやすくしています。

---

まとめ

第3回では、URL巡回と構造化の土台を完成させました。

• ProductRecord の導入
• タグ抽出を安全に行う共通関数
• issue判定ロジックの分離
• 失敗URLがあっても処理継続する実装

次回（第4回）は、DETAIL_BASE_URL + product_id1 の派生URLを作って
追加タグ（specific_tag1）を取得し、レコードに統合します。