From 1f727bcdd7d6a18bfbd1353d7c4b950dadeb5b81 Mon Sep 17 00:00:00 2001 From: chakcy <947105045@qq.com> Date: Fri, 14 Aug 2026 17:44:19 +0800 Subject: [PATCH] =?UTF-8?q?add:=20workers/=20=E8=BF=9B=E7=A8=8B=E6=8F=92?= =?UTF-8?q?=E4=BB=B6=E7=9B=AE=E5=BD=95=20+=20demo=20=E7=A4=BA=E4=BE=8B?= =?UTF-8?q?=EF=BC=88external=20=E6=A8=A1=E5=BC=8F=E7=BB=8F=20Zenoh=20?= =?UTF-8?q?=E6=B3=A8=E5=86=8C=E8=B7=AF=E7=94=B1=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 5 +++- src/chongming/__init__.py | 6 +++++ src/chongming/app/core/contracts.py | 6 ++++- src/chongming/scripts/gen_models.py | 20 +++++++++++++- uv.lock | 2 +- workers/__init__.py | 8 ++++++ workers/demo/__init__.py | 41 +++++++++++++++++++++++++++++ workers/demo/global_var.py | 32 ++++++++++++++++++++++ workers/demo/models/__init__.py | 14 ++++++++++ workers/demo/models/pydantic.py | 23 ++++++++++++++++ workers/demo/router/__init__.py | 5 ++++ workers/demo/router/hello.py | 25 ++++++++++++++++++ 12 files changed, 183 insertions(+), 4 deletions(-) create mode 100644 workers/__init__.py create mode 100644 workers/demo/__init__.py create mode 100644 workers/demo/global_var.py create mode 100644 workers/demo/models/__init__.py create mode 100644 workers/demo/models/pydantic.py create mode 100644 workers/demo/router/__init__.py create mode 100644 workers/demo/router/hello.py diff --git a/README.md b/README.md index 2fda4da..e600a3b 100644 --- a/README.md +++ b/README.md @@ -53,11 +53,13 @@ chongming-zenoh/ │ │ ├── api/generated/ # 自动生成 API 客户端 │ │ ├── views/ # 页面组件 │ │ └── router/ # Vue Router 配置 -│ └── plugins/ # Worker 插件源码 +│ └── plugins/ # Worker 插件源码（inline 模式） │ ├── auth/ # JWT 鉴权 │ ├── doubao/ # 豆包 AI（支持 AICC 安全通道） │ ├── aibox_sophon_adapter/ # 设备心跳 + 告警 │ └── example/ # 演示示例 +├── workers/ # 进程插件（external 模式，独立进程经 Zenoh 注册路由） +│ └── demo/ # 进程插件示例 ├── utils/ # 独立工具包 │ ├── bytedance/ # 字节跳动安全通信 SDK │ ├── config/ # TOML 配置加载 @@ -354,6 +356,7 @@ uv pip install -e utils/ # 安装/更新工具包 | **doubao** | `src/plugins/doubao/` | 豆包视觉理解，支持 AICC 安全通信通道 | | **aibox_sophon_adapter** | `src/plugins/aibox_sophon_adapter/` | AI Box 设备心跳监控、离线检测、告警上报与查询、视频上传 | | **example** | `src/plugins/example/` | 四则运算 + 心跳演示（含 MQTT 端点示例） | +| **demo**（进程插件示例） | `workers/demo/` | 独立进程运行（load_mode=external），经 Zenoh 注册路由 | --- diff --git a/src/chongming/__init__.py b/src/chongming/__init__.py index 7082f47..f89b63c 100644 --- a/src/chongming/__init__.py +++ b/src/chongming/__init__.py @@ -54,6 +54,12 @@ def worker_serve(): parser.add_argument("--config", default="config.toml", help="全局配置文件路径") args = parser.parse_args() + # 控制台脚本启动时 sys.path 不含项目根目录，显式加入以保证 + # plugins.（inline 插件）与 workers.（进程插件）均可导入 + project_root = os.path.abspath(os.getcwd()) + if project_root not in sys.path: + sys.path.insert(0, project_root) + config = load_config() logger = get_logger("app") diff --git a/src/chongming/app/core/contracts.py b/src/chongming/app/core/contracts.py index 16a1e5e..18d64c7 100644 --- a/src/chongming/app/core/contracts.py +++ b/src/chongming/app/core/contracts.py @@ -46,6 +46,7 @@ def _load_plugin_models(worker_type: str) -> Optional[Set[str]]: for mod_name in ( f"plugins.{worker_type}.models.pydantic", f"src.plugins.{worker_type}.models.pydantic", + f"workers.{worker_type}.models.pydantic", ): try: mod = importlib.import_module(mod_name) @@ -62,7 +63,10 @@ def _load_plugin_models(worker_type: str) -> Optional[Set[str]]: and id(obj) not in seen_objs ): seen_objs.add(id(obj)) - names.add(name) + # 用 obj.__name__ 而非 dir() 名称：gen_models 生成的 + # Response 别名与真实模型同一对象，别名保留原类名， + # 按对象去重后自动收敛到真实模型名（避免别名抢占排序位） + names.add(obj.__name__) return names return None diff --git a/src/chongming/scripts/gen_models.py b/src/chongming/scripts/gen_models.py index f0bf168..6c5cabe 100644 --- a/src/chongming/scripts/gen_models.py +++ b/src/chongming/scripts/gen_models.py @@ -101,6 +101,23 @@ def generate_pydantic_code( routes: list[dict[str, Any]], ) -> str: """生成 pydantic.py 文件内容""" + # 收集全部字段类型，按需生成 typing 导入（避免未使用导入触发 lint） + type_strings: list[str] = [] + has_optional_field = False + for route in routes: + for p in route.get("parsed_params", []): + type_strings.append(p.get("type", "str")) + for f in route.get("parsed_response", []): + type_strings.append(f.get("type", "str")) + # 非必填字段渲染为 Optional[...] + if not f.get("required", True): + has_optional_field = True + typing_imports = [] + if has_optional_field or any("Optional" in t for t in type_strings): + typing_imports.append("Optional") + if any("Any" in t for t in type_strings): + typing_imports.append("Any") + lines = [] lines.append('"""') lines.append(f"自动生成的 Pydantic 模型 — {plugin_name}") @@ -109,7 +126,8 @@ def generate_pydantic_code( lines.append("") lines.append("from __future__ import annotations") lines.append("") - lines.append("from typing import Any, Optional") + if typing_imports: + lines.append(f"from typing import {', '.join(typing_imports)}") lines.append("from pydantic import BaseModel, Field") lines.append("") diff --git a/uv.lock b/uv.lock index b9700b9..77a42c7 100644 --- a/uv.lock +++ b/uv.lock @@ -319,7 +319,7 @@ source = { git = "http://124.71.68.6:3000/chongming/chongming-stream-manager.git [[package]] name = "chongming-worker" version = "0.1.0" -source = { git = "http://124.71.68.6:3000/chongming/chongming-utils-worker.git#a58e92c21e5bf4b3ada55b2abc894e8ead7d17bc" } +source = { git = "http://124.71.68.6:3000/chongming/chongming-utils-worker.git#4908378e597eb200e255c71df92d16e272b4b12e" } dependencies = [ { name = "eclipse-zenoh" }, { name = "tomli" }, diff --git a/workers/__init__.py b/workers/__init__.py new file mode 100644 index 0000000..ec8816c --- /dev/null +++ b/workers/__init__.py @@ -0,0 +1,8 @@ +""" +进程插件目录（load_mode = "external" 的插件） + +与 src/plugins/（inline 插件）结构一致，区别仅在运行方式： + - 由 Docker/K8s 编排拉起独立进程（容器入口 uv run worker_serve --type ） + - 进程自读 config/workers// 配置、进程内契约校验，经 Zenoh + （chongming/routes/**）注册路由，网关动态挂载 HTTP 端点 +""" diff --git a/workers/demo/__init__.py b/workers/demo/__init__.py new file mode 100644 index 0000000..7c04d4e --- /dev/null +++ b/workers/demo/__init__.py @@ -0,0 +1,41 @@ +""" +Demo 进程插件 — 独立进程运行示例 + +与 inline 插件（src/plugins/）结构完全一致，区别仅在运行方式： + - config/workers/demo/config.toml 声明 load_mode = "external"、module = "workers.demo" + - 由 Docker/K8s 编排拉起独立进程: uv run worker_serve --type demo + - 启动流程: 自读 TOML → 进程内契约校验 → create_workers → + 经 Zenoh 发布路由描述符（网关 RouteRegistry 动态挂载 HTTP 路由）→ + 每 10s 周期重发布自愈 → SIGTERM 时摘除路由并 cleanup +""" +import logging + +from chongming_worker.plugin import Plugin + +from .router import * # noqa: F401,F403 触发 @rpc_method 注册 + +logger = logging.getLogger(__name__) + + +class DemoPlugin(Plugin): + """Demo 插件 — 无数据库的进程插件最小示例。""" + + type = "demo" + db_name = None # 无数据库 + auto_bind = True # 自动绑定 models/pydantic.py（支持 workers/ 路径） + + async def on_init(self): + """首次 create_workers 时调用一次：读取插件私有配置。""" + prefix = self.config.get("greeting_prefix", "Hello") + self.set_var("greeting_prefix", prefix) + logger.info("[demo] 初始化完成 (greeting_prefix=%s)", prefix) + + async def on_cleanup(self): + """cleanup 时调用（无外部资源需要释放）。""" + logger.info("[demo] 清理完成") + + +# 暴露给 worker_serve 的模块级入口（签名与 inline 插件一致） +plugin = DemoPlugin() +create_workers = plugin.create_workers +cleanup = plugin.cleanup diff --git a/workers/demo/global_var.py b/workers/demo/global_var.py new file mode 100644 index 0000000..dc67930 --- /dev/null +++ b/workers/demo/global_var.py @@ -0,0 +1,32 @@ +""" +Demo 进程插件全局运行时状态 — 与 inline 插件约定一致 + +约定: + - 模块只做"声明默认值"，不产生任何导入期副作用 + - 所有值统一在 on_init 中通过 set_global_var 显式写入 + - set_global_var 仅允许写入 _ALLOWED 白名单，拼写错误会立即抛错 +""" +from typing import Any + +# 允许写入的全局变量白名单（防止拼写错误静默创建无用的新变量） +_ALLOWED = frozenset({"greeting_prefix"}) + +# 显式声明所有运行时状态（默认值，运行时由 on_init 写入） +greeting_prefix: str = "Hello" + + +def set_global_var(name: str, value: Any) -> None: + """白名单校验版 setter — 设置模块级运行时状态。""" + if name not in _ALLOWED: + raise KeyError(f"global_var 未知变量: {name}（允许: {sorted(_ALLOWED)}）") + globals()[name] = value + + +def require(name: str) -> Any: + """取"必须已存在"的值，未初始化时抛清晰错误。""" + if name not in _ALLOWED: + raise KeyError(f"global_var 未知变量: {name}") + return globals()[name] + + +__all__ = ["greeting_prefix", "set_global_var", "require"] diff --git a/workers/demo/models/__init__.py b/workers/demo/models/__init__.py new file mode 100644 index 0000000..2959d2f --- /dev/null +++ b/workers/demo/models/__init__.py @@ -0,0 +1,14 @@ +""" +模型导出 — demo +自动生成，请勿手动修改 +""" + +from .pydantic import DemoResponse +from .pydantic import HelloGetRequest +from .pydantic import HelloGetResponse + +__all__ = [ + "DemoResponse", + "HelloGetRequest", + "HelloGetResponse", +] diff --git a/workers/demo/models/pydantic.py b/workers/demo/models/pydantic.py new file mode 100644 index 0000000..aab4062 --- /dev/null +++ b/workers/demo/models/pydantic.py @@ -0,0 +1,23 @@ +""" +自动生成的 Pydantic 模型 — demo +来源: config/workers/demo/ +""" + +from __future__ import annotations + +from typing import Optional +from pydantic import BaseModel, Field + +class HelloGetRequest(BaseModel): + """进程插件示例：问候 — 请求体""" + name: str = Field(...) + +class HelloGetResponse(BaseModel): + """进程插件示例：问候 — 响应体""" + greeting: str = Field(...) + worker_id: Optional[str] = Field(default='') + timestamp: float = Field(...) + + +# 所有响应模型的联合类型 +DemoResponse = HelloGetResponse diff --git a/workers/demo/router/__init__.py b/workers/demo/router/__init__.py new file mode 100644 index 0000000..5d8ee7e --- /dev/null +++ b/workers/demo/router/__init__.py @@ -0,0 +1,5 @@ +"""Demo 进程插件路由端点""" + +from . import hello # noqa: F401 触发 @rpc_method 注册 + +__all__ = ["hello"] diff --git a/workers/demo/router/hello.py b/workers/demo/router/hello.py new file mode 100644 index 0000000..b496896 --- /dev/null +++ b/workers/demo/router/hello.py @@ -0,0 +1,25 @@ +"""问候端点 — 演示框架参数注入（worker_id）与插件私有配置读取""" +import time + +from chongming_worker import rpc_method + +from .. import global_var as gv +from ..models.pydantic import HelloGetRequest, HelloGetResponse + + +@rpc_method( + subject="demo.hello", + method="GET", + summary="进程插件示例：问候", + response_model=HelloGetResponse, +) +async def hello( + hello_get_request: HelloGetRequest, + worker_id: str = "", +) -> HelloGetResponse: + """返回问候语 + 处理请求的 worker 副本 ID（证明请求经独立进程处理）""" + return HelloGetResponse( + greeting=f"{gv.greeting_prefix}, {hello_get_request.name}!", + worker_id=worker_id, + timestamp=time.time(), + )