特性	说明
FastAPI + Uvicorn	高性能异步 Web 框架，自动生成 OpenAPI 文档
Zenoh 服务发现	基于 liveliness 的 Worker 注册与 RPC 调用，负载均衡 + 超时重试
MQTT 集成	内嵌 amqtt Broker，MQTT ↔ Worker 双向桥接
动态路由生成	扫描 `config/workers/*.toml` 自动注册 HTTP 路由
Worker 框架	`@rpc_method` 装饰器 + 自动参数解析 + Pydantic 模型绑定
自动化模型生成	`gen_models` 根据 TOML 配置生成请求/响应模型
JWT 鉴权	Rust 实现，路由级 `auth_required` 控制
Argon2id 密码哈希	Rust 实现的高性能密码安全方案
异步 SQLite	aiosqlite + SQLModel，多进程并发安全
本地文件存储	内置文件上传/下载端点
流媒体代理	ffmpeg + MediaMTX RTSP 流转发
构建打包	PyInstaller / Nuitka 一键打包为独立可执行文件
Vue 3 前端	Element Plus UI、Pinia 状态管理、类型安全的 API 客户端

项目结构

chongming-zenoh/
├── config.toml                   # 应用主配置
├── pyproject.toml                # 项目依赖与脚本入口
├── config/                       # 配置文件目录
│   ├── zenoh_config.json5        # Zenoh 会话配置
│   └── workers/                  # Worker 路由定义（TOML）
│       ├── config.toml           # 插件注册列表
│       ├── auth/                 # 鉴权插件（login / register / me / …）
│       ├── doubao/               # 豆包视觉理解插件
│       ├── aibox_sophon_adapter/ # AI Box 设备心跳监控插件
│       └── example/              # 示例插件（四则运算 / MQTT 端点）
├── src/
│   ├── chongming/                # 核心应用
│   │   ├── app/
│   │   │   ├── core/             # WorkerManager, RouteGenerator, MQTTBridge, auth
│   │   │   └── routes/           # 内置路由（文件上传等）
│   │   └── scripts/              # build, gen_models
│   ├── chongming-web/            # Vue 3 前端
│   │   └── src/
│   │       ├── api/generated/    # 自动生成 API 客户端
│   │       ├── views/            # 页面组件
│   │       └── router/           # Vue Router 配置
│   └── plugins/                  # Worker 插件源码
│       ├── auth/                 # JWT 鉴权
│       ├── doubao/               # 豆包 AI（支持 AICC 安全通道）
│       ├── aibox_sophon_adapter/ # 设备心跳 + 告警
│       └── example/              # 演示示例
├── utils/                        # 独立工具包
│   ├── bytedance/   # 字节跳动安全通信 SDK
│   ├── config/      # TOML 配置加载
│   ├── database/    # 异步 SQLite 封装
│   ├── jwt/         # Rust JWT 编解码
│   ├── password/    # Rust Argon2id 哈希
│   ├── request/     # Zenoh RPC 客户端
│   ├── stream_manager/  # ffmpeg RTSP 流管理
│   └── worker/      # Worker 框架核心
├── data/
│   ├── storage/chongming/        # SQLite 数据库文件
│   └── uploads/                  # 文件上传目录
├── logs/                         # 日志目录
├── resources/                    # 模块银行打包资源
├── test_mqtt_e2e.py              # MQTT 端到端测试
├── publisher.py / subscriber.py  # MQTT 收发示例
└── README.md

快速开始

环境要求

Python >= 3.12
Rust 工具链（用于编译 jwt / password 扩展）
Node.js >= 22（前端开发）
uv 包管理器
ffmpeg（可选，stream_manager 需要）

安装与启动

# 同步依赖（自动编译 Rust 扩展 + 安装 Python 包）
uv sync

# 启动服务（开发模式，热重载）
uv run serve

服务启动后：

通道	地址	说明
HTTP	`http://localhost:8000`	REST API
Swagger	`http://localhost:8000/docs`	OpenAPI 交互文档
MQTT	`mqtt://localhost:1883`	内嵌 Broker

前端开发

cd src/chongming-web
npm install
npm run dev          # http://localhost:5173

架构与数据流

graph TB
    HTTP[HTTP 客户端] -->|REST API| FastAPI[FastAPI 路由<br/>TOML 自动生成]
    MQTT[MQTT 客户端] -->|chongming/+/+| Bridge[MQTT Bridge<br/>纯客户端模式]

    FastAPI --> WM[WorkerManager<br/>随机负载均衡 + 超时重试]
    Bridge --> WM

    WM -->|Zenoh Query| W1[Worker 副本 1]
    WM -->|Zenoh Query| W2[Worker 副本 2]
    WM -->|Zenoh Query| W3[Worker 副本 N]

    W1 & W2 & W3 -->|插件业务逻辑| DB[(SQLite)]
    W1 & W2 & W3 -->|AI 调用| AI[豆包 API]

