脱CSSセレクタ！Gemini 3 Flash × Pythonで実現する「視覚的」スクレイピング完全ガイド【メンテ不要】

import subprocess
import sys
import os
import time
import json
import base64
import asyncio
import importlib
import platform
from typing import List, Optional

# ==========================================
# 0. 自動セットアップ・パート (面倒な作業を全部やる)
# ==========================================
def install_and_import(package_name, import_name=None):
    if import_name is None:
        import_name = package_name
    try:
        importlib.import_module(import_name)
    except ImportError:
        print(f"🔧 {package_name} が見つからないからインストールするね... (ちょっと待ってて！)")
        try:
            subprocess.check_call([sys.executable, "-m", "pip", "install", package_name])
            print(f"✅ {package_name} のインストール完了！")
        except Exception as e:
            print(f"🚨 {package_name} のインストールに失敗しちゃった...: {e}")
            sys.exit(1)

# 必要なライブラリを片っ端からチェック＆インストール
# japanize-matplotlib は Python 3.12+ で死ぬのでリストラしました🙏
REQUIRED_PACKAGES = [
    ("playwright", "playwright"),
    ("google-genai", "google.genai"),
    ("pydantic", "pydantic"),
    ("python-dotenv", "dotenv"),
    ("pandas", "pandas"),
    ("matplotlib", "matplotlib")
]

print("🚗 自動セットアップ開始！シートベルト締めてね！")
for pkg, imp in REQUIRED_PACKAGES:
    install_and_import(pkg, imp)

# Playwrightのブラウザバイナリも確認してインストール
try:
    print("🌍 Playwrightブラウザの確認中...")
    subprocess.run([sys.executable, "-m", "playwright", "install", "chromium"], check=True)
except Exception as e:
    print(f"⚠️ ブラウザインストールでエラー出たけど、とりあえず進めるね: {e}")

# ここから本番のインポート
from playwright.async_api import async_playwright
from pydantic import BaseModel, Field
from google import genai
from google.genai import types
from dotenv import load_dotenv
import pandas as pd
import matplotlib.pyplot as plt

# 環境変数の読み込み
load_dotenv(encoding="utf-8")

# ==========================================
# 日本語フォント設定 (Python 3.12+ 対応版)
# ==========================================
def configure_japanese_font():
    system = platform.system()
    if system == "Windows":
        font_family = "MS Gothic"
    elif system == "Darwin": # Mac
        font_family = "AppleGothic"
    else: # Linux/Other
        font_family = "Noto Sans CJK JP" # 入ってないかもだけど一応

    try:
        plt.rcParams['font.family'] = font_family
        print(f"🎨 日本語フォントを '{font_family}' に設定したよ！")
    except Exception as e:
        print(f"⚠️ フォント設定に失敗しちゃった (文字化けするかも): {e}")

configure_japanese_font()

# ==========================================
# 1. 定義パート (Geminiに見させる型)
# ==========================================
class StatItem(BaseModel):
    title: str = Field(..., description="統計調査の名称や表のタイトル。")
    ministry: Optional[str] = Field(None, description="担当している府省名（例: 総務省, 厚生労働省）。")
    release_date: Optional[str] = Field(None, description="公表日や更新日。")
    description: Optional[str] = Field(None, description="統計の簡単な説明や、'新着' '更新' などのバッジ情報。")
    format_type: Optional[str] = Field(None, description="データの形式（例: 'DB', 'Excel', 'PDF', 'CSV'）。アイコンやリンク文字から推測。")

class StatList(BaseModel):
    items: List[StatItem] = Field(..., description="ページ内で検出された統計情報のリスト")

