Create app.py

app.py

@@ -0,0 +1,179 @@
+"""
+FSD Simulator 缓存中间层 (Hugging Face Space + Docker)
+=====================================================
+作用：挡在 Roblox 和 Turso 之间，做两件事
+  1) 写勤：接收 Roblox 批量上报的驾驶行为，直通写入 Turso（不缓存）
+  2) 读懒：统计查询结果缓存 12 小时，未命中才回源 Turso
+
+所有密钥从【环境变量】读取（在 HF Space 的 Settings -> Secrets 里填）：
+  TURSO_URL    形如 https://yourdb-yourorg.turso.io/v2/pipeline
+  TURSO_TOKEN  Turso 的 Bearer Token（建议用 INSERT/SELECT 权限的，不要全权）
+  APP_SECRET   保护本服务自己的口令，Roblox 请求时放进 X-App-Key 请求头
+"""
+
+import os
+import time
+import random
+from typing import Any, Optional
+
+import httpx
+from fastapi import FastAPI, Header, HTTPException, Query
+from pydantic import BaseModel
+
+# ---------- 从环境变量读配置（绝不把密钥硬编码进代码！）----------
+TURSO_URL = os.environ.get("TURSO_URL", "")
+TURSO_TOKEN = os.environ.get("TURSO_TOKEN", "")
+APP_SECRET = os.environ.get("APP_SECRET", "")
+
+# 缓存基础有效期：12 小时（单位：秒）
+CACHE_TTL = 12 * 60 * 60
+# 抖动范围 ±30 分钟，避免所有 key 同一刻过期造成"缓存雪崩"
+CACHE_JITTER = 30 * 60
+
+app = FastAPI(title="FSD Cache Layer")
+
+# ---------- 内存缓存（进程一重启就清空，所以只当加速层，不当唯一数据源）----------
+# 结构： { 缓存键: (数据, 过期时间戳) }
+_cache: dict[str, tuple[Any, float]] = {}
+
+
+def cache_get(key: str):
+    item = _cache.get(key)
+    if not item:
+        return None
+    value, expiry = item
+    if time.time() > expiry:          # 过期了
+        _cache.pop(key, None)
+        return None
+    return value
+
+
+def cache_set(key: str, value: Any):
+    jitter = random.randint(-CACHE_JITTER, CACHE_JITTER)
+    _cache[key] = (value, time.time() + CACHE_TTL + jitter)
+
+
+# ---------- Turso HTTP 封装 ----------
+def _to_arg(v: Any) -> dict:
+    """把 Python 值转成 Turso (Hrana) 参数格式。
+    注意：整数用字符串传，避免超大整数精度丢失。"""
+    if v is None:
+        return {"type": "null", "value": None}
+    if isinstance(v, bool):                       # SQLite 没有 bool，存成 0/1
+        return {"type": "integer", "value": str(int(v))}
+    if isinstance(v, int):
+        return {"type": "integer", "value": str(v)}
+    if isinstance(v, float):
+        return {"type": "float", "value": v}
+    return {"type": "text", "value": str(v)}
+
+
+async def turso_batch(statements: list[tuple[str, list]]) -> list:
+    """一次 HTTP 请求里执行多条 SQL（批量，省请求次数）。
+    statements: [(sql, [参数...]), ...]
+    返回：每条语句对应的 response 对象列表。"""
+    if not TURSO_URL or not TURSO_TOKEN:
+        raise HTTPException(500, "服务端未配置 TURSO_URL / TURSO_TOKEN")
+
+    requests = []
+    for sql, args in statements:
+        requests.append({
+            "type": "execute",
+            "stmt": {"sql": sql, "args": [_to_arg(a) for a in args]},
+        })
+    requests.append({"type": "close"})
+
+    async with httpx.AsyncClient(timeout=15) as client:
+        resp = await client.post(
+            TURSO_URL,
+            headers={
+                "Authorization": f"Bearer {TURSO_TOKEN}",
+                "Content-Type": "application/json",
+            },
+            json={"requests": requests},
+        )
+    resp.raise_for_status()
+    data = resp.json()
+
+    results = []
+    for item in data.get("results", []):
+        if item.get("type") == "error":
+            raise HTTPException(502, f"Turso 报错: {item.get('error')}")
+        results.append(item.get("response", {}))
+    return results
+
+
+def parse_rows(execute_response: dict) -> list[dict]:
+    """把 Turso 的 execute 结果转成 [{列名: 值}, ...] 这种好用的格式。"""
+    result = execute_response.get("result", {})
+    cols = [c.get("name") for c in result.get("cols", [])]
+    rows_out = []
+    for row in result.get("rows", []):
+        rows_out.append({col: cell.get("value") for col, cell in zip(cols, row)})
+    return rows_out
+
+
+# ---------- 鉴权：校验 Roblox 传来的口令 ----------
+def check_app_key(x_app_key: Optional[str]):
+    # 没设 APP_SECRET 就跳过校验（方便本地测试）；线上务必设置！
+    if APP_SECRET and x_app_key != APP_SECRET:
+        raise HTTPException(401, "X-App-Key 无效")
+
+
+# ==================== 路由 ====================
+
+@app.get("/")
+def health():
+    """保活端点：给 UptimeRobot 每 5 分钟 ping 一次，保持 Space 不休眠。
+    故意做得极轻——不碰 Turso、不碰缓存。"""
+    return {"status": "ok"}
+
+
+class Event(BaseModel):
+    user_id: int
+    scene: str
+    action: str
+
+
+class LogPayload(BaseModel):
+    events: list[Event]
+
+
+@app.post("/log")
+async def log_events(payload: LogPayload,
+                     x_app_key: Optional[str] = Header(default=None)):
+    """【写勤】Roblox 批量上报驾驶行为，直通写入 Turso（不缓存）。
+    玩家数据必须尽快落库，不能压在缓存里等 12 小时——Space 一重启就丢了。"""
+    check_app_key(x_app_key)
+    if not payload.events:
+        return {"written": 0}
+
+    statements = [
+        ("INSERT INTO driving_events (user_id, scene, action) VALUES (?, ?, ?)",
+         [e.user_id, e.scene, e.action])
+        for e in payload.events
+    ]
+    await turso_batch(statements)
+    return {"written": len(statements)}
+
+
+@app.get("/stats")
+async def scene_stats(scene: str = Query(...),
+                      x_app_key: Optional[str] = Header(default=None)):
+    """【读懒】查某个场景下各操作的人气统计，结果缓存 12 小时。
+    NPC 决策大量调这个，能命中缓存就绝不打扰 Turso。"""
+    check_app_key(x_app_key)
+
+    cache_key = f"stats:{scene}"
+    cached = cache_get(cache_key)
+    if cached is not None:
+        return {"scene": scene, "stats": cached, "cached": True}
+
+    # 未命中 → 回源 Turso 查一次，再写回缓存
+    responses = await turso_batch([
+        ("SELECT action, COUNT(*) AS cnt FROM driving_events "
+         "WHERE scene = ? GROUP BY action ORDER BY cnt DESC", [scene])
+    ])
+    rows = parse_rows(responses[0])
+    cache_set(cache_key, rows)
+    return {"scene": scene, "stats": rows, "cached": False}