スクレイピングの変化検知：SQLite スナップショットとハッシュ

シリーズ：Webスクレイピング講座 5/6（上級）。 取得・パース・ページ送りを礼儀正しく書けるようになった人が、「毎朝動かして、昨日から何が変わったかを知りたい」という段階に進むための回です。読み終えると、SQLite にスナップショットを保存し、コンテンツハッシュで価格と在庫の本当の変化だけを検知し、二重実行しても壊れない処理を書き、UTC のタイムスタンプで cron に載せられるようになります。

この記事は、価格比較や在庫監視のために一度だけ動かすスクリプトを書いたら、「じゃあ毎朝自動で回しておいて」と頼まれた人のためのものです。社内の定型作業の自動化でもよくある流れです。今日の価格を表示するだけなら簡単ですが、昨日を覚えていて、本当の変化だけを報告し、2回動いても平気で、ディスクを食いつぶさないスクリプトは別物で、その中身はほとんどデータベースの仕事です。

以下はすべて Apple Silicon Mac (M1 Pro) で、Python 3.9.6（標準ライブラリのみ）、SQLite 3.51.0（この Python に同梱のライブラリと sqlite3 コマンド）、そしてシリーズの練習用ショップを 127.0.0.1:18172 で配信する PHP 8.5.6 を使って実際に動かしました。パスは ~/demo/ に短縮しています。取得先は自分で立てた練習サイトだけです。

保存する前に：本当にスクレイピングでいいのか

毎日動くジョブは、すべてを掛け算にします。1回のリクエストは、1年で365回のリクエストです。だから第1回で確認したことが、ここで一段と重くなります。

