from fastapi import FastAPI, WebSocket, WebSocketDisconnect from fastapi.staticfiles import StaticFiles from fastapi.responses import FileResponse from pydantic import BaseModel from typing import Optional import sqlite3 import uuid from datetime import datetime import json app = FastAPI() app.mount("/static", StaticFiles(directory="."), name="static") DB = "wish.db" def db(): conn = sqlite3.connect(DB) conn.row_factory = sqlite3.Row conn.execute("PRAGMA journal_mode=WAL") return conn def init(): with db() as c: c.executescript(""" CREATE TABLE IF NOT EXISTS boards ( id TEXT PRIMARY KEY, name TEXT, color TEXT DEFAULT '#6366f1', created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); CREATE TABLE IF NOT EXISTS lists ( id TEXT PRIMARY KEY, board_id TEXT REFERENCES boards(id) ON DELETE CASCADE, title TEXT, position INTEGER DEFAULT 0 ); CREATE TABLE IF NOT EXISTS cards ( id TEXT PRIMARY KEY, list_id TEXT REFERENCES lists(id) ON DELETE CASCADE, board_id TEXT REFERENCES boards(id) ON DELETE CASCADE, title TEXT, description TEXT DEFAULT '', position INTEGER DEFAULT 0, image_url TEXT DEFAULT '', price REAL DEFAULT 0, url TEXT DEFAULT '', status TEXT DEFAULT 'want', category TEXT DEFAULT '', rating INTEGER DEFAULT 3, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, bought_at TIMESTAMP, comments TEXT DEFAULT '', archived INTEGER DEFAULT 0 ); CREATE INDEX IF NOT EXISTS idx_cards_status ON cards(status); CREATE INDEX IF NOT EXISTS idx_cards_category ON cards(category); CREATE INDEX IF NOT EXISTS idx_cards_board ON cards(board_id); """) init() class BoardIn(BaseModel): name: str color: str = "#6366f1" class ListIn(BaseModel): title: str class CardIn(BaseModel): title: str description: str = "" image_url: str = "" price: float = 0 url: str = "" status: str = "want" category: str = "" rating: int = 3 comments: str = "" class CardUpdate(BaseModel): title: Optional[str] = None description: Optional[str] = None image_url: Optional[str] = None price: Optional[float] = None url: Optional[str] = None status: Optional[str] = None category: Optional[str] = None rating: Optional[int] = None comments: Optional[str] = None archived: Optional[int] = None position: Optional[int] = None list_id: Optional[str] = None # Boards @app.get("/api/boards") def boards(): with db() as c: return [dict(r) for r in c.execute("SELECT * FROM boards ORDER BY created_at DESC").fetchall()] @app.post("/api/boards") def create_board(data: BoardIn): bid = str(uuid.uuid4()) with db() as c: c.execute("INSERT INTO boards VALUES (?,?,?,NULL)", (bid, data.name, data.color)) broadcast({"type": "board:create", "data": {"id": bid, "name": data.name, "color": data.color}}) return {"id": bid, "name": data.name, "color": data.color} @app.delete("/api/boards/{bid}") def delete_board(bid: str): with db() as c: c.execute("DELETE FROM boards WHERE id=?", (bid,)) broadcast({"type": "board:delete", "data": {"id": bid}}) return {"ok": True} # Lists @app.get("/api/boards/{bid}/lists") def lists(bid: str): with db() as c: return [dict(r) for r in c.execute("SELECT * FROM lists WHERE board_id=? ORDER BY position", (bid,)).fetchall()] @app.post("/api/boards/{bid}/lists") def create_list(bid: str, data: ListIn): lid = str(uuid.uuid4()) with db() as c: p = c.execute("SELECT COALESCE(MAX(position),-1) m FROM lists WHERE board_id=?", (bid,)).fetchone()["m"] c.execute("INSERT INTO lists VALUES (?,?,?,?)", (lid, bid, data.title, p+1)) broadcast({"type": "list:create", "data": {"id": lid, "board_id": bid, "title": data.title, "position": p+1}}) return {"id": lid, "board_id": bid, "title": data.title, "position": p+1} @app.patch("/api/lists/{lid}") def update_list(lid: str, data: dict): with db() as c: u, ps = [], [] if "title" in data: u.append("title=?"); ps.append(data["title"]) if "position" in data: u.append("position=?"); ps.append(data["position"]) if u: ps.append(lid) c.execute(f"UPDATE lists SET {', '.join(u)} WHERE id=?", ps) row = c.execute("SELECT * FROM lists WHERE id=?", (lid,)).fetchone() broadcast({"type": "list:update", "data": dict(row)}) return {"ok": True} @app.delete("/api/lists/{lid}") def delete_list(lid: str): with db() as c: c.execute("DELETE FROM lists WHERE id=?", (lid,)) broadcast({"type": "list:delete", "data": {"id": lid}}) return {"ok": True} # Cards @app.get("/api/lists/{lid}/cards") def cards(lid: str): with db() as c: return [dict(r) for r in c.execute("SELECT * FROM cards WHERE list_id=? ORDER BY position", (lid,)).fetchall()] @app.post("/api/lists/{lid}/cards") def create_card(lid: str, data: CardIn): cid = str(uuid.uuid4()) with db() as c: bid = c.execute("SELECT board_id FROM lists WHERE id=?", (lid,)).fetchone()["board_id"] p = c.execute("SELECT COALESCE(MAX(position),-1) m FROM cards WHERE list_id=?", (lid,)).fetchone()["m"] c.execute("INSERT INTO cards VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,0)", (cid, lid, bid, data.title, data.description, p+1, data.image_url, data.price, data.url, data.status, data.category, data.rating, data.comments)) broadcast({"type": "card:create", "data": dict(c.execute("SELECT * FROM cards WHERE id=?", (cid,)).fetchone())}) return dict(c.execute("SELECT * FROM cards WHERE id=?", (cid,)).fetchone()) @app.patch("/api/cards/{cid}") def update_card(cid: str, data: CardUpdate): with db() as c: u, ps = [], [] for k in ("title","description","image_url","price","url","status","category","rating","comments","archived","position","list_id"): if getattr(data, k, None) is not None: u.append(f"{k}=?"); ps.append(getattr(data, k)) if u: ps.append(cid) c.execute(f"UPDATE cards SET {', '.join(u)} WHERE id=?", ps) row = dict(c.execute("SELECT * FROM cards WHERE id=?", (cid,)).fetchone()) broadcast({"type": "card:update", "data": row}) return {"ok": True} @app.delete("/api/cards/{cid}") def delete_card(cid: str): with db() as c: c.execute("DELETE FROM cards WHERE id=?", (cid,)) broadcast({"type": "card:delete", "data": {"id": cid}}) return {"ok": True} # Stats @app.get("/api/stats/{bid}") def stats(bid: str): with db() as c: total = c.execute("SELECT COALESCE(SUM(price),0) s FROM cards WHERE board_id=? AND archived=0", (bid,)).fetchone()["s"] bought = c.execute("SELECT COUNT(*) c FROM cards WHERE board_id=? AND status='bought'", (bid,)).fetchone()["c"] want = c.execute("SELECT COUNT(*) c FROM cards WHERE board_id=? AND status='want'", (bid,)).fetchone()["c"] deferred = c.execute("SELECT COUNT(*) c FROM cards WHERE board_id=? AND status='deferred'", (bid,)).fetchone()["c"] in_progress = c.execute("SELECT COUNT(*) c FROM cards WHERE board_id=? AND status='in_progress'", (bid,)).fetchone()["c"] return {"total": round(total, 2), "bought": bought, "want": want, "deferred": deferred, "in_progress": in_progress} # Categories @app.get("/api/categories/{bid}") def categories(bid: str): with db() as c: rows = c.execute("SELECT DISTINCT category FROM cards WHERE board_id=? AND category != '' AND archived=0", (bid,)).fetchall() cats = [r["category"] for r in rows] return {"categories": cats} # Serve frontend @app.get("/") def index(): return FileResponse("index.html") # WebSocket broadcast clients = set() def broadcast(msg): payload = json.dumps(msg, ensure_ascii=False) for c in list(clients): try: import asyncio asyncio.get_event_loop().run_until_complete(c.send(payload)) except: clients.discard(c) @app.websocket("/ws") async def ws(websocket: WebSocket): await websocket.accept() clients.add(websocket) try: while True: await websocket.receive_text() except WebSocketDisconnect: clients.discard(websocket) if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=4000)