web-page-backend/backend/app/db.py

# backend/app/db.py
import os
import sqlite3
import json
import hashlib
from typing import Any, Dict, Optional, List

DB_PATH = "/app/data/lotto.db"

def _conn() -> sqlite3.Connection:
    os.makedirs(os.path.dirname(DB_PATH), exist_ok=True)
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row
    return conn

def _ensure_column(conn: sqlite3.Connection, table: str, col: str, ddl: str) -> None:
    cols = {r["name"] for r in conn.execute(f"PRAGMA table_info({table})").fetchall()}
    if col not in cols:
        conn.execute(ddl)

def init_db() -> None:
    with _conn() as conn:
        conn.execute(
            """
            CREATE TABLE IF NOT EXISTS draws (
              drw_no   INTEGER PRIMARY KEY,
              drw_date TEXT NOT NULL,
              n1       INTEGER NOT NULL,
              n2       INTEGER NOT NULL,
              n3       INTEGER NOT NULL,
              n4       INTEGER NOT NULL,
              n5       INTEGER NOT NULL,
              n6       INTEGER NOT NULL,
              bonus    INTEGER NOT NULL,
              updated_at TEXT NOT NULL DEFAULT (datetime('now'))
            );
            """
        )
        conn.execute("CREATE INDEX IF NOT EXISTS idx_draws_date ON draws(drw_date);")

        conn.execute(
            """
            CREATE TABLE IF NOT EXISTS recommendations (
              id INTEGER PRIMARY KEY AUTOINCREMENT,
              created_at TEXT NOT NULL DEFAULT (datetime('now')),
              based_on_draw INTEGER,
              numbers TEXT NOT NULL,
              params TEXT NOT NULL
            );
            """
        )
        conn.execute("CREATE INDEX IF NOT EXISTS idx_reco_created ON recommendations(created_at DESC);")

        # ✅ 확장 컬럼들(기존 DB에도 자동 추가)
        _ensure_column(conn, "recommendations", "numbers_sorted",
                       "ALTER TABLE recommendations ADD COLUMN numbers_sorted TEXT;")
        _ensure_column(conn, "recommendations", "dedup_hash",
                       "ALTER TABLE recommendations ADD COLUMN dedup_hash TEXT;")
        _ensure_column(conn, "recommendations", "favorite",
                       "ALTER TABLE recommendations ADD COLUMN favorite INTEGER NOT NULL DEFAULT 0;")
        _ensure_column(conn, "recommendations", "note",
                       "ALTER TABLE recommendations ADD COLUMN note TEXT NOT NULL DEFAULT '';")
        _ensure_column(conn, "recommendations", "tags",
                       "ALTER TABLE recommendations ADD COLUMN tags TEXT NOT NULL DEFAULT '[]';")

        # ✅ UNIQUE 인덱스(중복 저장 방지)
        conn.execute("CREATE UNIQUE INDEX IF NOT EXISTS uq_reco_dedup ON recommendations(dedup_hash);")

def upsert_draw(row: Dict[str, Any]) -> None:
    with _conn() as conn:
        conn.execute(
            """
            INSERT INTO draws (drw_no, drw_date, n1, n2, n3, n4, n5, n6, bonus)
            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
            ON CONFLICT(drw_no) DO UPDATE SET
              drw_date=excluded.drw_date,
              n1=excluded.n1, n2=excluded.n2, n3=excluded.n3,
              n4=excluded.n4, n5=excluded.n5, n6=excluded.n6,
              bonus=excluded.bonus,
              updated_at=datetime('now')
            """,
            (
                int(row["drw_no"]),
                str(row["drw_date"]),
                int(row["n1"]), int(row["n2"]), int(row["n3"]),
                int(row["n4"]), int(row["n5"]), int(row["n6"]),
                int(row["bonus"]),
            ),
        )

def upsert_many_draws(rows: List[Dict[str, Any]]) -> None:
    data = [
        (
            int(r["drw_no"]), str(r["drw_date"]),
            int(r["n1"]), int(r["n2"]), int(r["n3"]),
            int(r["n4"]), int(r["n5"]), int(r["n6"]),
            int(r["bonus"])
        ) for r in rows
    ]
    with _conn() as conn:
        conn.executemany(
            """
            INSERT INTO draws (drw_no, drw_date, n1, n2, n3, n4, n5, n6, bonus)
            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
            ON CONFLICT(drw_no) DO UPDATE SET
              drw_date=excluded.drw_date,
              n1=excluded.n1, n2=excluded.n2, n3=excluded.n3,
              n4=excluded.n4, n5=excluded.n5, n6=excluded.n6,
              bonus=excluded.bonus,
              updated_at=datetime('now')
            """,
            data
        )

def get_latest_draw() -> Optional[Dict[str, Any]]:
    with _conn() as conn:
        r = conn.execute("SELECT * FROM draws ORDER BY drw_no DESC LIMIT 1").fetchone()
        return dict(r) if r else None

