import os
import time
import json
import sqlite3
from typing import Optional, Dict, Any, List
from fastapi import FastAPI, Request, HTTPException, status, Depends
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel, Field
from starlette.responses import HTMLResponse
# =========================
# Config
# =========================
DB_PATH = os.getenv("DB_PATH", "messages.db")
API_KEY = os.getenv("API_KEY", "").strip() # set in Space Secrets for auth
TITLE = "Central Cards — Orders Feed (FastAPI)"
# =========================
# DB helpers (SQLite)
# =========================
def init_db():
con = sqlite3.connect(DB_PATH)
cur = con.cursor()
cur.execute(
"""
CREATE TABLE IF NOT EXISTS messages (
id INTEGER PRIMARY KEY AUTOINCREMENT,
kind TEXT NOT NULL, -- "order" | "suggestion" | "text"
text TEXT NOT NULL, -- human-readable / markdown summary
payload TEXT, -- raw JSON string (optional)
ts INTEGER NOT NULL -- epoch seconds
)
"""
)
con.commit()
con.close()
def db() -> sqlite3.Connection:
return sqlite3.connect(DB_PATH)
def insert_message(kind: str, text: str, payload_json: Optional[str] = None) -> int:
con = db()
cur = con.cursor()
cur.execute(
"INSERT INTO messages(kind, text, payload, ts) VALUES(?, ?, ?, ?)",
(kind, text, payload_json or None, int(time.time())),
)
con.commit()
mid = cur.lastrowid
con.close()
return mid
def list_messages(limit: int = 300) -> List[tuple]:
con = db()
cur = con.cursor()
cur.execute("SELECT id, kind, text, payload, ts FROM messages ORDER BY id DESC LIMIT ?", (limit,))
rows = cur.fetchall()
con.close()
return rows
def clear_messages():
con = db()
cur = con.cursor()
cur.execute("DELETE FROM messages")
con.commit()
con.close()
init_db()
# =========================
# API models
# =========================
class IngestText(BaseModel):
text: str = Field(..., min_length=1)
class StudentInfo(BaseModel):
name: str
room: str
email: str
class DrawnCard(BaseModel):
name: str
role: Optional[str] = ""
rarity: str
power: int
bio: Optional[str] = ""
powers: Optional[list[str]] = []
weaknesses: Optional[list[str]] = []
class OrderSummary(BaseModel):
counts: Dict[str, int]
v100: int
valueSum: int
class OrderPayload(BaseModel):
season: int
info: StudentInfo
picked: list[DrawnCard]
summary: OrderSummary
ts: Optional[int] = None
class SuggestionPayload(BaseModel):
name: str
role: str
why: str
powers: list[str] = []
weaknesses: list[str] = []
ts: Optional[int] = None
# =========================
# FastAPI app
# =========================
app = FastAPI(title=TITLE, version="1.0.0")
# CORS so your static site can call this
app.add_middleware(
CORSMiddleware,
allow_origins=["*"], # lock this down to your site if you want
allow_methods=["GET", "POST", "OPTIONS"],
allow_headers=["*"],
)
def require_api_key(request: Request):
"""If API_KEY is set, require X-API-Key to match."""
if not API_KEY:
return True
if request.headers.get("X-API-Key") != API_KEY:
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid API key")
return True
# ---------- Utility ----------
def fmt_ts(ts: int) -> str:
return time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(ts))
# ---------- Routes ----------
@app.get("/", response_class=HTMLResponse)
def root():
return f"""
{TITLE}
{TITLE}
Use the JSON endpoints from your frontend.
GET /healthGET /messagesPOST /ingest — {"{ text: string }"}POST /orders — order objectPOST /suggestions — suggestion object
If API_KEY is set, include header X-API-Key: <key> on POSTs.
"""
@app.get("/health")
def health():
return {"ok": True, "time": int(time.time())}
@app.get("/messages")
def get_messages(limit: int = 500):
rows = list_messages(limit=limit)
return {
"messages": [
{"id": r[0], "kind": r[1], "text": r[2], "payload": r[3], "ts": r[4], "ts_readable": fmt_ts(r[4])}
for r in rows
]
}
@app.post("/ingest")
def ingest_text(payload: IngestText, _: bool = Depends(require_api_key)):
mid = insert_message("text", payload.text.strip())
return {"ok": True, "id": mid}
@app.post("/orders")
def ingest_order(order: OrderPayload, _: bool = Depends(require_api_key)):
# server-side markdown-style summary for the feed
lines = []
lines.append(f"# 📦 Season {order.season} Order")
lines.append(f"**Name:** {order.info.name}")
lines.append(f"**Class:** {order.info.room}")
lines.append(f"**Email:** {order.info.email}")
lines.append("")
lines.append("## Results (x10)")
for i, c in enumerate(order.picked, 1):
lines.append(f"{i}. **{c.name}** — _{c.rarity}_, Power **{c.power}**")
counts = order.summary.counts
lines.append("")
lines.append(
f"**Rarity Breakdown:** C:{counts.get('Common',0)} • U:{counts.get('Uncommon',0)} • "
f"R:{counts.get('Rare',0)} • UR:{counts.get('Ultra Rare',0)} • L:{counts.get('Legendary',0)}"
)
lines.append(f"**Value Score:** {order.summary.v100}/100 (raw {order.summary.valueSum})")
summary_md = "\n".join(lines)
mid = insert_message("order", summary_md, json.dumps(order.dict(), ensure_ascii=False))
return {"ok": True, "id": mid}
@app.post("/suggestions")
def ingest_suggestion(s: SuggestionPayload, _: bool = Depends(require_api_key)):
lines = []
lines.append("# 💡 New Card Suggestion")
lines.append(f"**Name:** {s.name}")
lines.append(f"**Role:** {s.role}")
lines.append(f"**Why:** {s.why}")
lines.append(f"**Powers:** {', '.join(s.powers) if s.powers else '—'}")
lines.append(f"**Weaknesses:** {', '.join(s.weaknesses) if s.weaknesses else '—'}")
summary_md = "\n".join(lines)
mid = insert_message("suggestion", summary_md, json.dumps(s.dict(), ensure_ascii=False))
return {"ok": True, "id": mid}
@app.post("/admin/wipe")
def admin_wipe(_: bool = Depends(require_api_key)):
clear_messages()
return {"ok": True}