Spaces:

SalexAI
/

amazincards

Runtime error

File size: 6,815 Bytes

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"""
    <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> &mdash; <code>{"{ text: string }"}</code></li>
        <li><code>POST /orders</code> &mdash; order object</li>
        <li><code>POST /suggestions</code> &mdash; suggestion object</li>
      </ul>
      <p>If <code>API_KEY</code> is set, include header <code>X-API-Key: &lt;key&gt;</code> on POSTs.</p>
    </body></html>
    """

@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}