233 lines
8.7 KiB
Python
233 lines
8.7 KiB
Python
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 (id, list_id, board_id, title, description, position, image_url, price, url, status, category, rating, comments, bought_at) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,NULL)",
|
|
(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 = c.execute("SELECT * FROM cards WHERE id=?", (cid,)).fetchone()
|
|
if row:
|
|
broadcast({"type": "card:update", "data": dict(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)
|