技術概要

自動車情報プラットフォームの選車リストページから車両仕様、販売価格帯、ユーザー評価などの構造化データを自動取得する実装を示す。対象サイトはクライアントサイドレンダリングを採用しており、ページ遷移やスクロールに応じてXHRリクエストが実行される。さらに、数値情報に対してカスタムフォントマッピングを適用しているため、通信のインターセプトとレスポンステキストの復号処理を組み合わせる必要がある。

依存モジュール

import csv
import json
from typing import Any, Dict, List
import requests

スクレイピング制限の特性

• フォント置換による難読化: 価格や評価などの数値をPrivate Use Area（PUA）のUnicode文字列に置換し、DOM上の表示と内部的なデータ構造を意図的に分離している。
• スクリプト圧縮・変数名の難読化: 空白・改行の削除と無意味な変数名への置換により、静的解析を困難にし、データ生成ロジックの追跡を阻害する。
• 非同期データロード: 初期HTMLにはデータが含まれず、APIエンドポイントへのPOSTリクエストをトリガーとしてJSONペイロードを動的に展開する。

エンドポイント解析とパラメータ特定

1. 通信パターンの識別: ページ内スクロール時に追加データが読み込まれる動作から、ダイナミックコンテンツの取得と判定。
2. ネットワークトラフィックの監視: DevToolsのNetworkタブでFetch/XHRをフィルタリングし、車両リスト取得を担うリクエストを特定。
3. ペイロード構造の解読: 送信されるForm Dataより、sort_new（ソート基準）、city_name（対象地域）、limit（1リクエストあたりの取得件数）、page（ページインデックス）が制御パラメータとして機能していることを確認。
4. レスポンス階層の確認: 返却されるJSON構造を調査。車両一覧オブジェクトはdata.series配下に格納されており、各フィールドは文字列または数値として保持されている。フォント難読化解除後のシリアライズ/デシリアライズ処理が必須。

データ取得スクリプト

セッション維持、フォントグライフリプレイス、構造化出力をクラスベースで実装。エラーハンドリングとページ制御を統合している。

import csv
import json
import requests
from typing import Any, Dict, List


TARGET_ENDPOINT = "https://www.dongchedi.com/motor/pc/car/brand/select_series_v2"
BASE_QUERY = {"aid": 1839, "app_name": "auto_web_pc"}

PUA_TO_DIGIT = {
    "\ue420": "0", "\ue40d": "1", "\ue423": "2", "\ue409": "3",
    "\ue41d": "4", "\ue427": "5", "\ue419": "6", "\ue3fd": "7",
    "\ue428": "8", "\ue413": "9"
}


class AutoListExtractor:
    def __init__(self, city: str = "梅州", chunk_size: int = 30):
        self.session = requests.Session()
        self.city = city
        self.chunk_size = chunk_size
        self.session.headers.update({
            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
        })

    def pull_page_data(self, page_index: int) -> Dict[str, Any]:
        payload = {
            "sort_new": "hot_desc",
            "city_name": self.city,
            "limit": str(self.chunk_size),
            "page": str(page_index)
        }
        try:
            resp = self.session.post(TARGET_ENDPOINT, params=BASE_QUERY, data=payload, timeout=15)
            resp.raise_for_status()
            return resp.json()
        except requests.RequestException as err:
            print(f"[Fetch Error] Page {page_index}: {err}")
            return {}

    def decode_payload(self, raw_response: Dict[str, Any]) -> Dict[str, Any]:
        json_str = json.dumps(raw_response, ensure_ascii=False)
        for glyph, numeric in PUA_TO_DIGIT.items():
            json_str = json_str.replace(glyph, numeric)
        return json.loads(json_str)

    def structure_records(self, api_payload: Dict[str, Any]) -> List[Dict[str, str]]:
        vehicle_list = api_payload.get("data", {}).get("series", [])
        return [
            {
                "model_label": item.get("outter_name", "Unknown"),
                "price_span": item.get("dealer_price", "Unlisted"),
                "evaluation": item.get("dcar_score", "N/A")
            }
            for item in vehicle_list
        ]

    def persist_csv(self, records: List[Dict[str, str]], target_path: str = "vehicles_export.csv") -> None:
        if not records:
            return
        
        header_written = False
        with open(target_path, "a", encoding="utf-8-sig", newline="") as stream:
            if stream.tell() == 0:
                header_written = True
            
            writer = csv.DictWriter(stream, fieldnames=["model_label", "price_span", "evaluation"])
            if header_written:
                writer.writeheader()
            writer.writerows(records)

    def run(self, max_pages: int = 10) -> None:
        for current in range(1, max_pages + 1):
            print(f"Processing page {current}...")
            raw_json = self.pull_page_data(current)
            if not raw_json:
                print("Terminating: No data received or request failed.")
                break

            cleaned = self.decode_payload(raw_json)
            structured = self.structure_records(cleaned)
            
            if structured:
                self.persist_csv(structured)
                print(f"Saved {len(structured)} records successfully.")
            else:
                print("Empty dataset for this iteration.")


if __name__ == "__main__":
    extractor = AutoListExtractor()
    extractor.run(10)