스크래핑 변화 감지: SQLite 스냅숏과 해시로 가격 추적

시리즈: 웹 스크래핑 강좌 5/6 (고급). 가져오기, 파싱, 페이지 넘기기를 예의 있게 할 수 있게 된 사람이 "매일 돌려서 어제와 무엇이 달라졌는지 알고 싶다"는 단계로 넘어가는 회차입니다. 다 읽고 나면 SQLite 에 스냅숏을 쌓고, 콘텐츠 해시로 가격과 재고의 진짜 변화만 골라내고, 두 번 실행해도 망가지지 않는 코드를 쓰고, UTC 타임스탬프로 cron 에 올릴 수 있습니다.

이 글은 가격 비교용으로 한 번 돌릴 스크립트를 만들었다가 "그럼 이거 매일 아침 자동으로 돌려 줘"라는 말을 들은 사람을 위한 글입니다. 오늘 가격을 출력하는 건 쉽습니다. 어제를 기억하고, 진짜 변화만 알려 주고, 두 번 돌아도 괜찮고, 디스크를 채우지 않는 스크립트는 전혀 다른 일이고, 그 대부분은 데이터베이스 일입니다.

아래 내용은 모두 Apple Silicon Mac (M1 Pro) 에서 Python 3.9.6(표준 라이브러리만), SQLite 3.51.0(이 Python 에 들어 있는 라이브러리와 sqlite3 CLI), 그리고 시리즈의 연습용 쇼핑몰을 127.0.0.1:18172 에서 띄운 PHP 8.5.6 으로 직접 돌렸습니다. 경로는 ~/demo/ 로 줄였습니다. 수집 대상은 직접 띄운 연습 사이트뿐입니다.

저장하기 전에: 정말 스크래핑이 맞을까

매일 도는 작업은 모든 것을 곱셈으로 만듭니다. 요청 한 번은 1년이면 365번입니다. 그래서 1강에서 확인한 것들이 여기서 더 무거워집니다.

• 공식 API 가 먼저입니다. 공공데이터라면 스크래핑보다 공공데이터포털의 오픈 API 를 먼저 찾아보세요. 형식이 안정적이고 이용 조건이 명확하며, 갱신 주기도 적혀 있습니다. 쇼핑몰도 상품 피드나 제휴 API 를 제공하는 곳이 많습니다.
• robots.txt 와 Crawl-delay. 연습 쇼핑몰의 robots.txt 는 Crawl-delay: 1 이라서, 감시 스크립트는 페이지 사이에 1초를 기다립니다. 10페이지에 약 9초. 하루 한 번이면 아무 문제 없는, 손님으로서 당연한 비용입니다.
• 약관과 개인정보. 가격과 재고는 보통 개인정보가 아니지만, 작성자 이름이 붙은 리뷰는 개인정보입니다. 수집한 데이터를 모아 두는 것 자체가 데이터베이스 관련 권리나 약관 문제와 얽힐 수도 있습니다. 저장을 시작하기 전에 그 사이트의 약관과 해당 지역의 법을 확인하고, 필요한 필드만 저장하세요. 이 글은 법률 자문이 아닙니다.
• 정직한 User-Agent. 감시 스크립트는 shopwatch/1.0 (+https://example.com/bot-contact) 를 보냅니다. 이름과, 사이트 운영자가 연락할 수 있는 곳입니다.

데이터 모델: products 와 snapshots

연습 쇼핑몰에는 10페이지에 상품 200개가 있습니다. HTML 에서 상품 하나는 이렇게 보입니다.

<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>

서버의 상태 파일로 "날짜"를 넘기면, 하루에 가격 약 4% 가 바뀌고 상품 약 3% 의 재고가 바뀝니다(random.seed(42) 로 한 번 생성).

테이블은 두 개입니다. products 는 상품마다 한 행이고 마지막으로 본 상태를 가집니다. 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));

여기에는 결정 세 가지가 숨어 있습니다.

1. 금액은 정수 센트. $66.79 는 6679 로 저장합니다. 원화는 처음부터 정수지만, 해외 직구 가격처럼 달러를 다루는 순간 부동소수점 문제가 바로 나옵니다. 실제 버그는 아래에서 봅니다.
2. 타임스탬프는 UTC 문자열. 2026-10-02T03:00:00Z 처럼 ISO 8601 끝에 Z 를 붙입니다. 이 형식은 문자열 그대로 올바르게 정렬되고 비교되며 sqlite3 CLI 에서도 읽힙니다.
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일차"를 두 번 연속 돌리고, 이어서 "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일차의 두 번째 실행은 아무것도 바꾸지 않았습니다. 중복 행도, 잘못된 알림도 없습니다. 이 성질을 멱등성이라고 합니다. 같은 작업을 두 번 실행해도 한 번 실행한 것과 결과가 같다는 뜻입니다. cron 재시도, 잠자기에서 깨어난 노트북, 실수로 한 번 더 누른 버튼. 멱등한 작업이라면 전부 무해합니다.

