chore: initial backend + travel-proxy dockerized setup for NAS deployment
This commit is contained in:
239
backend/app/db.py
Normal file
239
backend/app/db.py
Normal file
@@ -0,0 +1,239 @@
|
||||
# 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 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
|
||||
|
||||
Reference in New Issue
Block a user