Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
be647bbaba |
+1236
-194
File diff suppressed because it is too large
Load Diff
@@ -3,14 +3,15 @@
|
||||
<head>
|
||||
<meta charset="UTF-8" />
|
||||
<meta name="viewport" content="width=device-width, initial-scale=1" />
|
||||
<title>xiaobai-datahub 管理后台</title>
|
||||
<title>xiaobai-datahub 管理后台 · 数据中枢</title>
|
||||
<link rel="stylesheet" href="/admin/styles.css" />
|
||||
</head>
|
||||
<body>
|
||||
<div id="app">
|
||||
<section id="login-view" class="panel auth-panel">
|
||||
<span class="badge-sim">内网 · 8766</span>
|
||||
<h1>数据中枢</h1>
|
||||
<p class="muted">内网管理后台，用于查看源状态、调度和盘后发布批次。</p>
|
||||
<p class="muted">四路来源持续汇流、调度、发布与审计的运转空间。</p>
|
||||
<form id="login-form">
|
||||
<label>账号 <input name="username" value="hub_admin" autocomplete="username" /></label>
|
||||
<label>密码 <input name="password" type="password" autocomplete="current-password" /></label>
|
||||
@@ -30,22 +31,46 @@
|
||||
</section>
|
||||
|
||||
<section id="shell" hidden>
|
||||
<header class="top">
|
||||
<strong>xiaobai-datahub</strong>
|
||||
<header id="topbar">
|
||||
<div class="logo">小白复盘 <em>·</em> 数据中枢</div>
|
||||
<span class="crumb" id="crumb">8766 · 四源汇流 · 持续运转</span>
|
||||
<span id="phase" class="pill"></span>
|
||||
<span id="who" class="muted"></span>
|
||||
<span class="spacer"></span>
|
||||
<span id="who" class="muted who"></span>
|
||||
<button type="button" id="theme-btn" class="ghost">夜间</button>
|
||||
<button type="button" id="logout-btn" class="ghost">退出</button>
|
||||
</header>
|
||||
<nav>
|
||||
<button data-page="overview" class="active">总览</button>
|
||||
<button data-page="sources">数据源</button>
|
||||
<button data-page="jobs">调度任务</button>
|
||||
<button data-page="release">盘后发布</button>
|
||||
<button data-page="datasets">数据集</button>
|
||||
<button data-page="audit">审计</button>
|
||||
</nav>
|
||||
<main id="page"></main>
|
||||
|
||||
<!-- 动效版：一台连续空间舞台，六幕滚动切换，同一核心与流道贯穿始终 -->
|
||||
<div id="stageWrap">
|
||||
<canvas id="scene"></canvas>
|
||||
<div id="cabin"></div>
|
||||
<div id="legend">
|
||||
<span class="lampico li-link"></span>LINK 链路常亮（低亮）<br>
|
||||
<span class="lampico li-act"></span>ACT 活动灯 · 事件成簇短闪
|
||||
</div>
|
||||
<div id="rail"></div>
|
||||
<div id="hint">滚 动 推 进 镜 头</div>
|
||||
<div id="detail" class="closed">
|
||||
<div class="dtag"></div>
|
||||
<button id="detailClose" type="button" aria-label="关闭详情">×</button>
|
||||
<div id="detailBody"></div>
|
||||
</div>
|
||||
</div>
|
||||
<div id="track" aria-hidden="true">
|
||||
<section data-scene="0"></section>
|
||||
<section data-scene="1"></section>
|
||||
<section data-scene="2"></section>
|
||||
<section data-scene="3"></section>
|
||||
<section data-scene="4"></section>
|
||||
<section data-scene="5"></section>
|
||||
</div>
|
||||
|
||||
<!-- 减少动态效果版：六幕静态空间图 + 固定文字状态，信息与功能完全等价 -->
|
||||
<main id="staticShell" hidden>
|
||||
<nav id="staticNav"></nav>
|
||||
<div id="staticScenes"></div>
|
||||
</main>
|
||||
</section>
|
||||
</div>
|
||||
<script src="/admin/app.js"></script>
|
||||
|
||||
@@ -1,51 +1,286 @@
|
||||
/* xiaobai-datahub 数据中枢后台 —— 第七版「轨道机芯」
|
||||
统一动效 Token：进入/退出用强 ease-out，屏内移动用强 ease-in-out，持续流动用 linear。
|
||||
禁止 ease-in、禁止 transition:all —— 全文件遵守。 */
|
||||
:root {
|
||||
color-scheme: light;
|
||||
--bg: #f4f5f7;
|
||||
--surface: #ffffff;
|
||||
--text: #1f2329;
|
||||
--muted: #646a73;
|
||||
--line: #dee0e3;
|
||||
--action: #3370ff;
|
||||
--danger: #e04536;
|
||||
--ok: #16a34a;
|
||||
--warn: #b45309;
|
||||
--radius: 8px;
|
||||
--bar-h: 52px;
|
||||
--radius: 14px;
|
||||
--pad: 16px;
|
||||
font-family: "Segoe UI", "PingFang SC", "Noto Sans SC", sans-serif;
|
||||
--font-cn: "Segoe UI", "PingFang SC", "Noto Sans SC", "Microsoft YaHei", sans-serif;
|
||||
--font-mono: "JetBrains Mono", "DejaVu Sans Mono", Consolas, monospace;
|
||||
--ease-out: cubic-bezier(.22, 1, .36, 1);
|
||||
--ease-in-out: cubic-bezier(.65, 0, .35, 1);
|
||||
--ease-linear: linear;
|
||||
|
||||
/* day (default) */
|
||||
--bg0: #EDF0F7;
|
||||
--bg1: #F8FAFF;
|
||||
--surface: #ffffff;
|
||||
--stage-edge: rgba(47, 99, 214, .16);
|
||||
--text: #1B2547;
|
||||
--muted: #5B6785;
|
||||
--faint: #93A0BD;
|
||||
--line: rgba(47, 99, 214, .18);
|
||||
--action: #2F63D6;
|
||||
--cyan: #0A9C93;
|
||||
--amber: #B97A0E;
|
||||
--danger: #D64F4F;
|
||||
--ok: #1E9E6A;
|
||||
--warn: #B97A0E;
|
||||
--glass: rgba(255, 255, 255, .82);
|
||||
--chip: rgba(47, 99, 214, .08);
|
||||
--face: rgba(120, 160, 235,);
|
||||
--edge: rgba(47, 99, 214,);
|
||||
--slab: rgba(10, 156, 147,);
|
||||
--ring: rgba(47, 99, 214,);
|
||||
--chan1: rgba(47, 99, 214,);
|
||||
--chan2: rgba(10, 124, 146,);
|
||||
--chan3: rgba(109, 93, 214,);
|
||||
--chan4: rgba(190, 120, 10,);
|
||||
}
|
||||
:root[data-theme="night"] {
|
||||
color-scheme: dark;
|
||||
--bg: #111318;
|
||||
--surface: #1b1e24;
|
||||
--text: #e8eaed;
|
||||
--muted: #9aa0a6;
|
||||
--line: #2a2f38;
|
||||
--action: #5b8cff;
|
||||
--bg0: #05080F;
|
||||
--bg1: #0A1020;
|
||||
--surface: #12172A;
|
||||
--stage-edge: rgba(90, 140, 255, .14);
|
||||
--text: #E8EEFC;
|
||||
--muted: #8B97B8;
|
||||
--faint: #59637F;
|
||||
--line: rgba(120, 160, 255, .16);
|
||||
--action: #4A86FF;
|
||||
--cyan: #2FD8CE;
|
||||
--amber: #FFB84D;
|
||||
--danger: #FF6B6B;
|
||||
--ok: #3ED598;
|
||||
--warn: #FFB84D;
|
||||
--glass: rgba(13, 20, 38, .72);
|
||||
--chip: rgba(74, 134, 255, .10);
|
||||
--face: rgba(58, 96, 180,);
|
||||
--edge: rgba(140, 190, 255,);
|
||||
--slab: rgba(47, 216, 206,);
|
||||
--ring: rgba(120, 170, 255,);
|
||||
--chan1: rgba(150, 205, 255,);
|
||||
--chan2: rgba(120, 180, 255,);
|
||||
--chan3: rgba(185, 175, 255,);
|
||||
--chan4: rgba(255, 214, 160,);
|
||||
}
|
||||
|
||||
* { box-sizing: border-box; }
|
||||
body { margin: 0; background: var(--bg); color: var(--text); }
|
||||
.panel, header.top, nav, main { background: var(--surface); }
|
||||
.auth-panel { max-width: 420px; margin: 12vh auto; padding: 28px; border-radius: var(--radius); border: 1px solid var(--line); }
|
||||
label { display: block; margin: 12px 0; }
|
||||
input, select { width: 100%; padding: 8px 10px; border: 1px solid var(--line); border-radius: 4px; background: var(--bg); color: var(--text); }
|
||||
button { background: var(--action); color: #fff; border: 0; border-radius: 4px; padding: 8px 14px; cursor: pointer; }
|
||||
button.ghost { background: transparent; color: var(--text); border: 1px solid var(--line); }
|
||||
button.danger { background: var(--danger); }
|
||||
html, body { margin: 0; background: var(--bg0); color: var(--text); font-family: var(--font-cn); }
|
||||
body { overflow-x: hidden; }
|
||||
.mono { font-family: var(--font-mono); font-variant-numeric: tabular-nums; }
|
||||
.muted { color: var(--muted); }
|
||||
.error { color: var(--danger); }
|
||||
.top { display: flex; gap: 12px; align-items: center; padding: 10px var(--pad); border-bottom: 1px solid var(--line); }
|
||||
nav { display: flex; gap: 4px; padding: 8px var(--pad); border-bottom: 1px solid var(--line); }
|
||||
nav button { background: transparent; color: var(--muted); }
|
||||
nav button.active { color: var(--action); background: transparent; font-weight: 600; }
|
||||
main { padding: var(--pad); min-height: calc(100vh - 96px); }
|
||||
.cards { display: grid; grid-template-columns: repeat(auto-fit, minmax(180px, 1fr)); gap: 12px; margin-bottom: 16px; }
|
||||
.card { border: 1px solid var(--line); border-radius: var(--radius); padding: 12px; }
|
||||
table { width: 100%; border-collapse: collapse; font-size: 13px; }
|
||||
th, td { text-align: left; padding: 8px; border-bottom: 1px solid var(--line); vertical-align: top; }
|
||||
.pill { font-size: 12px; padding: 2px 8px; border-radius: 999px; border: 1px solid var(--line); }
|
||||
.ok { color: var(--ok); }
|
||||
.warn { color: var(--warn); }
|
||||
.fail { color: var(--danger); }
|
||||
.toolbar { display: flex; gap: 8px; flex-wrap: wrap; margin: 12px 0; align-items: end; }
|
||||
.toolbar label { margin: 0; }
|
||||
|
||||
/* ---------- 登录 / 改密（未进入舞台的前置流程） ---------- */
|
||||
.auth-panel {
|
||||
max-width: 420px; margin: 12vh auto; padding: 30px 32px; border-radius: var(--radius);
|
||||
background: var(--glass); border: 1px solid var(--line); backdrop-filter: blur(14px);
|
||||
box-shadow: 0 30px 70px -32px rgba(20, 30, 60, .35);
|
||||
}
|
||||
.auth-panel h1 { margin: 0 0 6px; font-size: 21px; }
|
||||
.auth-panel label { display: block; margin: 12px 0; font-size: 13px; color: var(--muted); }
|
||||
.auth-panel input {
|
||||
width: 100%; margin-top: 6px; padding: 9px 11px; border: 1px solid var(--line); border-radius: 8px;
|
||||
background: var(--chip); color: var(--text); font-family: var(--font-cn);
|
||||
}
|
||||
.auth-panel input:focus { outline: 2px solid var(--action); outline-offset: 1px; }
|
||||
.auth-panel .badge-sim { display: inline-block; margin-bottom: 10px; }
|
||||
|
||||
button { font-family: var(--font-cn); cursor: pointer; }
|
||||
.btn, button.primary, .auth-panel button[type="submit"] {
|
||||
background: linear-gradient(135deg, var(--action), var(--cyan)); color: #fff; border: 0; border-radius: 8px;
|
||||
padding: 8px 16px; font-size: 13px; transition: transform .12s var(--ease-out), opacity .12s var(--ease-out);
|
||||
}
|
||||
.btn:active, button:active { transform: translateY(1px); }
|
||||
.btn.ghost, button.ghost {
|
||||
background: var(--chip); color: var(--text); border: 1px solid var(--line);
|
||||
}
|
||||
.btn.warn { background: transparent; color: var(--warn); border: 1px solid var(--warn); }
|
||||
.btn.danger, button.danger { background: var(--danger); color: #fff; border: 0; }
|
||||
.btn.pri { background: linear-gradient(135deg, var(--action), var(--cyan)); color: #fff; border: 0; }
|
||||
.btn { padding: 6px 13px; border-radius: 8px; font-size: 12px; }
|
||||
|
||||
.badge-sim {
|
||||
font-size: 11px; padding: 3px 10px; border-radius: 99px; letter-spacing: .1em;
|
||||
color: var(--amber); border: 1px solid var(--amber); opacity: .85;
|
||||
}
|
||||
|
||||
/* ---------- 通用表格 / 卡片（cabin、detail、reduced 三处共用同一来源） ---------- */
|
||||
table { width: 100%; border-collapse: collapse; font-size: 12.5px; }
|
||||
th, td { text-align: left; padding: 7px 6px; border-bottom: 1px dashed var(--line); vertical-align: top; }
|
||||
th { color: var(--faint); font-weight: 500; font-size: 11px; letter-spacing: .04em; }
|
||||
.pill { font-size: 12px; padding: 2px 10px; border-radius: 999px; border: 1px solid var(--line); color: var(--muted); }
|
||||
.cards { display: grid; grid-template-columns: repeat(auto-fit, minmax(170px, 1fr)); gap: 10px; margin-bottom: 14px; }
|
||||
.card { border: 1px solid var(--line); border-radius: 10px; padding: 10px 12px; background: var(--chip); }
|
||||
.card strong { font-size: 18px; }
|
||||
.toolbar { display: flex; gap: 8px; flex-wrap: wrap; margin: 10px 0; align-items: center; }
|
||||
.toolbar label { font-size: 12px; color: var(--muted); display: flex; align-items: center; gap: 6px; }
|
||||
.toolbar input { padding: 6px 9px; border: 1px solid var(--line); border-radius: 6px; background: var(--chip); color: var(--text); }
|
||||
h2 { font-size: 16px; margin: 4px 0 10px; }
|
||||
h3 { font-size: 13px; margin: 14px 0 6px; color: var(--muted); }
|
||||
|
||||
/* ---------- 顶部总控条 ---------- */
|
||||
#topbar {
|
||||
height: var(--bar-h); display: flex; align-items: center; gap: 14px; padding: 0 22px;
|
||||
border-bottom: 1px solid var(--line); background: linear-gradient(180deg, var(--bg1), transparent);
|
||||
backdrop-filter: blur(10px);
|
||||
position: sticky; top: 0; z-index: 4;
|
||||
}
|
||||
#topbar .logo { font-weight: 700; font-size: 15px; letter-spacing: .04em; }
|
||||
#topbar .logo em { font-style: normal; color: var(--cyan); }
|
||||
#topbar .crumb { font-size: 12px; color: var(--faint); }
|
||||
#topbar .spacer { flex: 1; }
|
||||
#topbar .who { font-size: 12px; }
|
||||
|
||||
/* ---------- 固定空间舞台（76vh，六幕连续场景） ----------
|
||||
#track 的六段高幕（每段 130vh）在它身后正常参与文档流把页面撑高，
|
||||
#stageWrap 用 sticky 钉在视口里，滚动时只有 #track 的高度被“划过”，
|
||||
舞台本体在这段距离内始终可见、位置不变，直到 #track 撑出的空间耗尽才随之离场。 */
|
||||
#stageWrap {
|
||||
flex: none; height: 76vh; min-height: 480px; margin: 0 18px; border-radius: 18px;
|
||||
overflow: hidden; border: 1px solid var(--stage-edge);
|
||||
box-shadow: 0 30px 80px -30px rgba(0, 0, 0, .45), inset 0 0 120px rgba(0, 0, 0, .08);
|
||||
position: sticky; top: var(--bar-h); z-index: 2;
|
||||
}
|
||||
#scene { position: absolute; inset: 0; width: 100%; height: 100%; display: block; cursor: crosshair; }
|
||||
|
||||
/* ---------- 控制舱：当前场景的关键数据与操作 ---------- */
|
||||
#cabin {
|
||||
position: absolute; left: 22px; top: 22px; width: 300px; z-index: 5; max-height: calc(100% - 44px);
|
||||
overflow: auto; background: var(--glass); border: 1px solid var(--line); border-radius: 14px;
|
||||
backdrop-filter: blur(14px); padding: 16px 18px;
|
||||
transition: opacity .45s var(--ease-out), transform .45s var(--ease-out);
|
||||
}
|
||||
#cabin.hide { opacity: 0; transform: translateX(-14px); pointer-events: none; }
|
||||
#cabin .eyebrow { font-size: 10px; letter-spacing: .22em; color: var(--cyan); margin-bottom: 6px; }
|
||||
#cabin h2 { font-size: 18px; margin: 0 0 4px; }
|
||||
#cabin .sub { font-size: 11.5px; color: var(--muted); line-height: 1.6; margin-bottom: 10px; }
|
||||
#cabin .rows { display: flex; flex-direction: column; gap: 7px; margin-bottom: 4px; }
|
||||
#cabin .row { display: flex; align-items: center; gap: 8px; font-size: 12px; color: var(--muted); }
|
||||
#cabin .row b { color: var(--text); font-weight: 600; }
|
||||
#cabin .row .fill { flex: 1; }
|
||||
#cabin .dotlamp { width: 7px; height: 7px; border-radius: 99px; background: var(--ok); box-shadow: 0 0 8px var(--ok); flex: none; }
|
||||
#cabin .dotlamp.warn { background: var(--warn); box-shadow: 0 0 8px var(--warn); }
|
||||
#cabin .dotlamp.fail { background: var(--danger); box-shadow: 0 0 8px var(--danger); }
|
||||
#cabin .actions { display: flex; gap: 8px; margin-top: 12px; flex-wrap: wrap; }
|
||||
|
||||
/* ---------- 底部章节轨（导航等同滚动跳转） ---------- */
|
||||
#rail {
|
||||
position: absolute; left: 50%; bottom: 16px; transform: translateX(-50%); z-index: 5;
|
||||
display: flex; gap: 6px; align-items: center; padding: 8px 12px; border-radius: 99px;
|
||||
background: var(--glass); border: 1px solid var(--line); backdrop-filter: blur(14px);
|
||||
}
|
||||
#rail .stop {
|
||||
display: flex; align-items: center; gap: 7px; padding: 5px 11px; border-radius: 99px; cursor: pointer;
|
||||
font-size: 11.5px; color: var(--faint); background: transparent; border: 0;
|
||||
transition: color .25s var(--ease-in-out), background .25s var(--ease-in-out);
|
||||
font-family: var(--font-cn); white-space: nowrap;
|
||||
}
|
||||
#rail .stop i { width: 6px; height: 6px; border-radius: 99px; background: var(--faint); transition: background .25s var(--ease-in-out), box-shadow .25s var(--ease-in-out); display: inline-block; }
|
||||
#rail .stop.on { color: var(--text); background: var(--chip); }
|
||||
#rail .stop.on i { background: var(--cyan); box-shadow: 0 0 10px var(--cyan); }
|
||||
#rail .sep { width: 12px; height: 1px; background: var(--line); }
|
||||
|
||||
/* ---------- 图例 ---------- */
|
||||
#legend {
|
||||
position: absolute; right: 22px; top: 22px; z-index: 5; font-size: 10.5px; color: var(--muted);
|
||||
background: var(--glass); border: 1px solid var(--line); border-radius: 12px; padding: 10px 13px;
|
||||
backdrop-filter: blur(14px); line-height: 1.9;
|
||||
}
|
||||
#legend .lampico { display: inline-block; width: 14px; height: 5px; border-radius: 99px; vertical-align: middle; margin-right: 6px; }
|
||||
#legend .li-link { background: rgba(47, 216, 206, .5); box-shadow: 0 0 6px rgba(47, 216, 206, .6); }
|
||||
#legend .li-act { background: var(--amber); box-shadow: 0 0 8px var(--amber); }
|
||||
|
||||
/* ---------- 详情面板：从端口/来源空间位置展开的功能抽屉 ---------- */
|
||||
#detail {
|
||||
position: absolute; z-index: 8; width: 360px; max-width: calc(100% - 40px); max-height: calc(100% - 40px);
|
||||
overflow: auto; pointer-events: auto; background: var(--glass); border: 1px solid var(--line);
|
||||
border-radius: 14px; backdrop-filter: blur(16px); padding: 16px 18px;
|
||||
transition: transform .4s var(--ease-out), opacity .3s var(--ease-out);
|
||||
box-shadow: 0 24px 60px -20px rgba(0, 0, 0, .45);
|
||||
}
|
||||
#detail.closed { transform: scale(.14); opacity: 0; pointer-events: none; }
|
||||
#detail h3 { font-size: 14px; margin: 0 0 2px; color: var(--text); }
|
||||
#detail .dsub { font-size: 10.5px; color: var(--faint); margin-bottom: 10px; letter-spacing: .04em; }
|
||||
#detailClose {
|
||||
position: absolute; right: 10px; top: 10px; width: 24px; height: 24px; border-radius: 8px;
|
||||
border: 1px solid var(--line); background: transparent; color: var(--muted); cursor: pointer; font-size: 13px; line-height: 1;
|
||||
}
|
||||
#detail .dtag {
|
||||
position: absolute; left: -7px; top: 26px; width: 14px; height: 14px; transform: rotate(45deg);
|
||||
background: var(--glass); border-left: 1px solid var(--line); border-bottom: 1px solid var(--line);
|
||||
}
|
||||
|
||||
/* ---------- 滚动提示 ---------- */
|
||||
#hint {
|
||||
position: absolute; left: 50%; bottom: 100px; transform: translateX(-50%); z-index: 5;
|
||||
font-size: 11px; letter-spacing: .3em; color: var(--faint); transition: opacity .5s var(--ease-out);
|
||||
}
|
||||
#hint.off { opacity: 0; }
|
||||
|
||||
/* 滚动章节占位：仅提供滚动行程，无可见内容 */
|
||||
#track { position: relative; z-index: 1; pointer-events: none; }
|
||||
#track section { height: 130vh; }
|
||||
|
||||
/* ---------- 危险操作 / 确认弹层，复用 dialog ---------- */
|
||||
dialog { border: 1px solid var(--line); border-radius: var(--radius); background: var(--surface); color: var(--text); padding: 20px; }
|
||||
|
||||
/* ================================================================
|
||||
减少动态效果（prefers-reduced-motion）：舞台切换为六幕静态空间图 + 固定文字状态
|
||||
信息与功能与动效版完全一致，只是去掉滚动镜头、旋转与闪烁。
|
||||
================================================================ */
|
||||
#staticShell { display: none; }
|
||||
html.reduced #stageWrap, html.reduced #track, html.reduced #hint { display: none !important; }
|
||||
html.reduced #staticShell { display: block; }
|
||||
|
||||
#staticNav {
|
||||
display: flex; gap: 4px; padding: 8px var(--pad); border-bottom: 1px solid var(--line);
|
||||
background: var(--surface); flex-wrap: wrap; position: sticky; top: 0; z-index: 3;
|
||||
}
|
||||
#staticNav button {
|
||||
background: transparent; color: var(--muted); border: 0; padding: 7px 12px; border-radius: 8px; font-size: 13px;
|
||||
}
|
||||
#staticNav button.active { color: var(--action); font-weight: 600; background: var(--chip); }
|
||||
|
||||
.scene-static { padding: var(--pad); border-bottom: 1px solid var(--line); }
|
||||
.scene-static .scene-head { display: flex; align-items: baseline; gap: 10px; margin-bottom: 10px; }
|
||||
.scene-static .scene-head .eyebrow { font-size: 11px; letter-spacing: .18em; color: var(--cyan); }
|
||||
.scene-static .diagram {
|
||||
width: 100%; max-width: 620px; height: 190px; border-radius: 12px; border: 1px solid var(--line);
|
||||
background: var(--chip); display: block; margin-bottom: 12px;
|
||||
}
|
||||
.scene-static .lamprow { display: flex; flex-wrap: wrap; gap: 10px; margin-bottom: 10px; }
|
||||
.lampchip {
|
||||
display: inline-flex; align-items: center; gap: 6px; font-size: 11.5px; color: var(--muted);
|
||||
border: 1px solid var(--line); border-radius: 999px; padding: 4px 10px; background: var(--chip);
|
||||
}
|
||||
.lampchip .lk, .lampchip .ac { width: 7px; height: 7px; border-radius: 99px; background: var(--faint); display: inline-block; }
|
||||
.lampchip .lk.on { background: var(--cyan); box-shadow: 0 0 5px var(--cyan); }
|
||||
.lampchip .ac.on { background: var(--amber); box-shadow: 0 0 6px var(--amber); }
|
||||
.lampchip .ac.recent { outline: 1px solid var(--amber); transition: outline-color .8s var(--ease-out); }
|
||||
|
||||
/* ---------- 主内容区（main/reduced 内容容器统一样式，供动效版 detail 与静态版共用） ---------- */
|
||||
main#page { padding: var(--pad); min-height: calc(100vh - 96px); background: var(--surface); }
|
||||
|
||||
/* ---------- 响应式：1440 / 1280 / 1024 常见桌面宽 ---------- */
|
||||
@media (max-width: 1280px) {
|
||||
#cabin { width: 260px; }
|
||||
#detail { width: 320px; }
|
||||
}
|
||||
@media (max-width: 1100px) {
|
||||
#stageWrap { margin: 0 10px; }
|
||||
#cabin { width: 230px; padding: 13px 14px; }
|
||||
#legend { display: none; }
|
||||
}
|
||||
|
||||
/* ---------- 系统级兜底：即便 JS 未及时接管，也不留半成品动效 ---------- */
|
||||
@media (prefers-reduced-motion: reduce) {
|
||||
#stageWrap, #track, #hint { display: none !important; }
|
||||
#staticShell { display: block !important; }
|
||||
* { animation-duration: .001ms !important; animation-iteration-count: 1 !important; transition-duration: .001ms !important; }
|
||||
}
|
||||
|
||||
@@ -6,12 +6,9 @@ from typing import Any
|
||||
from datahub.adapters import RESERVED
|
||||
from datahub.auth import AuthService
|
||||
from datahub.db import HubDB
|
||||
from datahub import lineage as lineage_module
|
||||
from datahub import observability
|
||||
from datahub.pipeline import OFFICIAL_DATASETS, STOCKS_DATASET, Pipeline
|
||||
from datahub.scheduler import Scheduler
|
||||
from datahub.serving import ApiError
|
||||
from datahub import source_catalog
|
||||
from datahub.timeutil import isoformat, now_shanghai, session_phase, yyyymmdd
|
||||
|
||||
|
||||
@@ -108,53 +105,6 @@ class AdminAPI:
|
||||
raise ApiError("INVALID_ARGUMENT", f"unknown provider: {provider}")
|
||||
return adapter.probe()
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# HEL-543: read-only side-channel status/catalog/lineage. These never
|
||||
# change routing, credentials, or adapters; they only read the
|
||||
# provider_call_log/provider_health tables (observability.py) plus the
|
||||
# static registries in source_catalog.py / lineage.py.
|
||||
# ------------------------------------------------------------------
|
||||
def providers_status(self, provider: str = "", limit: int = 50) -> dict[str, Any]:
|
||||
if not observability.is_enabled(self.db):
|
||||
return {"enabled": False, "health": [], "recent_calls": []}
|
||||
limit = max(1, min(int(limit or 50), 200))
|
||||
health_sql = "SELECT * FROM provider_health"
|
||||
params: tuple[Any, ...] = ()
|
||||
if provider:
|
||||
health_sql += " WHERE provider = ?"
|
||||
params = (provider,)
|
||||
health_sql += " ORDER BY provider, interface"
|
||||
health = self.db.fetchall(health_sql, params)
|
||||
calls_sql = "SELECT * FROM provider_call_log"
|
||||
if provider:
|
||||
calls_sql += " WHERE provider = ?"
|
||||
calls_sql += " ORDER BY id DESC LIMIT ?"
|
||||
recent = self.db.fetchall(calls_sql, (*params, limit))
|
||||
return {"enabled": True, "health": health, "recent_calls": recent}
|
||||
|
||||
def source_catalog(self) -> dict[str, Any]:
|
||||
if not observability.is_enabled(self.db):
|
||||
return {"enabled": False, "items": []}
|
||||
return {"enabled": True, "items": source_catalog.snapshot(self.db, self.auth)}
|
||||
|
||||
def lineage(self, trade_date: str = "") -> dict[str, Any]:
|
||||
day = yyyymmdd(trade_date) if trade_date else yyyymmdd(now_shanghai())
|
||||
if not observability.is_enabled(self.db):
|
||||
return {"enabled": False, "trade_date": day, "items": []}
|
||||
return {"enabled": True, "trade_date": day, "items": lineage_module.snapshot(self.db, day)}
|
||||
|
||||
def lineage_affected(self, provider: str = "", interface: str = "") -> dict[str, Any]:
|
||||
if not provider:
|
||||
raise ApiError("INVALID_ARGUMENT", "provider is required")
|
||||
if not observability.is_enabled(self.db):
|
||||
return {"enabled": False, "provider": provider, "interface": interface, "items": []}
|
||||
return {
|
||||
"enabled": True,
|
||||
"provider": provider,
|
||||
"interface": interface,
|
||||
"items": lineage_module.affected(self.db, provider, interface),
|
||||
}
|
||||
|
||||
def jobs(self) -> dict[str, Any]:
|
||||
runs = self.db.fetchall("SELECT * FROM job_runs ORDER BY id DESC LIMIT 100")
|
||||
stocks_times = "/".join(self.pipeline.settings.stocks_refresh_times) or "20:00"
|
||||
|
||||
@@ -8,7 +8,6 @@ from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
from datahub.datasets_ext import EXTENDED_DATASET_TABLES, EXTENDED_SCHEMA
|
||||
from datahub.observability import OBS_SCHEMA
|
||||
from datahub.timeutil import isoformat
|
||||
|
||||
_BASE_SCHEMA = """
|
||||
@@ -297,7 +296,7 @@ CREATE INDEX IF NOT EXISTS idx_eod_bars_date ON eod_bars(trade_date, batch_id);
|
||||
CREATE INDEX IF NOT EXISTS idx_calendar_open ON trade_calendar(is_open, cal_date);
|
||||
"""
|
||||
|
||||
SCHEMA = _BASE_SCHEMA + EXTENDED_SCHEMA + OBS_SCHEMA
|
||||
SCHEMA = _BASE_SCHEMA + EXTENDED_SCHEMA
|
||||
|
||||
DATASET_TABLES = {
|
||||
"daily": ("eod_bars", "staging_bars"),
|
||||
|
||||
@@ -122,26 +122,6 @@ class HubRequestHandler(BaseHTTPRequestHandler):
|
||||
if path == "/admin/api/sources" and method == "GET":
|
||||
self._json(self.hub.admin.sources(), HTTPStatus.OK)
|
||||
return
|
||||
if path == "/admin/api/providers/status" and method == "GET":
|
||||
query = parse_query(urlparse(self.path).query)
|
||||
provider = (query.get("provider") or [""])[0]
|
||||
limit = (query.get("limit") or ["50"])[0]
|
||||
self._json(self.hub.admin.providers_status(provider, int(limit or 50)), HTTPStatus.OK)
|
||||
return
|
||||
if path == "/admin/api/source-catalog" and method == "GET":
|
||||
self._json(self.hub.admin.source_catalog(), HTTPStatus.OK)
|
||||
return
|
||||
if path == "/admin/api/lineage" and method == "GET":
|
||||
query = parse_query(urlparse(self.path).query)
|
||||
date = (query.get("date") or [""])[0]
|
||||
self._json(self.hub.admin.lineage(date), HTTPStatus.OK)
|
||||
return
|
||||
if path == "/admin/api/lineage/affected" and method == "GET":
|
||||
query = parse_query(urlparse(self.path).query)
|
||||
provider = (query.get("provider") or [""])[0]
|
||||
interface = (query.get("interface") or [""])[0]
|
||||
self._json(self.hub.admin.lineage_affected(provider, interface), HTTPStatus.OK)
|
||||
return
|
||||
if path.startswith("/admin/api/sources/") and path.endswith("/probe") and method == "POST":
|
||||
provider = path.split("/")[4]
|
||||
self._json(self.hub.admin.probe(provider), HTTPStatus.OK)
|
||||
|
||||
@@ -24,12 +24,6 @@ class Hub:
|
||||
raise SystemExit("DATAHUB_ENCRYPTION_KEY 未配置")
|
||||
self.settings = settings
|
||||
self.db = HubDB(settings.db_path)
|
||||
# HEL-543: carry the observability kill switch on the db handle so
|
||||
# every call site that already threads `db` through (pipeline,
|
||||
# realtime_serve, steward, admin_api) picks it up for free with no
|
||||
# extra plumbing. Missing this attribute (e.g. a bare HubDB built
|
||||
# directly in tests) defaults to enabled — see observability.is_enabled.
|
||||
self.db.observability_enabled = settings.observability_enabled
|
||||
self.vault = SecretVault(settings.encryption_key)
|
||||
self.auth = AuthService(self.db, self.vault, settings.api_token, settings.admin_password)
|
||||
token = settings.tushare_token or self.auth.load_credential("tushare_token")
|
||||
|
||||
@@ -1,258 +0,0 @@
|
||||
"""Read-only lineage/impact inventory (HEL-543).
|
||||
|
||||
Answers, without changing any routing decision: for a given main-site data
|
||||
item, which datahub dataset backs it, which provider/interface currently
|
||||
serves it (primary and backup), and — when a provider/interface is
|
||||
unhealthy — which datasets and, best-effort, which main-site consumers are
|
||||
affected.
|
||||
|
||||
Every row cites where it was verified so a reviewer does not have to trust
|
||||
a paraphrase:
|
||||
|
||||
- ``v1_endpoint``/``primary_source``/``backup_source`` are taken verbatim
|
||||
from ``datahub/serving.py`` (the ``source=`` string literal passed to
|
||||
``_published_rows``/``_official_meta``) or from the provider/interface
|
||||
pairs wired into ``datahub/realtime_serve.py`` for HEL-543.
|
||||
- ``known_consumers`` lists only call sites this round actually found via
|
||||
code search in the ``xiaobai-review`` website tree (cited as
|
||||
``file:line`` in the comment above each dataset). Anything not backed by
|
||||
a citation is left out rather than guessed; a fuller page-by-page map is
|
||||
tracked separately (HEL-549) and can extend this table later without
|
||||
touching its shape.
|
||||
|
||||
This module never talks to a provider and never mutates anything; it only
|
||||
reads ``provider_health``/``provider_call_log`` (HEL-543) and the existing
|
||||
``publications``/``batches`` tables to attach live status to each row.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
# Verified against backend/features/screener/data_sync.py (calendar,
|
||||
# stock_basic, daily, daily_basic, index_daily-as-benchmark, stk_auction,
|
||||
# moneyflow, ths_hot, dc_hot all called via `self.client.query(...)`) and
|
||||
# backend/features/heaven/market_context.py (stock_basic, index_daily via
|
||||
# `self._tushare_client().query(...)` / `client.query(...)`).
|
||||
DATASETS: list[dict[str, Any]] = [
|
||||
{
|
||||
"dataset": "calendar",
|
||||
"tier": "official",
|
||||
"v1_endpoint": "/v1/calendar",
|
||||
"primary_source": "tushare:trade_cal",
|
||||
"backup_source": None,
|
||||
"known_consumers": ["智能选股(data_sync.py 交易日历解析)", "问天(交易日推算)"],
|
||||
},
|
||||
{
|
||||
"dataset": "stocks",
|
||||
"tier": "official",
|
||||
"v1_endpoint": "/v1/stocks",
|
||||
"primary_source": "tushare:stock_basic",
|
||||
"backup_source": None,
|
||||
"known_consumers": ["智能选股(股票主档)", "问天(market_context.py 股票代码/名称解析)"],
|
||||
},
|
||||
{
|
||||
"dataset": "daily",
|
||||
"tier": "official",
|
||||
"v1_endpoint": "/v1/bars/daily",
|
||||
"primary_source": "tushare:daily",
|
||||
"backup_source": None,
|
||||
"known_consumers": ["智能选股(data_sync.py 日K因子)", "交易复盘/个股详情日K图表"],
|
||||
},
|
||||
{
|
||||
"dataset": "valuation",
|
||||
"tier": "official",
|
||||
"v1_endpoint": "/v1/valuation",
|
||||
"primary_source": "tushare:daily_basic",
|
||||
"backup_source": None,
|
||||
"known_consumers": ["智能选股(data_sync.py 估值因子)"],
|
||||
},
|
||||
{
|
||||
"dataset": "moneyflow",
|
||||
"tier": "official",
|
||||
"v1_endpoint": "/v1/moneyflow",
|
||||
"primary_source": "tushare:moneyflow",
|
||||
"backup_source": None,
|
||||
"known_consumers": ["智能选股(data_sync.py 资金流因子)", "个股详情资金流"],
|
||||
},
|
||||
{
|
||||
"dataset": "auction",
|
||||
"tier": "official",
|
||||
"v1_endpoint": "/v1/auction",
|
||||
"primary_source": "tushare:stk_auction",
|
||||
"backup_source": None,
|
||||
"known_consumers": ["智能选股(data_sync.py 竞价快照)", "竞价板块"],
|
||||
},
|
||||
{
|
||||
"dataset": "index_daily",
|
||||
"tier": "official",
|
||||
"v1_endpoint": "/v1/indexes/bars",
|
||||
"primary_source": "tushare:index_daily",
|
||||
"backup_source": None,
|
||||
"known_consumers": ["问天(market_context.py 指数近20日走势)", "智能选股(基准回看)"],
|
||||
},
|
||||
{
|
||||
"dataset": "limit_events",
|
||||
"tier": "official",
|
||||
"v1_endpoint": "/v1/limit-events",
|
||||
"primary_source": "tushare:limit_list_d",
|
||||
"backup_source": None,
|
||||
"known_consumers": ["涨停梯队(历史/盘后视图)"],
|
||||
},
|
||||
{
|
||||
"dataset": "popularity",
|
||||
"tier": "official",
|
||||
"v1_endpoint": "/v1/popularity",
|
||||
"primary_source": "tushare:ths_hot+dc_hot",
|
||||
"backup_source": None,
|
||||
"known_consumers": ["人气榜", "智能选股(data_sync.py 人气因子)"],
|
||||
},
|
||||
{
|
||||
"dataset": "dragon_tiger",
|
||||
"tier": "official",
|
||||
"v1_endpoint": "/v1/dragon-tiger",
|
||||
"primary_source": "tushare:hm_detail",
|
||||
"backup_source": None,
|
||||
"known_consumers": ["龙虎榜"],
|
||||
},
|
||||
{
|
||||
"dataset": "sector_daily",
|
||||
"tier": "official",
|
||||
"v1_endpoint": "/v1/sectors",
|
||||
"primary_source": "tushare:ths_daily+dc_index+sw_daily",
|
||||
"backup_source": None,
|
||||
"known_consumers": ["主题轮动", "板块梯队"],
|
||||
},
|
||||
{
|
||||
"dataset": "quotes_latest",
|
||||
"tier": "provisional",
|
||||
"v1_endpoint": "/v1/quotes/latest",
|
||||
"primary_source": "eastmoney:ulist/clist",
|
||||
"backup_source": "tencent:qt",
|
||||
"known_consumers": ["竞价/股票池盘中价格", "情绪周期盘中快照"],
|
||||
},
|
||||
{
|
||||
"dataset": "index_quotes",
|
||||
"tier": "provisional",
|
||||
"v1_endpoint": "/v1/indexes/quotes",
|
||||
"primary_source": "eastmoney:ulist",
|
||||
"backup_source": "tencent:qt",
|
||||
"known_consumers": ["首页大盘指数条"],
|
||||
},
|
||||
{
|
||||
"dataset": "sectors_quote",
|
||||
"tier": "provisional",
|
||||
"v1_endpoint": "/v1/sectors/quote",
|
||||
"primary_source": "eastmoney:sw",
|
||||
"backup_source": None,
|
||||
"known_consumers": ["主题轮动盘中板块报价"],
|
||||
},
|
||||
{
|
||||
"dataset": "limit_pool",
|
||||
"tier": "provisional",
|
||||
"v1_endpoint": "/v1/limit-pool",
|
||||
"primary_source": "eastmoney:zt_pool",
|
||||
"backup_source": None,
|
||||
"known_consumers": ["涨停梯队盘中视图"],
|
||||
},
|
||||
{
|
||||
"dataset": "intraday_points",
|
||||
"tier": "provisional",
|
||||
"v1_endpoint": "/v1/intraday/points",
|
||||
"primary_source": "eastmoney:trends2",
|
||||
"backup_source": None,
|
||||
"known_consumers": ["个股详情分时图"],
|
||||
},
|
||||
{
|
||||
"dataset": "ifind_wencai",
|
||||
"tier": "licensed",
|
||||
"v1_endpoint": "/v1/query (api_name=ifind_wencai)",
|
||||
"primary_source": "ifind:smart_stock_picking",
|
||||
"backup_source": None,
|
||||
"known_consumers": ["问师(自然语言选股，需 iFinD 凭证)"],
|
||||
},
|
||||
]
|
||||
|
||||
_KNOWN_PROVIDERS_BY_SOURCE_PREFIX = ("tushare", "eastmoney", "tencent", "ifind")
|
||||
|
||||
|
||||
def _providers_for(primary_source: str, backup_source: str | None) -> list[str]:
|
||||
providers: list[str] = []
|
||||
for source in (primary_source, backup_source or ""):
|
||||
for provider in _KNOWN_PROVIDERS_BY_SOURCE_PREFIX:
|
||||
if source.startswith(provider) and provider not in providers:
|
||||
providers.append(provider)
|
||||
return providers
|
||||
|
||||
|
||||
def snapshot(db: Any, trade_date: str = "") -> list[dict[str, Any]]:
|
||||
"""Attach live status to the static lineage table. Read-only; never
|
||||
raises (a per-row status lookup failure just leaves that row's status
|
||||
empty rather than failing the whole snapshot)."""
|
||||
result: list[dict[str, Any]] = []
|
||||
for entry in DATASETS:
|
||||
row = dict(entry)
|
||||
providers = _providers_for(entry["primary_source"], entry.get("backup_source"))
|
||||
row["providers"] = providers
|
||||
live: list[dict[str, Any]] = []
|
||||
try:
|
||||
if db is not None and providers:
|
||||
placeholders = ",".join("?" for _ in providers)
|
||||
live = db.fetchall(
|
||||
f"SELECT provider, interface, state, last_error, last_fallback_reason, "
|
||||
f"consec_failures, updated_at FROM provider_health WHERE provider IN ({placeholders})",
|
||||
tuple(providers),
|
||||
)
|
||||
except Exception:
|
||||
live = []
|
||||
row["live_provider_health"] = live
|
||||
if entry["tier"] == "official":
|
||||
pub = None
|
||||
try:
|
||||
if db is not None and trade_date:
|
||||
pub = db.fetchone(
|
||||
"SELECT dataset, trade_date, state, published_at FROM publications "
|
||||
"WHERE dataset = ? AND trade_date = ?",
|
||||
(entry["dataset"], trade_date),
|
||||
)
|
||||
except Exception:
|
||||
pub = None
|
||||
row["publication"] = pub
|
||||
result.append(row)
|
||||
return result
|
||||
|
||||
|
||||
def affected(db: Any, provider: str = "", interface: str = "") -> list[dict[str, Any]]:
|
||||
"""Read-only: which datasets/pages are impacted by a given provider (and,
|
||||
optionally, a specific interface) right now. Does not change routing."""
|
||||
provider = str(provider or "").strip()
|
||||
interface = str(interface or "").strip()
|
||||
result: list[dict[str, Any]] = []
|
||||
for entry in DATASETS:
|
||||
providers = _providers_for(entry["primary_source"], entry.get("backup_source"))
|
||||
if provider and provider not in providers:
|
||||
continue
|
||||
row = dict(entry)
|
||||
row["providers"] = providers
|
||||
health: list[dict[str, Any]] = []
|
||||
try:
|
||||
if db is not None:
|
||||
if interface:
|
||||
health = db.fetchall(
|
||||
"SELECT provider, interface, state, last_error, last_fallback_reason, "
|
||||
"consec_failures, updated_at FROM provider_health "
|
||||
"WHERE provider = ? AND interface = ?",
|
||||
(provider, interface),
|
||||
)
|
||||
elif providers:
|
||||
placeholders = ",".join("?" for _ in providers)
|
||||
health = db.fetchall(
|
||||
f"SELECT provider, interface, state, last_error, last_fallback_reason, "
|
||||
f"consec_failures, updated_at FROM provider_health WHERE provider IN ({placeholders})",
|
||||
tuple(providers),
|
||||
)
|
||||
except Exception:
|
||||
health = []
|
||||
row["live_provider_health"] = health
|
||||
result.append(row)
|
||||
return result
|
||||
@@ -1,319 +0,0 @@
|
||||
"""Side-channel provider-call observability (HEL-543).
|
||||
|
||||
This module is additive-only and must never change what any existing call
|
||||
returns or raises. It exists purely to answer, after the fact and without
|
||||
touching routing: which provider/interface was called, whether it
|
||||
succeeded, how stale/complete the payload looked, and why a fallback fired.
|
||||
|
||||
Hard rules enforced here:
|
||||
|
||||
- Every public entry point (`observe`, `record_call`) is wrapped so that a
|
||||
database failure, a classifier bug, or any other internal error is
|
||||
swallowed and logged at DEBUG level. It never raises into the caller and
|
||||
never delays/blocks the caller's real data path beyond a best-effort
|
||||
timing measurement.
|
||||
- `observe()` always returns exactly what `fn()` returned, and re-raises
|
||||
exactly what `fn()` raised (same exception object, unmodified). It does
|
||||
not retry, does not change ordering, and does not add new failure modes.
|
||||
- No mock data is ever produced or returned by this module.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
import time
|
||||
from typing import Any, Callable, TypeVar
|
||||
|
||||
from datahub.timeutil import isoformat, now_shanghai
|
||||
|
||||
_T = TypeVar("_T")
|
||||
|
||||
# Additive-only schema: two new tables, no changes to any existing table.
|
||||
# Merged into datahub.db.SCHEMA the same way EXTENDED_SCHEMA is.
|
||||
OBS_SCHEMA = """
|
||||
CREATE TABLE IF NOT EXISTS provider_call_log (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
provider TEXT NOT NULL,
|
||||
interface TEXT NOT NULL,
|
||||
fetched_at TEXT NOT NULL,
|
||||
latency_ms INTEGER,
|
||||
status TEXT NOT NULL,
|
||||
error TEXT,
|
||||
fallback_reason TEXT,
|
||||
data_age_seconds INTEGER,
|
||||
created_at TEXT NOT NULL
|
||||
);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS provider_health (
|
||||
provider TEXT NOT NULL,
|
||||
interface TEXT NOT NULL,
|
||||
state TEXT NOT NULL,
|
||||
last_ok_at TEXT,
|
||||
last_error TEXT,
|
||||
last_fallback_reason TEXT,
|
||||
consec_failures INTEGER NOT NULL DEFAULT 0,
|
||||
last_latency_ms INTEGER,
|
||||
last_data_age_seconds INTEGER,
|
||||
updated_at TEXT NOT NULL,
|
||||
PRIMARY KEY (provider, interface)
|
||||
);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS idx_provider_call_log_created ON provider_call_log(created_at);
|
||||
CREATE INDEX IF NOT EXISTS idx_provider_call_log_provider ON provider_call_log(provider, interface, created_at);
|
||||
"""
|
||||
|
||||
DEFAULT_STALE_SECONDS = 300
|
||||
|
||||
_BLOCKED_MARKERS = (
|
||||
"<!doctype", "<html", "expecting value", "verify you are human",
|
||||
"unusual traffic", "captcha", "安全验证", "访问异常", "请完成验证",
|
||||
"拦截", "禁止访问", "forbidden",
|
||||
)
|
||||
|
||||
# Matches provider messages like "Eastmoney returned 0/3 indices" or
|
||||
# "Tencent returned 2/3 indices" (see adapters/eastmoney.py, adapters/tencent.py).
|
||||
_COUNT_MISMATCH_RE = re.compile(r"returned (\d+)\s*/\s*(\d+)")
|
||||
|
||||
|
||||
def _logger():
|
||||
from datahub.logutil import get_logger
|
||||
|
||||
return get_logger()
|
||||
|
||||
|
||||
def is_enabled(db: Any) -> bool:
|
||||
"""Runtime kill switch (``Settings.observability_enabled`` /
|
||||
``DATAHUB_OBSERVABILITY``, wired onto the db handle in ``Hub.__init__``).
|
||||
|
||||
Defaults to enabled when the attribute is absent — e.g. a bare ``HubDB``
|
||||
built directly in a test, or any call site that predates HEL-543 — so
|
||||
this can never accidentally disable an existing deployment. Never
|
||||
raises.
|
||||
"""
|
||||
try:
|
||||
return bool(getattr(db, "observability_enabled", True))
|
||||
except Exception: # pragma: no cover - defensive, must never raise
|
||||
return True
|
||||
|
||||
|
||||
def classify_error(message: str) -> tuple[str, str]:
|
||||
"""Best-effort, side-reading classification of an exception message.
|
||||
|
||||
Never raises. Unknown shapes fall back to a generic ``error`` status so a
|
||||
classifier miss can never be mistaken for a healthy call.
|
||||
"""
|
||||
try:
|
||||
lower = (message or "").lower()
|
||||
if any(marker in lower for marker in _BLOCKED_MARKERS):
|
||||
return "blocked", "response_looks_like_intercept_page"
|
||||
if "timeout" in lower or "timed out" in lower:
|
||||
return "timeout", "request_timeout"
|
||||
mismatch = _COUNT_MISMATCH_RE.search(lower)
|
||||
if mismatch and int(mismatch.group(1)) == 0:
|
||||
return "empty", "empty_or_incomplete_response"
|
||||
if mismatch:
|
||||
return "degraded", "partial_or_mismatched_response"
|
||||
if "empty" in lower or "no intraday chart data" in lower or "missing" in lower:
|
||||
return "empty", "empty_or_incomplete_response"
|
||||
if "too small" in lower or "incomplete" in lower or "mismatch" in lower:
|
||||
return "degraded", "partial_or_mismatched_response"
|
||||
return "error", ""
|
||||
except Exception: # pragma: no cover - defensive, must never raise
|
||||
return "error", ""
|
||||
|
||||
|
||||
def classify_rows(
|
||||
rows: Any,
|
||||
*,
|
||||
required_fields: tuple[str, ...] | None = None,
|
||||
freshness_field: str | None = "quote_time_epoch",
|
||||
max_age_seconds: int = DEFAULT_STALE_SECONDS,
|
||||
) -> tuple[str, str, int | None]:
|
||||
"""Read-only classification of an already-successful payload.
|
||||
|
||||
Only ever called on a value a caller is about to use as-is; this never
|
||||
mutates ``rows`` and a classifier bug always degrades to ``("ok", "",
|
||||
None)`` rather than mislabeling a real success as a failure.
|
||||
"""
|
||||
try:
|
||||
if isinstance(rows, dict):
|
||||
items = [rows] if rows else []
|
||||
elif isinstance(rows, (list, tuple)):
|
||||
items = [item for item in rows if isinstance(item, dict)]
|
||||
else:
|
||||
items = []
|
||||
if not items:
|
||||
return "empty", "no_rows_returned", None
|
||||
if required_fields:
|
||||
missing: set[str] = set()
|
||||
for item in items:
|
||||
for field in required_fields:
|
||||
if item.get(field) in (None, ""):
|
||||
missing.add(field)
|
||||
if missing:
|
||||
return "missing_fields", "missing:" + ",".join(sorted(missing)), None
|
||||
data_age: int | None = None
|
||||
if freshness_field:
|
||||
now_epoch = time.time()
|
||||
ages: list[int] = []
|
||||
for item in items:
|
||||
raw = item.get(freshness_field)
|
||||
try:
|
||||
epoch = int(raw or 0)
|
||||
except (TypeError, ValueError):
|
||||
epoch = 0
|
||||
if epoch > 0:
|
||||
ages.append(max(0, int(now_epoch - epoch)))
|
||||
if ages:
|
||||
data_age = max(ages)
|
||||
if data_age > max_age_seconds:
|
||||
return "stale", "data_age_exceeds_threshold", data_age
|
||||
return "ok", "", data_age
|
||||
except Exception: # pragma: no cover - defensive, must never raise
|
||||
return "ok", "", None
|
||||
|
||||
|
||||
def record_call(
|
||||
db: Any,
|
||||
provider: str,
|
||||
interface: str,
|
||||
*,
|
||||
status: str,
|
||||
latency_ms: int | None = None,
|
||||
error: str = "",
|
||||
fallback_reason: str = "",
|
||||
data_age_seconds: int | None = None,
|
||||
) -> None:
|
||||
"""Fail-open recorder. Never raises; a write failure here must never be
|
||||
able to take down a real, otherwise-successful data path."""
|
||||
if db is None or not is_enabled(db):
|
||||
return
|
||||
try:
|
||||
now = isoformat(now_shanghai())
|
||||
ok = status == "ok"
|
||||
error_text = (error or "")[:500]
|
||||
reason_text = (fallback_reason or "")[:200]
|
||||
with db.write() as connection:
|
||||
connection.execute(
|
||||
"INSERT INTO provider_call_log("
|
||||
"provider, interface, fetched_at, latency_ms, status, error, "
|
||||
"fallback_reason, data_age_seconds, created_at) "
|
||||
"VALUES (?,?,?,?,?,?,?,?,?)",
|
||||
(provider, interface, now, latency_ms, status, error_text, reason_text, data_age_seconds, now),
|
||||
)
|
||||
connection.execute(
|
||||
"""
|
||||
INSERT INTO provider_health(
|
||||
provider, interface, state, last_ok_at, last_error, last_fallback_reason,
|
||||
consec_failures, last_latency_ms, last_data_age_seconds, updated_at)
|
||||
VALUES (?,?,?,?,?,?,?,?,?,?)
|
||||
ON CONFLICT(provider, interface) DO UPDATE SET
|
||||
state = excluded.state,
|
||||
last_ok_at = CASE WHEN excluded.state = 'ok' THEN excluded.last_ok_at ELSE provider_health.last_ok_at END,
|
||||
last_error = CASE WHEN excluded.state = 'ok' THEN '' ELSE excluded.last_error END,
|
||||
last_fallback_reason = excluded.last_fallback_reason,
|
||||
consec_failures = CASE WHEN excluded.state = 'ok' THEN 0 ELSE provider_health.consec_failures + 1 END,
|
||||
last_latency_ms = excluded.last_latency_ms,
|
||||
last_data_age_seconds = excluded.last_data_age_seconds,
|
||||
updated_at = excluded.updated_at
|
||||
""",
|
||||
(
|
||||
provider,
|
||||
interface,
|
||||
status,
|
||||
now if ok else None,
|
||||
"" if ok else (error_text or reason_text or "unknown_error"),
|
||||
reason_text,
|
||||
0 if ok else 1,
|
||||
latency_ms,
|
||||
data_age_seconds,
|
||||
now,
|
||||
),
|
||||
)
|
||||
except Exception: # pragma: no cover - defensive, must never raise
|
||||
try:
|
||||
_logger().debug(
|
||||
"observability record_call failed (fail-open)",
|
||||
extra={"hub": {"provider": provider, "interface": interface}},
|
||||
exc_info=True,
|
||||
)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
||||
def _safe_record_failure(db: Any, provider: str, interface: str, latency_ms: int, exc: BaseException) -> None:
|
||||
if db is None:
|
||||
return
|
||||
try:
|
||||
message = str(exc)
|
||||
status, reason = classify_error(message)
|
||||
record_call(
|
||||
db, provider, interface,
|
||||
status=status, latency_ms=latency_ms, error=message, fallback_reason=reason,
|
||||
)
|
||||
except Exception: # pragma: no cover - defensive, must never raise
|
||||
pass
|
||||
|
||||
|
||||
def _safe_record_success(
|
||||
db: Any,
|
||||
provider: str,
|
||||
interface: str,
|
||||
latency_ms: int,
|
||||
result: Any,
|
||||
classify: Callable[[Any], tuple[str, str, int | None] | None] | None,
|
||||
) -> None:
|
||||
if db is None:
|
||||
return
|
||||
status, reason, data_age = "ok", "", None
|
||||
if classify is not None:
|
||||
try:
|
||||
classified = classify(result)
|
||||
if classified:
|
||||
status, reason, data_age = classified
|
||||
except Exception: # pragma: no cover - defensive, must never raise
|
||||
# A classifier bug must never mislabel (or hide) a real success;
|
||||
# degrade to a plain "ok" call rather than skipping the log.
|
||||
status, reason, data_age = "ok", "", None
|
||||
try:
|
||||
record_call(
|
||||
db, provider, interface,
|
||||
status=status, latency_ms=latency_ms, fallback_reason=reason, data_age_seconds=data_age,
|
||||
)
|
||||
except Exception: # pragma: no cover - defensive, must never raise
|
||||
pass
|
||||
|
||||
|
||||
def observe(
|
||||
db: Any,
|
||||
provider: str,
|
||||
interface: str,
|
||||
fn: Callable[[], _T],
|
||||
*,
|
||||
classify: Callable[[Any], tuple[str, str, int | None] | None] | None = None,
|
||||
) -> _T:
|
||||
"""Call ``fn()`` and record a side-channel status row.
|
||||
|
||||
Returns exactly what ``fn()`` returns and re-raises exactly what
|
||||
``fn()`` raises. ``db`` may be ``None`` (e.g. in call sites that are not
|
||||
wired to a database yet); in that case this is a transparent passthrough
|
||||
with no recording at all. Same when the ``DATAHUB_OBSERVABILITY`` kill
|
||||
switch is off (see ``is_enabled``): this becomes ``return fn()`` with no
|
||||
timing, no classification, and no db access whatsoever.
|
||||
"""
|
||||
if db is not None and not is_enabled(db):
|
||||
return fn()
|
||||
started = time.perf_counter()
|
||||
try:
|
||||
result = fn()
|
||||
except Exception:
|
||||
latency_ms = round((time.perf_counter() - started) * 1000)
|
||||
import sys
|
||||
|
||||
exc = sys.exc_info()[1]
|
||||
if exc is not None:
|
||||
_safe_record_failure(db, provider, interface, latency_ms, exc)
|
||||
raise
|
||||
latency_ms = round((time.perf_counter() - started) * 1000)
|
||||
_safe_record_success(db, provider, interface, latency_ms, result, classify)
|
||||
return result
|
||||
@@ -20,7 +20,6 @@ from datahub.governance.ratelimit import TokenBucket
|
||||
from datahub.governance.retry import RetryError, retry_call
|
||||
from datahub.logutil import get_logger
|
||||
from datahub.normalize import finite_number, normalize_daily
|
||||
from datahub import observability
|
||||
from datahub.revision import (
|
||||
compare_fields,
|
||||
diff_published_vs_upstream,
|
||||
@@ -1635,7 +1634,6 @@ class Pipeline:
|
||||
deleted += cur.rowcount
|
||||
connection.execute("DELETE FROM job_runs WHERE started_at < ?", (cutoff_jobs,))
|
||||
connection.execute("DELETE FROM src_calls WHERE created_at < ?", (cutoff_jobs,))
|
||||
connection.execute("DELETE FROM provider_call_log WHERE created_at < ?", (cutoff_jobs,))
|
||||
return {"staging_deleted": deleted}
|
||||
|
||||
def audit(self, actor: str, action: str, target: str = "", detail: str = "") -> None:
|
||||
@@ -1770,18 +1768,6 @@ class Pipeline:
|
||||
"INSERT INTO src_calls(provider, endpoint, ok, latency_ms, error, created_at) VALUES (?,?,?,?,?,?)",
|
||||
("tushare", endpoint, 1 if ok else 0, latency_ms, error, isoformat(self.clock())),
|
||||
)
|
||||
# HEL-543 side channel: unified cross-provider call log/health. Kept
|
||||
# strictly additive and fail-open; the src_calls insert above (the
|
||||
# existing, already-compatible Tushare record) is unaffected either
|
||||
# way.
|
||||
if ok:
|
||||
status, reason = "ok", ""
|
||||
else:
|
||||
status, reason = observability.classify_error(error)
|
||||
observability.record_call(
|
||||
self.db, "tushare", endpoint,
|
||||
status=status, latency_ms=latency_ms, error=error, fallback_reason=reason,
|
||||
)
|
||||
|
||||
def _persist_health(self, state: str, error: str = "") -> None:
|
||||
snap = self.breaker.snapshot()
|
||||
|
||||
@@ -18,7 +18,6 @@ from datahub.adapters.tencent import TencentAdapter
|
||||
from datahub.codes import resolve_code
|
||||
from datahub.db import HubDB
|
||||
from datahub.governance.lkg import LastKnownGood
|
||||
from datahub import observability
|
||||
from datahub.timeutil import isoformat, now_shanghai, yyyymmdd
|
||||
|
||||
QUOTE_TTL = 60
|
||||
@@ -26,23 +25,6 @@ INDEX_TTL = 60
|
||||
INTRADAY_TTL = 20
|
||||
QUOTE_BATCH = 60
|
||||
|
||||
# HEL-543 side-channel classifiers. These only *read* an already-successful
|
||||
# payload to decide what to log; they never change the payload itself and a
|
||||
# classifier exception always degrades to "ok" (see observability.classify_rows).
|
||||
|
||||
|
||||
def _classify_quote_rows(rows: Any) -> tuple[str, str, int | None]:
|
||||
return observability.classify_rows(rows, freshness_field="quote_time_epoch")
|
||||
|
||||
|
||||
def _classify_rows_no_freshness(rows: Any) -> tuple[str, str, int | None]:
|
||||
return observability.classify_rows(rows, freshness_field=None)
|
||||
|
||||
|
||||
def _classify_intraday_payload(data: Any) -> tuple[str, str, int | None]:
|
||||
points = data.get("points") if isinstance(data, dict) else None
|
||||
return observability.classify_rows(points or [], freshness_field=None)
|
||||
|
||||
|
||||
class RealtimeApiError(RuntimeError):
|
||||
def __init__(self, code: str, message: str) -> None:
|
||||
@@ -62,17 +44,12 @@ def fetch_index_quotes(db: HubDB) -> dict[str, Any]:
|
||||
cached = _read_cache(db, cache_key)
|
||||
if cached is not None:
|
||||
return cached
|
||||
eastmoney = EastmoneyAdapter()
|
||||
try:
|
||||
rows = observability.observe(
|
||||
db, "eastmoney", "indices", lambda: EastmoneyAdapter().fetch_indices(),
|
||||
classify=_classify_quote_rows,
|
||||
)
|
||||
rows = eastmoney.fetch_indices()
|
||||
source = "eastmoney:ulist"
|
||||
except Exception:
|
||||
rows = observability.observe(
|
||||
db, "tencent", "indices", lambda: TencentAdapter().fetch_indices(),
|
||||
classify=_classify_quote_rows,
|
||||
)
|
||||
rows = TencentAdapter().fetch_indices()
|
||||
source = "tencent:qt"
|
||||
if len(rows) < 3:
|
||||
raise RealtimeApiError("SOURCE_UNAVAILABLE", "index quotes incomplete")
|
||||
@@ -100,10 +77,7 @@ def fetch_market_quotes(db: HubDB) -> dict[str, Any]:
|
||||
rows: list[dict[str, Any]] = []
|
||||
source = ""
|
||||
try:
|
||||
rows = observability.observe(
|
||||
db, "eastmoney", "market_quotes", lambda: EastmoneyAdapter().fetch_market_quotes(),
|
||||
classify=_classify_quote_rows,
|
||||
)
|
||||
rows = EastmoneyAdapter().fetch_market_quotes()
|
||||
source = "eastmoney:clist"
|
||||
except Exception as exc:
|
||||
errors.append(f"eastmoney:{exc}")
|
||||
@@ -111,10 +85,7 @@ def fetch_market_quotes(db: HubDB) -> dict[str, Any]:
|
||||
listed = _listed_ts_codes(db)
|
||||
if not listed:
|
||||
raise AdapterError("no local stock master for tencent market snapshot")
|
||||
rows = observability.observe(
|
||||
db, "tencent", "market_quotes_fallback", lambda: _tencent_named_quotes(listed),
|
||||
classify=_classify_quote_rows,
|
||||
)
|
||||
rows = _tencent_named_quotes(listed)
|
||||
if len(rows) < 200:
|
||||
raise AdapterError(f"Tencent market snapshot too small: {len(rows)}")
|
||||
source = "tencent:qt"
|
||||
@@ -159,10 +130,7 @@ def fetch_quotes(db: HubDB, codes: list[str]) -> dict[str, Any]:
|
||||
|
||||
missing = [code for code in resolved if code not in by_code]
|
||||
try:
|
||||
rows = observability.observe(
|
||||
db, "eastmoney", "named_quotes", lambda: _eastmoney_named_quotes(missing),
|
||||
classify=_classify_quote_rows,
|
||||
)
|
||||
rows = _eastmoney_named_quotes(missing)
|
||||
by_code.update(_quote_map(rows, missing))
|
||||
if rows:
|
||||
sources.append("eastmoney:ulist")
|
||||
@@ -172,10 +140,7 @@ def fetch_quotes(db: HubDB, codes: list[str]) -> dict[str, Any]:
|
||||
missing = [code for code in resolved if code not in by_code]
|
||||
if missing:
|
||||
try:
|
||||
rows = observability.observe(
|
||||
db, "tencent", "named_quotes", lambda: _tencent_named_quotes(missing),
|
||||
classify=_classify_quote_rows,
|
||||
)
|
||||
rows = _tencent_named_quotes(missing)
|
||||
by_code.update(_quote_map(rows, missing))
|
||||
if rows:
|
||||
sources.append("tencent:qt")
|
||||
@@ -238,10 +203,7 @@ def fetch_sector_quote(db: HubDB, code: str, expected_date: str = "") -> dict[st
|
||||
return cached
|
||||
errors: list[str] = []
|
||||
try:
|
||||
row = observability.observe(
|
||||
db, "eastmoney", "sector_quote", lambda: EastmoneyAdapter().fetch_shenwan_quote(ts_code),
|
||||
classify=_classify_quote_rows,
|
||||
)
|
||||
row = EastmoneyAdapter().fetch_shenwan_quote(ts_code)
|
||||
if not _sector_row_matches(row, canonical_name):
|
||||
raise AdapterError(
|
||||
f"industry name mismatch: expected {canonical_name}, got {row.get('name') or '--'}"
|
||||
@@ -308,10 +270,7 @@ def fetch_limit_pool(db: HubDB, trade_date: str = "") -> dict[str, Any]:
|
||||
if cached is not None:
|
||||
return cached
|
||||
try:
|
||||
rows = observability.observe(
|
||||
db, "eastmoney", "limit_pool", lambda: EastmoneyAdapter().fetch_limit_pool(day),
|
||||
classify=_classify_rows_no_freshness,
|
||||
)
|
||||
rows = EastmoneyAdapter().fetch_limit_pool(day)
|
||||
source = "eastmoney:zt_pool"
|
||||
except Exception as exc:
|
||||
recovered = _load_quotes_lkg(db, cache_key)
|
||||
@@ -571,10 +530,7 @@ def warm_realtime(db: HubDB) -> dict[str, Any]:
|
||||
if master_code and master_name:
|
||||
canonical_names.setdefault(master_code, master_name)
|
||||
codes = list(canonical_names)
|
||||
fetched_sector_rows = observability.observe(
|
||||
db, "eastmoney", "sector_quotes_batch", lambda: _eastmoney_sector_quotes(codes),
|
||||
classify=_classify_quote_rows,
|
||||
)
|
||||
fetched_sector_rows = _eastmoney_sector_quotes(codes)
|
||||
for row in fetched_sector_rows:
|
||||
if _row_quote_date(row, today) != today:
|
||||
continue
|
||||
@@ -632,10 +588,7 @@ def fetch_intraday(db: HubDB, code: str, date: str = "") -> dict[str, Any]:
|
||||
return cached
|
||||
adapter = EastmoneyAdapter()
|
||||
try:
|
||||
payload_data = observability.observe(
|
||||
db, "eastmoney", "intraday", lambda: adapter.fetch_intraday(ts_code, date),
|
||||
classify=_classify_intraday_payload,
|
||||
)
|
||||
payload_data = adapter.fetch_intraday(ts_code, date)
|
||||
source = "eastmoney:trends2"
|
||||
except Exception as exc:
|
||||
recovered = _load_intraday_lkg(db, ts_code, date)
|
||||
|
||||
@@ -33,11 +33,6 @@ class Settings:
|
||||
quality: dict[str, Any] = field(default_factory=dict)
|
||||
log_level: str = "INFO"
|
||||
scheduler_enabled: bool = True
|
||||
# HEL-543 kill switch: off disables the provider_call_log/provider_health
|
||||
# side channel entirely (observe()/record_call() become no-ops and the
|
||||
# new read-only admin endpoints report {"enabled": false}). Default on;
|
||||
# existing routing/fetch/publish behavior is identical either way.
|
||||
observability_enabled: bool = True
|
||||
|
||||
@property
|
||||
def tushare_rate_per_minute(self) -> int:
|
||||
@@ -131,5 +126,4 @@ def load_settings(
|
||||
quality=_load_quality(quality_path),
|
||||
log_level=environ.get("DATAHUB_LOG_LEVEL") or "INFO",
|
||||
scheduler_enabled=str(environ.get("DATAHUB_SCHEDULER") or "1") not in {"0", "false", "False"},
|
||||
observability_enabled=str(environ.get("DATAHUB_OBSERVABILITY") or "1") not in {"0", "false", "False", "off", "OFF"},
|
||||
)
|
||||
|
||||
@@ -1,165 +0,0 @@
|
||||
"""Minimal, read-only source directory (HEL-543).
|
||||
|
||||
Registers what already exists: providers, their concrete interfaces, what
|
||||
capability/dataset each interface serves, and whether the provider plays a
|
||||
primary or backup role. This module only *describes* the current adapters
|
||||
and datasets already wired in `datahub/hub.py`, `datahub/serving.py`, and
|
||||
`datahub/realtime_serve.py`; it does not add a way to configure or add a new
|
||||
source without code, and it never changes routing, retries, or fallback
|
||||
order.
|
||||
|
||||
Every ``interfaces`` entry below is a docs-as-code mirror of a real call
|
||||
site, cross-referenced in comments so a reviewer can verify each row is
|
||||
accurate rather than aspirational:
|
||||
|
||||
- tushare interfaces mirror ``datahub/serving.py``'s ``_official_meta``/``source=``
|
||||
strings and ``datahub/steward.py``'s live/published dataset table.
|
||||
- eastmoney/tencent interfaces mirror the ``observability.observe(...)``
|
||||
call sites added in ``datahub/realtime_serve.py`` for HEL-543.
|
||||
- ifind interfaces mirror ``datahub/steward.py``'s ``IFIND_APIS`` table.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
CATALOG: list[dict[str, Any]] = [
|
||||
{
|
||||
"provider": "tushare",
|
||||
"label": "Tushare",
|
||||
"role": "official_primary",
|
||||
"credential_key": "tushare_token",
|
||||
"status_source": "src_health (legacy, kept) + provider_health (unified, HEL-543)",
|
||||
"interfaces": [
|
||||
{"interface": "trade_cal", "capability": "交易日历", "datasets": ["calendar"]},
|
||||
{"interface": "stock_basic", "capability": "股票主档", "datasets": ["stocks"]},
|
||||
{"interface": "daily", "capability": "个股日K", "datasets": ["daily"]},
|
||||
{"interface": "adj_factor", "capability": "复权因子", "datasets": ["daily"]},
|
||||
{"interface": "daily_basic", "capability": "估值", "datasets": ["valuation"]},
|
||||
{"interface": "index_daily", "capability": "指数日K", "datasets": ["index_daily"]},
|
||||
{"interface": "moneyflow", "capability": "资金流", "datasets": ["moneyflow"]},
|
||||
{"interface": "stk_auction", "capability": "集合竞价", "datasets": ["auction"]},
|
||||
{"interface": "limit_list_d", "capability": "涨跌停池", "datasets": ["limit_events"]},
|
||||
{"interface": "ths_hot", "capability": "同花顺人气榜", "datasets": ["popularity"]},
|
||||
{"interface": "dc_hot", "capability": "东方财富人气榜", "datasets": ["popularity"]},
|
||||
{"interface": "hm_detail", "capability": "龙虎榜游资明细", "datasets": ["dragon_tiger"]},
|
||||
{"interface": "ths_daily", "capability": "同花顺概念行情", "datasets": ["sector_daily"]},
|
||||
{"interface": "dc_index", "capability": "东方财富板块行情", "datasets": ["sector_daily"]},
|
||||
{"interface": "sw_daily", "capability": "申万行业行情", "datasets": ["sector_daily"]},
|
||||
],
|
||||
},
|
||||
{
|
||||
"provider": "eastmoney",
|
||||
"label": "东方财富",
|
||||
"role": "provisional_primary",
|
||||
"credential_key": None,
|
||||
"status_source": "provider_health (unified, HEL-543)",
|
||||
"interfaces": [
|
||||
{"interface": "indices", "capability": "指数实时报价", "datasets": ["index_quotes"]},
|
||||
{"interface": "market_quotes", "capability": "全市场实时快照", "datasets": ["quotes_latest"]},
|
||||
{"interface": "named_quotes", "capability": "指定个股实时报价", "datasets": ["quotes_latest"]},
|
||||
{"interface": "sector_quote", "capability": "申万板块实时报价(单个)", "datasets": ["sectors_quote"]},
|
||||
{"interface": "sector_quotes_batch", "capability": "申万板块批量报价(预热)", "datasets": ["sectors_quote"]},
|
||||
{"interface": "limit_pool", "capability": "涨停/炸板池(盘中)", "datasets": ["limit_pool"]},
|
||||
{"interface": "intraday", "capability": "分时走势", "datasets": ["intraday_points"]},
|
||||
],
|
||||
},
|
||||
{
|
||||
"provider": "tencent",
|
||||
"label": "腾讯行情",
|
||||
"role": "provisional_backup",
|
||||
"credential_key": None,
|
||||
"status_source": "provider_health (unified, HEL-543)",
|
||||
"interfaces": [
|
||||
{"interface": "indices", "capability": "指数实时报价(东财失败时备用)", "datasets": ["index_quotes"]},
|
||||
{
|
||||
"interface": "market_quotes_fallback",
|
||||
"capability": "全市场快照(备用；按本地股票主档逐只请求拼接)",
|
||||
"datasets": ["quotes_latest"],
|
||||
},
|
||||
{"interface": "named_quotes", "capability": "指定个股实时报价(东财失败时备用)", "datasets": ["quotes_latest"]},
|
||||
],
|
||||
},
|
||||
{
|
||||
"provider": "ifind",
|
||||
"label": "同花顺 iFinD",
|
||||
"role": "licensed_optional",
|
||||
"credential_key": "ifind_refresh_token",
|
||||
"status_source": "provider_health (unified, HEL-543) + adapter.status()",
|
||||
"interfaces": [
|
||||
{"interface": "wencai", "capability": "问财自然语言选股", "datasets": ["ifind_wencai"]},
|
||||
{"interface": "snapshots", "capability": "快照", "datasets": ["ifind_snapshots"]},
|
||||
{"interface": "history", "capability": "历史行情", "datasets": ["ifind_history"]},
|
||||
{"interface": "realtime", "capability": "实时行情", "datasets": ["ifind_realtime"]},
|
||||
{"interface": "intraday", "capability": "分时(高频)", "datasets": ["ifind_intraday"]},
|
||||
],
|
||||
},
|
||||
{
|
||||
"provider": "ths",
|
||||
"label": "同花顺(预留)",
|
||||
"role": "reserved",
|
||||
"credential_key": None,
|
||||
"status_source": "adapter.probe()（占位，本阶段未接入真实数据）",
|
||||
"interfaces": [],
|
||||
},
|
||||
{
|
||||
"provider": "xgb",
|
||||
"label": "选股宝(预留)",
|
||||
"role": "reserved",
|
||||
"credential_key": None,
|
||||
"status_source": "adapter.probe()（占位，本阶段未接入真实数据）",
|
||||
"interfaces": [],
|
||||
},
|
||||
{
|
||||
"provider": "akshare",
|
||||
"label": "AKShare(预留)",
|
||||
"role": "reserved",
|
||||
"credential_key": None,
|
||||
"status_source": "adapter.probe()（占位，本阶段未接入真实数据）",
|
||||
"interfaces": [],
|
||||
},
|
||||
]
|
||||
|
||||
|
||||
def snapshot(db: Any, auth: Any = None) -> list[dict[str, Any]]:
|
||||
"""Merge the static catalog with live credential/health facts.
|
||||
|
||||
Purely read-only: never touches routing, credentials, or adapters. Any
|
||||
failure while enriching one entry only degrades that entry's live data;
|
||||
it never drops the entry or raises, so a directory read can never break
|
||||
on a partially-unhealthy database.
|
||||
"""
|
||||
result: list[dict[str, Any]] = []
|
||||
for entry in CATALOG:
|
||||
item: dict[str, Any] = {
|
||||
"provider": entry["provider"],
|
||||
"label": entry.get("label", entry["provider"]),
|
||||
"role": entry["role"],
|
||||
"status_source": entry["status_source"],
|
||||
"interfaces": [dict(i) for i in entry.get("interfaces", [])],
|
||||
}
|
||||
cred_key = entry.get("credential_key")
|
||||
if cred_key:
|
||||
cred = None
|
||||
try:
|
||||
if auth is not None:
|
||||
cred = auth.credential_status(cred_key)
|
||||
except Exception:
|
||||
cred = None
|
||||
item["credential"] = cred or {"configured": False, "last4": "", "updated_at": ""}
|
||||
else:
|
||||
item["credential"] = {"configured": True, "last4": "", "updated_at": "", "note": "无需凭证"}
|
||||
health_rows: list[dict[str, Any]] = []
|
||||
try:
|
||||
if db is not None:
|
||||
health_rows = db.fetchall(
|
||||
"SELECT interface, state, last_ok_at, last_error, last_fallback_reason, "
|
||||
"consec_failures, last_latency_ms, last_data_age_seconds, updated_at "
|
||||
"FROM provider_health WHERE provider = ? ORDER BY interface",
|
||||
(entry["provider"],),
|
||||
)
|
||||
except Exception:
|
||||
health_rows = []
|
||||
item["live_interfaces"] = health_rows
|
||||
result.append(item)
|
||||
return result
|
||||
@@ -12,7 +12,6 @@ from typing import Any
|
||||
|
||||
from datahub.adapters.base import AdapterError
|
||||
from datahub.adapters.tushare import TUSHARE_FIELDS
|
||||
from datahub import observability
|
||||
from datahub.numbers import finite_number
|
||||
from datahub.realtime_serve import (
|
||||
RealtimeApiError,
|
||||
@@ -153,12 +152,8 @@ def _ifind_query(api, api_name: str, params: dict[str, Any], fields: str) -> dic
|
||||
)
|
||||
if not adapter.configured:
|
||||
raise ApiError("SOURCE_UNAVAILABLE", "iFinD 尚未配置")
|
||||
db = getattr(api, "db", None)
|
||||
try:
|
||||
rows = observability.observe(
|
||||
db, "ifind", dataset, lambda: adapter.fetch(dataset, dict(params)),
|
||||
classify=lambda r: observability.classify_rows(r, freshness_field=None),
|
||||
)
|
||||
rows = adapter.fetch(dataset, dict(params))
|
||||
except AdapterError as exc:
|
||||
raise ApiError("SOURCE_UNAVAILABLE", str(exc)) from exc
|
||||
return envelope(
|
||||
|
||||
@@ -1,180 +0,0 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import tempfile
|
||||
import threading
|
||||
import unittest
|
||||
from http.server import ThreadingHTTPServer
|
||||
from pathlib import Path
|
||||
from urllib.request import Request, urlopen
|
||||
|
||||
from datahub.adapters.tushare import TushareAdapter
|
||||
from datahub.crypto import SecretVault
|
||||
from datahub.httpapp import make_handler
|
||||
from datahub.hub import Hub
|
||||
from datahub.settings import Settings
|
||||
from tests.fixtures import TRADE_DATE, fake_transport
|
||||
|
||||
|
||||
class AdminObservabilityApiTests(unittest.TestCase):
|
||||
"""HEL-543: new read-only admin endpoints for provider status, source
|
||||
catalog and lineage. These must never require write access and must
|
||||
never touch the existing routing/publish logic."""
|
||||
|
||||
def setUp(self) -> None:
|
||||
self.tmp = tempfile.TemporaryDirectory()
|
||||
settings = Settings(
|
||||
encryption_key=SecretVault.generate_key(),
|
||||
api_token="z" * 32,
|
||||
admin_password="StartPass1",
|
||||
tushare_token="real-tushare-token-abcdef",
|
||||
db_path=Path(self.tmp.name) / "hub.db",
|
||||
scheduler_enabled=False,
|
||||
)
|
||||
self.hub = Hub(settings, adapter=TushareAdapter("real-tushare-token-abcdef", transport=fake_transport))
|
||||
handler = make_handler(self.hub)
|
||||
self.server = ThreadingHTTPServer(("127.0.0.1", 0), handler)
|
||||
threading.Thread(target=self.server.serve_forever, daemon=True).start()
|
||||
self.base = f"http://127.0.0.1:{self.server.server_address[1]}"
|
||||
|
||||
_, body, cookie_header = self._json(
|
||||
"/admin/api/login", "POST", {"username": "hub_admin", "password": "StartPass1"}
|
||||
)
|
||||
cookie = cookie_header.split(";")[0]
|
||||
csrf = body["csrf"]
|
||||
self._json(
|
||||
"/admin/api/change-password",
|
||||
"POST",
|
||||
{"current": "StartPass1", "new_password": "NewPass123"},
|
||||
cookie=cookie,
|
||||
csrf=csrf,
|
||||
)
|
||||
self.cookie = cookie
|
||||
self.csrf = csrf
|
||||
|
||||
def tearDown(self) -> None:
|
||||
self.server.shutdown()
|
||||
self.server.server_close()
|
||||
self.tmp.cleanup()
|
||||
|
||||
def _json(self, path, method="GET", body=None, cookie="", csrf=""):
|
||||
data = None if body is None else json.dumps(body).encode()
|
||||
headers = {"Content-Type": "application/json"}
|
||||
if cookie:
|
||||
headers["Cookie"] = cookie
|
||||
if csrf:
|
||||
headers["X-CSRF-Token"] = csrf
|
||||
req = Request(self.base + path, data=data, headers=headers, method=method)
|
||||
with urlopen(req, timeout=5) as resp:
|
||||
set_cookie = resp.headers.get("Set-Cookie", "")
|
||||
return resp.status, json.loads(resp.read().decode()), set_cookie
|
||||
|
||||
def _get(self, path):
|
||||
return self._json(path, cookie=self.cookie, csrf=self.csrf)
|
||||
|
||||
def test_providers_status_reflects_real_pipeline_activity(self) -> None:
|
||||
pipeline = self.hub.pipeline
|
||||
pipeline.ingest_reference(TRADE_DATE)
|
||||
pipeline.run_dataset("daily", TRADE_DATE)
|
||||
|
||||
status, body, _ = self._get("/admin/api/providers/status")
|
||||
self.assertEqual(status, 200)
|
||||
health = body["health"]
|
||||
self.assertTrue(any(item["provider"] == "tushare" and item["interface"] == "daily" for item in health))
|
||||
row = next(item for item in health if item["provider"] == "tushare" and item["interface"] == "daily")
|
||||
self.assertEqual(row["state"], "ok")
|
||||
recent = body["recent_calls"]
|
||||
self.assertTrue(any(item["provider"] == "tushare" and item["interface"] == "daily" for item in recent))
|
||||
|
||||
def test_providers_status_filters_by_provider(self) -> None:
|
||||
pipeline = self.hub.pipeline
|
||||
pipeline.ingest_reference(TRADE_DATE)
|
||||
pipeline.run_dataset("daily", TRADE_DATE)
|
||||
|
||||
status, body, _ = self._get("/admin/api/providers/status?provider=tushare")
|
||||
self.assertEqual(status, 200)
|
||||
self.assertTrue(all(item["provider"] == "tushare" for item in body["health"]))
|
||||
self.assertTrue(all(item["provider"] == "tushare" for item in body["recent_calls"]))
|
||||
|
||||
status, body, _ = self._get("/admin/api/providers/status?provider=eastmoney")
|
||||
self.assertEqual(status, 200)
|
||||
self.assertEqual(body["health"], [])
|
||||
self.assertEqual(body["recent_calls"], [])
|
||||
|
||||
def test_source_catalog_lists_known_providers_without_leaking_secrets(self) -> None:
|
||||
status, body, _ = self._get("/admin/api/source-catalog")
|
||||
self.assertEqual(status, 200)
|
||||
blob = json.dumps(body)
|
||||
self.assertNotIn("real-tushare-token-abcdef", blob)
|
||||
providers = {item["provider"] for item in body["items"]}
|
||||
self.assertIn("tushare", providers)
|
||||
self.assertIn("eastmoney", providers)
|
||||
self.assertIn("tencent", providers)
|
||||
self.assertIn("ifind", providers)
|
||||
|
||||
def test_lineage_snapshot_and_affected_query(self) -> None:
|
||||
status, body, _ = self._get("/admin/api/lineage")
|
||||
self.assertEqual(status, 200)
|
||||
self.assertTrue(len(body["items"]) > 0)
|
||||
datasets = {item["dataset"] for item in body["items"]}
|
||||
self.assertIn("stocks", datasets)
|
||||
|
||||
status, body, _ = self._get("/admin/api/lineage/affected?provider=tushare&interface=daily")
|
||||
self.assertEqual(status, 200)
|
||||
self.assertEqual(body["provider"], "tushare")
|
||||
self.assertEqual(body["interface"], "daily")
|
||||
|
||||
def test_disabled_kill_switch_reports_enabled_false_with_empty_structure(self) -> None:
|
||||
# HEL-543 total-review 🔴: flip the runtime kill switch the same way
|
||||
# Hub.__init__ wires Settings.observability_enabled onto the db
|
||||
# handle, then confirm every new endpoint reports disabled with an
|
||||
# explicit empty structure rather than silently going quiet.
|
||||
self.hub.db.observability_enabled = False
|
||||
try:
|
||||
pipeline = self.hub.pipeline
|
||||
pipeline.ingest_reference(TRADE_DATE)
|
||||
pipeline.run_dataset("daily", TRADE_DATE) # must still fully succeed
|
||||
|
||||
status, body, _ = self._get("/admin/api/providers/status")
|
||||
self.assertEqual(status, 200)
|
||||
self.assertEqual(body, {"enabled": False, "health": [], "recent_calls": []})
|
||||
|
||||
status, body, _ = self._get("/admin/api/source-catalog")
|
||||
self.assertEqual(status, 200)
|
||||
self.assertEqual(body, {"enabled": False, "items": []})
|
||||
|
||||
status, body, _ = self._get("/admin/api/lineage")
|
||||
self.assertEqual(status, 200)
|
||||
self.assertFalse(body["enabled"])
|
||||
self.assertEqual(body["items"], [])
|
||||
|
||||
status, body, _ = self._get("/admin/api/lineage/affected?provider=tushare")
|
||||
self.assertEqual(status, 200)
|
||||
self.assertEqual(
|
||||
body, {"enabled": False, "provider": "tushare", "interface": "", "items": []}
|
||||
)
|
||||
|
||||
# Nothing was ever written while disabled.
|
||||
self.assertEqual(self.hub.db.fetchall("SELECT * FROM provider_call_log"), [])
|
||||
finally:
|
||||
self.hub.db.observability_enabled = True
|
||||
|
||||
def test_must_change_password_blocks_new_endpoints_too(self) -> None:
|
||||
from urllib.error import HTTPError
|
||||
|
||||
_, body, cookie_header = self._json(
|
||||
"/admin/api/login", "POST", {"username": "hub_admin", "password": "NewPass123"}
|
||||
)
|
||||
# Freshly logged-in user has already changed password in setUp, so
|
||||
# this login should not require a change; verify the endpoint is
|
||||
# reachable with a valid, non-must-change session (regression guard
|
||||
# against accidentally bypassing the must-change gate for these new
|
||||
# routes).
|
||||
cookie = cookie_header.split(";")[0]
|
||||
csrf = body["csrf"]
|
||||
status, _, _ = self._json("/admin/api/source-catalog", cookie=cookie, csrf=csrf)
|
||||
self.assertEqual(status, 200)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -1,105 +0,0 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
from datahub.adapters.ifind import IfindAdapter
|
||||
from datahub.db import HubDB
|
||||
from datahub.serving import ApiError
|
||||
from datahub.steward import steward_query
|
||||
|
||||
|
||||
class _Resp:
|
||||
def __init__(self, payload: dict, status: int = 200) -> None:
|
||||
import json
|
||||
|
||||
self.status = status
|
||||
self._raw = json.dumps(payload).encode("utf-8")
|
||||
|
||||
def read(self):
|
||||
return self._raw
|
||||
|
||||
def __enter__(self):
|
||||
return self
|
||||
|
||||
def __exit__(self, *args):
|
||||
return False
|
||||
|
||||
|
||||
class IfindObservabilityTests(unittest.TestCase):
|
||||
def setUp(self) -> None:
|
||||
self.tmp = tempfile.TemporaryDirectory()
|
||||
self.db = HubDB(Path(self.tmp.name) / "hub.db")
|
||||
|
||||
def tearDown(self) -> None:
|
||||
self.tmp.cleanup()
|
||||
|
||||
def _adapter_with_urlopen(self, urlopen) -> IfindAdapter:
|
||||
return IfindAdapter(refresh_token="rt", access_token="at", urlopen=urlopen)
|
||||
|
||||
def test_successful_fetch_is_logged_without_changing_rows(self) -> None:
|
||||
def urlopen(request, timeout=None):
|
||||
return _Resp(
|
||||
{
|
||||
"errorcode": 0,
|
||||
"tables": [{"thscode": ["000001.SZ"], "table": {"涨停原因": ["重组"]}}],
|
||||
}
|
||||
)
|
||||
|
||||
adapter = self._adapter_with_urlopen(urlopen)
|
||||
|
||||
class _Api:
|
||||
ifind = adapter
|
||||
db = self.db
|
||||
|
||||
payload = steward_query(
|
||||
_Api(), {"api_name": "ifind_wencai", "params": {"query": "涨停"}}
|
||||
)
|
||||
self.assertEqual(payload["data"][0]["thscode"], "000001.SZ")
|
||||
log = self.db.fetchone("SELECT * FROM provider_call_log WHERE provider = 'ifind'")
|
||||
self.assertIsNotNone(log)
|
||||
self.assertEqual(log["interface"], "wencai")
|
||||
self.assertEqual(log["status"], "ok")
|
||||
|
||||
def test_failed_fetch_reraises_and_logs_error(self) -> None:
|
||||
def urlopen(request, timeout=None):
|
||||
return _Resp({"errorcode": -9999, "errmsg": "quota exceeded"})
|
||||
|
||||
adapter = self._adapter_with_urlopen(urlopen)
|
||||
|
||||
class _Api:
|
||||
ifind = adapter
|
||||
db = self.db
|
||||
|
||||
with self.assertRaises(ApiError) as ctx:
|
||||
steward_query(_Api(), {"api_name": "ifind_wencai", "params": {"query": "涨停"}})
|
||||
self.assertEqual(ctx.exception.code, "SOURCE_UNAVAILABLE")
|
||||
log = self.db.fetchone("SELECT * FROM provider_call_log WHERE provider = 'ifind'")
|
||||
self.assertIsNotNone(log)
|
||||
self.assertEqual(log["status"], "error")
|
||||
|
||||
def test_status_check_alone_does_not_dial_or_log_a_fetch_call(self) -> None:
|
||||
class _Api:
|
||||
ifind = IfindAdapter()
|
||||
db = self.db
|
||||
|
||||
payload = steward_query(_Api(), {"api_name": "ifind_status", "params": {}})
|
||||
self.assertFalse(payload["data"][0]["configured"])
|
||||
log = self.db.fetchall("SELECT * FROM provider_call_log WHERE provider = 'ifind'")
|
||||
self.assertEqual(log, [])
|
||||
|
||||
def test_api_double_without_db_attribute_still_works(self) -> None:
|
||||
# Mirrors tests/test_ifind_adapter.py's `_Api` double, which has no
|
||||
# `db` attribute at all. Observability must not require it.
|
||||
class _Api:
|
||||
ifind = IfindAdapter()
|
||||
|
||||
payload = steward_query(_Api(), {"api_name": "ifind_status", "params": {}})
|
||||
self.assertFalse(payload["data"][0]["configured"])
|
||||
with self.assertRaises(ApiError):
|
||||
steward_query(_Api(), {"api_name": "ifind_wencai", "params": {"query": "涨停"}})
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -1,306 +0,0 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import tempfile
|
||||
import unittest
|
||||
from contextlib import contextmanager
|
||||
from pathlib import Path
|
||||
|
||||
from datahub import observability
|
||||
from datahub.db import HubDB
|
||||
|
||||
|
||||
class _BrokenDB:
|
||||
"""A db double whose write() always raises, to prove fail-open."""
|
||||
|
||||
@contextmanager
|
||||
def write(self):
|
||||
raise RuntimeError("disk is full")
|
||||
yield None # pragma: no cover - unreachable, keeps this a generator
|
||||
|
||||
def fetchall(self, sql, params=()):
|
||||
raise RuntimeError("disk is full")
|
||||
|
||||
def fetchone(self, sql, params=()):
|
||||
raise RuntimeError("disk is full")
|
||||
|
||||
|
||||
class ClassifyRowsTests(unittest.TestCase):
|
||||
def test_empty_list_is_flagged_empty(self):
|
||||
status, reason, age = observability.classify_rows([])
|
||||
self.assertEqual(status, "empty")
|
||||
self.assertEqual(reason, "no_rows_returned")
|
||||
self.assertIsNone(age)
|
||||
|
||||
def test_empty_dict_result_is_flagged_empty(self):
|
||||
status, reason, _ = observability.classify_rows({})
|
||||
self.assertEqual(status, "empty")
|
||||
self.assertEqual(reason, "no_rows_returned")
|
||||
|
||||
def test_missing_required_field_is_flagged(self):
|
||||
rows = [{"ts_code": "600000.SH", "close": 10.2}, {"ts_code": "000001.SZ"}]
|
||||
status, reason, _ = observability.classify_rows(rows, required_fields=("close",), freshness_field=None)
|
||||
self.assertEqual(status, "missing_fields")
|
||||
self.assertIn("close", reason)
|
||||
|
||||
def test_fresh_rows_are_ok(self):
|
||||
import time
|
||||
|
||||
rows = [{"ts_code": "600000.SH", "close": 10.2, "quote_time_epoch": int(time.time())}]
|
||||
status, reason, age = observability.classify_rows(rows)
|
||||
self.assertEqual(status, "ok")
|
||||
self.assertEqual(reason, "")
|
||||
self.assertIsNotNone(age)
|
||||
self.assertLess(age, 5)
|
||||
|
||||
def test_stale_rows_are_flagged(self):
|
||||
import time
|
||||
|
||||
rows = [{"ts_code": "600000.SH", "close": 10.2, "quote_time_epoch": int(time.time()) - 3600}]
|
||||
status, reason, age = observability.classify_rows(rows, max_age_seconds=300)
|
||||
self.assertEqual(status, "stale")
|
||||
self.assertEqual(reason, "data_age_exceeds_threshold")
|
||||
self.assertGreaterEqual(age, 3600 - 5)
|
||||
|
||||
def test_classifier_never_raises_on_garbage_input(self):
|
||||
status, reason, age = observability.classify_rows(object())
|
||||
self.assertEqual(status, "empty")
|
||||
self.assertIsNone(age)
|
||||
# Malformed rows inside a list must not raise either.
|
||||
status, _, _ = observability.classify_rows(["not-a-dict", 123, None])
|
||||
self.assertEqual(status, "empty")
|
||||
|
||||
|
||||
class ClassifyErrorTests(unittest.TestCase):
|
||||
def test_blocked_page_markers_are_detected(self):
|
||||
status, reason = observability.classify_error("eastmoney request failed: Expecting value: line 1 column 1")
|
||||
self.assertEqual(status, "blocked")
|
||||
self.assertEqual(reason, "response_looks_like_intercept_page")
|
||||
|
||||
def test_timeout_is_detected(self):
|
||||
status, _ = observability.classify_error("tencent request failed: timed out")
|
||||
self.assertEqual(status, "timeout")
|
||||
|
||||
def test_generic_error_falls_back(self):
|
||||
status, reason = observability.classify_error("connection reset by peer")
|
||||
self.assertEqual(status, "error")
|
||||
self.assertEqual(reason, "")
|
||||
|
||||
def test_never_raises_on_none(self):
|
||||
status, reason = observability.classify_error(None)
|
||||
self.assertEqual(status, "error")
|
||||
|
||||
|
||||
class RecordCallTests(unittest.TestCase):
|
||||
def setUp(self) -> None:
|
||||
self.tmp = tempfile.TemporaryDirectory()
|
||||
self.db = HubDB(Path(self.tmp.name) / "hub.db")
|
||||
|
||||
def tearDown(self) -> None:
|
||||
self.tmp.cleanup()
|
||||
|
||||
def test_record_call_writes_log_and_health(self):
|
||||
observability.record_call(self.db, "eastmoney", "indices", status="ok", latency_ms=42)
|
||||
log_rows = self.db.fetchall("SELECT * FROM provider_call_log")
|
||||
self.assertEqual(len(log_rows), 1)
|
||||
self.assertEqual(log_rows[0]["provider"], "eastmoney")
|
||||
self.assertEqual(log_rows[0]["interface"], "indices")
|
||||
self.assertEqual(log_rows[0]["status"], "ok")
|
||||
health = self.db.fetchone(
|
||||
"SELECT * FROM provider_health WHERE provider = ? AND interface = ?",
|
||||
("eastmoney", "indices"),
|
||||
)
|
||||
self.assertIsNotNone(health)
|
||||
self.assertEqual(health["state"], "ok")
|
||||
self.assertEqual(health["consec_failures"], 0)
|
||||
|
||||
def test_consecutive_failures_increment_and_reset(self):
|
||||
observability.record_call(self.db, "tencent", "named_quotes", status="error", error="boom")
|
||||
observability.record_call(self.db, "tencent", "named_quotes", status="error", error="boom again")
|
||||
health = self.db.fetchone(
|
||||
"SELECT * FROM provider_health WHERE provider = ? AND interface = ?",
|
||||
("tencent", "named_quotes"),
|
||||
)
|
||||
self.assertEqual(health["consec_failures"], 2)
|
||||
self.assertEqual(health["state"], "error")
|
||||
observability.record_call(self.db, "tencent", "named_quotes", status="ok")
|
||||
health = self.db.fetchone(
|
||||
"SELECT * FROM provider_health WHERE provider = ? AND interface = ?",
|
||||
("tencent", "named_quotes"),
|
||||
)
|
||||
self.assertEqual(health["consec_failures"], 0)
|
||||
self.assertEqual(health["state"], "ok")
|
||||
|
||||
def test_none_db_is_a_silent_noop(self):
|
||||
# Must not raise even though there is nowhere to write.
|
||||
observability.record_call(None, "ifind", "wencai", status="ok")
|
||||
|
||||
def test_broken_db_write_does_not_raise(self):
|
||||
observability.record_call(_BrokenDB(), "eastmoney", "indices", status="ok")
|
||||
|
||||
|
||||
class ObserveTests(unittest.TestCase):
|
||||
def setUp(self) -> None:
|
||||
self.tmp = tempfile.TemporaryDirectory()
|
||||
self.db = HubDB(Path(self.tmp.name) / "hub.db")
|
||||
|
||||
def tearDown(self) -> None:
|
||||
self.tmp.cleanup()
|
||||
|
||||
def test_returns_exact_success_value_unmodified(self):
|
||||
sentinel = {"ts_code": "600000.SH", "close": 10.2}
|
||||
result = observability.observe(self.db, "eastmoney", "indices", lambda: sentinel)
|
||||
self.assertIs(result, sentinel)
|
||||
rows = self.db.fetchall("SELECT * FROM provider_call_log")
|
||||
self.assertEqual(len(rows), 1)
|
||||
self.assertEqual(rows[0]["status"], "ok")
|
||||
|
||||
def test_reraises_exact_exception_on_failure(self):
|
||||
boom = ValueError("upstream exploded")
|
||||
|
||||
def fn():
|
||||
raise boom
|
||||
|
||||
with self.assertRaises(ValueError) as ctx:
|
||||
observability.observe(self.db, "eastmoney", "indices", fn)
|
||||
self.assertIs(ctx.exception, boom)
|
||||
rows = self.db.fetchall("SELECT * FROM provider_call_log")
|
||||
self.assertEqual(len(rows), 1)
|
||||
self.assertEqual(rows[0]["status"], "error")
|
||||
self.assertIn("upstream exploded", rows[0]["error"])
|
||||
|
||||
def test_classify_downgrades_success_to_stale_without_changing_return_value(self):
|
||||
sentinel = [{"ts_code": "600000.SH", "quote_time_epoch": 1}]
|
||||
result = observability.observe(
|
||||
self.db, "eastmoney", "indices", lambda: sentinel,
|
||||
classify=lambda rows: observability.classify_rows(rows),
|
||||
)
|
||||
self.assertIs(result, sentinel)
|
||||
row = self.db.fetchone("SELECT * FROM provider_call_log")
|
||||
self.assertEqual(row["status"], "stale")
|
||||
|
||||
def test_broken_db_never_breaks_a_successful_call(self):
|
||||
sentinel = {"ok": True}
|
||||
result = observability.observe(_BrokenDB(), "eastmoney", "indices", lambda: sentinel)
|
||||
self.assertIs(result, sentinel)
|
||||
|
||||
def test_broken_db_never_masks_a_real_failure(self):
|
||||
def fn():
|
||||
raise RuntimeError("real upstream failure")
|
||||
|
||||
with self.assertRaises(RuntimeError) as ctx:
|
||||
observability.observe(_BrokenDB(), "eastmoney", "indices", fn)
|
||||
self.assertEqual(str(ctx.exception), "real upstream failure")
|
||||
|
||||
def test_classifier_exception_does_not_break_the_call(self):
|
||||
sentinel = {"ok": True}
|
||||
|
||||
def bad_classify(_result):
|
||||
raise KeyError("classifier bug")
|
||||
|
||||
result = observability.observe(self.db, "eastmoney", "indices", lambda: sentinel, classify=bad_classify)
|
||||
self.assertIs(result, sentinel)
|
||||
row = self.db.fetchone("SELECT * FROM provider_call_log")
|
||||
# A classifier bug must degrade to "ok", never silently drop the row
|
||||
# nor claim the call failed when it did not.
|
||||
self.assertEqual(row["status"], "ok")
|
||||
|
||||
def test_none_db_is_transparent_passthrough(self):
|
||||
sentinel = object()
|
||||
result = observability.observe(None, "eastmoney", "indices", lambda: sentinel)
|
||||
self.assertIs(result, sentinel)
|
||||
|
||||
|
||||
class _ToggleDB(HubDB):
|
||||
"""A real HubDB subclass so we can flip the HEL-543 kill switch the same
|
||||
way Hub.__init__ does, without needing a full Hub/Settings wiring."""
|
||||
|
||||
|
||||
class KillSwitchTests(unittest.TestCase):
|
||||
"""HEL-543 total-review 🔴: the observability side channel must be
|
||||
disable-able at runtime, and disabling it must leave existing behavior
|
||||
completely unchanged (pure passthrough, zero db access)."""
|
||||
|
||||
def setUp(self) -> None:
|
||||
self.tmp = tempfile.TemporaryDirectory()
|
||||
self.db = _ToggleDB(Path(self.tmp.name) / "hub.db")
|
||||
|
||||
def tearDown(self) -> None:
|
||||
self.tmp.cleanup()
|
||||
|
||||
def test_is_enabled_defaults_true_when_attribute_absent(self):
|
||||
# A bare HubDB (as used throughout the rest of this test suite, and
|
||||
# by any pre-HEL-543 call site) must default to enabled.
|
||||
self.assertTrue(observability.is_enabled(self.db))
|
||||
self.assertTrue(observability.is_enabled(None))
|
||||
|
||||
def test_disabled_record_call_writes_nothing(self):
|
||||
self.db.observability_enabled = False
|
||||
observability.record_call(self.db, "eastmoney", "indices", status="ok", latency_ms=1)
|
||||
self.assertEqual(self.db.fetchall("SELECT * FROM provider_call_log"), [])
|
||||
self.assertEqual(self.db.fetchall("SELECT * FROM provider_health"), [])
|
||||
|
||||
def test_disabled_observe_is_a_pure_passthrough_on_success(self):
|
||||
self.db.observability_enabled = False
|
||||
sentinel = {"ts_code": "600000.SH"}
|
||||
calls = {"n": 0}
|
||||
|
||||
def fn():
|
||||
calls["n"] += 1
|
||||
return sentinel
|
||||
|
||||
result = observability.observe(self.db, "eastmoney", "indices", fn)
|
||||
self.assertIs(result, sentinel)
|
||||
self.assertEqual(calls["n"], 1)
|
||||
self.assertEqual(self.db.fetchall("SELECT * FROM provider_call_log"), [])
|
||||
|
||||
def test_disabled_observe_still_reraises_the_exact_exception(self):
|
||||
self.db.observability_enabled = False
|
||||
boom = RuntimeError("upstream exploded")
|
||||
|
||||
def fn():
|
||||
raise boom
|
||||
|
||||
with self.assertRaises(RuntimeError) as ctx:
|
||||
observability.observe(self.db, "eastmoney", "indices", fn)
|
||||
self.assertIs(ctx.exception, boom)
|
||||
self.assertEqual(self.db.fetchall("SELECT * FROM provider_call_log"), [])
|
||||
|
||||
def test_re_enabling_resumes_recording(self):
|
||||
self.db.observability_enabled = False
|
||||
observability.observe(self.db, "eastmoney", "indices", lambda: {"ok": True})
|
||||
self.assertEqual(self.db.fetchall("SELECT * FROM provider_call_log"), [])
|
||||
self.db.observability_enabled = True
|
||||
observability.observe(self.db, "eastmoney", "indices", lambda: {"ok": True})
|
||||
self.assertEqual(len(self.db.fetchall("SELECT * FROM provider_call_log")), 1)
|
||||
|
||||
|
||||
class SettingsToggleTests(unittest.TestCase):
|
||||
"""The kill switch follows the same env-var pattern as DATAHUB_SCHEDULER."""
|
||||
|
||||
def test_defaults_to_enabled(self):
|
||||
from datahub.settings import load_settings
|
||||
|
||||
settings = load_settings(env={})
|
||||
self.assertTrue(settings.observability_enabled)
|
||||
|
||||
def test_datahub_observability_zero_disables(self):
|
||||
from datahub.settings import load_settings
|
||||
|
||||
settings = load_settings(env={"DATAHUB_OBSERVABILITY": "0"})
|
||||
self.assertFalse(settings.observability_enabled)
|
||||
|
||||
def test_datahub_observability_off_disables(self):
|
||||
from datahub.settings import load_settings
|
||||
|
||||
settings = load_settings(env={"DATAHUB_OBSERVABILITY": "off"})
|
||||
self.assertFalse(settings.observability_enabled)
|
||||
|
||||
def test_datahub_observability_one_keeps_enabled(self):
|
||||
from datahub.settings import load_settings
|
||||
|
||||
settings = load_settings(env={"DATAHUB_OBSERVABILITY": "1"})
|
||||
self.assertTrue(settings.observability_enabled)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -1,82 +0,0 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import unittest
|
||||
|
||||
from datahub.pipeline import RetryError
|
||||
from tests.fixtures import TRADE_DATE
|
||||
from tests.test_pipeline import make_pipeline
|
||||
|
||||
|
||||
class PipelineObservabilityTests(unittest.TestCase):
|
||||
"""HEL-543: Tushare calls must keep writing the existing `src_calls`
|
||||
record unchanged, while also feeding the new cross-provider
|
||||
`provider_call_log` / `provider_health` side channel."""
|
||||
|
||||
def test_successful_fetch_logs_to_both_src_calls_and_provider_call_log(self) -> None:
|
||||
pipe, db = make_pipeline()
|
||||
pipe.ingest_reference(TRADE_DATE)
|
||||
result = pipe.run_dataset("daily", TRADE_DATE)
|
||||
self.assertEqual(result["state"], "published")
|
||||
|
||||
src_calls = db.fetchall("SELECT * FROM src_calls WHERE provider = 'tushare' AND endpoint = 'daily'")
|
||||
self.assertTrue(any(row["ok"] == 1 for row in src_calls))
|
||||
|
||||
log = db.fetchall(
|
||||
"SELECT * FROM provider_call_log WHERE provider = 'tushare' AND interface = 'daily'"
|
||||
)
|
||||
self.assertTrue(len(log) >= 1)
|
||||
self.assertEqual(log[-1]["status"], "ok")
|
||||
|
||||
health = db.fetchone(
|
||||
"SELECT * FROM provider_health WHERE provider = 'tushare' AND interface = 'daily'"
|
||||
)
|
||||
self.assertIsNotNone(health)
|
||||
self.assertEqual(health["state"], "ok")
|
||||
self.assertEqual(health["consec_failures"], 0)
|
||||
|
||||
def test_failed_fetch_logs_error_to_both_channels_and_still_raises(self) -> None:
|
||||
pipe, db = make_pipeline()
|
||||
pipe.ingest_reference(TRADE_DATE)
|
||||
|
||||
def boom(dataset, params):
|
||||
raise RuntimeError("tushare upstream 500")
|
||||
|
||||
pipe.adapter.fetch = boom # type: ignore[assignment]
|
||||
|
||||
with self.assertRaises(RetryError):
|
||||
pipe.run_dataset("daily", TRADE_DATE, attempts=1)
|
||||
|
||||
src_calls = db.fetchall("SELECT * FROM src_calls WHERE provider = 'tushare' AND endpoint = 'daily' AND ok = 0")
|
||||
self.assertTrue(len(src_calls) >= 1)
|
||||
self.assertIn("tushare upstream 500", src_calls[-1]["error"])
|
||||
|
||||
log = db.fetchall(
|
||||
"SELECT * FROM provider_call_log WHERE provider = 'tushare' AND interface = 'daily' AND status != 'ok'"
|
||||
)
|
||||
self.assertTrue(len(log) >= 1)
|
||||
self.assertIn("tushare upstream 500", log[-1]["error"])
|
||||
|
||||
health = db.fetchone(
|
||||
"SELECT * FROM provider_health WHERE provider = 'tushare' AND interface = 'daily'"
|
||||
)
|
||||
self.assertIsNotNone(health)
|
||||
self.assertNotEqual(health["state"], "ok")
|
||||
self.assertGreaterEqual(health["consec_failures"], 1)
|
||||
|
||||
def test_provider_call_log_is_purged_by_existing_cleanup_job(self) -> None:
|
||||
pipe, db = make_pipeline()
|
||||
pipe.ingest_reference(TRADE_DATE)
|
||||
pipe.run_dataset("daily", TRADE_DATE)
|
||||
self.assertTrue(db.fetchall("SELECT * FROM provider_call_log"))
|
||||
|
||||
# Force everything to look ancient so cleanup() sweeps it.
|
||||
db.execute("UPDATE provider_call_log SET created_at = '2000-01-01T00:00:00+08:00'")
|
||||
db.execute("UPDATE src_calls SET created_at = '2000-01-01T00:00:00+08:00'")
|
||||
db.execute("UPDATE job_runs SET started_at = '2000-01-01T00:00:00+08:00'")
|
||||
|
||||
pipe.cleanup()
|
||||
self.assertEqual(db.fetchall("SELECT * FROM provider_call_log"), [])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -1,163 +0,0 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import tempfile
|
||||
import unittest
|
||||
from contextlib import contextmanager
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch
|
||||
|
||||
from datahub.adapters.base import AdapterError
|
||||
from datahub.db import HubDB
|
||||
from datahub.realtime_serve import fetch_index_quotes, fetch_intraday, fetch_quotes
|
||||
|
||||
|
||||
class _WriteBreaksDB:
|
||||
"""Wraps a real HubDB but breaks only the write path, to prove the
|
||||
real serving path (reads/caches) is untouched by an observability
|
||||
failure while still exercising real fetch/cache code around it."""
|
||||
|
||||
def __init__(self, real: HubDB) -> None:
|
||||
self._real = real
|
||||
|
||||
def fetchall(self, sql, params=()):
|
||||
return self._real.fetchall(sql, params)
|
||||
|
||||
def fetchone(self, sql, params=()):
|
||||
return self._real.fetchone(sql, params)
|
||||
|
||||
def execute(self, sql, params=()):
|
||||
return self._real.execute(sql, params)
|
||||
|
||||
def executemany(self, sql, rows):
|
||||
return self._real.executemany(sql, rows)
|
||||
|
||||
@contextmanager
|
||||
def write(self):
|
||||
raise RuntimeError("db is not writable right now")
|
||||
yield None # pragma: no cover
|
||||
|
||||
|
||||
class RealtimeObservabilityTests(unittest.TestCase):
|
||||
def setUp(self) -> None:
|
||||
self.tmp = tempfile.TemporaryDirectory()
|
||||
self.db = HubDB(Path(self.tmp.name) / "hub.db")
|
||||
|
||||
def tearDown(self) -> None:
|
||||
self.tmp.cleanup()
|
||||
|
||||
def test_eastmoney_success_is_logged_without_changing_payload(self) -> None:
|
||||
rows = [
|
||||
{"ts_code": "000001.SH", "code": "000001", "name": "上证指数", "price": 3000.0,
|
||||
"previous_close": 2990.0, "quote_time_epoch": 0, "source": "eastmoney_push2"},
|
||||
{"ts_code": "399001.SZ", "code": "399001", "name": "深证成指", "price": 9000.0,
|
||||
"previous_close": 8990.0, "quote_time_epoch": 0, "source": "eastmoney_push2"},
|
||||
{"ts_code": "399006.SZ", "code": "399006", "name": "创业板指", "price": 1800.0,
|
||||
"previous_close": 1790.0, "quote_time_epoch": 0, "source": "eastmoney_push2"},
|
||||
]
|
||||
with patch("datahub.realtime_serve.EastmoneyAdapter") as mocked:
|
||||
mocked.return_value.fetch_indices.return_value = rows
|
||||
payload = fetch_index_quotes(self.db)
|
||||
self.assertEqual(payload["data"], rows)
|
||||
self.assertEqual(payload["meta"]["source"], "eastmoney:ulist")
|
||||
log = self.db.fetchall("SELECT * FROM provider_call_log WHERE provider = 'eastmoney'")
|
||||
self.assertEqual(len(log), 1)
|
||||
self.assertEqual(log[0]["interface"], "indices")
|
||||
self.assertEqual(log[0]["status"], "ok")
|
||||
health = self.db.fetchone(
|
||||
"SELECT * FROM provider_health WHERE provider = 'eastmoney' AND interface = 'indices'"
|
||||
)
|
||||
self.assertEqual(health["state"], "ok")
|
||||
|
||||
def test_eastmoney_failure_falls_back_to_tencent_and_logs_both(self) -> None:
|
||||
tencent_rows = [
|
||||
{"ts_code": "000001.SH", "code": "000001", "name": "上证指数", "price": 3000.0,
|
||||
"previous_close": 2990.0, "quote_time_epoch": 0, "source": "tencent_qt"},
|
||||
{"ts_code": "399001.SZ", "code": "399001", "name": "深证成指", "price": 9000.0,
|
||||
"previous_close": 8990.0, "quote_time_epoch": 0, "source": "tencent_qt"},
|
||||
{"ts_code": "399006.SZ", "code": "399006", "name": "创业板指", "price": 1800.0,
|
||||
"previous_close": 1790.0, "quote_time_epoch": 0, "source": "tencent_qt"},
|
||||
]
|
||||
with patch("datahub.realtime_serve.EastmoneyAdapter") as eastmoney, patch(
|
||||
"datahub.realtime_serve.TencentAdapter"
|
||||
) as tencent:
|
||||
eastmoney.return_value.fetch_indices.side_effect = AdapterError("Eastmoney returned 0/3 indices")
|
||||
tencent.return_value.fetch_indices.return_value = tencent_rows
|
||||
payload = fetch_index_quotes(self.db)
|
||||
self.assertEqual(payload["meta"]["source"], "tencent:qt")
|
||||
self.assertEqual(payload["data"], tencent_rows)
|
||||
east_log = self.db.fetchone("SELECT * FROM provider_call_log WHERE provider = 'eastmoney'")
|
||||
self.assertEqual(east_log["status"], "empty")
|
||||
tencent_log = self.db.fetchone("SELECT * FROM provider_call_log WHERE provider = 'tencent'")
|
||||
self.assertEqual(tencent_log["status"], "ok")
|
||||
|
||||
def test_observability_db_failure_never_breaks_a_real_successful_fetch(self) -> None:
|
||||
rows = [
|
||||
{"ts_code": "000001.SH", "price": 3000.0, "previous_close": 2990.0, "quote_time_epoch": 0},
|
||||
{"ts_code": "399001.SZ", "price": 9000.0, "previous_close": 8990.0, "quote_time_epoch": 0},
|
||||
{"ts_code": "399006.SZ", "price": 1800.0, "previous_close": 1790.0, "quote_time_epoch": 0},
|
||||
]
|
||||
broken = _WriteBreaksDB(self.db)
|
||||
with patch("datahub.realtime_serve.EastmoneyAdapter") as mocked:
|
||||
mocked.return_value.fetch_indices.return_value = rows
|
||||
payload = fetch_index_quotes(broken)
|
||||
self.assertEqual(payload["data"], rows)
|
||||
self.assertEqual(payload["meta"]["source"], "eastmoney:ulist")
|
||||
|
||||
def test_observability_db_failure_never_masks_a_real_source_outage(self) -> None:
|
||||
broken = _WriteBreaksDB(self.db)
|
||||
with patch("datahub.realtime_serve.EastmoneyAdapter") as eastmoney, patch(
|
||||
"datahub.realtime_serve.TencentAdapter"
|
||||
) as tencent:
|
||||
eastmoney.return_value.fetch_indices.side_effect = AdapterError("down")
|
||||
tencent.return_value.fetch_indices.side_effect = AdapterError("also down")
|
||||
with self.assertRaises(Exception):
|
||||
fetch_index_quotes(broken)
|
||||
|
||||
def test_named_quotes_records_both_providers_on_partial_merge(self) -> None:
|
||||
with patch("datahub.realtime_serve.EastmoneyAdapter") as eastmoney, patch(
|
||||
"datahub.realtime_serve.TencentAdapter"
|
||||
) as tencent:
|
||||
eastmoney.return_value.fetch_quotes.return_value = [
|
||||
{"ts_code": "000001.SZ", "close": 10, "pre_close": 9, "quote_date": "20260907"},
|
||||
]
|
||||
tencent.return_value.fetch_quotes.return_value = [
|
||||
{"ts_code": "000002.SZ", "close": 20, "pre_close": 19, "quote_date": "20260907"},
|
||||
]
|
||||
payload = fetch_quotes(self.db, ["000001.SZ", "000002.SZ"])
|
||||
self.assertEqual(payload["meta"]["complete"], True)
|
||||
east_log = self.db.fetchone(
|
||||
"SELECT * FROM provider_call_log WHERE provider = 'eastmoney' AND interface = 'named_quotes'"
|
||||
)
|
||||
self.assertEqual(east_log["status"], "ok")
|
||||
tencent_log = self.db.fetchone(
|
||||
"SELECT * FROM provider_call_log WHERE provider = 'tencent' AND interface = 'named_quotes'"
|
||||
)
|
||||
self.assertEqual(tencent_log["status"], "ok")
|
||||
|
||||
def test_intraday_success_is_logged_as_ok(self) -> None:
|
||||
payload_data = {
|
||||
"entity_type": "stock", "ts_code": "601318.SH", "trade_date": "2026-09-07",
|
||||
"previous_close": 55.8, "points": [{"date": "2026-09-07", "time": "09:30", "close": 55.9}],
|
||||
}
|
||||
with patch("datahub.realtime_serve.EastmoneyAdapter") as mocked:
|
||||
mocked.return_value.fetch_intraday.return_value = payload_data
|
||||
payload = fetch_intraday(self.db, "601318.SH")
|
||||
self.assertEqual(payload["data"], payload_data)
|
||||
log = self.db.fetchone(
|
||||
"SELECT * FROM provider_call_log WHERE provider = 'eastmoney' AND interface = 'intraday'"
|
||||
)
|
||||
self.assertEqual(log["status"], "ok")
|
||||
|
||||
def test_intraday_failure_is_logged_as_empty(self) -> None:
|
||||
with patch("datahub.realtime_serve.EastmoneyAdapter") as mocked:
|
||||
mocked.return_value.fetch_intraday.side_effect = AdapterError("No intraday chart data returned")
|
||||
with self.assertRaises(Exception):
|
||||
fetch_intraday(self.db, "000001.SZ")
|
||||
log = self.db.fetchone(
|
||||
"SELECT * FROM provider_call_log WHERE provider = 'eastmoney' AND interface = 'intraday'"
|
||||
)
|
||||
self.assertEqual(log["status"], "empty")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user