from __future__ import annotations
import json
import time
import uuid
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
from zoneinfo import ZoneInfo
from aiohttp import web
SHANGHAI_TZ = ZoneInfo("Asia/Shanghai")
SIGNAL_PAGE = """
LoL Signal
A
B
ready
"""
class SignalStore:
def __init__(self, output_root: Path) -> None:
self.output_root = output_root
self.sequence = 0
def write(self, event_slug: str, row: dict[str, Any]) -> None:
safe_slug = "".join(ch for ch in event_slug if ch.isalnum() or ch in {"-", "_"})
if not safe_slug:
safe_slug = "unknown"
path = self.output_root / safe_slug / "signals.jsonl"
path.parent.mkdir(parents=True, exist_ok=True)
with path.open("a", encoding="utf-8") as f:
f.write(json.dumps(row, ensure_ascii=False, sort_keys=True))
f.write("\n")
f.flush()
def _utc_now() -> datetime:
return datetime.now(timezone.utc)
def _json_error(message: str, status: int = 400) -> web.Response:
return web.json_response({"ok": False, "error": message}, status=status)
async def index(_request: web.Request) -> web.Response:
config = _request.app["page_config"]
page = SIGNAL_PAGE.replace(
"__CONFIG__",
json.dumps(config, ensure_ascii=False, separators=(",", ":")),
)
return web.Response(text=page, content_type="text/html")
async def health(request: web.Request) -> web.Response:
return web.json_response(
{
"ok": True,
"server_time": _utc_now().isoformat(),
"output_root": str(request.app["store"].output_root),
"page_config": {
key: value
for key, value in request.app["page_config"].items()
if key != "token"
},
}
)
async def signal(request: web.Request) -> web.Response:
token = request.app["token"]
if token:
provided = request.headers.get("X-Signal-Token", "")
if provided != token:
return _json_error("unauthorized", status=401)
try:
payload = await request.json()
except json.JSONDecodeError:
return _json_error("invalid json")
if not isinstance(payload, dict):
return _json_error("payload must be object")
event_slug = str(payload.get("event_slug") or "").strip()
signal_name = str(payload.get("signal") or "").strip()
if not event_slug:
return _json_error("event_slug is required")
if not signal_name:
return _json_error("signal is required")
store: SignalStore = request.app["store"]
store.sequence += 1
received_at = _utc_now()
source = request.headers.get("X-Forwarded-For") or request.remote or ""
row = {
"sequence": store.sequence,
"event_slug": event_slug,
"market": str(payload.get("market") or ""),
"team_key": str(payload.get("team_key") or ""),
"team": str(payload.get("team") or ""),
"team_token_id": str(payload.get("team_token_id") or ""),
"signal": signal_name,
"client_ts": str(payload.get("client_ts") or ""),
"received_at_wall": received_at.isoformat(),
"received_at_wall_cn": received_at.astimezone(SHANGHAI_TZ).isoformat(),
"received_at_monotonic_ns": time.monotonic_ns(),
"nonce": str(payload.get("nonce") or uuid.uuid4()),
"source": source,
"user_agent": request.headers.get("User-Agent", ""),
"raw_json": payload,
}
store.write(event_slug, row)
return web.json_response(
{
"ok": True,
"sequence": store.sequence,
"received_at_wall": row["received_at_wall"],
"received_at_wall_cn": row["received_at_wall_cn"],
}
)
def make_app(
*,
output_root: Path,
token: str = "",
event_slug: str = "",
team_a: str = "A",
team_b: str = "B",
team_a_token_id: str = "",
team_b_token_id: str = "",
markets: dict[str, Any] | None = None,
default_market: str = "game1",
lock_config: bool = True,
expose_token: bool = True,
) -> web.Application:
app = web.Application(client_max_size=1024 * 1024)
app["store"] = SignalStore(output_root)
app["token"] = token
app["page_config"] = {
"event_slug": event_slug,
"team_a": team_a,
"team_b": team_b,
"team_a_token_id": team_a_token_id,
"team_b_token_id": team_b_token_id,
"markets": markets
or {
"moneyline": {
"slug": event_slug,
"teams": [
{"name": team_a, "token_id": team_a_token_id},
{"name": team_b, "token_id": team_b_token_id},
],
}
},
"default_market": default_market,
"token": token if expose_token else "",
"lock_config": lock_config,
"hide_token": bool(token and expose_token),
}
app.router.add_get("/", index)
app.router.add_get("/health", health)
app.router.add_post("/api/signal", signal)
return app