# 昔ながらのスクレイピング
from bs4 import BeautifulSoup
import requests

soup = BeautifulSoup(requests.get("https://example-shop.com/products").text, "html.parser")
prices = soup.select("div.product-card > span.price")  # このセレクタが命綱

[毎朝9時] GitHub Actions
     ↓
[自然言語クエリ] Firecrawlエージェント
     ↓ URLなし！プロンプトだけ！
[競合10サイトの価格データ] JSON
     ↓
[前日比較・差分検出] Python
     ↓
[価格変動レポート] Slack通知

入力: "日本のECサイトで売られているワイヤレスイヤホンの価格情報"
  ↓
LLMが分解:
  - タスク1: "ワイヤレスイヤホン 価格比較 日本" で検索
  - タスク2: Amazon.co.jp, 楽天, ヨドバシなどの候補URLを特定
  - タスク3: 各ページの価格情報を抽出

# 依存パッケージをインストール
pip3 install firecrawl-py httpx python-dotenv pydantic

price-monitor/
├── .env                    # APIキーはここに集約
├── collector.py            # エージェントクエリ＋バリデーション
├── notifier.py             # Slack通知
├── schemas.py              # Pydanticモデル定義
├── cache/
│   └── prices.json         # 前日データのキャッシュ
└── .github/
    └── workflows/
        └── price-monitor.yml  # GitHub Actionsのワークフロー

# .env
FIRECRAWL_API_KEY=fc-xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
SLACK_WEBHOOK_URL=https://hooks.slack.com/services/T00000/B00000/xxxxxxxx

# init_check.py（動作確認用）
import os
from dotenv import load_dotenv
from firecrawl import FirecrawlApp

load_dotenv()

app = FirecrawlApp(api_key=os.environ["FIRECRAWL_API_KEY"])

# 動作確認：シンプルな1クエリテスト
result = app.extract(
    prompt="Sony WH-1000XM5 ワイヤレスイヤホンの最安値を1件だけ教えてください",
    schema={
        "type": "object",
        "properties": {
            "product_name": {"type": "string"},
            "price": {"type": "number"},
            "currency": {"type": "string"},
            "source_url": {"type": "string"}
        },
        "required": ["product_name", "price", "currency"]
    }
)

print(result)

# schemas.py
from pydantic import BaseModel, Field, field_validator
from typing import Optional
from datetime import datetime, timezone


class PriceRecord(BaseModel):
    """価格レコードの型定義"""
    product_name: str = Field(description="商品の正式名称")
    price: float = Field(description="販売価格（税込、数値のみ）", gt=0)
    currency: str = Field(default="JPY", description="通貨コード（ISO 4217）")
    retailer: str = Field(description="販売店名（例: Amazon, 楽天市場）")
    source_url: Optional[str] = Field(default=None, description="商品ページのURL")
    fetched_at: str = Field(
        default_factory=lambda: datetime.now(timezone.utc).isoformat(),
        description="データ取得日時（ISO 8601）"
    )

    @field_validator("price", mode="before")
    @classmethod
    def clean_price(cls, v):
        """価格表記ゆれを正規化（「¥3,980」→ 3980.0）"""
        if isinstance(v, str):
            cleaned = v.replace("¥", "").replace(",", "").replace("円", "").strip()
            return float(cleaned)
        return float(v)

    @field_validator("currency")
    @classmethod
    def normalize_currency(cls, v):
        """通貨表記を正規化（「円」→「JPY」）"""
        mapping = {"円": "JPY", "yen": "JPY", "￥": "JPY"}
        return mapping.get(v, v.upper())


class PriceReport(BaseModel):
    """複数商品の価格レポート"""
    items: list[PriceRecord]
    query_summary: Optional[str] = Field(
        default=None, description="エージェントが実行したクエリの概要"
    )

# NG: 曖昧すぎて精度が出ない
bad_prompt = "ワイヤレスイヤホンの価格"

# OK: 対象・条件・出力形式を明示
good_prompt = """
以下の条件でワイヤレスイヤホンの価格情報を収集してください：

【対象商品】
- Sony WH-1000XM5
- Apple AirPods Pro（第2世代）
- Bose QuietComfort Ultra Headphones

【収集条件】
- 日本国内ECサイト（Amazon.co.jp、楽天市場、ヨドバシ.com、ビックカメラ.com）
- 税込価格
- 最新価格

【出力形式】
各商品について、商品名・価格（数値のみ）・販売店名・商品ページURLを返してください
"""

# collector.py
import os
import json
import logging
from pathlib import Path
from datetime import datetime, timezone

from dotenv import load_dotenv
from firecrawl import FirecrawlApp
from schemas import PriceReport, PriceRecord

load_dotenv()

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger(__name__)

CACHE_FILE = Path("cache/prices.json")
CHANGES_FILE = Path("cache/changes.json")
CACHE_FILE.parent.mkdir(exist_ok=True)

