diff --git a/backend/app/db.py b/backend/app/db.py index ed6e0f1..30afb0a 100644 --- a/backend/app/db.py +++ b/backend/app/db.py @@ -162,6 +162,25 @@ def init_db() -> None: "CREATE INDEX IF NOT EXISTS idx_todos_created ON todos(created_at DESC);" ) + # ── blog_posts 테이블 ────────────────────────────────────────────────── + conn.execute( + """ + CREATE TABLE IF NOT EXISTS blog_posts ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + title TEXT NOT NULL, + body TEXT NOT NULL DEFAULT '', + excerpt TEXT NOT NULL DEFAULT '', + tags TEXT NOT NULL DEFAULT '[]', + date TEXT NOT NULL DEFAULT (date('now','localtime')), + created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now')), + updated_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now')) + ); + """ + ) + conn.execute( + "CREATE INDEX IF NOT EXISTS idx_blog_date ON blog_posts(date DESC);" + ) + # ── todos CRUD ─────────────────────────────────────────────────────────────── @@ -230,6 +249,68 @@ def delete_done_todos() -> int: return cur.rowcount +# ── blog_posts CRUD ────────────────────────────────────────────────────────── + +def _post_row_to_dict(r) -> Dict[str, Any]: + return { + "id": r["id"], + "title": r["title"], + "body": r["body"], + "excerpt": r["excerpt"], + "tags": json.loads(r["tags"]) if r["tags"] else [], + "date": r["date"], + "created_at": r["created_at"], + "updated_at": r["updated_at"], + } + + +def get_all_posts() -> List[Dict[str, Any]]: + with _conn() as conn: + rows = conn.execute( + "SELECT * FROM blog_posts ORDER BY date DESC, id DESC" + ).fetchall() + return [_post_row_to_dict(r) for r in rows] + + +def create_post(title: str, body: str, excerpt: str, tags: List[str], date: str) -> Dict[str, Any]: + with _conn() as conn: + conn.execute( + "INSERT INTO blog_posts (title, body, excerpt, tags, date) VALUES (?, ?, ?, ?, ?)", + (title, body, excerpt, json.dumps(tags), date), + ) + row = conn.execute( + "SELECT * FROM blog_posts WHERE rowid = last_insert_rowid()" + ).fetchone() + return _post_row_to_dict(row) + + +def update_post(post_id: int, fields: Dict[str, Any]) -> Optional[Dict[str, Any]]: + allowed = {"title", "body", "excerpt", "tags", "date"} + updates = {k: v for k, v in fields.items() if k in allowed} + if not updates: + with _conn() as conn: + row = conn.execute("SELECT * FROM blog_posts WHERE id = ?", (post_id,)).fetchone() + return _post_row_to_dict(row) if row else None + + if "tags" in updates: + updates["tags"] = json.dumps(updates["tags"]) + + set_clauses = ", ".join(f"{k} = ?" for k in updates) + set_clauses += ", updated_at = strftime('%Y-%m-%dT%H:%M:%fZ','now')" + args = list(updates.values()) + [post_id] + + with _conn() as conn: + conn.execute(f"UPDATE blog_posts SET {set_clauses} WHERE id = ?", args) + row = conn.execute("SELECT * FROM blog_posts WHERE id = ?", (post_id,)).fetchone() + return _post_row_to_dict(row) if row else None + + +def delete_post(post_id: int) -> bool: + with _conn() as conn: + cur = conn.execute("DELETE FROM blog_posts WHERE id = ?", (post_id,)) + return cur.rowcount > 0 + + def upsert_draw(row: Dict[str, Any]) -> None: with _conn() as conn: conn.execute( diff --git a/backend/app/main.py b/backend/app/main.py index c8e8549..2c721cb 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -12,6 +12,8 @@ from .db import ( get_best_picks, get_simulation_runs, get_simulation_candidates, # todos get_all_todos, create_todo, update_todo, delete_todo, delete_done_todos, + # blog + get_all_posts, create_post, update_post, delete_post, ) from .recommender import recommend_numbers, recommend_with_heatmap from .collector import sync_latest, sync_ensure_all @@ -621,3 +623,50 @@ def api_todos_delete(todo_id: str): if not ok: raise HTTPException(status_code=404, detail="Todo not found") return {"ok": True} + + +# ── Blog API ────────────────────────────────────────────────────────────────── + +class BlogPostCreate(BaseModel): + title: str + body: str = "" + excerpt: str = "" + tags: List[str] = [] + date: str = "" # 빈 문자열이면 오늘 날짜 사용 + + +class BlogPostUpdate(BaseModel): + title: Optional[str] = None + body: Optional[str] = None + excerpt: Optional[str] = None + tags: Optional[List[str]] = None + date: Optional[str] = None + + +@app.get("/api/blog/posts") +def api_blog_list(): + return {"posts": get_all_posts()} + + +@app.post("/api/blog/posts", status_code=201) +def api_blog_create(body: BlogPostCreate): + from datetime import date as _date + post_date = body.date if body.date else _date.today().isoformat() + post = create_post(body.title, body.body, body.excerpt, body.tags, post_date) + return post + + +@app.put("/api/blog/posts/{post_id}") +def api_blog_update(post_id: int, body: BlogPostUpdate): + updated = update_post(post_id, body.model_dump(exclude_none=True)) + if updated is None: + raise HTTPException(status_code=404, detail="Post not found") + return updated + + +@app.delete("/api/blog/posts/{post_id}") +def api_blog_delete(post_id: int): + ok = delete_post(post_id) + if not ok: + raise HTTPException(status_code=404, detail="Post not found") + return {"ok": True}