From e7615fb9e6c5fff828f38d7e1ff772108f4b3af1 Mon Sep 17 00:00:00 2001 From: shy3130 Date: Wed, 1 Jul 2026 23:02:37 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E6=89=A9=E5=B1=95=E6=95=B0=E6=8D=AE?= =?UTF-8?q?=E6=8B=89=E5=8F=96=E4=BF=AE=E5=A4=8D=E4=B8=8E=E8=BF=9E=E6=9D=BF?= =?UTF-8?q?=E6=A2=AF=E9=98=9FqueryKey=E4=BF=AE=E5=A4=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/app/api/ext_data.py | 24 ++ backend/app/services/ext_data.py | 15 +- backend/app/services/ext_pull.py | 191 ++++++++-- .../components/ext-data/ExtDataPullPanel.tsx | 342 ++++++++++++------ frontend/src/lib/api.ts | 1 + frontend/src/pages/LimitUpLadder.tsx | 6 +- 6 files changed, 426 insertions(+), 153 deletions(-) diff --git a/backend/app/api/ext_data.py b/backend/app/api/ext_data.py index 71368f2..a2b3ff2 100644 --- a/backend/app/api/ext_data.py +++ b/backend/app/api/ext_data.py @@ -575,6 +575,13 @@ def configure_pull(request: Request, config_id: str, body: PullConfigReq): # 刷新调度器 pull_scheduler.refresh(_data_dir(request)) + # 关闭定时拉取时清理残留的 next_run, 避免前端展示一个永不执行的"下次" + if not config.pull.enabled: + cleared = store.get(config_id) + if cleared and cleared.pull and cleared.pull.next_run: + cleared.pull.next_run = None + store.upsert(cleared) + return {"status": "ok", "pull": config.pull.to_dict()} @@ -630,8 +637,25 @@ async def run_pull(request: Request, config_id: str): try: n, d = await fetch_and_ingest(config, _data_dir(request)) _refresh_views(request) + # 写回执行状态, 让前端"上次执行"面板立即反映 + updated = store.get(config_id) + if updated and updated.pull: + from datetime import datetime, timezone + updated.pull.last_run = datetime.now(timezone.utc).isoformat() + updated.pull.last_status = "success" + updated.pull.last_message = f"{n} rows @ {d}" + updated.pull.last_rows = n + store.upsert(updated) return {"status": "ok", "rows": n, "date": d} except Exception as e: + # 失败也写回状态, 记录错误信息 + failed = store.get(config_id) + if failed and failed.pull: + from datetime import datetime, timezone + failed.pull.last_run = datetime.now(timezone.utc).isoformat() + failed.pull.last_status = "error" + failed.pull.last_message = str(e)[:200] + store.upsert(failed) raise HTTPException(400, f"拉取失败: {e}") from e diff --git a/backend/app/services/ext_data.py b/backend/app/services/ext_data.py index 63cd8be..7e7695d 100644 --- a/backend/app/services/ext_data.py +++ b/backend/app/services/ext_data.py @@ -38,6 +38,7 @@ class PullConfig: "url", "method", "headers", "body", "response_path", "field_map", "schedule_minutes", "enabled", "last_run", "last_status", "last_message", "last_rows", + "next_run", ) def __init__( @@ -54,6 +55,7 @@ class PullConfig: last_status: str | None = None, last_message: str | None = None, last_rows: int | None = None, + next_run: str | None = None, ) -> None: self.url = url self.method = method # GET | POST @@ -67,6 +69,7 @@ class PullConfig: self.last_status = last_status # "success" | "error" self.last_message = last_message self.last_rows = last_rows + self.next_run = next_run # 下次预计运行 (ISO, 调度器写入) def to_dict(self) -> dict: return { @@ -82,6 +85,7 @@ class PullConfig: "last_status": self.last_status, "last_message": self.last_message, "last_rows": self.last_rows, + "next_run": self.next_run, } @classmethod @@ -101,6 +105,7 @@ class PullConfig: last_status=d.get("last_status"), last_message=d.get("last_message"), last_rows=d.get("last_rows"), + next_run=d.get("next_run"), ) @@ -407,8 +412,10 @@ def write_ext_parquet( existing = pl.read_parquet(out_path) key = "symbol" if "symbol" in df.columns else df.columns[0] df = pl.concat([existing, df]).unique(subset=[key], keep="last") - except Exception: - pass + except Exception as e: + # schema 不一致 (列不同) 时 concat 失败 → 直接用新 df 覆盖。 + # 记日志而非静默吞掉, 便于排查"数据结构错乱"类问题。 + logger.warning("扩展表 %s 合并去重失败, 将覆盖写入: %s", config.id, e) else: # 时序: timeseries/ 下按日期分区 out_dir = cfg_dir / "timeseries" / f"date={snap}" @@ -421,8 +428,8 @@ def write_ext_parquet( existing = pl.read_parquet(out_path) key = "symbol" if "symbol" in df.columns else df.columns[0] df = pl.concat([existing, df]).unique(subset=[key], keep="last") - except Exception: - pass + except Exception as e: + logger.warning("扩展表 %s 合并去重失败, 将覆盖写入: %s", config.id, e) df = cast_df_to_schema(df, config.fields) df.write_parquet(out_path) diff --git a/backend/app/services/ext_pull.py b/backend/app/services/ext_pull.py index d15a638..5f04ea7 100644 --- a/backend/app/services/ext_pull.py +++ b/backend/app/services/ext_pull.py @@ -75,6 +75,19 @@ def _apply_field_map(rows: list[dict], field_map: dict[str, str]) -> list[dict]: # 拉取执行 # --------------------------------------------------------------------------- +def _apply_preset_flatten(config_id: str, rows: list[dict]) -> list[dict]: + """对内置预设 (概念/行业) 应用结构转换, 与 fetch_preset 保持一致。 + + 延迟导入避免与 ext_presets 形成循环依赖。 + 非预设 id 原样返回。 + """ + if config_id not in ("ext_gn_ths", "ext_hy_ths"): + return rows + from app.services.ext_presets import _flatten_concept_rows, _flatten_industry_rows + flatten = _flatten_concept_rows if config_id == "ext_gn_ths" else _flatten_industry_rows + return flatten(rows) + + async def fetch_and_ingest( config: ExtConfig, data_dir, @@ -111,6 +124,12 @@ async def fetch_and_ingest( if not rows: raise ValueError("提取到的行数为 0") + # 内置预设 (概念/行业): 应用结构转换, 让产出 schema 与分析页一致。 + # 否则 raw 接口列 (concepts/industries 数组、name) 会直接覆盖正确的 part.parquet, + # 导致分析页因找不到维度字段 (所属概念/所属同花顺行业) 而"数据消失"。 + # 见 ext_presets._flatten_* —— 手动拉取 / 定时拉取都必须走同一套转换。 + rows = _apply_preset_flatten(config.id, rows) + # 字段映射 rows = _apply_field_map(rows, pull.field_map) @@ -129,21 +148,46 @@ async def fetch_and_ingest( # --------------------------------------------------------------------------- class PullScheduler: - """后台调度器：为每个启用了 pull 的 ExtConfig 维护定时任务。""" + """后台调度器：为每个启用了 pull 的 ExtConfig 维护定时任务。 + + 线程安全说明: + refresh()/stop() 可能从主事件循环 (lifespan startup) 或同步路由的 + worker 线程 (configure_pull 是 def 而非 async def, FastAPI 丢进线程池) + 调用。worker 线程里没有 running loop, 直接 asyncio.create_task 会抛 + "no running event loop"。因此对 task 的增删一律通过 + call_soon_threadsafe 提交到主循环执行 —— 同一套代码两种调用场景都安全。 + """ def __init__(self) -> None: self._tasks: dict[str, asyncio.Task] = {} self._running = False - self._lock = threading.Lock() + self._loop: asyncio.AbstractEventLoop | None = None def start(self, data_dir) -> None: - """启动调度（在 lifespan startup 调用）。""" + """启动调度（在 lifespan startup 调用，主事件循环内）。""" self._running = True self._data_dir = data_dir + try: + self._loop = asyncio.get_running_loop() + except RuntimeError: + self._loop = None logger.info("PullScheduler started") + def _submit(self, fn, *args) -> None: + """把一个 callable 提交到主事件循环执行 (线程安全)。 + + startup 在主循环内调用时 fn 立即排队; worker 线程调用时跨线程排队。 + 两者都通过 call_soon_threadsafe, 保证 _tasks 字典的读写只在主循环里发生。 + """ + loop = self._loop + if loop is None or loop.is_closed(): + raise RuntimeError( + "PullScheduler: 事件循环不可用 (start() 未在事件循环中调用?)" + ) + loop.call_soon_threadsafe(fn, *args) + def stop(self) -> None: - """停止所有任务。""" + """停止所有任务 (从 shutdown 调用)。""" self._running = False for task in self._tasks.values(): task.cancel() @@ -151,47 +195,61 @@ class PullScheduler: logger.info("PullScheduler stopped") def refresh(self, data_dir) -> None: - """重新加载配置，更新调度任务（增/删/改）。""" + """重新加载配置，更新调度任务（增/删/改）。线程安全。""" self._data_dir = data_dir store = ExtConfigStore(data_dir) configs = store.load_all() active_ids: set[str] = set() + new_configs: list[ExtConfig] = [] for config in configs: if not config.pull or not config.pull.enabled or not config.pull.url: continue active_ids.add(config.id) if config.id not in self._tasks: - # 新增调度 - task = asyncio.create_task(self._run_loop(config)) - self._tasks[config.id] = task - logger.info("PullScheduler: scheduled %s (every %d min)", config.id, config.pull.schedule_minutes) + new_configs.append(config) - # 移除不再活跃的 - for cid in list(self._tasks): - if cid not in active_ids: - self._tasks[cid].cancel() - del self._tasks[cid] - logger.info("PullScheduler: removed %s", cid) + # 需要移除的 id (快照当前 task 字典的键, 避免遍历时改字典) + remove_ids = [cid for cid in list(self._tasks) if cid not in active_ids] + + # 所有对 _tasks 的修改都提交到主循环里执行, 保证线程安全 + def _apply() -> None: + for config in new_configs: + if config.id not in self._tasks: # 二次校验, 防重复 + self._tasks[config.id] = self._loop.create_task( + self._run_loop(config) + ) + logger.info( + "PullScheduler: scheduled %s (every %d min)", + config.id, config.pull.schedule_minutes, + ) + for cid in remove_ids: + task = self._tasks.pop(cid, None) + if task is not None: + task.cancel() + logger.info("PullScheduler: removed %s", cid) + + self._submit(_apply) async def _run_loop(self, config: ExtConfig) -> None: - """单个配置的定时拉取循环。""" + """单个配置的定时拉取循环。 + + 策略: 启用后立即执行一次, 之后按 interval 循环。 + 每次循环重读最新配置 (fresh), interval 取自 fresh.pull.schedule_minutes, + 这样用户中途修改间隔也能立即生效 (无需重启)。 + """ try: while self._running: - pull = config.pull - if not pull: - break - interval = max(pull.schedule_minutes * 60, 60) # 至少 60s - await asyncio.sleep(interval) - if not self._running: + # 每轮重读最新配置 — 用户可能修改了 url / interval / enabled + store = ExtConfigStore(self._data_dir) + fresh = store.get(config.id) + if not fresh or not fresh.pull or not fresh.pull.enabled: break + pull = fresh.pull + + # 先执行一次 (启用即拉取, 让用户立刻看到生效) try: - # 重新加载最新配置（用户可能中途修改） - store = ExtConfigStore(self._data_dir) - fresh = store.get(config.id) - if not fresh or not fresh.pull or not fresh.pull.enabled: - break n, d = await fetch_and_ingest(fresh, self._data_dir) fresh.pull.last_run = datetime.now(timezone.utc).isoformat() fresh.pull.last_status = "success" @@ -200,14 +258,79 @@ class PullScheduler: store.upsert(fresh) logger.info("PullScheduler: %s success, %d rows", config.id, n) except Exception as e: - store = ExtConfigStore(self._data_dir) - fresh = store.get(config.id) - if fresh and fresh.pull: - fresh.pull.last_run = datetime.now(timezone.utc).isoformat() - fresh.pull.last_status = "error" - fresh.pull.last_message = str(e)[:200] - store.upsert(fresh) + fresh2 = store.get(config.id) + if fresh2 and fresh2.pull: + fresh2.pull.last_run = datetime.now(timezone.utc).isoformat() + fresh2.pull.last_status = "error" + fresh2.pull.last_message = str(e)[:200] + store.upsert(fresh2) logger.warning("PullScheduler: %s error: %s", config.id, e) + + # 间隔取自最新配置 (每次重新读取, 修复改间隔不生效) + interval = max(pull.schedule_minutes * 60, 60) # 至少 60s + # 预告下次运行时间, 供前端展示 + next_dt = datetime.now(timezone.utc).timestamp() + interval + latest = store.get(config.id) + if latest and latest.pull: + latest.pull.next_run = datetime.fromtimestamp( + next_dt, tz=timezone.utc + ).isoformat() + store.upsert(latest) + + await asyncio.sleep(interval) + if not self._running: + break + except asyncio.CancelledError: + pass + + async def _run_loop(self, config: ExtConfig) -> None: + """单个配置的定时拉取循环。 + + 策略: 启用后立即执行一次, 之后按 interval 循环。 + 每次循环重读最新配置 (fresh), interval 取自 fresh.pull.schedule_minutes, + 这样用户中途修改间隔也能立即生效 (无需重启)。 + """ + try: + while self._running: + # 每轮重读最新配置 — 用户可能修改了 url / interval / enabled + store = ExtConfigStore(self._data_dir) + fresh = store.get(config.id) + if not fresh or not fresh.pull or not fresh.pull.enabled: + break + pull = fresh.pull + + # 先执行一次 (启用即拉取, 让用户立刻看到生效) + try: + n, d = await fetch_and_ingest(fresh, self._data_dir) + fresh.pull.last_run = datetime.now(timezone.utc).isoformat() + fresh.pull.last_status = "success" + fresh.pull.last_message = f"{n} rows @ {d}" + fresh.pull.last_rows = n + store.upsert(fresh) + logger.info("PullScheduler: %s success, %d rows", config.id, n) + except Exception as e: + fresh2 = store.get(config.id) + if fresh2 and fresh2.pull: + fresh2.pull.last_run = datetime.now(timezone.utc).isoformat() + fresh2.pull.last_status = "error" + fresh2.pull.last_message = str(e)[:200] + store.upsert(fresh2) + logger.warning("PullScheduler: %s error: %s", config.id, e) + + # 间隔取自最新配置 (每次重新读取, 修复改间隔不生效) + interval = max(pull.schedule_minutes * 60, 60) # 至少 60s + # 预告下次运行时间, 供前端展示 + next_dt = datetime.now(timezone.utc).timestamp() + interval + latest = store.get(config.id) + if latest and latest.pull: + latest.pull.next_run = datetime.fromtimestamp( + next_dt, tz=timezone.utc + ).isoformat() + store.upsert(latest) + + await asyncio.sleep(interval) + if not self._running: + break except asyncio.CancelledError: pass diff --git a/frontend/src/components/ext-data/ExtDataPullPanel.tsx b/frontend/src/components/ext-data/ExtDataPullPanel.tsx index 36ad379..6ef6350 100644 --- a/frontend/src/components/ext-data/ExtDataPullPanel.tsx +++ b/frontend/src/components/ext-data/ExtDataPullPanel.tsx @@ -1,6 +1,7 @@ import { useState } from 'react' -import { Loader2, Search, RefreshCw, Check } from 'lucide-react' +import { Loader2, Search, Check, Clock, Zap, Settings2, AlertCircle, CheckCircle2, Calendar } from 'lucide-react' import { api, type ExtDataConfig } from '@/lib/api' +import { toast } from '@/components/Toast' export function ExtDataPullPanel({ config, onSaved }: { config: ExtDataConfig @@ -22,157 +23,284 @@ export function ExtDataPullPanel({ config, onSaved }: { const [saving, setSaving] = useState(false) const [testing, setTesting] = useState(false) const [running, setRunning] = useState(false) + const [runResult, setRunResult] = useState<{ rows: number; date: string } | null>(null) const [testResult, setTestResult] = useState<{ total_rows: number; preview: Record[]; has_symbol: boolean } | null>(null) const [error, setError] = useState('') - const handleSave = () => { - let headers: Record | undefined - if (headerStr.trim()) { - try { headers = JSON.parse(headerStr) } - catch { setError('Headers 不是有效 JSON'); return } - } - let field_map: Record | undefined - if (fieldMapStr.trim()) { - try { field_map = JSON.parse(fieldMapStr) } - catch { setError('字段映射不是有效 JSON'); return } - } - setSaving(true); setError('') - api.extDataPullConfig(config.id, { + // 解析 JSON 输入, 失败时设置 error 并返回 null + const parseJson = (str: string, label: string): Record | undefined | null => { + if (!str.trim()) return undefined + try { return JSON.parse(str) } + catch { setError(`${label} 不是有效 JSON`); return null } + } + + // 构建保存 payload (复用当前编辑态), enabledOverride 用于开关自动保存 + const buildPayload = (enabledOverride?: boolean) => { + const headers = parseJson(headerStr, 'Headers') + if (headers === null) return null + const field_map = parseJson(fieldMapStr, '字段映射') + if (field_map === null) return null + return { url, method, headers, body: body || undefined, response_path: responsePath, field_map, - schedule_minutes: schedule, enabled, - }).then(() => onSaved()) + schedule_minutes: schedule, enabled: enabledOverride ?? enabled, + } + } + + const handleSave = (silent = false) => { + const payload = buildPayload() + if (!payload) return + setSaving(true); setError('') + api.extDataPullConfig(config.id, payload) + .then(() => { + onSaved() + if (!silent) toast('配置已保存', 'success') + }) .catch(e => setError(e.message || '保存失败')) .finally(() => setSaving(false)) } const handleTest = () => { setTesting(true); setError(''); setTestResult(null) - let headers: Record | undefined - if (headerStr.trim()) { - try { headers = JSON.parse(headerStr) } - catch { setError('Headers 不是有效 JSON'); setTesting(false); return } - } - let field_map: Record | undefined - if (fieldMapStr.trim()) { - try { field_map = JSON.parse(fieldMapStr) } - catch { setError('字段映射不是有效 JSON'); setTesting(false); return } - } - api.extDataPullConfig(config.id, { - url, method, headers, body: body || undefined, - response_path: responsePath, field_map, - schedule_minutes: schedule, enabled, - }).then(() => api.extDataPullTest(config.id)) + const payload = buildPayload() + if (!payload) { setTesting(false); return } + api.extDataPullConfig(config.id, payload) + .then(() => api.extDataPullTest(config.id)) .then(r => { setTestResult(r); onSaved() }) .catch(e => setError(e.message || '测试失败')) .finally(() => setTesting(false)) } const handleRun = () => { - setRunning(true); setError('') + setRunning(true); setError(''); setRunResult(null) api.extDataPullRun(config.id) - .then(() => onSaved()) + .then(r => { + setRunResult({ rows: r.rows, date: r.date }) + onSaved() + toast(`拉取成功 · ${r.rows} 行`, 'success') + }) .catch(e => setError(e.message || '执行失败')) .finally(() => setRunning(false)) } + // 开关 toggle: 自动保存全量配置 (切换 enabled), 后端 refresh 后立即首次拉取 + const [toggling, setToggling] = useState(false) + const handleToggle = (next: boolean) => { + if (toggling) return + if (next && !url.trim()) { + toast('请先填写拉取 URL', 'error') + return + } + const payload = buildPayload(next) + if (!payload) return + setToggling(true); setError(''); setEnabled(next) + api.extDataPullConfig(config.id, payload) + .then(() => { + onSaved() + toast(next ? '定时拉取已启用 · 立即执行首次拉取' : '定时拉取已关闭', 'success') + }) + .catch(e => { + setEnabled(!next) // 回滚 + setError(e.message || '切换失败') + }) + .finally(() => setToggling(false)) + } + + // 格式化时间显示 + const fmtTime = (iso: string | null | undefined) => { + if (!iso) return null + const d = new Date(iso) + if (isNaN(d.getTime())) return null + const mm = String(d.getMonth() + 1).padStart(2, '0') + const dd = String(d.getDate()).padStart(2, '0') + const hh = String(d.getHours()).padStart(2, '0') + const mi = String(d.getMinutes()).padStart(2, '0') + return `${mm}-${dd} ${hh}:${mi}` + } + return ( -
-
- - setUrl(e.target.value)} - placeholder="https://api.example.com/data" - className="flex-1 min-w-0 rounded-md border border-border bg-elevated px-2.5 py-1.5 text-[11px] font-mono text-foreground placeholder:text-muted/50" - /> -
+
+ {/* ===== 分区 ①: 请求配置 ===== */} +
+
+ + 请求配置 +
-
-
Headers (JSON，可选)
-