# 監視対象の設定
WATCH_CONFIG = [
    {
        "category": "ワイヤレスイヤホン",
        "products": [
            "Sony WH-1000XM5",
            "Apple AirPods Pro 第2世代",
            "Bose QuietComfort Ultra Headphones",
        ],
        "retailers": ["Amazon.co.jp", "楽天市場", "ヨドバシ.com", "ビックカメラ.com"],
    }
]


def load_cache() -> dict[str, float]:
    """前回実行時の価格キャッシュを読み込む。ファイルが存在しない場合は空dictを返す。"""
    if not CACHE_FILE.exists():
        logger.info("キャッシュファイルが存在しません。初回実行として扱います。")
        return {}
    try:
        with CACHE_FILE.open("r", encoding="utf-8") as f:
            raw: list[dict] = json.load(f)
        # キー: "{retailer}::{product_name}", 値: price
        return {f"{item['retailer']}::{item['product_name']}": item["price"] for item in raw}
    except (json.JSONDecodeError, KeyError) as e:
        logger.warning("キャッシュの読み込みに失敗しました: %s", e)
        return {}


def save_cache(records: list[PriceRecord]) -> None:
    """今回取得した価格データをキャッシュとして保存する。"""
    payload = [record.model_dump() for record in records]
    with CACHE_FILE.open("w", encoding="utf-8") as f:
        json.dump(payload, f, ensure_ascii=False, indent=2)
    logger.info("キャッシュを保存しました: %d件", len(payload))


def build_prompt(category: str, products: list[str], retailers: list[str]) -> str:
    """監視設定からFirecrawlエージェント向けプロンプトを生成する。"""
    product_list = "\n".join(f"- {p}" for p in products)
    retailer_list = "、".join(retailers)
    return f"""
以下の条件で{category}の価格情報を収集してください。

【対象商品】
{product_list}

【収集対象サイト】
{retailer_list}

【収集条件】
- 日本国内の税込価格（円）
- 最新の販売価格（セール価格がある場合はセール価格を優先）

【出力形式】
各商品・販売店の組み合わせについて、以下を返してください：
- product_name: 商品名（正式名称）
- price: 価格（数値のみ、カンマ・記号なし）
- currency: 通貨コード（JPY）
- retailer: 販売店名
- source_url: 商品ページのURL
""".strip()


def collect_prices() -> list[PriceRecord]:
    """Firecrawlエージェントを呼び出して全カテゴリの価格を収集する。"""
    app = FirecrawlApp(api_key=os.environ["FIRECRAWL_API_KEY"])
    all_records: list[PriceRecord] = []

    for config in WATCH_CONFIG:
        prompt = build_prompt(
            category=config["category"],
            products=config["products"],
            retailers=config["retailers"],
        )
        logger.info("エージェントクエリを実行中: カテゴリ=%s", config["category"])

        try:
            result = app.extract(
                prompt=prompt,
                schema=PriceReport.model_json_schema(),
                timeout=120,  # エージェントは複数ページを探索するため長めに設定
            )
            report = PriceReport.model_validate(result)
            all_records.extend(report.items)
            logger.info("取得完了: %d件", len(report.items))
        except Exception as e:
            logger.error("価格収集中にエラーが発生しました: %s", e)
            raise

    return all_records


def detect_changes(
    current: list[PriceRecord],
    cache: dict[str, float],
) -> list[dict]:
    """現在の価格と前回キャッシュを比較して変動を検出する。

    Returns:
        変動があったレコードのリスト。各要素は product_name / retailer /
        old_price / new_price / diff / diff_pct を持つ。
    """
    changes = []
    for record in current:
        key = f"{record.retailer}::{record.product_name}"
        old_price = cache.get(key)
        if old_price is None:
            continue  # 初回取得または新規追加商品はスキップ
        diff = record.price - old_price
        diff_pct = diff / old_price * 100
        if abs(diff_pct) >= 1.0:  # 1%以上の変動を検出対象とする
            changes.append(
                {
                    "product_name": record.product_name,
                    "retailer": record.retailer,
                    "old_price": old_price,
                    "new_price": record.price,
                    "diff": diff,
                    "diff_pct": round(diff_pct, 2),
                    "source_url": record.source_url,
                    "fetched_at": record.fetched_at,
                }
            )
    return changes


if __name__ == "__main__":
    logger.info("価格収集を開始します")
    cache = load_cache()
    records = collect_prices()
    changes = detect_changes(records, cache)
    save_cache(records)

    # notifier.py が読み込む変動データを書き出す
    CHANGES_FILE.write_text(
        json.dumps(changes, ensure_ascii=False, indent=2), encoding="utf-8"
    )
    logger.info("変動データを保存しました: %d件", len(changes))

# .github/workflows/price-monitor.yml
name: Price Monitor

on:
  schedule:
    # 毎朝9時（JST = UTC 0:00）に実行
    - cron: "0 0 * * *"
  workflow_dispatch:
    # 手動実行も可能にしておく

permissions:
  contents: write