• 公式 API やフィードが先。 多くの EC サイトやモールは商品フィードやアフィリエイト API を用意しています。HTML より安定していて、利用条件もはっきりしています。
• robots.txt と Crawl-delay。 練習ショップの robots.txt は Crawl-delay: 1 なので、ページの間で1秒待ちます。10ページで約9秒。1日1回なら何の問題もない、お客として当然のコストです。
• アクセスの頻度は相手のサーバーの問題でもある。 日本では2010年に、公共図書館の蔵書検索システムへの自動アクセスがきっかけで、作成者が逮捕にまで至った事例がありました。本人に悪意はなく、相手側システムの作りにも問題があったと言われていますが、「自分のスクリプトがどれだけの負荷をかけるか」は、取得する側が責任を持って抑えるべきだという教訓は変わりません。
• 利用規約と個人情報。 価格や在庫は通常は個人情報ではありませんが、名前付きのレビューは個人情報です。著作権法や利用規約との関係はケースによって違うので、保存を始める前にそのサイトの利用規約とお住まいの地域の法律を確認し、必要な項目だけを保存してください。この記事は法的助言ではありません。
• 正直な User-Agent。 ウォッチャーは shopwatch/1.0 (+https://example.com/bot-contact) を送ります。名前と、サイト運営者が連絡できる場所です。

データモデル：products と snapshots

練習ショップには10ページに200商品があります。HTML では1商品がこう見えます。

<li class="product" data-sku="SKU-0020"><a href="/product/SKU-0020">Cable clip 2</a>
  <span class="price">$66.79</span> <span class="stock">In stock</span></li>

サーバー側の状態ファイルで「日付」を進めると、1日あたり約4%の価格が変わり、約3%の商品の在庫が切り替わります（random.seed(42) で一度だけ生成）。

テーブルは2つです。products は商品ごとに1行で、最後に見た状態を持ちます。snapshots は何かが変わったときだけ行が増える、価格の履歴です。

CREATE TABLE IF NOT EXISTS products (
  sku TEXT PRIMARY KEY, name TEXT NOT NULL, url TEXT NOT NULL,
  first_seen TEXT NOT NULL, last_seen TEXT NOT NULL, last_hash TEXT NOT NULL);
CREATE TABLE IF NOT EXISTS snapshots (
  sku TEXT NOT NULL REFERENCES products(sku), seen_at TEXT NOT NULL,
  price_cents INTEGER NOT NULL, in_stock INTEGER NOT NULL, hash TEXT NOT NULL,
  PRIMARY KEY (sku, seen_at));

ここには3つの判断が隠れています。

1. 金額は整数のセント。 $66.79 は 6679 として保存します。浮動小数点数とお金は相性が悪く、その実例は後で見ます。円なら最初から整数ですが、ドルやユーロを扱う比較サイトでは必ず出てくる問題です。
2. タイムスタンプは UTC の文字列。 2026-10-02T03:00:00Z のように ISO 8601 の末尾に Z を付けます。この形式なら文字列のままで正しく並び、比較でき、sqlite3 コマンドでも読めます。
3. 自然キー（sku）を主キーにします。これが upsert を可能にします。安定した ID がないサイトなら、商品 URL が次の候補です。

ウォッチャー：取得・正規化・ハッシュ・upsert

スクリプト全体です。第2回の html.parser の書き方と、第3回の礼儀正しいページ送りをそのまま使っています。

"""shopwatch.py - snapshot a practice shop into SQLite and report price/stock changes.
Python 3.9+ stdlib only. Point it ONLY at a site you run yourself (php -S 127.0.0.1:18172).
"""
import hashlib, json, os, sqlite3, sys, time
import urllib.request, urllib.robotparser
from datetime import datetime, timezone
from html.parser import HTMLParser
from urllib.parse import urljoin

BASE = os.environ.get("SHOP", "http://127.0.0.1:18172")
UA = "shopwatch/1.0 (+https://example.com/bot-contact)"   # honest name + contact page

SCHEMA = """
CREATE TABLE IF NOT EXISTS products (
  sku TEXT PRIMARY KEY, name TEXT NOT NULL, url TEXT NOT NULL,
  first_seen TEXT NOT NULL, last_seen TEXT NOT NULL, last_hash TEXT NOT NULL);
CREATE TABLE IF NOT EXISTS snapshots (
  sku TEXT NOT NULL REFERENCES products(sku), seen_at TEXT NOT NULL,
  price_cents INTEGER NOT NULL, in_stock INTEGER NOT NULL, hash TEXT NOT NULL,
  PRIMARY KEY (sku, seen_at));
"""

class ListParser(HTMLParser):
    def __init__(self):
        super().__init__(); self.items, self.next, self.cur, self.field = [], None, None, None
    def handle_starttag(self, tag, attrs):
        a = dict(attrs)
        if tag == "li" and a.get("class") == "product":
            self.cur = {"sku": a.get("data-sku"), "name": "", "price": None, "stock": None, "url": None}
        elif self.cur is not None and tag == "a":
            self.cur["url"], self.field = a.get("href"), "name"
        elif self.cur is not None and tag == "span":
            self.field = a.get("class")
        elif tag == "a" and a.get("rel") == "next":
            self.next = a.get("href")
    def handle_data(self, data):
        if self.cur is not None and self.field in ("name", "price", "stock"):
            self.cur[self.field] = (self.cur[self.field] or "") + data
    def handle_endtag(self, tag):
        if tag in ("a", "span"): self.field = None
        if tag == "li" and self.cur is not None:
            self.items.append(self.cur); self.cur = None

def normalize(raw, page_url):
    price = raw["price"].strip().lstrip("$").replace(",", "")
    return {"sku": raw["sku"], "name": " ".join(raw["name"].split()),
            "url": urljoin(page_url, raw["url"]),
            "price_cents": round(float(price) * 100),          # money as integer cents
            "in_stock": raw["stock"].strip() == "In stock"}

def content_hash(item):
    # hash only the fields we care about, in a fixed order - never the raw HTML
    key = json.dumps([item["name"], item["price_cents"], item["in_stock"]], separators=(",", ":"))
    return hashlib.sha256(key.encode()).hexdigest()[:16]

def fetch(url):
    req = urllib.request.Request(url, headers={"User-Agent": UA})
    with urllib.request.urlopen(req, timeout=10) as r:
        return r.read().decode(r.headers.get_content_charset() or "utf-8")

def crawl(delay=None):
    rp = urllib.robotparser.RobotFileParser(BASE + "/robots.txt"); rp.read()
    delay = (rp.crawl_delay(UA) or 1) if delay is None else delay
    url, items, pages = BASE + "/products?page=1", [], 0
    while url:
        if not rp.can_fetch(UA, url):
            print("robots.txt disallows", url, file=sys.stderr); break
        p = ListParser(); p.feed(fetch(url)); pages += 1
        items += [normalize(i, url) for i in p.items]
        url = urljoin(url, p.next) if p.next else None
        if url: time.sleep(delay)
    return items, pages

def utc_now():
    fake = os.environ.get("SHOPWATCH_NOW")       # for tests and the 30-day simulation
    return fake or datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")

def store(db, items, now):
    changes = []
    with db:                                      # one transaction per run
        for it in items:
            h = content_hash(it)
            old = db.execute("SELECT last_hash FROM products WHERE sku=?", (it["sku"],)).fetchone()
            db.execute("""INSERT INTO products (sku, name, url, first_seen, last_seen, last_hash)
                          VALUES (?, ?, ?, ?, ?, ?)
                          ON CONFLICT(sku) DO UPDATE SET name=excluded.name, url=excluded.url,
                            last_seen=excluded.last_seen, last_hash=excluded.last_hash""",
                       (it["sku"], it["name"], it["url"], now, now, h))
            if old is None or old[0] != h:        # new product or something changed
                prev = db.execute("""SELECT price_cents, in_stock FROM snapshots WHERE sku=?
                                     ORDER BY seen_at DESC LIMIT 1""", (it["sku"],)).fetchone()
                db.execute("INSERT OR IGNORE INTO snapshots VALUES (?, ?, ?, ?, ?)",
                           (it["sku"], now, it["price_cents"], int(it["in_stock"]), h))
                if prev: changes.append((it["sku"], prev, (it["price_cents"], int(it["in_stock"]))))
    return changes

if __name__ == "__main__":
    db = sqlite3.connect(os.environ.get("SHOPWATCH_DB", "shop.db"), timeout=5)
    db.executescript(SCHEMA)
    t0 = time.perf_counter()
    items, pages = crawl()
    changes = store(db, items, utc_now())
    for sku, (p0, s0), (p1, s1) in changes:
        what = f"price {p0/100:.2f} -> {p1/100:.2f}" if p0 != p1 else f"stock {s0} -> {s1}"
        print(f"CHANGE {sku} {what}")
    print(f"{utc_now()} pages={pages} items={len(items)} changes={len(changes)} "
          f"secs={time.perf_counter() - t0:.1f}")

「0日目」を2回続けて実行し、次に「1日目」を実行しました。

$ SHOPWATCH_NOW=2026-11-01T03:00:00Z python3 shopwatch.py
2026-11-01T03:00:00Z pages=10 items=200 changes=0 secs=9.2
$ SHOPWATCH_NOW=2026-11-01T03:00:00Z python3 shopwatch.py
2026-11-01T03:00:00Z pages=10 items=200 changes=0 secs=9.1
$ echo 1 > ~/demo/shop/state/day
$ SHOPWATCH_NOW=2026-11-02T03:00:00Z python3 shopwatch.py
CHANGE SKU-0020 price 66.79 -> 63.49
CHANGE SKU-0029 price 84.09 -> 88.29
CHANGE SKU-0036 price 74.89 -> 67.39
CHANGE SKU-0039 stock 1 -> 0
...
CHANGE SKU-0170 price 56.79 -> 53.99
2026-11-02T03:00:00Z pages=10 items=200 changes=10 secs=9.1
$ sqlite3 shop.db "select count(*) from products; select count(*) from snapshots;"
200
210

0日目の2回目は何も変えていません。重複行も誤報もありません。この性質を冪等性（べきとうせい）と呼びます。同じジョブを2回動かしても、1回動かしたのと同じ結果になることです。cron の再実行、スリープ明けのノート PC、うっかりもう一度押したボタン。冪等なジョブなら、どれも害になりません。

よくある間違いと、実際に起きること

間違い1：HTML 全体をハッシュする。 一見うまい方法です。ページごとにハッシュを取り、変わったら通知する。しかし練習ショップは、多くの実サイトと同じく、毎回の生成時刻とリクエスト ID をページに出しています。後述の30日分の実行で、HTML 全体のハッシュは 290 回中 290 回変わりました。本当に価格か在庫が変わった商品を含むページは 217 回だけです。対策は content_hash() のように、正規化した項目だけをハッシュすること。項目の順序を固定する（ランダムな順の dict ではなく JSON のリスト）と、ハッシュが安定します。

間違い2：商品行を素の INSERT で書く。 2回目の実行で止まります。

IntegrityError: UNIQUE constraint failed: products.sku

間違い3：それを INSERT OR REPLACE で「直す」。 動きはしますが、REPLACE の意味は「古い行を消して新しい行を入れる」です。10月1日に SKU-0001 を入れ、10月2日に REPLACE しました。

after OR REPLACE: (2, 'SKU-0001', 'Desk lamp 1', '2026-10-02T03:00:00Z', '2026-10-02T03:00:00Z')
after upsert:     (2, 'SKU-0001', 'Desk lamp 1', '2026-10-02T03:00:00Z', '2026-10-03T03:00:00Z')

rowid が 1 から 2 に飛び、first_seen は10月2日になりました。履歴が黙って間違っています。外部キーを有効にしていれば、削除が連鎖したり失敗したりもします。対策は INSERT ... ON CONFLICT(sku) DO UPDATE SET ...（SQLite 3.24 以降）で、変えるべき列だけを書くこと。first_seen は触られません。

間違い4：int(float(...)) でお金を変換する。 int(float("4.35") * 100) は 434 です。4.35 * 100 が 434.99999999999994 になり、int() が切り捨てるからです。1.00 から 99.99 までの全価格で試すと、9,900 件中 570 件が1セント少なくなりました。対策は上のコードのように round() を使うか、計算もするなら decimal.Decimal です。

間違い5：ローカル時刻をデータベースに入れる。 プロセスのタイムゾーンを5時間ずれた固定値（TZ=EST5）にすると、3つの時計が食い違います。

python datetime.now()          : 2026-11-14 18:16:36 (naive local)
python datetime.now(utc)       : 2026-11-14T23:16:36Z
sqlite datetime('now')         : 2026-11-14 23:16:36

SQLite の datetime('now') は UTC です。Python のタイムゾーンなしのローカル時刻と混ぜると、「直近24時間の変化」が何時間もずれます。夏時間のある地域のサーバーなら、年に2回さらにずれます。対策は、どこでも Z 付きの UTC で書き、表示するときだけローカル時刻に変換することです。

履歴を問い合わせる

変化だけの履歴でも、すべての質問に答えられます。ある時点 T の状態は、T 以前で最新のスナップショットだからです。

SELECT price_cents, in_stock, seen_at FROM snapshots
WHERE sku='SKU-0020' AND seen_at <= '2026-10-20T23:59:59Z'
ORDER BY seen_at DESC LIMIT 1;
-- 4589 | 1 | 2026-10-17T03:00:00Z

「今日の実行で値下がり幅が大きい商品」はウィンドウ関数1つです（SQLite 3.25 以降）。

WITH ordered AS (
  SELECT sku, seen_at, price_cents,
         LAG(price_cents) OVER (PARTITION BY sku ORDER BY seen_at) AS before
  FROM snapshots)
SELECT sku, before, price_cents AS now, ROUND(100.0*(price_cents-before)/before,1) AS pct
FROM ordered WHERE seen_at = '2026-10-30T03:00:00Z' AND price_cents < before
ORDER BY pct LIMIT 3;

sku       before  now   pct
--------  ------  ----  -----
SKU-0036  5439    4619  -15.1
SKU-0162  8629    7329  -15.1
SKU-0111  5469    4649  -15.0

WHERE・ORDER BY・LIMIT が初めてなら、SQL 入門：SELECT・WHERE・ORDER BY から読んでください。

実測：容量の増え方、重複排除、コミットのコスト

練習ショップで30日分。 シミュレーション上の1日に1回、30日間ウォッチャーを動かしました（クロール間隔1秒、1回約9秒）。同じパース結果から、毎回全商品を保存する2つ目のデータベースも作りました。30回後の結果です。

594 行は、初回の 200 商品と 394 回の本当の変化です。

1年分を HTTP なしで。 規模が大きいときの増え方を見るため、2,000 商品 × 365 日分を直接挿入しました（1日7%が変化、random.seed(7)）。

dbstat ビューで見ると、77 MB のうち 33.6 MB は PRIMARY KEY (sku, seen_at) のための自動インデックスでした。通常のテーブルはキーをインデックスとテーブル本体の2か所に持つからです。WITHOUT ROWID テーブルにすると、行そのものがキーの順で保存されます。

コミットのコスト。 もう1つ驚かれる数字です。ファイル DB への 10,000 件の upsert、5回の中央値です。

1行ごとにコミットすると、既定のロールバックジャーナルで 3,143.2 ms、WAL モードで 258.4 ms。1つのトランザクションにまとめると 7.4 ms と 7.2 ms でした。コミットのたびにディスクへの書き込みを待つからで、1回の実行につき1回で十分です。store() が実行全体を with db: で包んでいるのはそのためで、途中でクラッシュしても「今日の半分」ではなく「昨日のデータ」が丸ごと残ります。

cron で UTC のまま定期実行する

このマシンには何も設定していません。自分で管理している Linux サーバーや Mac で crontab -e に書くなら、この1行です。

15 3 * * * cd "$HOME/demo/shopwatch" && /usr/bin/python3 shopwatch.py >> logs/shopwatch.log 2>&1

• インタープリタは絶対パスで。 cron の PATH は最小限なので、python3 だけでは見つからないことがあります。
• 出力はすべてログへ。 >> ... 2>&1 でトレースバックも残します。ローテーションや終了コードはシェルスクリプトの堅牢化・cron・ログで扱っています。
• cron のスケジュールはマシンのローカル時刻。 実装によっては（多くの Linux の cronie など）CRON_TZ=UTC が使えますが、ここでは試していません。どちらにしても、データはスクリプトが UTC で書くので UTC です。
• 実行の重なり。 1回が長引いて次が始まると、書き込みが2つぶつかります。1つ目の接続が書き込みトランザクションを持ったまま、2つ目が INSERT しようとする状況を再現しました。

timeout=0: OperationalError: database is locked after 0.00 s
timeout=2: OperationalError: database is locked after 2.11 s

timeout は SQLite にロックを待たせるだけで、衝突そのものは消しません。実行間隔を処理時間より十分長くし、ロックを置き（たとえば開始時に mkdir "$HOME/demo/shopwatch/.lock"、終了時に rmdir、mkdir が失敗したらその回はスキップ）、ジョブを冪等に保てば、スキップや二重実行があっても害はありません。

どの保存方法をいつ使うか

正直な限界とバージョンの注意

• Python 3.9.6 と SQLite 3.51.0 で確認しました。ON CONFLICT ... DO UPDATE は SQLite 3.24（2018年）、LAG() は 3.25 からです。本番サーバーの sqlite3.sqlite_version は手元より古いことがあるので確認してください。
• 練習ショップの SKU は安定しています。実サイトでは ID が変わり、バリエーションが統合され、地域やログイン状態で価格が違います。キーの選び方がいちばん難しい部分です。
• このマシンには requests がたまたま入っていましたが、スクリプトは標準ライブラリだけを使っています。requests なら取得は requests.get(url, headers=..., timeout=10).text ですが、ここでは実行していません。
• ファイルサイズはページサイズ、キーの長さ、SQLite のバージョンで変わります。ディスク容量の見積もりは自分のデータで測り直してください。
• SHA-256 を16桁（64ビット）に切り詰めたハッシュは、数千行の変化検知には十分ですが、セキュリティ対策ではありません。

演習

1. 「消えた商品」のレポートを追加してください。products にはあるのに、今日の実行に現れなかった商品です。どの列がすでにそれを教えてくれるでしょう。
2. 2026-10-23T03:00:00Z の実行で在庫切れになった商品を一覧する SQL を書いてください。

解答

1. last_seen です。store() の後に、今日のタイムスタンプで SELECT sku FROM products WHERE last_seen < ? を実行すれば、今回見えなかった商品が出ます。すぐ削除はしないでください。1日だけ消えた商品は、たいていページの取得失敗です（第6回）。
2. 価格のクエリと同じく LAG(in_stock) を使います。WITH o AS (SELECT sku, seen_at, in_stock, LAG(in_stock) OVER (PARTITION BY sku ORDER BY seen_at) AS before FROM snapshots) SELECT sku FROM o WHERE seen_at='2026-10-23T03:00:00Z' AND before=1 AND in_stock=0; 今回のデータでは SKU-0020 と SKU-0136 が返りました。

シリーズの案内

• 前の回：Webスクレイピング講座 4/6：セッション・フォーム・Cookie・CSRF
• 次の回：Webスクレイピング講座 6/6：堅牢化・監視・レイアウト変更
• シリーズの最初：第1回：robots.txt と urllib
• 時刻に依存するコードのテスト：不安定なテスト：時刻・乱数・並行性
• すべての講座：プログラミング学習ロードマップ