198 lines
8.7 KiB
Python
198 lines
8.7 KiB
Python
"""FastAPI app: 纯 /v1 JSON API(2026-05-15 切换 — 详见 DESIGN §7.9)。
|
|
|
|
设计要点:
|
|
- 所有路由 `/v1/*` 前缀,响应 JSON;模板 / HTMX / 服务端 markdown 渲染全删
|
|
- SSE 事件 payload 是 JSON dict 而非 HTML 片段(`event: <type>` + `data: <json>`)
|
|
- Auth: PLATFORM_KEY → JWT 兑换(§7 D' 过渡形态,见 web/auth.py);OIDC 替换时只动 /v1/auth/login 内部
|
|
- 所有 /v1/tasks* 路由 Depends(require_user),按 user_id 隔离数据
|
|
- 豁免:/healthz、/docs、/openapi.json、/、/v1/auth/login、/static/*
|
|
- CORS allow_origins=["*"] 本地宽松;真发布按 platform 域名收紧
|
|
- `GET /` 302 → /static/dev.html(本地 dev SPA)
|
|
|
|
本文件只剩两件事(2026-07-23 拆分后):
|
|
- `create_app` 工厂:装配 auth 依赖 → 注册 11 个 router 模块(web/routers/)+ admin
|
|
- `lifespan` 编排:线程池接管 / 孤儿收割 / 6 类后台协程起停 / 优雅 drain
|
|
(协程本体在 web/background.py + scheduler_runner.py + wechat_runner.py)
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import mimetypes
|
|
import os
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
from contextlib import asynccontextmanager
|
|
from pathlib import Path
|
|
|
|
from fastapi import FastAPI
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
|
|
from core import __version__
|
|
|
|
from .auth import (
|
|
REFRESHED_TOKEN_HEADER,
|
|
TOKEN_EXPIRES_HEADER,
|
|
AuthConfig,
|
|
make_require_admin,
|
|
make_require_user,
|
|
)
|
|
from .admin import register_admin_routes
|
|
from .background import (
|
|
cancel_and_wait,
|
|
drain_inflight,
|
|
init_sandbox,
|
|
reap_stale_runs,
|
|
start_disk_scanner,
|
|
start_proc_sweeper,
|
|
start_stats_logger,
|
|
start_toolfail_scanner,
|
|
)
|
|
from .broker import broker
|
|
from .routers.asr import register_asr_routes
|
|
from .routers.authroutes import register_auth_routes
|
|
from .routers.files import register_file_routes
|
|
from .routers.kb import register_kb_routes
|
|
from .routers.messages import register_message_routes
|
|
from .routers.misc import register_misc_routes
|
|
from .routers.models import register_model_routes
|
|
from .routers.schedules import register_schedule_routes
|
|
from .routers.skills_memory import register_skill_memory_routes
|
|
from .routers.tasks import register_task_routes
|
|
from .routers.wechat import register_wechat_routes
|
|
from .scheduler_runner import start_scheduler
|
|
from .static_files import NoCacheStaticFiles
|
|
from .wechat_runner import start_wechat_inbound
|
|
|
|
# ────────────────────── App 工厂 ──────────────────────
|
|
|
|
# web/static 目录路径 — /static 静态挂载用,dev.html 也放这
|
|
_STATIC_DIR = Path(__file__).parent / "static"
|
|
|
|
|
|
def create_app() -> FastAPI:
|
|
# fail-fast:env 缺失直接抛,不裸跑无密
|
|
auth_cfg = AuthConfig.from_env()
|
|
require_user = make_require_user(auth_cfg)
|
|
require_admin = make_require_admin(auth_cfg)
|
|
|
|
@asynccontextmanager
|
|
async def lifespan(app: FastAPI):
|
|
loop = asyncio.get_running_loop()
|
|
broker.bind_loop(loop)
|
|
|
|
# ── 接管默认线程池 executor(§8.4)──────────────────────────────
|
|
# run 走 asyncio.to_thread(用 loop 默认 executor);默认是匿名的,读不到大小、
|
|
# 不可调。显式建一个同尺寸(复刻 Python 默认 min(32, cpu+4))接管,好处:① 监控
|
|
# 能读 max_workers 判断有没有排队 ② 并发不够时改 ZCBOT_RUN_MAX_WORKERS 调大不改码。
|
|
# 注:run 与 disk scan / pptx 转换 / reaper 共享此池(同原默认行为);真要隔离
|
|
# 长任务再另开 run 专用池,那是后话。
|
|
run_max_workers = int(
|
|
os.getenv("ZCBOT_RUN_MAX_WORKERS") or min(32, (os.cpu_count() or 1) + 4)
|
|
)
|
|
run_executor = ThreadPoolExecutor(
|
|
max_workers=run_max_workers, thread_name_prefix="run"
|
|
)
|
|
loop.set_default_executor(run_executor)
|
|
app.state.run_executor = run_executor
|
|
app.state.run_max_workers = run_max_workers
|
|
print(f"[startup] run executor: max_workers={run_max_workers} "
|
|
f"(override via ZCBOT_RUN_MAX_WORKERS)")
|
|
|
|
from core.agent_builder import load_config
|
|
_cfg = load_config()
|
|
|
|
# 优雅 drain 状态(SIGTERM / systemctl restart 兜底,见下方 finally):
|
|
# draining 置位后 POST /messages 返 503;inflight 登记在跑的 BG run task,
|
|
# 关停时 await 它们收尾。inflight 同时给 create_task 持强引用,防被 GC 中途回收。
|
|
app.state.draining = asyncio.Event()
|
|
app.state.inflight = {} # dict[asyncio.Task, UUID(task_id)]
|
|
_shutdown_cfg = _cfg.get("shutdown") or {}
|
|
drain_timeout = int(_shutdown_cfg.get("drain_timeout_seconds") or 90)
|
|
cancel_grace = int(_shutdown_cfg.get("cancel_grace_seconds") or 15)
|
|
|
|
# 启动钩子 + 后台协程群(本体见 web/background.py 等;None=该项未启用)
|
|
reap_stale_runs()
|
|
disk_scanner_task = start_disk_scanner(_cfg)
|
|
stats_logger_task = start_stats_logger(app, run_max_workers)
|
|
toolfail_task = start_toolfail_scanner()
|
|
scheduler_task = start_scheduler(app, _cfg)
|
|
wechat_task, wechat_stop = start_wechat_inbound(app)
|
|
sandbox_reaper_task = init_sandbox(app, _cfg)
|
|
proc_sweeper_task = start_proc_sweeper(_cfg)
|
|
|
|
try:
|
|
yield
|
|
finally:
|
|
# 先拒新 run + drain in-flight(细节见 background.drain_inflight)
|
|
app.state.draining.set()
|
|
await drain_inflight(app, drain_timeout, cancel_grace)
|
|
|
|
await cancel_and_wait(disk_scanner_task)
|
|
await cancel_and_wait(stats_logger_task)
|
|
await cancel_and_wait(toolfail_task)
|
|
await cancel_and_wait(scheduler_task)
|
|
if wechat_task is not None:
|
|
wechat_stop.set()
|
|
await cancel_and_wait(wechat_task)
|
|
await cancel_and_wait(sandbox_reaper_task)
|
|
await cancel_and_wait(proc_sweeper_task)
|
|
|
|
pool = getattr(app.state, "sandbox_pool", None)
|
|
if pool is not None:
|
|
try:
|
|
pool.shutdown_all()
|
|
except Exception as e:
|
|
print(f"[shutdown] sandbox shutdown_all error: {type(e).__name__}: {e}")
|
|
|
|
# broker 收尾(redis 版停 pubsub reader + 断连;local no-op)。放在 drain
|
|
# 之后 —— drain 期间 run 还要 emit/close。
|
|
try:
|
|
await broker.shutdown()
|
|
except Exception as e:
|
|
print(f"[shutdown] broker shutdown error: {type(e).__name__}: {e}")
|
|
|
|
# drain 已 await inflight 收尾、run 线程退完;非阻塞关池(进程在退出,保守清理)
|
|
run_executor.shutdown(wait=False)
|
|
|
|
app = FastAPI(
|
|
title="zcbot api",
|
|
version=__version__,
|
|
description=(
|
|
"zcbot 后端 — /v1 JSON API + SSE。Auth: PLATFORM_KEY → JWT(§7 D' 过渡)。"
|
|
"本地 dev SPA: /static/dev.html。"
|
|
),
|
|
lifespan=lifespan,
|
|
)
|
|
app.add_middleware(
|
|
CORSMiddleware,
|
|
allow_origins=["*"], # 本地宽松,部署 platform 时按域名收紧
|
|
allow_credentials=False,
|
|
allow_methods=["*"],
|
|
allow_headers=["*"],
|
|
# 滑动续签的新 token 走响应头带回;expose 才能让浏览器 fetch 读到(默认不暴露自定义头)
|
|
expose_headers=[REFRESHED_TOKEN_HEADER, TOKEN_EXPIRES_HEADER],
|
|
)
|
|
|
|
if _STATIC_DIR.is_dir():
|
|
# Windows 上 mimetypes 偶尔把 .js 判成 text/plain,会令 <script type="module"> 被浏览器拒执行;
|
|
# 显式兜底,保证静态 ES module 以正确 MIME 下发。
|
|
mimetypes.add_type("text/javascript", ".js")
|
|
app.mount("/static", NoCacheStaticFiles(directory=str(_STATIC_DIR)), name="static")
|
|
|
|
# ───────────── 路由模块(与 register_admin_routes 同范式)─────────────
|
|
register_misc_routes(app, require_user=require_user)
|
|
register_wechat_routes(app, require_user=require_user, auth_cfg=auth_cfg)
|
|
register_model_routes(app, require_user=require_user)
|
|
register_auth_routes(app, require_user=require_user, auth_cfg=auth_cfg)
|
|
register_skill_memory_routes(app, require_user=require_user)
|
|
register_kb_routes(app, require_user=require_user)
|
|
register_schedule_routes(app, require_user=require_user)
|
|
register_file_routes(app, require_user=require_user)
|
|
register_asr_routes(app, require_user=require_user, auth_cfg=auth_cfg)
|
|
register_task_routes(app, require_user=require_user)
|
|
register_message_routes(app, require_user=require_user)
|
|
|
|
# ───────────── 管理后台(admin-only)─────────────
|
|
register_admin_routes(app, require_admin)
|
|
|
|
return app
|