jobs:
  collect-and-notify:
    runs-on: ubuntu-latest
    timeout-minutes: 10

    steps:
      - name: Checkout repository
        uses: actions/checkout@v4

      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: "3.11"

      - name: Restore price cache
        uses: actions/cache@v4
        with:
          path: cache/prices.json
          key: price-cache-${{ github.run_id }}
          restore-keys: |
            price-cache-

      - name: Install dependencies
        run: pip install firecrawl-py httpx python-dotenv pydantic

      - name: Run price collector
        env:
          FIRECRAWL_API_KEY: ${{ secrets.FIRECRAWL_API_KEY }}
        run: python collector.py

      - name: Send Slack notification
        if: success()
        env:
          SLACK_WEBHOOK_URL: ${{ secrets.SLACK_WEBHOOK_URL }}
        run: python notifier.py

      - name: Save updated cache
        uses: actions/cache@v4
        with:
          path: cache/prices.json
          key: price-cache-${{ github.run_id }}

# 動作確認（ターミナルで実行）
curl -X POST "$SLACK_WEBHOOK_URL" \
  -H "Content-Type: application/json" \
  -d '{"text": "テスト通知です！price-monitorの設定確認中"}'

# notifier.py
import os
import json
import logging
from pathlib import Path
from datetime import datetime, timezone

import httpx
from dotenv import load_dotenv

load_dotenv()

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger(__name__)

CACHE_FILE = Path("cache/prices.json")
CHANGES_FILE = Path("cache/changes.json")


def load_changes() -> list[dict]:
    """collector.pyが検出した変動データを読み込む。"""
    if not CHANGES_FILE.exists():
        return []
    with CHANGES_FILE.open("r", encoding="utf-8") as f:
        return json.load(f)


def format_change_block(change: dict) -> dict:
    """1件の価格変動をSlack Block Kit形式に変換する。"""
    sign = "📈" if change["diff"] > 0 else "📉"
    direction = "値上がり" if change["diff"] > 0 else "値下がり"
    diff_abs = abs(change["diff"])
    diff_pct = abs(change["diff_pct"])
    url_text = f"\n<{change['source_url']}|商品ページを見る>" if change.get("source_url") else ""

    return {
        "type": "section",
        "text": {
            "type": "mrkdwn",
            "text": (
                f"{sign} *{change['product_name']}* @ {change['retailer']}\n"
                f"¥{change['old_price']:,.0f} → *¥{change['new_price']:,.0f}*"
                f"（{direction} ¥{diff_abs:,.0f} / {diff_pct:.1f}%）{url_text}"
            ),
        },
    }


def build_slack_payload(changes: list[dict], total_records: int) -> dict:
    """Slackに送信するペイロードを構築する。"""
    now = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M UTC")

    if not changes:
        return {
            "blocks": [
                {
                    "type": "section",
                    "text": {
                        "type": "mrkdwn",
                        "text": (
                            f"✅ *価格モニタリング完了* ({now})\n"
                            f"監視件数: {total_records}件 ｜ 価格変動: なし"
                        ),
                    },
                }
            ]
        }

    blocks = [
        {
            "type": "header",
            "text": {
                "type": "plain_text",
                "text": f"💰 価格変動アラート（{len(changes)}件）",
            },
        },
        {
            "type": "context",
            "elements": [
                {
                    "type": "mrkdwn",
                    "text": f"実行日時: {now} ｜ 監視件数: {total_records}件",
                }
            ],
        },
        {"type": "divider"},
    ]

    for change in changes:
        blocks.append(format_change_block(change))

    blocks.append({"type": "divider"})
    blocks.append(
        {
            "type": "context",
            "elements": [
                {
                    "type": "mrkdwn",
                    "text": "1%以上の変動を検出した商品のみを表示しています",
                }
            ],
        }
    )

    return {"blocks": blocks}


def send_notification(payload: dict) -> None:
    """SlackのIncoming Webhook URLにペイロードを送信する。"""
    webhook_url = os.environ["SLACK_WEBHOOK_URL"]
    response = httpx.post(webhook_url, json=payload, timeout=10)
    response.raise_for_status()
    logger.info("Slack通知を送信しました")


def count_records() -> int:
    """キャッシュファイルに保存されたレコード数を返す。"""
    if not CACHE_FILE.exists():
        return 0
    with CACHE_FILE.open("r", encoding="utf-8") as f:
        return len(json.load(f))


if __name__ == "__main__":
    changes = load_changes()
    total = count_records()
    payload = build_slack_payload(changes, total)
    send_notification(payload)
    logger.info("通知完了: 変動%d件 / 監視%d件", len(changes), total)

# テスト用のchanges.jsonを手動作成
mkdir -p cache
cat > cache/changes.json << 'EOF'
[
  {
    "product_name": "Sony WH-1000XM5",
    "retailer": "Amazon.co.jp",
    "old_price": 39800,
    "new_price": 35800,
    "diff": -4000,
    "diff_pct": -10.05,
    "source_url": "https://www.amazon.co.jp/dp/B09XS7JWHH",
    "fetched_at": "2026-03-28T00:00:00+00:00"
  }
]
EOF

# 通知テスト実行
python notifier.py