def get_draw(drw_no: int) -> Optional[Dict[str, Any]]:
    with _conn() as conn:
        r = conn.execute("SELECT * FROM draws WHERE drw_no = ?", (drw_no,)).fetchone()
        return dict(r) if r else None

def count_draws() -> int:
    with _conn() as conn:
        r = conn.execute("SELECT COUNT(*) AS c FROM draws").fetchone()
        return int(r["c"])

def get_all_draw_numbers():
    with _conn() as conn:
        rows = conn.execute(
            "SELECT drw_no, n1, n2, n3, n4, n5, n6 FROM draws ORDER BY drw_no ASC"
        ).fetchall()
        return [(int(r["drw_no"]), [int(r["n1"]), int(r["n2"]), int(r["n3"]), int(r["n4"]), int(r["n5"]), int(r["n6"])]) for r in rows]

# ---------- ✅ recommendation helpers ----------

def _canonical_params(params: dict) -> str:
    return json.dumps(params, sort_keys=True, separators=(",", ":"))

def _numbers_sorted_str(numbers: List[int]) -> str:
    return ",".join(str(x) for x in sorted(numbers))

def _dedup_hash(based_on_draw: Optional[int], numbers: List[int], params: dict) -> str:
    s = f"{based_on_draw or ''}|{_numbers_sorted_str(numbers)}|{_canonical_params(params)}"
    return hashlib.sha1(s.encode("utf-8")).hexdigest()

def save_recommendation_dedup(based_on_draw: Optional[int], numbers: List[int], params: dict) -> Dict[str, Any]:
    """
    ✅ 동일 추천(번호+params+based_on_draw)이면 중복 저장 없이 기존 id 반환
    """
    ns = _numbers_sorted_str(numbers)
    h = _dedup_hash(based_on_draw, numbers, params)

    with _conn() as conn:
        # 이미 있으면 반환
        r = conn.execute("SELECT id FROM recommendations WHERE dedup_hash = ?", (h,)).fetchone()
        if r:
            return {"id": int(r["id"]), "saved": False, "deduped": True}

        cur = conn.execute(
            """
            INSERT INTO recommendations (based_on_draw, numbers, params, numbers_sorted, dedup_hash)
            VALUES (?, ?, ?, ?, ?)
            """,
            (based_on_draw, json.dumps(numbers), json.dumps(params), ns, h),
        )
        return {"id": int(cur.lastrowid), "saved": True, "deduped": False}

def list_recommendations_ex(
    limit: int = 30,
    offset: int = 0,
    favorite: Optional[bool] = None,
    tag: Optional[str] = None,
    q: Optional[str] = None,
    sort: str = "id_desc",  # id_desc|created_desc|favorite_desc
) -> List[Dict[str, Any]]:
    import json

    where = []
    args: list[Any] = []

    if favorite is not None:
        where.append("favorite = ?")
        args.append(1 if favorite else 0)

    if q:
        where.append("note LIKE ?")
        args.append(f"%{q}%")

    # tags는 JSON 문자열이므로 단순 LIKE로 처리(가볍게 시작)
    if tag:
        where.append("tags LIKE ?")
        args.append(f"%{tag}%")

    where_sql = ("WHERE " + " AND ".join(where)) if where else ""

    if sort == "created_desc":
        order = "created_at DESC"
    elif sort == "favorite_desc":
        # favorite(1)이 먼저, 그 다음 최신
        order = "favorite DESC, id DESC"
    else:
        order = "id DESC"

    sql = f"""
    SELECT id, created_at, based_on_draw, numbers, params, favorite, note, tags
    FROM recommendations
    {where_sql}
    ORDER BY {order}
    LIMIT ? OFFSET ?
    """
    args.extend([int(limit), int(offset)])

    with _conn() as conn:
        rows = conn.execute(sql, args).fetchall()

    out = []
    for r in rows:
        out.append({
            "id": int(r["id"]),
            "created_at": r["created_at"],
            "based_on_draw": r["based_on_draw"],
            "numbers": json.loads(r["numbers"]),
            "params": json.loads(r["params"]),
            "favorite": bool(r["favorite"]) if r["favorite"] is not None else False,
            "note": r["note"],
            "tags": json.loads(r["tags"]) if r["tags"] else [],
        })
    return out

def update_recommendation(rec_id: int, favorite: Optional[bool] = None, note: Optional[str] = None, tags: Optional[List[str]] = None) -> bool:
    fields = []
    args: list[Any] = []

    if favorite is not None:
        fields.append("favorite = ?")
        args.append(1 if favorite else 0)
    if note is not None:
        fields.append("note = ?")
        args.append(note)
    if tags is not None:
        fields.append("tags = ?")
        args.append(json.dumps(tags))

    if not fields:
        return False

    args.append(rec_id)

    with _conn() as conn:
        cur = conn.execute(
            f"UPDATE recommendations SET {', '.join(fields)} WHERE id = ?",
            args,
        )
        return cur.rowcount > 0

def delete_recommendation(rec_id: int) -> bool:
    with _conn() as conn:
        cur = conn.execute("DELETE FROM recommendations WHERE id = ?", (rec_id,))
        return cur.rowcount > 0