흔한 실수와 실제로 일어나는 일

실수 1: HTML 전체를 해시한다. 그럴듯해 보입니다. 페이지마다 해시를 떠 두고 바뀌면 알린다. 하지만 연습 쇼핑몰은 실제 사이트 대부분처럼 매 요청마다 생성 시각과 요청 ID 를 페이지에 찍습니다. 아래 30일 실행에서 HTML 전체 해시는 290번 중 290번 바뀌었습니다. 실제로 가격이나 재고가 바뀐 상품이 있는 페이지는 217번뿐이었습니다. 해결책은 content_hash() 처럼 정규화한 필드만 해시하는 것입니다. 필드 순서를 고정하면(순서가 흔들리는 dict 가 아니라 JSON 리스트) 해시가 안정됩니다.

실수 2: 상품 행을 그냥 INSERT 한다. 두 번째 실행에서 멈춥니다.

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: 로컬 시각을 DB 에 넣는다. 프로세스 시간대를 5시간 차이 나는 고정값(TZ=EST5)으로 두면 시계 세 개가 서로 다릅니다.

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시간 동안의 변화"가 몇 시간씩 어긋납니다. 서머타임이 있는 지역의 서버라면 1년에 두 번 더 어긋납니다. 해결책은 어디서나 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

"오늘 실행에서 가장 많이 내린 상품"은 윈도 함수 하나면 됩니다(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일. 시뮬레이션 하루에 한 번씩 30일 동안 감시 스크립트를 돌렸습니다(크롤 간격 1초, 1회 약 9초). 같은 파싱 결과로 매번 모든 상품을 저장하는 두 번째 DB 도 만들었습니다. 30회 후 결과입니다.

594행은 처음 본 상품 200개와 진짜 변화 394건입니다.

HTTP 없이 1년치. 규모가 커졌을 때를 보려고 상품 2,000개 × 365일치를 직접 넣었습니다(하루 7% 변화, random.seed(7)).

dbstat 뷰로 보니 77 MB 중 33.6 MB 가 PRIMARY KEY (sku, seen_at) 를 위한 자동 인덱스였습니다. 일반 테이블은 키를 인덱스와 테이블 본체 두 곳에 갖기 때문입니다. WITHOUT ROWID 테이블로 선언하면 행 자체가 키 순서로 저장됩니다.

커밋 비용. 사람들이 놀라는 또 하나의 숫자입니다. 파일 DB 에 upsert 10,000건, 5회 중앙값입니다.

행마다 커밋하면 기본 롤백 저널에서 3,143.2 ms, WAL 모드에서 258.4 ms 였고, 트랜잭션 하나로 묶으면 7.4 ms 와 7.2 ms 였습니다. 커밋할 때마다 디스크 기록을 기다리기 때문이고, 실행 한 번에 커밋 한 번이면 충분합니다. store() 가 실행 전체를 with db: 로 감싸는 이유이고, 덕분에 중간에 죽어도 "오늘 데이터 절반"이 아니라 "어제 데이터 전부"가 남습니다.

cron 으로 UTC 그대로 정기 실행

이 테스트 머신에는 아무것도 설치하지 않았습니다. 직접 관리하는 Linux 서버나 Mac 에서 crontab -e 로 넣는다면 이 한 줄입니다.

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 입니다.
• 실행이 겹칠 때. 한 번이 길어져 다음 실행이 시작되면 쓰기 두 개가 부딪힙니다. 첫 번째 연결이 쓰기 트랜잭션을 잡은 채로 두 번째가 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 < ? 를 실행하면 이번에 안 보인 상품이 나옵니다. 바로 지우지는 마세요. 하루만 사라진 상품은 대개 페이지 가져오기 실패입니다(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 이 나왔습니다.

시리즈 안내

• 이전 강의: 웹 스크래핑 강좌 4/6: 세션, 폼, 쿠키, CSRF
• 다음 강의: 웹 스크래핑 강좌 6/6: 견고함, 모니터링, 레이아웃 변경
• 시리즈 처음: 1강: robots.txt 와 urllib
• 시간에 의존하는 코드 테스트: 불안정한 테스트: 시간, 난수, 동시성
• 전체 강좌: 프로그래밍 학습 로드맵