From 580743853a6ae59c9bfa3d3f3d84991b42938efa Mon Sep 17 00:00:00 2001 From: fish Date: Mon, 6 Jul 2026 12:04:46 +0800 Subject: [PATCH] =?UTF-8?q?serve=20=E8=AE=BE=E4=B8=BA=E5=8F=AA=E8=AF=BB?= =?UTF-8?q?=E6=A8=A1=E5=BC=8F=EF=BC=8C=E7=A6=81=E7=94=A8=E5=A4=96=E9=83=A8?= =?UTF-8?q?=E6=95=B0=E6=8D=AE=E6=8B=89=E5=8F=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit serve 端不再启动 pull_scheduler 和 financial_scheduler， 阻塞 ext_data 和 financials 的外部获取/同步 API 端点， 数据仅通过 local→serve 同步推送。 Co-Authored-By: Claude --- serve/backend/app/api/ext_data.py | 120 ++-------------------------- serve/backend/app/api/financials.py | 23 +----- serve/backend/app/main.py | 21 +---- 3 files changed, 9 insertions(+), 155 deletions(-) diff --git a/serve/backend/app/api/ext_data.py b/serve/backend/app/api/ext_data.py index a2b3ff2..3648257 100644 --- a/serve/backend/app/api/ext_data.py +++ b/serve/backend/app/api/ext_data.py @@ -26,7 +26,6 @@ from app.services.ext_data import ( write_ext_parquet, rows_to_parquet, ) -from app.services.ext_pull import fetch_and_ingest, pull_scheduler logger = logging.getLogger(__name__) router = APIRouter(prefix="/api/ext-data", tags=["ext-data"]) @@ -330,20 +329,8 @@ async def fetch_preset_data(request: Request, config_id: str): """手动触发内置预设 (概念/行业) 的数据拉取。 注意: 必须在 /{config_id}/... 动态路由之前声明, 否则 'presets' 会被当成 config_id。 - 与通用 pull/run 不同: 走 ext_presets 的结构转换 (接口的 concepts/industries - 数组 → 拼接成字符串), 保证 schema 与现有数据一致。 """ - from app.services.ext_presets import fetch_preset - - try: - n = await fetch_preset(config_id, _data_dir(request)) - except ValueError as e: - raise HTTPException(404, str(e)) from e - except Exception as e: - raise HTTPException(400, f"拉取失败: {e}") from e - - _refresh_views(request) - return {"status": "ok", "rows": n} + raise HTTPException(403, "serve 端为只读模式，数据仅通过 local→serve 同步") @router.post("") @@ -548,115 +535,20 @@ def ingest_data(request: Request, config_id: str, body: IngestReq): @router.put("/{config_id}/pull") def configure_pull(request: Request, config_id: str, body: PullConfigReq): - """配置（或更新）定时拉取。""" - store = _store(request) - config = store.get(config_id) - if not config: - raise HTTPException(404, f"配置 '{config_id}' 不存在") - - # 保留历史状态字段 - old_pull = config.pull - config.pull = PullConfig( - url=body.url, - method=body.method, - headers=body.headers, - body=body.body, - response_path=body.response_path, - field_map=body.field_map, - schedule_minutes=body.schedule_minutes, - enabled=body.enabled, - last_run=old_pull.last_run if old_pull else None, - last_status=old_pull.last_status if old_pull else None, - last_message=old_pull.last_message if old_pull else None, - last_rows=old_pull.last_rows if old_pull else None, - ) - store.upsert(config) - - # 刷新调度器 - 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()} + """配置（或更新）定时拉取 — serve 端为只读模式，已禁用。""" + raise HTTPException(403, "serve 端为只读模式，数据仅通过 local→serve 同步") @router.post("/{config_id}/pull/test") async def test_pull(request: Request, config_id: str): - """测试拉取：请求外部 API 并返回预览数据，不写入。""" - store = _store(request) - config = store.get(config_id) - if not config: - raise HTTPException(404, f"配置 '{config_id}' 不存在") - if not config.pull or not config.pull.url: - raise HTTPException(400, "拉取未配置或 URL 为空") - - # 临时构建一个带新配置的 config 用于测试 - from app.services.ext_pull import _extract_rows, _apply_field_map - import httpx - - pull = config.pull - try: - async with httpx.AsyncClient(timeout=30) as client: - headers = pull.headers or {} - kwargs: dict = {"headers": headers} - if pull.method.upper() == "POST" and pull.body: - kwargs["content"] = pull.body - if "content-type" not in {k.lower() for k in headers}: - kwargs["headers"]["Content-Type"] = "application/json" - resp = await client.request(pull.method.upper(), pull.url, **kwargs) - resp.raise_for_status() - data = resp.json() - - rows = _extract_rows(data, pull.response_path) - preview = _apply_field_map(rows[:5], pull.field_map) - return { - "status": "ok", - "total_rows": len(rows), - "preview": preview, - "has_symbol": bool(rows and "symbol" in rows[0]), - } - except Exception as e: - raise HTTPException(400, f"测试失败: {e}") from e + """测试拉取 — serve 端为只读模式，已禁用。""" + raise HTTPException(403, "serve 端为只读模式，数据仅通过 local→serve 同步") @router.post("/{config_id}/pull/run") async def run_pull(request: Request, config_id: str): """手动触发一次拉取并写入。""" - store = _store(request) - config = store.get(config_id) - if not config: - raise HTTPException(404, f"配置 '{config_id}' 不存在") - if not config.pull or not config.pull.url: - raise HTTPException(400, "拉取未配置或 URL 为空") - - 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 + raise HTTPException(403, "serve 端为只读模式，数据仅通过 local→serve 同步") # --------------------------------------------------------------------------- diff --git a/serve/backend/app/api/financials.py b/serve/backend/app/api/financials.py index ac39676..ba884ec 100644 --- a/serve/backend/app/api/financials.py +++ b/serve/backend/app/api/financials.py @@ -98,27 +98,8 @@ 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 - 同步在后台线程执行,全量同步需数分钟。本接口立即返回 started 状态, - 前端通过轮询 GET /status 的 syncing 字段观察进度。 - """ - capset = request.app.state.capabilities - capset.require(Cap.FINANCIAL) - - valid_tables = {"metrics", "income", "balance_sheet", "cash_flow", "all"} - if table not in valid_tables: - raise HTTPException(400, f"invalid table: {table}, expected one of {valid_tables}") - - fs = getattr(request.app.state, "financial_scheduler", None) - if not fs: - return {"status": "error", "message": "FinancialScheduler not available"} - - target = None if table == "all" else table - result = fs.trigger(target) - - return {"status": "ok", "synced": result} + """手动触发同步 — serve 端为只读模式，已禁用。""" + raise HTTPException(403, "serve 端为只读模式，数据仅通过 local→serve 同步") class AnalyzeRequest(BaseModel): diff --git a/serve/backend/app/main.py b/serve/backend/app/main.py index a72b325..a784cf5 100644 --- a/serve/backend/app/main.py +++ b/serve/backend/app/main.py @@ -63,26 +63,13 @@ async def lifespan(app: FastAPI): logger.warning("scheduler not started: %s", e) app.state.scheduler = None - # 扩展数据定时拉取 - from app.services.ext_pull import pull_scheduler - pull_scheduler.start(store.data_dir) - pull_scheduler.refresh(store.data_dir) - app.state.pull_scheduler = pull_scheduler - - # 内置扩展表 (概念/行业): 只创建 config (含拉取配置), 不自动拉数据 - # 数据获取由用户在概念/行业页点「获取数据」手动触发 (POST /api/ext-data/presets/{id}/fetch) + # 内置扩展表 (概念/行业): 只创建 config 供 sync 解压时用, 不拉数据。 try: from app.services.ext_presets import ensure_builtin_presets await ensure_builtin_presets(store.data_dir) except Exception as e: # noqa: BLE001 logger.warning("内置扩展表初始化失败 (不影响启动): %s", e) - # 财务数据 (需 Expert 套餐): 仅初始化调度器供 /api/financials/sync/* 手动同步, - # 不启动自动调度——用户在「财务分析」页点「同步」手动拉取。 - from app.services.financial_sync import financial_scheduler - financial_scheduler.start(store.data_dir, capset) - app.state.financial_scheduler = financial_scheduler - # 策略引擎 from app.strategy.engine import StrategyEngine from app.services.screener import ScreenerService @@ -105,12 +92,6 @@ async def lifespan(app: FastAPI): if app.state.scheduler: app.state.scheduler.shutdown(wait=False) - ps = getattr(app.state, "pull_scheduler", None) - if ps: - ps.stop() - fsc = getattr(app.state, "financial_scheduler", None) - if fsc: - fsc.stop() logger.info("shutdown")