diff --git a/backend/app/api/financials.py b/backend/app/api/financials.py index 2cdf433..8e6427d 100644 --- a/backend/app/api/financials.py +++ b/backend/app/api/financials.py @@ -109,7 +109,12 @@ def get_cash_flow(request: Request, symbol: str | None = None): @router.post("/sync/{table}") def sync_table(request: Request, table: str): - """手动触发同步。table: metrics / income / balance_sheet / cash_flow / all""" + """手动触发同步(立即返回,后台异步执行)。 + + table: metrics / income / balance_sheet / cash_flow / all + 同步在后台线程执行,全量同步需数分钟。本接口立即返回 started 状态, + 前端通过轮询 GET /status 的 syncing 字段观察进度。 + """ capset = request.app.state.capabilities capset.require(Cap.FINANCIAL) @@ -122,6 +127,6 @@ def sync_table(request: Request, table: str): return {"status": "error", "message": "FinancialScheduler not available"} target = None if table == "all" else table - result = fs.run_now(target) + result = fs.trigger(target) return {"status": "ok", "synced": result} diff --git a/backend/app/services/financial_sync.py b/backend/app/services/financial_sync.py index d8f900b..2f9fe14 100644 --- a/backend/app/services/financial_sync.py +++ b/backend/app/services/financial_sync.py @@ -243,8 +243,39 @@ class FinancialScheduler: except asyncio.CancelledError: pass + def _run_body(self, table: str | None) -> dict[str, int]: + """同步逻辑本体(不加锁,假设调用方已持有 _is_syncing)。 + + table=None 同步全部 4 张表;否则只同步指定表。 + 每张表完成立即更新 last_sync,让前端轮询 /status 能看到进度递增。 + """ + if table: + fn = { + "metrics": sync_metrics, + "income": sync_income, + "balance_sheet": sync_balance_sheet, + "cash_flow": sync_cash_flow, + }.get(table) + if not fn: + return {} + rows = fn(self._data_dir, self._capset) + self._last_sync[table] = datetime.now(timezone.utc).isoformat() + return {table: rows} + # 全部同步 + symbols = _get_symbols(self._data_dir) + result: dict[str, int] = {} + for t in FINANCIAL_TABLES: + result[t] = _sync_table(t, symbols, self._data_dir, self._capset, latest_only=True) + self._last_sync[t] = datetime.now(timezone.utc).isoformat() + _refresh_financials_views(self._data_dir) + return result + def run_now(self, table: str | None = None) -> dict[str, int]: - """手动触发同步。table=None 同步全部。 + """同步执行一次同步(阻塞调用线程)。 + + ⚠ 全量同步需数分钟,务必在后台线程调用,不要直接在 HTTP 请求线程里阻塞, + 否则请求会长时间 pending 直至被浏览器/代理超时掐断(表现为"点击无反应")。 + HTTP 接口应调用 trigger() 立即返回,再让前端轮询 /status.syncing 看进度。 用 _is_syncing 标志防并发:若已有同步在进行,本次直接跳过, 避免重复请求拖慢服务端 / 触发上游限流。 @@ -257,32 +288,46 @@ class FinancialScheduler: return {"_skipped": 1} self._is_syncing = True try: - if table: - fn = { - "metrics": sync_metrics, - "income": sync_income, - "balance_sheet": sync_balance_sheet, - "cash_flow": sync_cash_flow, - }.get(table) - if not fn: - return {} - rows = fn(self._data_dir, self._capset) - self._last_sync[table] = datetime.now(timezone.utc).isoformat() - return {table: rows} - else: - # 全部同步: 逐表执行, 每张完成立即更新 last_sync, - # 让前端轮询 /status 能看到进度递增 (而非等全部完成才一次性更新)。 - symbols = _get_symbols(self._data_dir) - result: dict[str, int] = {} - for t in FINANCIAL_TABLES: - result[t] = _sync_table(t, symbols, self._data_dir, self._capset, latest_only=True) - self._last_sync[t] = datetime.now(timezone.utc).isoformat() - _refresh_financials_views(self._data_dir) - return result + return self._run_body(table) finally: with self._lock: self._is_syncing = False + def trigger(self, table: str | None = None) -> dict[str, int]: + """触发一次同步(非阻塞,立即返回)。 + + 在后台线程执行同步体,HTTP 请求无需等待。 + 返回 {"started": True/False}: + - False = 能力不足或已有同步在进行(被防并发跳过) + - True = 已在后台开始,前端应轮询 /status.syncing 观察进度 + + ⚠ _is_syncing 在此处置 True(持锁),确保 trigger 返回时前端轮询 + /status 已能看到 syncing=True,无竞态窗口;同时防止快速重复点击 + 启动多个后台线程。后台线程复用 _run_body 执行真正的同步逻辑。 + """ + if not self._capset or not self._capset.has(Cap.FINANCIAL): + return {"started": False, "reason": "no FINANCIAL capability"} + with self._lock: + if self._is_syncing: + logger.info("financial sync trigger skipped: already running") + return {"started": False, "reason": "already running"} + # 持锁置位:保证 trigger 返回前 syncing 已为 True + self._is_syncing = True + + def _bg() -> None: + try: + self._run_body(table) + except Exception as e: # noqa: BLE001 + logger.exception("background financial sync failed: %s", e) + finally: + with self._lock: + self._is_syncing = False + + t = threading.Thread(target=_bg, name="financial-sync", daemon=True) + t.start() + logger.info("financial sync triggered in background: table=%s", table or "all") + return {"started": True} + @property def is_syncing(self) -> bool: """手动同步是否正在进行(供 /status 返回,前端据此显示"同步中")。""" diff --git a/frontend/src/lib/api.ts b/frontend/src/lib/api.ts index 298aa4e..f778756 100644 --- a/frontend/src/lib/api.ts +++ b/frontend/src/lib/api.ts @@ -1166,8 +1166,9 @@ export const api = { `/api/financials/cash-flow${symbol ? `?symbol=${encodeURIComponent(symbol)}` : ''}`, ), + /** 触发财务数据同步(后台异步执行,接口立即返回 started 状态) */ financialSync: (table: string) => - request<{ status: string; synced: Record }>( + request<{ status: string; synced: { started: boolean; reason?: string } }>( `/api/financials/sync/${table}`, { method: 'POST' }, ), diff --git a/frontend/src/pages/Financials.tsx b/frontend/src/pages/Financials.tsx index 9f655e7..c7e1cfc 100644 --- a/frontend/src/pages/Financials.tsx +++ b/frontend/src/pages/Financials.tsx @@ -1,4 +1,4 @@ -import { useState, useRef } from 'react' +import { useState, useEffect } from 'react' import { RefreshCw, Lock, Loader2, X, Search, FileText, Database, Clock, CheckCircle2, Hourglass } from 'lucide-react' import { PageHeader } from '@/components/PageHeader' import { EmptyState } from '@/components/EmptyState' @@ -7,6 +7,7 @@ import { useFinancialStatus, useFinancialSync } from '@/lib/useFinancials' import { StockFinancialSearch } from '@/components/financials/StockFinancialSearch' import { StockFinancialDetail } from '@/components/financials/StockFinancialDetail' import { fmtBigNum } from '@/lib/format' +import { toast } from '@/components/Toast' const TABLE_LABELS: Record = { metrics: '核心指标', @@ -27,13 +28,24 @@ export function Financials() { const hasFinancial = caps?.capabilities?.['financial'] != null const { data: status, isLoading } = useFinancialStatus() const syncMut = useFinancialSync() - // 同步状态: 服务端 syncing 真值优先, 兜底本地 mutation pending + // 同步进行中 = 服务端真值(status.syncing)或本地乐观态(请求已发出待确认)。 + // 乐观窗口:点击后到 invalidate 触发的 refetch 返回之间,status.syncing 暂为 false, + // 用 syncMut.isPending 覆盖,让按钮立即置灰、避免重复点击。 + // 后端 trigger() 返回时 syncing 已为 true,refetch 到达后 status.syncing 接管。 const syncing = (status?.syncing ?? false) || syncMut.isPending // 本次同步开始时间戳(ms): 用于判断每张表的 last_sync 是否属于本次同步 // (后端每张表完成即更新 last_sync, 前端轮询时对比时间戳得到精确进度) - const syncStartedAtRef = useRef(null) + const [syncStartedAt, setSyncStartedAt] = useState(null) // 单表同步时记录表名 (null = 全量同步), 用于区分卡片状态 - const syncSingleTableRef = useRef(null) + const [syncSingleTable, setSyncSingleTable] = useState(null) + // 同步自然结束(服务端 syncing 由 true→false):清空本次同步记录。 + // 这是可靠的收尾时机 —— 不依赖 mutation 的 onSettled(它现在瞬间触发,会误清)。 + useEffect(() => { + if (!syncing && syncStartedAt !== null) { + setSyncStartedAt(null) + setSyncSingleTable(null) + } + }, [syncing, syncStartedAt]) // 选中的个股(模糊搜索结果);null 时显示搜索引导 const [selected, setSelected] = useState<{ symbol: string; name: string } | null>(null) @@ -57,15 +69,27 @@ export function Financials() { } const handleSync = (table: string) => { - // 防重复点击:syncing 中不再触发(后端 run_now 也有 is_syncing 兜底) + // 防重复点击:syncing 中不再触发(后端 trigger 也有 _is_syncing 兜底) if (syncing) return // 记录开始时间: 全量同步判断所有 4 张表, 单表同步只判断这一张 - syncStartedAtRef.current = Date.now() - syncSingleTableRef.current = table === 'all' ? null : table + setSyncStartedAt(Date.now()) + setSyncSingleTable(table === 'all' ? null : table) syncMut.mutate(table, { - onSettled: () => { - syncStartedAtRef.current = null - syncSingleTableRef.current = null + onSuccess: (r) => { + // 后端 trigger 立即返回 started 状态;若被防并发跳过(已有同步在进行), + // 给用户明确反馈,并清空本次误设的记录。 + if (!r.synced?.started) { + if (r.synced?.reason === 'already running') { + toast('财务数据正在同步中,请稍候', 'success') + } + setSyncStartedAt(null) + setSyncSingleTable(null) + } + }, + onError: () => { + // 请求失败:清空本次记录(request 已弹错误 toast) + setSyncStartedAt(null) + setSyncSingleTable(null) }, }) } @@ -74,8 +98,6 @@ export function Financials() { const available = status?.available ?? false const lastSync = status?.last_sync ?? {} // 本次同步进度: 仅当 syncStartedAt 存在且 syncing 时, 按 last_sync 时间戳判断 - const syncStartedAt = syncStartedAtRef.current - const syncSingleTable = syncSingleTableRef.current const isFullSync = syncing && syncStartedAt && !syncSingleTable // 全量同步 const isSingleSync = syncing && syncStartedAt && !!syncSingleTable // 单表同步 const TABLE_ORDER = ['metrics', 'income', 'balance_sheet', 'cash_flow'] as const