# ==========================================
# 2. スクレイピング・クラス (Playwright + Gemini)
# ==========================================
class VisualScraper:
    def __init__(self, api_key: str, model_name: str = "models/gemini-3-flash-preview"):
        self.client = genai.Client(api_key=api_key)
        self.model_name = model_name
        print(f"✨ VisualScraper Initialized with {model_name} ✨")

    async def fetch_page_data(self, url: str):
        print(f"🌍 Accessing URL: {url} ...")
        async with async_playwright() as p:
            browser = await p.chromium.launch(headless=True)
            page = await browser.new_page(viewport={"width": 1280, "height": 1200})
            try:
                await page.goto(url, wait_until="networkidle", timeout=60000) # タイムアウト長めに
                await page.evaluate("window.scrollTo(0, 500)")
                await asyncio.sleep(3) 
                screenshot_bytes = await page.screenshot(full_page=False)
                screenshot_b64 = base64.b64encode(screenshot_bytes).decode("utf-8")
                text_content = await page.inner_text("body")
                print("📸 Screenshot captured & Text extracted!")
                return screenshot_b64, text_content
            except Exception as e:
                print(f"❌ Browser Error: {e}")
                return None, None
            finally:
                await browser.close()

    def analyze_with_gemini(self, screenshot_b64: str, text_content: str) -> Optional[StatList]:
        print("💎 Sending data to Gemini...")
        system_instruction = """
        あなたは政府統計データの専門アナリストです。
        提供されたWebページの「スクリーンショット」と「テキスト情報」をもとに、
        主要な統計情報（新着情報やランキングなど）を抽出し、指定されたJSON形式で出力してください。
        視覚的に認識できるアイコン（Excelなど）の情報も重要です。
        """
        prompt = "このページに表示されている統計データをリストアップしてください。"
        
        try:
            response = self.client.models.generate_content(
                model=self.model_name,
                contents=[
                    types.Content(
                        role="user",
                        parts=[
                            types.Part(text=prompt),
                            types.Part(
                                inline_data=types.Blob(
                                    mime_type="image/png",
                                    data=base64.b64decode(screenshot_b64)
                                )
                            )
                        ]
                    )
                ],
                config=types.GenerateContentConfig(
                    system_instruction=system_instruction,
                    response_mime_type="application/json",
                    response_schema=StatList,
                    temperature=0.1
                )
            )
            if response.text:
                return StatList(**json.loads(response.text))
            return None
        except Exception as e:
            print(f"🔥 Gemini API Error: {e}")
            return None

# ==========================================
# 3. 可視化パート (Pandas + Matplotlib)
# ==========================================
def visualize_results(json_file: str, output_dir: str):
    if not os.path.exists(json_file):
        return

    print("📊 グラフを作成中...")
    with open(json_file, "r", encoding="utf-8") as f:
        data = json.load(f)
    
    df = pd.DataFrame(data["items"]) if "items" in data else pd.DataFrame(data)
    os.makedirs(output_dir, exist_ok=True)

    # 省庁ランキング
    if "ministry" in df.columns:
        plt.figure(figsize=(10, 6))
        ministry_counts = df["ministry"].dropna().value_counts()
        if not ministry_counts.empty:
            ministry_counts.sort_values().plot(kind="barh", color="skyblue", edgecolor="black")
            plt.title("省庁別：統計データ公表数ランキング", fontsize=16)
            plt.tight_layout()
            plt.savefig(os.path.join(output_dir, "ministry_ranking.png"))
            print("📈 省庁ランキング保存完了！")

    # ファイル形式円グラフ
    if "format_type" in df.columns:
        plt.figure(figsize=(8, 8))
        format_counts = df["format_type"].dropna().value_counts()
        if not format_counts.empty:
            format_counts.plot(kind="pie", autopct="%1.1f%%", startangle=90, cmap="Pastel1")
            plt.title("統計データのファイル形式内訳", fontsize=16)
            plt.ylabel("")
            plt.savefig(os.path.join(output_dir, "format_pie_chart.png"))
            print("🍕 円グラフ保存完了！")

# ==========================================
# メイン実行 (指揮者)
# ==========================================
async def main():
    TARGET_URL = "https://www.e-stat.go.jp/"
    JSON_FILE = "estat_result.json"
    GRAPH_DIR = "graphs"

    api_key = os.getenv("GEMINI_API_KEY")
    if not api_key:
        # キーがない場合、入力させる親切設計
        print("\n🚨 GEMINI_API_KEY が .env に見つかりません！")
        api_key = input("🔑 ここにAPIキーを直接貼り付けてEnter押してね: ").strip()
        if not api_key:
            print("終了します...")
            return

    # 1. スクレイピング実行
    scraper = VisualScraper(api_key=api_key)
    screenshot, text = await scraper.fetch_page_data(TARGET_URL)
    
    if screenshot:
        result = scraper.analyze_with_gemini(screenshot, text)
        if result:
            with open(JSON_FILE, "w", encoding="utf-8") as f:
                f.write(result.model_dump_json(indent=2, ensure_ascii=False))
            print(f"\n💾 データ取得成功: {len(result.items)}件のデータを {JSON_FILE} に保存したよ！")
            
            # 2. グラフ化実行
            visualize_results(JSON_FILE, GRAPH_DIR)
            print(f"\n🎉 完全勝利！ {GRAPH_DIR} フォルダを確認してね！")
        else:
            print("☁️ Geminiの解析に失敗しちゃった...")
    else:
        print("💀 ページが開けなかったよ...")

if __name__ == "__main__":
    asyncio.run(main())