Amazon の購入履歴を Gemini API で分類して可視化してみた

コピー

PERSONAL_COLUMNS = {
    "Billing Address",
    "Shipping Address",
    "Gift Message",
    "Gift Recipient Contact",
    "Gift Sender Name",
    "Item Serial Number",
}

# 該当カラムを除外して新しい CSV に書き出す
fieldnames = [col for col in reader.fieldnames if col not in PERSONAL_COLUMNS]
rows = [
    {k: v for k, v in row.items() if k not in PERSONAL_COLUMNS}
    for row in reader
]

コピー

import csv, json, os, time
from dotenv import load_dotenv
from google import genai
from google.genai.types import GenerateContentConfig

load_dotenv()

CATEGORIES = [
    "食品・飲料", "日用品・消耗品", "キッチン用品",
    "ガジェット・電子機器", "本・雑誌", "Blu-ray・映像作品",
    "ファッション・衣類", "家具・収納", "フィットネス用品",
    "雑貨・インテリア", "その他",
]

SYSTEM_PROMPT = f"""あなたはAmazonの購入商品をカテゴリに分類するアシスタントです。
カテゴリ一覧から最も適切なものを1つ選んでください。

カテゴリ一覧:
{chr(10).join(f"- {c}" for c in CATEGORIES)}

「日用品・消耗品」には美容・スキンケア・ヘアケア・掃除用品なども含みます。
「Blu-ray・映像作品」はBlu-ray/DVD/映像ソフト単体の商品に使います。
「ガジェット・電子機器」はBlu-ray/DVDプレーヤーなどの再生機器を含みますが、映像ソフト自体は含みません。

入力: 商品名のJSONオブジェクト（インデックス → 商品名）
出力: 各インデックスとカテゴリのJSONオブジェクト

例:
入力: {"0": "アイリスオーヤマ 無洗米 5kg", "1": "クワトロボタニコ 洗顔料"}
出力: {"0": "食品・飲料", "1": "日用品・消耗品"}
"""

def classify_batch(client, batch):
    response = client.models.generate_content(
        model="gemini-3.1-flash-lite",
        config=GenerateContentConfig(
            system_instruction=SYSTEM_PROMPT,
            response_mime_type="application/json",
        ),
        contents=json.dumps(batch, ensure_ascii=False),
    )
    return json.loads(response.text)

def classify_all(input_path, output_path):
    client = genai.Client(api_key=os.environ["GEMINI_API_KEY"])
    with open(input_path, encoding="utf-8-sig", newline="") as f:
        reader = csv.DictReader(f)
        rows = list(reader)
        fieldnames = list(reader.fieldnames) + ["Category"]

    results = {}
    for start in range(0, len(rows), 20):
        batch = {str(i): row["Product Name"]
                 for i, row in enumerate(rows[start:start + 20], start=start)}
        classified = classify_batch(client, batch)
        results.update({int(k): v for k, v in classified.items()})
        time.sleep(5)

    for i, row in enumerate(rows):
        row["Category"] = results.get(i, "その他")

    with open(output_path, "w", encoding="utf-8", newline="") as f:
        writer = csv.DictWriter(f, fieldnames=fieldnames)
        writer.writeheader()
        writer.writerows(rows)