수집 데이터 차분 감지 및 엑셀/CSV 적재

[ 새 수집 데이터 도착: 상품 A = 45,000원 ]
                │
                ▼
┌───────────────────────────────────────────────┐
│ SQLite DB 직전 기록 조회                      │
│ SELECT price, in_stock FROM products WHERE id=A │
└───────────────────────────────────────────────┘
                │
         ┌──────┴────────────────────────┐
      (변동 없음)                      (변동 감지!)
         │                               │
         ▼                               ▼
    [알림 생략]                 ┌─────────────────────────────────┐
  (스팸 방지 100%)              │ 1. DB 가격 갱신 (45,000원)      │
                                │ 2. 시계열 히스토리 로그 INSERT  │
                                │ 3. 텔레그램/디스코드 특가 발송  │
                                │    "🚨 50,000원 ➔ 45,000원 할인!"│
                                └─────────────────────────────────┘

import os
import sqlite3
import csv
import logging
from datetime import datetime
from pathlib import Path
import requests
 
logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger("PricePipeline")
 
# 영구 디스크 경로
DATA_DIR = Path("/app/data")
DATA_DIR.mkdir(parents=True, exist_ok=True)
DB_PATH = DATA_DIR / "monitoring.db"
CSV_DIR = DATA_DIR / "csv_exports"
CSV_DIR.mkdir(parents=True, exist_ok=True)
 
DISCORD_WEBHOOK_URL = os.environ.get("DISCORD_WEBHOOK_URL")
 
def init_db():
    with sqlite3.connect(DB_PATH) as conn:
        conn.execute("PRAGMA journal_mode = WAL;")
        # 최신 상태 테이블
        conn.execute("""
            CREATE TABLE IF NOT EXISTS product_latest (
                product_id TEXT PRIMARY KEY,
                product_name TEXT,
                price INTEGER,
                in_stock INTEGER,
                updated_at TIMESTAMP
            );
        """)
        # 시계열 가격 변동 이력 테이블
        conn.execute("""
            CREATE TABLE IF NOT EXISTS price_history (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                product_id TEXT,
                price INTEGER,
                in_stock INTEGER,
                recorded_at TIMESTAMP
            );
        """)
        conn.commit()
 
def process_scraped_product(product_id: str, name: str, price: int, in_stock: bool):
    """차분 감지 및 데이터 저장 엔진"""
    with sqlite3.connect(DB_PATH) as conn:
        cursor = conn.cursor()
        
        # 1. 직전 상태 조회
        cursor.execute("SELECT price, in_stock FROM product_latest WHERE product_id = ?", (product_id,))
        previous = cursor.fetchone()
        
        is_changed = False
        change_reason = ""
 
        if previous is None:
            # 신규 상품 최초 등록
            is_changed = True
            change_reason = "신규 상품 추적 시작"
        else:
            old_price, old_stock = previous
            current_stock_int = 1 if in_stock else 0
            
            # 가격 하락 감지
            if price < old_price:
                diff = old_price - price
                rate = (diff / old_price) * 100
                is_changed = True
                change_reason = f"📉 가격 인하! {old_price:,}원 ➔ {price:,}원 ({rate:.1f}% 할인)"
            # 품절 해제(재입고) 감지
            elif old_stock == 0 and current_stock_int == 1:
                is_changed = True
                change_reason = "🎉 품절 해제! 지금 구매 가능"
            # 가격 인상
            elif price > old_price:
                is_changed = True
                change_reason = f"📈 가격 인상 ({old_price:,}원 ➔ {price:,}원)"
 
        # 2. 상태 갱신 및 시계열 로그 기록
        now_str = datetime.now().isoformat()
        cursor.execute("""
            INSERT INTO product_latest (product_id, product_name, price, in_stock, updated_at)
            VALUES (?, ?, ?, ?, ?)
            ON CONFLICT(product_id) DO UPDATE SET
                price = excluded.price,
                in_stock = excluded.in_stock,
                updated_at = excluded.updated_at;
        """, (product_id, name, price, 1 if in_stock else 0, now_str))
 
        cursor.execute("""
            INSERT INTO price_history (product_id, price, in_stock, recorded_at)
            VALUES (?, ?, ?, ?);
        """, (product_id, price, 1 if in_stock else 0, now_str))
        conn.commit()
 
        # 3. 변동이 있을 때만 정밀 웹훅 알림 발송
        if is_changed:
            logger.info(f"변동 감지 [{name}]: {change_reason}")
            send_discord_alert(name, change_reason, price)
 
def send_discord_alert(name: str, reason: str, price: int):
    if not DISCORD_WEBHOOK_URL:
        return
    payload = {
        "content": f"**[DashCloud 알림] {name}**\n{reason}\n현재가: **{price:,}원**",
        "username": "가격 감시 봇"
    }
    try:
        requests.post(DISCORD_WEBHOOK_URL, json=payload, timeout=5)
    except Exception as e:
        logger.error(f"디스코드 웹훅 발송 실패: {e}")
 
def export_daily_csv():
    """매일 자정 시계열 데이터를 CSV로 일자별 내보내기"""
    today_str = datetime.now().strftime("%Y%m%d")
    csv_file = CSV_DIR / f"price_history_{today_str}.csv"
 
    with sqlite3.connect(DB_PATH) as conn:
        cursor = conn.cursor()
        cursor.execute("""
            SELECT p.product_name, h.price, h.in_stock, h.recorded_at 
            FROM price_history h
            JOIN product_latest p ON h.product_id = p.product_id
            ORDER BY h.recorded_at DESC;
        """)
        rows = cursor.fetchall()
 
    with open(csv_file, "w", newline="", encoding="utf-8-sig") as f: # utf-8-sig: 엑셀 한글 깨짐 방지
        writer = csv.writer(f)
        writer.writerow(["상품명", "가격(원)", "재고상태", "수집시각"])
        writer.writerows(rows)
 
    logger.info(f"일일 CSV 데이터 내보내기 완료: {csv_file}")
 
if __name__ == "__main__":
    init_db()
    # 테스트 시뮬레이션
    process_scraped_product("PROD-101", "에어팟 프로 2세대", 299000, True)
    # 가격 인하 시뮬레이션
    process_scraped_product("PROD-101", "에어팟 프로 2세대", 269000, True)
    export_daily_csv()