FinDataHub
cb11a3d445
## 现象
装机后测受保护端点，文件 key → 401、乱码 key → 401、无 key → 401。
全 401 说明不是「没鉴权」，而是**鉴权用的 key 与客户端手里的不是同一个**。
## 根因（与 T6 同族，又一处双入口分叉）
app/main.py → _load_or_create_api_key() 读持久化文件
Scripts/bin/newsflash_backend.py → secrets.token_urlsafe(32) 现生成随机值
生产走后者，每次启动换一把钥匙。客户端读 ~/Library/Application Support/
findatahub/.api_key 去请求，永远对不上；两边都不报错，只表现为「点了没反应」。
## 修复
把加载逻辑收进 app/dependencies.py::load_or_create_api_key()，两个入口共用：
- 文件已存在 → 复用，**不覆盖**（覆盖会让客户端手里的旧 key 立即失效）
- 文件存在但内容为空 → 补生成（空 key 会让 compare_digest("", "") 恒真）
路径用 platformdirs.user_config_dir("findatahub")，macOS 上即
~/Library/Application Support/findatahub，与 Swift 端读取位置一致。
## 部署位实测
文件 key → HTTP 200 （修复前 401）
乱码 key → HTTP 401
无 key → HTTP 401
## 防回退锁
test_app_factory.py::TestApiKeySource 3 条：共享加载函数存在、复用时不被覆盖/
空内容补生成、两个入口源码都必须出现 load_or_create_api_key 且都不得出现
secrets.token_urlsafe。
backend 443 → 446。
## 这一族的教训
T6 收敛了「装配」，但 lifespan 里还藏着同类分叉。凡是「两个入口都该有且必须一致」
的逻辑，都该像 key 加载这样收成单一函数，而不是指望两边同步。
443 lines
17 KiB
Python
443 lines
17 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
FinDataHub 快讯后端 — FastAPI V2 (来自 finance-terminal/kuaixun-service)
|
|
支持 API + WebSocket + 内置前端静态文件
|
|
"""
|
|
import os, sys, time, json, secrets
|
|
from pathlib import Path
|
|
|
|
from common import KXS_HOME as KX_SERVICE, BUNDLED_KXS_HOME, BIN_DIR, BACKEND_PORT, MYSQL_CFG, setup_kuaixun_paths
|
|
setup_kuaixun_paths()
|
|
_KX_SERVICE = BUNDLED_KXS_HOME or KX_SERVICE
|
|
sys.path.insert(0, _KX_SERVICE)
|
|
|
|
# 前端静态文件路径：依次尝试 bundled -> 源码树 dev 路径。
|
|
# 不要写死绝对路径——项目重组后旧路径会失效，改为由本文件位置推导。
|
|
_FRONTEND_CANDIDATES = [
|
|
Path(BIN_DIR).parent / "kuaixun-frontend" / "dist", # 打包后：Resources/
|
|
Path(BIN_DIR).parent.parent / "Services" / "kuaixun-frontend" / "dist", # 源码树
|
|
]
|
|
FRONTEND_DIR = str(next(
|
|
(p for p in _FRONTEND_CANDIDATES if p.is_dir()),
|
|
_FRONTEND_CANDIDATES[0],
|
|
))
|
|
|
|
import uvicorn
|
|
from contextlib import asynccontextmanager
|
|
from fastapi import FastAPI, Request, Response, Depends
|
|
from fastapi.staticfiles import StaticFiles
|
|
from starlette.responses import FileResponse, JSONResponse
|
|
from starlette.middleware.base import BaseHTTPMiddleware
|
|
import threading
|
|
import signal
|
|
import logging
|
|
|
|
# 连接池
|
|
from db_pool import get_conn, put_conn
|
|
|
|
try:
|
|
import pymysql
|
|
HAS_PYMYSQL = True
|
|
except ImportError:
|
|
HAS_PYMYSQL = False
|
|
|
|
# ── 结构化日志 ──
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
|
|
datefmt="%Y-%m-%d %H:%M:%S",
|
|
)
|
|
logger = logging.getLogger("findatahub")
|
|
|
|
import app.dependencies as deps
|
|
from app.dependencies import require_auth
|
|
# 基础路由由 app/factory.py 统一注册，本文件不再直接 import 各 router——
|
|
# 否则新增路由时两处都要改，且漏改只会表现为某个端点 404。
|
|
|
|
# ── 优雅关闭 ──
|
|
@asynccontextmanager
|
|
async def lifespan(app):
|
|
# 确保 API_KEY 非空，防止 compare_digest("", "") 鉴权绕过。
|
|
# 必须读**持久化**的 key（与客户端共用的 ~/Library/Application Support/findatahub/.api_key），
|
|
# 不能现生成随机值——历史实现就是这样，导致客户端拿文件里的 key 请求受保护端点恒 401，
|
|
# 两边都不报错、只表现为「点了没反应」。2026-10-27 实测 /api/v1/admin/keys 401。
|
|
if not deps.API_KEY:
|
|
deps.API_KEY = deps.load_or_create_api_key()
|
|
logger.info("API_KEY 已从持久化文件加载")
|
|
logger.info("FinDataHub 后端启动 (port=%s)", os.getenv("BACKEND_PORT", BACKEND_PORT))
|
|
yield
|
|
logger.info("FinDataHub 后端关闭中...")
|
|
_response_cache.clear()
|
|
|
|
# ⚠️ 本文件**不再自建 FastAPI 实例**。
|
|
# 装配（中间件顺序 + 基础路由）统一由 `app/factory.py::create_app` 提供，
|
|
# 与 `app/main.py` 共用同一份定义——两处各写一遍的代价是：只改 app/main.py 时
|
|
# 生产完全空转且不报错（2026-10-27 实测 CORS 预检 405、/health 只有两字段）。
|
|
# app 在下方中间件类定义之后创建（create_app 需要类对象）。
|
|
from app.factory import create_app # noqa: E402 — 必须在 setup_kuaixun_paths 之后
|
|
|
|
# ── 响应缓存中间件：对慢变更 GET 接口做 TTL 缓存，减少 DB 查询 ──
|
|
_CACHE_TTL = {
|
|
"/api/v1/news/tags": 300, # 标签列表 5 分钟
|
|
"/api/v1/news/heat": 60, # 热度分布 1 分钟
|
|
"/api/v1/news/stats": 30, # 统计数据 30 秒
|
|
"/api/v1/news/summary": 300, # 当日总结 5 分钟
|
|
"/api/v1/sources": 600, # 来源列表 10 分钟
|
|
"/api/v1/analysis/overview": 60, # 总览 1 分钟
|
|
"/api/v1/analysis/trend": 300, # 趋势 5 分钟
|
|
"/api/v1/analysis/hourly": 600, # 24小时分布 10 分钟
|
|
"/api/v1/analysis/top_tags": 300, # 标签排行 5 分钟
|
|
"/api/v1/analysis/top_sources": 300,# 来源排行 5 分钟
|
|
"/api/v1/analysis/importance": 120, # 热度分布 2 分钟
|
|
}
|
|
_response_cache: dict[str, tuple[float, bytes, str]] = {} # path → (expire_ts, body, content_type)
|
|
|
|
|
|
class ResponseCacheMiddleware(BaseHTTPMiddleware):
|
|
"""对指定 GET 接口缓存 JSON 响应，减少重复 DB 查询"""
|
|
|
|
async def dispatch(self, request: Request, call_next):
|
|
path = request.url.path
|
|
method = request.method
|
|
|
|
if method != "GET" or path not in _CACHE_TTL:
|
|
return await call_next(request)
|
|
|
|
# 缓存 key 包含查询字符串，避免不同参数的响应互相覆盖
|
|
cache_key = f"{path}?{request.url.query}" if request.url.query else path
|
|
|
|
# 检查缓存
|
|
cached = _response_cache.get(cache_key)
|
|
if cached:
|
|
expire_ts, body, content_type = cached
|
|
if time.time() < expire_ts:
|
|
return Response(content=body, media_type=content_type)
|
|
else:
|
|
_response_cache.pop(cache_key, None)
|
|
|
|
# 执行请求
|
|
response = await call_next(request)
|
|
|
|
# 仅缓存 200 响应
|
|
if response.status_code == 200:
|
|
ttl = _CACHE_TTL[path]
|
|
body_bytes = b""
|
|
async for chunk in response.body_iterator:
|
|
body_bytes += chunk
|
|
_response_cache[cache_key] = (time.time() + ttl, body_bytes, "application/json")
|
|
return Response(content=body_bytes, media_type="application/json")
|
|
|
|
return response
|
|
|
|
|
|
# ── 请求日志中间件 ──
|
|
class RequestLogMiddleware(BaseHTTPMiddleware):
|
|
"""慢请求告警 + 请求日志"""
|
|
|
|
async def dispatch(self, request: Request, call_next):
|
|
t0 = time.time()
|
|
response = await call_next(request)
|
|
elapsed = (time.time() - t0) * 1000
|
|
path = request.url.path
|
|
if elapsed > 3000:
|
|
logger.warning("慢请求 %s %s → %.0fms", request.method, path, elapsed)
|
|
elif path.startswith("/api/"):
|
|
logger.info("%s %s → %d (%.0fms)", request.method, path, response.status_code, elapsed)
|
|
return response
|
|
|
|
|
|
# ══════════════════════════════════════════════════════════
|
|
# 装配（必须在中间件类定义之后：create_app 需要拿到类对象）
|
|
#
|
|
# 中间件顺序与基础路由**只在 app/factory.py 定义一次**，本文件不得自行
|
|
# apply_* 或 include_router。历史教训：本文件曾自建 FastAPI 并自己排中间件，
|
|
# 而 CORS/安全头只加在 app/main.py，两者互不相干 —— 生产跨源请求全部失败
|
|
# 且不报错（2026-10-27 实测 CORS 预检 405）。
|
|
# ══════════════════════════════════════════════════════════
|
|
app = create_app(
|
|
title="FinDataHub 快讯",
|
|
service_name="newsflash_backend",
|
|
lifespan=lifespan,
|
|
# 顺序即层序（先传入的更内层）：缓存 → 请求日志
|
|
inner_middlewares=(ResponseCacheMiddleware, RequestLogMiddleware),
|
|
)
|
|
|
|
# 健康检查（在 catch-all 之前注册）
|
|
#
|
|
# 载荷构建走 app/health.py 的共享实现，不要在此另写一份。
|
|
# 本文件是**生产实际入口**（不 import app.main，但走同一个 create_app 装配），
|
|
# 载荷实现早已收敛到 app/health.py。历史教训：此前两处各写一份，导致
|
|
# app/main.py 已返回 db 状态与 timebase_drift_seconds，而部署位 /health
|
|
# 仍只有 {status, time} 两个字段。
|
|
@app.get("/health")
|
|
def health():
|
|
from app.health import build_health_payload
|
|
return build_health_payload()
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════
|
|
# 分析 API
|
|
# ═══════════════════════════════════════════════════════════
|
|
|
|
def _query_all(sql, params=None):
|
|
"""执行查询，返回字典列表"""
|
|
if not HAS_PYMYSQL:
|
|
return []
|
|
conn = None
|
|
cur = None
|
|
try:
|
|
conn = get_conn()
|
|
cur = conn.cursor(pymysql.cursors.DictCursor)
|
|
cur.execute(sql, params or ())
|
|
return cur.fetchall()
|
|
except Exception as e:
|
|
logger.error(f"[sql error] {e}")
|
|
return []
|
|
finally:
|
|
# 必须无条件归还连接：异常路径若不归还，池容量（8）会被逐个耗尽，
|
|
# 之后所有请求都阻塞在 _pool.get() 上，后端表现为永久挂死。
|
|
if cur is not None:
|
|
try:
|
|
cur.close()
|
|
except Exception:
|
|
pass
|
|
if conn is not None:
|
|
put_conn(conn)
|
|
|
|
|
|
def _query_one(sql, params=None):
|
|
"""执行查询，返回第一行"""
|
|
rows = _query_all(sql, params)
|
|
return rows[0] if rows else None
|
|
|
|
|
|
@app.get("/api/v1/analysis/overview")
|
|
def analysis_overview():
|
|
"""总览统计：今日/总条数/来源数/活跃标签数"""
|
|
today_row = _query_one("""
|
|
SELECT COUNT(*) as cnt FROM news_flash
|
|
WHERE published_at > 0
|
|
AND published_at >= UNIX_TIMESTAMP(CURDATE()) * 1000
|
|
""")
|
|
total_row = _query_one("SELECT COUNT(*) as cnt FROM news_flash WHERE published_at > 0")
|
|
sources_row = _query_one("SELECT COUNT(*) as cnt FROM news_sources WHERE is_active = 1")
|
|
tags_row = _query_one("""
|
|
SELECT COUNT(DISTINCT tag) as cnt FROM (
|
|
SELECT JSON_UNQUOTE(JSON_EXTRACT(tags, CONCAT('$[', n, ']'))) as tag
|
|
FROM news_flash
|
|
JOIN (SELECT 0 as n UNION SELECT 1 UNION SELECT 2 UNION SELECT 3 UNION SELECT 4) nums
|
|
WHERE JSON_LENGTH(tags) > n AND published_at > 0
|
|
) t WHERE tag IS NOT NULL AND tag != ''
|
|
""")
|
|
return {
|
|
"today_total": int(today_row["cnt"]) if today_row else 0,
|
|
"total_count": int(total_row["cnt"]) if total_row else 0,
|
|
"active_sources": int(sources_row["cnt"]) if sources_row else 0,
|
|
"active_tags": int(tags_row["cnt"]) if tags_row else 0,
|
|
}
|
|
|
|
|
|
@app.get("/api/v1/analysis/trend")
|
|
def analysis_trend(days: int = 7):
|
|
"""近N天每日快讯趋势"""
|
|
days = min(max(days, 1), 30)
|
|
rows = _query_all(f"""
|
|
SELECT DATE(FROM_UNIXTIME(published_at/1000)) as day,
|
|
COUNT(*) as count
|
|
FROM news_flash
|
|
WHERE published_at > 0
|
|
AND published_at >= UNIX_TIMESTAMP(DATE_SUB(CURDATE(), INTERVAL {days - 1} DAY)) * 1000
|
|
GROUP BY day
|
|
ORDER BY day ASC
|
|
""")
|
|
return {"daily": rows}
|
|
|
|
|
|
@app.get("/api/v1/analysis/hourly")
|
|
def analysis_hourly():
|
|
"""24小时分布（每小时发布量）"""
|
|
rows = _query_all("""
|
|
SELECT HOUR(FROM_UNIXTIME(published_at/1000)) as hour,
|
|
COUNT(*) as count
|
|
FROM news_flash
|
|
WHERE published_at > 0
|
|
AND published_at >= UNIX_TIMESTAMP(DATE_SUB(CURDATE(), INTERVAL 6 DAY)) * 1000
|
|
GROUP BY hour
|
|
ORDER BY hour ASC
|
|
""")
|
|
return {"hourly": rows}
|
|
|
|
|
|
@app.get("/api/v1/analysis/top_tags")
|
|
def analysis_top_tags(limit: int = 10):
|
|
"""Top 标签排行"""
|
|
limit = min(max(limit, 5), 30)
|
|
rows = _query_all(f"""
|
|
SELECT tag, COUNT(*) as count
|
|
FROM (
|
|
SELECT JSON_UNQUOTE(JSON_EXTRACT(tags, CONCAT('$[', n, ']'))) as tag
|
|
FROM news_flash
|
|
JOIN (SELECT 0 as n UNION SELECT 1 UNION SELECT 2 UNION SELECT 3 UNION SELECT 4) nums
|
|
WHERE JSON_LENGTH(tags) > n AND published_at > 0
|
|
AND published_at >= UNIX_TIMESTAMP(DATE_SUB(CURDATE(), INTERVAL 6 DAY)) * 1000
|
|
) t
|
|
WHERE tag IS NOT NULL AND tag != ''
|
|
GROUP BY tag
|
|
ORDER BY count DESC
|
|
LIMIT {limit}
|
|
""")
|
|
return {"tags": rows}
|
|
|
|
|
|
@app.get("/api/v1/analysis/top_sources")
|
|
def analysis_top_sources(limit: int = 10):
|
|
"""Top 来源排行（近7天）"""
|
|
limit = min(max(limit, 5), 20)
|
|
rows = _query_all(f"""
|
|
SELECT s.code as source_code, s.name as source_name, COUNT(*) as count
|
|
FROM news_flash f
|
|
LEFT JOIN news_sources s ON f.source_id = s.id
|
|
WHERE f.published_at > 0
|
|
AND f.published_at >= UNIX_TIMESTAMP(DATE_SUB(CURDATE(), INTERVAL 6 DAY)) * 1000
|
|
GROUP BY s.code, s.name
|
|
ORDER BY count DESC
|
|
LIMIT {limit}
|
|
""")
|
|
return {"sources": rows}
|
|
|
|
|
|
@app.get("/api/v1/analysis/importance")
|
|
def analysis_importance():
|
|
"""热度分布（近7天）"""
|
|
rows = _query_all("""
|
|
SELECT importance, COUNT(*) as count
|
|
FROM news_flash
|
|
WHERE published_at > 0 AND importance >= 0
|
|
AND published_at >= UNIX_TIMESTAMP(DATE_SUB(CURDATE(), INTERVAL 6 DAY)) * 1000
|
|
GROUP BY importance
|
|
ORDER BY importance DESC
|
|
""")
|
|
return {"importance_dist": {str(r["importance"]): int(r["count"]) for r in rows}}
|
|
|
|
# 挂载前端静态文件
|
|
app.mount("/assets", StaticFiles(directory=f"{FRONTEND_DIR}/assets"), name="assets")
|
|
|
|
# SPA 回退：所有非 API 请求返回 index.html
|
|
@app.get("/{full_path:path}")
|
|
async def serve_frontend(full_path: str):
|
|
if full_path.startswith("api/") or full_path.startswith("ws/"):
|
|
from fastapi.responses import JSONResponse
|
|
return JSONResponse({"error": "not_found"}, status_code=404)
|
|
index = f"{FRONTEND_DIR}/index.html"
|
|
if os.path.exists(index):
|
|
return FileResponse(index)
|
|
return {"error": "frontend not built"}
|
|
|
|
|
|
# 测试数据源
|
|
@app.post("/api/v1/source/test")
|
|
async def api_test_source(body: dict, _auth=Depends(require_auth)):
|
|
"""测试指定数据源是否可用
|
|
body: {"source_code": "21jingji"}
|
|
返回: {"ok": True, "count": 20, "latest_title": "...", "error": null}
|
|
"""
|
|
source_code = body.get("source_code", "")
|
|
if not source_code:
|
|
return {"ok": False, "count": 0, "latest_title": "", "error": "source_code 不能为空"}
|
|
|
|
try:
|
|
import asyncio as _asyncio
|
|
import importlib.util
|
|
from pathlib import Path as _Path
|
|
|
|
bin_dir = str(_Path(__file__).resolve().parent)
|
|
fetcher_map = {}
|
|
|
|
# OmniData 数据源
|
|
omnidata_map = {
|
|
"21jingji": "Jingji21Fetcher",
|
|
"yicai": "YicaiFetcher",
|
|
"futunn": "FutunnFetcher",
|
|
"hexun": "HexunFetcher",
|
|
"investing": "InvestingFetcher",
|
|
"jrj": "JrjFetcher",
|
|
"jiemian": "JiemianFetcher",
|
|
"xuangubao": "XuangubaoFetcher",
|
|
"xueqiu": "XueqiuFetcher",
|
|
}
|
|
if source_code in omnidata_map:
|
|
spec = importlib.util.spec_from_file_location("omnidata_sources_fetcher", f"{bin_dir}/omnidata_sources_fetcher.py")
|
|
mod = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(mod)
|
|
cls = getattr(mod, omnidata_map[source_code])
|
|
fetcher = cls()
|
|
items = await fetcher.fetch()
|
|
|
|
# 新数据源（东方财富、同花顺、金十）
|
|
elif source_code in ["eastmoney", "10jqka", "jin10"]:
|
|
spec = importlib.util.spec_from_file_location("new_sources_fetcher", f"{bin_dir}/new_sources_fetcher.py")
|
|
mod = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(mod)
|
|
cls_map = {
|
|
"eastmoney": "EastMoneyFetcher",
|
|
"10jqka": "TonghuashunFetcher",
|
|
"jin10": "Jin10Fetcher",
|
|
}
|
|
cls = getattr(mod, cls_map[source_code])
|
|
fetcher = cls()
|
|
items = await fetcher.fetch()
|
|
|
|
# 核心数据源（从 kuaixun-old 加载）
|
|
elif source_code in ["sina", "cls", "wallstreetcn", "gelonghui", "36kr", "gnews", "newsapi"]:
|
|
KX_HOME = str(_Path.home() / "Projects/项目/archive/kuaixun-old")
|
|
import sys as _sys
|
|
if KX_HOME not in _sys.path:
|
|
_sys.path.insert(0, KX_HOME)
|
|
if source_code == "sina":
|
|
from fetchers.sina_fetcher import SinaFetcher
|
|
fetcher = SinaFetcher()
|
|
elif source_code == "cls":
|
|
from fetchers.cls_fetcher import ClsFetcher
|
|
fetcher = ClsFetcher()
|
|
elif source_code == "wallstreetcn":
|
|
from fetchers.wallstreetcn_fetcher import WallstreetcnFetcher
|
|
fetcher = WallstreetcnFetcher()
|
|
else:
|
|
return {"ok": False, "count": 0, "latest_title": "", "error": f"暂不支持测试 {source_code}"}
|
|
items = await fetcher.fetch()
|
|
|
|
else:
|
|
return {"ok": False, "count": 0, "latest_title": "", "error": f"未知数据源: {source_code}"}
|
|
|
|
if items:
|
|
latest = items[0]
|
|
return {
|
|
"ok": True,
|
|
"count": len(items),
|
|
"latest_title": latest.title[:80] if latest.title else "",
|
|
"latest_time": latest.published_at,
|
|
"error": None,
|
|
}
|
|
else:
|
|
return {"ok": False, "count": 0, "latest_title": "", "error": "未获取到数据"}
|
|
|
|
except Exception as e:
|
|
return {"ok": False, "count": 0, "latest_title": "", "error": str(e)[:200]}
|
|
|
|
|
|
# 优雅关闭端点（需内部鉴权）
|
|
@app.post("/api/shutdown")
|
|
async def api_shutdown(_auth=Depends(require_auth)):
|
|
os.kill(os.getpid(), signal.SIGTERM)
|
|
return {"shutdown": True}
|
|
|
|
|
|
def run_server(port: int = BACKEND_PORT):
|
|
# 安全实践：绑定 127.0.0.1 仅限本地访问，避免暴露到外部网络
|
|
uvicorn.run(app, host="127.0.0.1", port=port, log_level="warning")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
port = int(os.getenv("BACKEND_PORT", str(BACKEND_PORT)))
|
|
run_server(port)
|