Spaces:
Runtime error
Runtime error
| 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 ---------- | |
| def root(): | |
| return f""" | |
| <html><head><meta charset="utf-8"/><title>{TITLE}</title></head> | |
| <body style="font-family: ui-sans-serif, system-ui; max-width: 920px; margin: 40px auto; line-height:1.5"> | |
| <h1>{TITLE}</h1> | |
| <p>Use the JSON endpoints from your frontend.</p> | |
| <ul> | |
| <li><code>GET /health</code></li> | |
| <li><code>GET /messages</code></li> | |
| <li><code>POST /ingest</code> — <code>{"{ text: string }"}</code></li> | |
| <li><code>POST /orders</code> — order object</li> | |
| <li><code>POST /suggestions</code> — suggestion object</li> | |
| </ul> | |
| <p>If <code>API_KEY</code> is set, include header <code>X-API-Key: <key></code> on POSTs.</p> | |
| </body></html> | |
| """ | |
| def health(): | |
| return {"ok": True, "time": int(time.time())} | |
| 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 | |
| ] | |
| } | |
| def ingest_text(payload: IngestText, _: bool = Depends(require_api_key)): | |
| mid = insert_message("text", payload.text.strip()) | |
| return {"ok": True, "id": mid} | |
| 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} | |
| 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} | |
| def admin_wipe(_: bool = Depends(require_api_key)): | |
| clear_messages() | |
| return {"ok": True} | |