mirror of
https://github.com/shy3130/tick-stock-panel.git
synced 2026-11-02 10:26:13 +00:00
shy3130
f20772d7b7
二开闭环补齐三块: 数据进得来 (写入)、事件推得出 (SSE)、契约有人守 (快照+CI)。
- write:ext scope (第六档): POST /api/ext-data/{id}/ingest 对 Token 开放,
复用管理端同一校验/落盘/视图刷新路径; 结构配置/上传/拉取仍不开放,
外部只能写行数据, 不能改写表结构
- SSE 短期票据 (V2 收尾): POST /api/events/ticket 换 60s 一次性票据 →
GET /api/events?ticket=… 订阅事件流; 票据只继承 Token scope 不放大权限;
进程内事件总线 (services/events.py, 慢消费者丢旧保新), 告警落盘点已挂
(alert_store.append/append_many → bus), 事件按 scope 过滤双向验证
- 契约守护: test_openapi_contract.py 冻结 tier=a 开放面 (61 端点),
增删须显式更新快照; 新增 ci.yml (后端全量测试 + 前端构建) 把契约
变化拦在 PR 上
- examples/open-api: 四个零依赖可运行示例 (行情/写入/回测/事件流) + README
- 菜单设置架构标注: 每页核心/扩展徽标 + 悬停原因 + 图例 (按 open-platform-plan
§2 依赖方向分类, 核心 8 页/扩展 11 页); 顺带补上 MenuSettings 缺失的 模拟盘
- 测试: 网关 write矩阵/票据语义 + 票据生命周期/总线/SSE帧/告警联动 + 契约快照,
全量 2408 passed / 102s; 线上实测: ingest 200 / 票据一次性 / SSE hello+
alert 推送 / scope 过滤双向 / 契约 61 路径
59 lines
2.1 KiB
Python
59 lines
2.1 KiB
Python
"""开放接口示例脚本 — 公共工具 (纯标准库, 无第三方依赖)。
|
|
|
|
使用前:
|
|
1. 面板 → 设置 → 开放接口 → 新建 Token (按需勾选 scope)
|
|
2. 设置环境变量:
|
|
TSP_TOKEN=tsp_xxx (必填)
|
|
TSP_BASE=http://127.0.0.1:3018 (默认本机开发端口, 按实际改)
|
|
3. 逐个运行: python 01_pull_market.py
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import urllib.error
|
|
import urllib.request
|
|
|
|
BASE = os.environ.get("TSP_BASE", "http://127.0.0.1:3018").rstrip("/")
|
|
TOKEN = os.environ.get("TSP_TOKEN", "")
|
|
|
|
|
|
def call(method: str, path: str, body: dict | None = None, headers: dict[str, str] | None = None):
|
|
"""调开放接口; 非 2xx 抛 RuntimeError 并带出服务端 detail。"""
|
|
if not TOKEN:
|
|
raise SystemExit("请先设置环境变量 TSP_TOKEN (设置 → 开放接口 → 新建 Token)")
|
|
req = urllib.request.Request(
|
|
BASE + path,
|
|
data=json.dumps(body).encode() if body is not None else None,
|
|
method=method,
|
|
headers={
|
|
"Authorization": f"Bearer {TOKEN}",
|
|
"Content-Type": "application/json",
|
|
**(headers or {}),
|
|
},
|
|
)
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=30) as resp:
|
|
return json.loads(resp.read().decode())
|
|
except urllib.error.HTTPError as e:
|
|
detail = e.read().decode()
|
|
raise RuntimeError(f"{method} {path} → HTTP {e.code}: {detail}") from e
|
|
|
|
|
|
def sse(path: str):
|
|
"""GET 一个 SSE 流, 逐行 yield (demo 用; 生产建议浏览器 EventSource)。"""
|
|
req = urllib.request.Request(BASE + path, headers={"Accept": "text/event-stream"})
|
|
with urllib.request.urlopen(req, timeout=None) as resp:
|
|
event: dict[str, str] = {}
|
|
for raw in resp:
|
|
line = raw.decode().rstrip("\n")
|
|
if not line:
|
|
if event:
|
|
yield event
|
|
event = {}
|
|
continue
|
|
if line.startswith(":"):
|
|
continue # 心跳注释帧
|
|
key, _, val = line.partition(": ")
|
|
event[key] = val
|