启动流程：

uv run serve → FastAPI lifespan:
  1. setup_storage()           — 本地文件系统存储
  2. setup_zenoh()             — Zenoh peer 会话
  3. setup_worker_manager()    — WorkerManager + 定期 liveliness 刷新
  4. load_workers()            — 动态 import 插件 → 启动 Worker 协程
  5. register_routes()         — 扫描 config/workers/ 生成 HTTP 路由
  6. setup_mqtt()              — 内嵌 Broker（子进程）+ MQTT Bridge

端点类型与路由注册

端点类型

TOML 文件中的节类型决定端点暴露方式：

节类型	HTTP 路由	OpenAPI 文档	MQTT	请求模型命名
`[GET]`	✅	✅	✅	`XxxGetRequest / XxxGetResponse`
`[POST]`	✅	✅	✅	`XxxPostRequest / XxxPostResponse`
`[MQTT]`	❌	❌	✅	`XxxMqttRequest`（无 response_model）

两种路由注册方式

方式一：TOML 动态路由（标准 JSON 端点）

通过 config/workers/<插件>/*.toml 定义端点，自动生成 FastAPI 路由。适用于参数和响应可被 Pydantic 模型描述的标准接口。

# config/workers/example/add.toml
[GET]
summary = "加法运算"
params = ["a: float", "b: float"]

[GET.response_model]
result = { type = "float", required = true }

方式二：插件自定义路由（特殊端点）

当端点无法用 TOML 描述时（如 multipart/form-data 上传），插件暴露 APIRouter 实例，网关自动发现并注册。

# src/plugins/<name>/routes.py
from fastapi import APIRouter, UploadFile, File

router = APIRouter(prefix="/<prefix>", tags=["<worker_type>"])

@router.post("/upload")
async def upload(file: UploadFile = File(...)):
    ...

约定：插件 __init__.py 中导入 router 实例（from .routes import router），网关通过 getattr(module, "router", None) 自动发现。

MQTT 协议

Topic 约定

方向	Topic	说明
请求	`chongming/<type>/<op>`	MQTT → Worker
响应	`chongming/<type>/<op>/response`	Worker → MQTT（`None` 时跳过）

自定义订阅主题

除内置的 chongming/<type>/<op> 接口外，可将外部 MQTT 主题桥接到任意 Worker。推荐在插件自己的配置 config/workers/<type>/config.toml 中通过 [[workers.mqtt_topics]] 声明（目标 worker 即该插件本身）：

# config/workers/aibox_sophon_adapter/config.toml
[[workers.mqtt_topics]]
topic = "/board_ping"                       # 支持 +（单级）/ #（多级，须在末尾）通配符
subject = "aibox_sophon_adapter.board_ping" # 可选，{1} {2}... 对应通配符捕获值
response_topic = "/board_ping/response"     # 可选，响应主题；默认 "<实际主题>/response"

subject 缺省为 <插件类型>.<topic 最后一段>；response_topic 缺省为 <实际匹配主题>/response。
网关发现插件时自动收集所有 [[workers.mqtt_topics]]，MQTT Bridge 一并订阅并路由。
请求负载同样支持 _method / _timeout / _cacheable 控制字段。
多条自定义主题按配置顺序匹配，先匹配者优先；内置 chongming/... 接口最后兜底。
全局 [<env>.mqtt].custom_topics 仍可作为通用兜底（不依赖具体插件时使用，可选）。

快速测试

# 终端 1：启动服务
uv run serve

# 终端 2：端到端测试
python test_mqtt_e2e.py

# 手动测试
python subscriber.py                             # 监听所有响应
python publisher.py example add '{"a":3,"b":5}'  # 调用 add

开发指南

添加新 Worker 插件

1. 创建 src/plugins/<name>/ — __init__.py, global_var.py, router/, models/
2. 注册到 config/workers/config.toml 的 [[workers]]
3. 创建 config/workers/<name>/*.toml 端点定义
4. 运行 uv run python -m chongming.scripts.gen_models 生成模型
5. 实现 router/<op>.py，引用 models/pydantic.py 中的模型
6. 更新 router/__init__.py 导入

为现有插件添加新端点

1. 创建 config/workers/<name>/<op>.toml
2. 运行 uv run python -m chongming.scripts.gen_models
3. 创建 router/<op>.py → 引用生成的模型
4. 更新 router/__init__.py
5. 重启服务

常用命令

uv run serve                                    # 启动开发服务
uv run python -m chongming.scripts.gen_models   # 生成 Pydantic 模型
uv run python -m chongming.scripts.build        # 构建打包（PyInstaller/Nuitka）
uv pip install -e utils/<name>                  # 安装/更新工具包

插件列表

插件	路径	功能
auth	`src/plugins/auth/`	用户注册/登录/资料管理，JWT 鉴权，管理员初始化
doubao	`src/plugins/doubao/`	豆包视觉理解，支持 AICC 安全通信通道
aibox_sophon_adapter	`src/plugins/aibox_sophon_adapter/`	AI Box 设备心跳监控、离线检测、告警上报与查询、视频上传
example	`src/plugins/example/`	四则运算 + 心跳演示（含 MQTT 端点示例）

工具包列表

包名	路径	语言	说明
chongming-config	`utils/config/`	Python	TOML 配置加载，TypedDict 类型安全
chongming-database	`utils/database/`	Python	aiosqlite + SQLModel 异步数据库封装
chongming-jwt	`utils/jwt/`	Rust + PyO3	JWT 编解码，支持 HS256/384/512 + RS256/384/512
chongming-password	`utils/password/`	Rust + PyO3	Argon2id 密码哈希与验证
chongming-request	`utils/request/`	Python	Zenoh RPC 直连客户端
chongming-worker	`utils/worker/`	Python	Worker 框架核心：装饰器、运行时、配置加载
chongming-stream-manager	`utils/stream_manager/`	Python	ffmpeg + MediaMTX RTSP 流代理管理
bytedance	`utils/bytedance/`	Python	字节跳动 AICC 安全通信 SDK（Jeddak 等）

配置说明

应用配置：根目录 config.toml，包含环境、服务器、安全、MQTT 等全局设置
Zenoh 配置：config/zenoh_config.json5
Worker 注册：config/workers/config.toml，定义加载的插件及其副本数
路由定义：config/workers/<name>/*.toml，每个端点一个文件

详细配置说明见 config/README.md。

构建部署

# PyInstaller 打包
uv run python -m chongming.scripts.build pyinstaller

# Nuitka 打包
uv run python -m chongming.scripts.build nuitka

构建过程包括：模块加密打包 → 资源复制 → PyInstaller/Nuitka 打包 → 数据库初始化 → 前端编译 → VFS 挂载。

详见插件开发指南。

常用命令

uv run serve                              # 启动开发服务器
uv run python -m chongming.scripts.gen_models  # 生成 Pydantic 模型
uv run build pyinstaller                  # PyInstaller 打包
uv run build nuitka                       # Nuitka 打包
uv pip install -e utils/<name>            # 安装/更新工具包子包

TOML 配置字段速查

字段	位置	说明
`count`	`workers/config.toml`	Worker 副本数
`router_prefix`	`<插件>/config.toml`	API 路由前缀
`defaults.timeout`	`<插件>/config.toml`	默认超时（秒）
`defaults.cacheable`	`<插件>/config.toml`	是否缓存 RPC 结果
`params`	`<操作>.toml`	参数列表（`"name: type"` 格式）
`auth_required`	`<操作>.toml`	是否需要 JWT 鉴权
`internal`	`<操作>.toml`	`true` 则仅限内部 RPC，不注册 HTTP 路由
`cacheable`	`<操作>.toml`	端点是否可缓存
`summary`	`<操作>.toml`	端点描述（用于 OpenAPI / 日志）

内置插件

插件	HTTP 前缀	MQTT 主题前缀	说明
auth	`/auth`	`auth.*`	注册、登录、JWT 鉴权、个人信息、改密
doubao	`/doubao`	`doubao.*`	火山引擎豆包视觉理解模型
example	`/calc`	`example.*`	四则运算 + 访问计数 + MQTT 心跳示例
aibox_sophon_adapter	`/aibox`	`aibox_sophon_adapter.*`	AI Box 设备心跳适配 + 视频上传

工具包

包名	说明	技术栈
`chongming-config`	TOML 配置加载、TypedDict 类型定义	Python
`chongming-database`	异步 SQLite（aiosqlite + SQLModel）	Python
`chongming-jwt`	JWT 编码/验证（HS256/384/512, RS256/384/512）	Rust + PyO3
`chongming-password`	Argon2id 密码哈希与验证	Rust + PyO3
`chongming-request`	Zenoh Worker 客户端（负载均衡、超时重试）	Python
`chongming-worker`	Worker 框架核心（`@rpc_method`、运行时）	Python

README.md Unescape Escape

Chongming-Zenoh

核心特性

项目结构

快速开始

环境要求

安装与启动

前端开发

架构与数据流

端点类型与路由注册

端点类型

两种路由注册方式

方式一：TOML 动态路由（标准 JSON 端点）

方式二：插件自定义路由（特殊端点）

MQTT 协议

Topic 约定

自定义订阅主题

快速测试

开发指南

添加新 Worker 插件

为现有插件添加新端点

常用命令

插件列表

工具包列表

配置说明

构建部署

常用命令

TOML 配置字段速查

内置插件

工具包

更多文档

README.md