diff --git a/aibox_sophon_adapter/routes.py b/aibox_sophon_adapter/routes.py index fcbf76d..3c36dcc 100644 --- a/aibox_sophon_adapter/routes.py +++ b/aibox_sophon_adapter/routes.py @@ -500,10 +500,19 @@ async def alarm_report( # 2. ip 音柱 TTS 播放(可通过配置开关关闭),失败时仅记录日志,不影响主流程 from .global_var import ip_sound_column_volume, ip_sound_column_tts + + properties = full_data.get("Result", {}).get("Properties", []) + if len(properties) > 0: + properties_dict = { + prop.get("property", ""): prop.get("value", "") for prop in properties + } + area = f"{properties_dict.get("IntrudeRegionId", "")}," + else: + area = "" tts_text = ip_sound_column_tts.get( full_data.get("Summary", ""), - f"告警: {full_data.get('Summary', '')}" + f"告警: {area}{full_data.get('TaskDesc', '')}".strip() ) tts_payload = { diff --git a/auth/README.md b/auth/README.md deleted file mode 100644 index b387995..0000000 --- a/auth/README.md +++ /dev/null @@ -1,205 +0,0 @@ -# Auth Worker 插件 — 用户管理 - -auth 域拆分后的**用户管理插件**(可替换的业务部分)。 -登录 / token 签发 / admin 种子已**内置在网关**(管理平面底线,不依赖本插件是否 -加载,本插件缩容到 0 或全部副本故障时登录与 admin API 仍可用)。 - -共享契约归 `chongming-auth` 包(`utils/auth/`):Users 模型(唯一权威定义)、 -建库、登录校验、JWT 签发。本插件与网关内置登录**共用同一张 users 表**。 - ---- - -## 功能 - -- **用户注册**:用户名 + 邮箱 + 密码注册,自动校验唯一性 -- **获取用户信息**:根据 JWT Token 获取完整用户档案 -- **修改密码**:旧密码验证 + 新密码更新 -- **更新资料**:更新全名、邮箱、头像、手机号等信息 - -(登录 `POST /auth/login` 由网关内置提供,路径与响应 shape 不变) - ---- - -## 数据库 - -使用 SQLite(通过 `chongming-database` 异步封装),数据库文件位于 -`data/auth.db`(`[workers.config].data_dir` 下);配置 `database.url` 时走 -PostgreSQL。库位置与网关内置登录读同一份配置(本插件 `config.toml` 的 -`[workers.config]`),指向同一张 users 表。 - -### Users 表结构 - -权威定义见 `chongming_auth.models.Users`(本仓库 `models/auth.py` 仅为 re-export): - -| 字段 | 类型 | 说明 | -| ----------------- | ----------------- | --------------------------------- | -| `id` | `str` (PK) | UUID 格式主键 | -| `username` | `str(50)` UNIQUE | 登录用户名 | -| `email` | `str(255)` UNIQUE | 电子邮箱 | -| `hashed_password` | `str(255)` | Argon2id 加密密码 | -| `full_name` | `str(100)` | 用户全名(可选) | -| `avatar_url` | `str(500)` | 头像链接(可选) | -| `phone` | `str(20)` | 手机号码(可选) | -| `is_active` | `bool` | 账户启用状态(默认 `true`) | -| `is_verified` | `bool` | 邮箱/手机验证状态(默认 `false`) | -| `is_superuser` | `bool` | 超级管理员标识(默认 `false`) | -| `created_at` | `datetime` | 创建时间 | -| `updated_at` | `datetime` | 最后更新时间 | -| `last_login_at` | `datetime` | 最后登录时间(可选) | - -密码使用 Rust 实现的 **Argon2id** 哈希,内存成本 19456 KiB,迭代 2 次,并行度 1。 - ---- - -## API 端点 - -所有端点路由前缀为 `/auth`。 - -### `POST /auth/login` — 登录(网关内置) - -无需认证。由网关核心 `chongming/app/routes/auth.py` 提供(非本插件), -路径与响应 shape 与拆分前一致。JWT claims 契约见 `chongming-auth` 包 README。 - -```json -// 请求 -{ - "username": "admin", - "password": "admin" -} - -// 响应 -{ - "result": "success", - "token": "eyJhbGciOiJIUzI1NiIs...", - "message": "登录成功", - "timestamp": 1720000000.123 -} -``` - -### `POST /auth/register` — 用户注册 - -无需认证。 - -```json -// 请求 -{ - "username": "john", - "password": "secret123", - "email": "john@example.com", - "full_name": "John Doe", - "phone": "13800138000" -} - -// 响应 -{ - "result": "success", - "message": "注册成功", - "user_id": "a1b2c3d4_12345_1720000000.123", - "timestamp": 1720000000.123 -} -``` - -### `GET /auth/me?user_id=` — 获取用户信息 - -需要 JWT 鉴权(`Authorization: Bearer `)。 - -```json -// 响应 -{ - "result": "success", - "id": "a1b2c3d4_12345_1720000000.123", - "username": "john", - "email": "john@example.com", - "full_name": "John Doe", - "avatar_url": null, - "phone": "13800138000", - "is_active": true, - "is_verified": false, - "is_superuser": false, - "created_at": "2024-01-01T00:00:00+00:00", - "updated_at": "2024-01-01T00:00:00+00:00", - "last_login_at": null -} -``` - -### `POST /auth/change_password` — 修改密码 - -需要 JWT 鉴权。 - -```json -// 请求 -{ - "user_id": "a1b2c3d4_12345_1720000000.123", - "old_password": "secret123", - "new_password": "newSecret456" -} - -// 响应 -{ - "result": "success", - "message": "密码修改成功", - "timestamp": 1720000000.123 -} -``` - -### `POST /auth/update_profile` — 更新用户资料 - -需要 JWT 鉴权。仅更新非空字段。 - -```json -// 请求 -{ - "user_id": "a1b2c3d4_12345_1720000000.123", - "full_name": "John Updated", - "email": "", - "avatar_url": "https://example.com/avatar.png", - "phone": "" -} - -// 响应 -{ - "result": "success", - "message": "资料更新成功", - "timestamp": 1720000000.123 -} -``` - ---- - -## 默认管理员账户 - -admin 种子逻辑随登录一并内置到网关(网关 lifespan 的 `setup_auth_store` -→ `chongming_auth.seed_admin`,幂等,多进程并发启动安全),不再依赖本插件。 -数据库 `auth.db` 本插件仍由 `Plugin` 基类统一初始化并在 cleanup 时释放。 - -首次启动时自动创建: - -| 字段 | 值 | -| ------------ | --------------------- | -| **用户名** | `admin` | -| **密码** | `admin` | -| **邮箱** | `admin@example.com` | -| **超管权限** | `is_superuser = true` | - -> ⚠️ **生产环境务必修改默认密码!** - ---- - -## 安全说明 - -- 密码使用 **Argon2id**(Rust 实现),对抗 GPU/ASIC 暴力破解 -- JWT 使用 **HS256** 算法,密钥在 `config.toml` 的 `[security]` 节配置 -- JWT 有效期 3 小时,过期后需重新登录 -- 支持用户软禁用(`is_active = false`) -- 接口级鉴权控制:每个端点可通过 TOML 的 `auth_required` 独立配置 - ---- - -## 依赖 - -| 工具包 | 用途 | -| -------------------- | ------------------------------------ | -| `chongming-auth` | Users 模型 / 建库 / 登录校验 / 签发(共享契约) | -| `chongming-database` | 异步 SQLite 数据库操作 | -| `chongming-config` | 全局配置加载 | -| `chongming-worker` | `@rpc_method` 装饰器 + Worker 运行时 | diff --git a/auth/__init__.py b/auth/__init__.py deleted file mode 100644 index 8d63fb7..0000000 --- a/auth/__init__.py +++ /dev/null @@ -1,45 +0,0 @@ -""" -Auth Worker 插件 — 用户管理(注册 / 资料 / 改密) - -auth 域拆分后本插件只保留用户管理功能(可替换的业务部分); -登录 / token 签发 / admin 种子已内置在网关(chongming_auth 包 + -chongming.app.routes.auth),不依赖本插件是否加载。 - -基于 Plugin 基类统一生命周期:数据库由基类初始化(auth.db,与网关内置 -登录共用同一张 users 表),on_cleanup 无需重复 dispose。 -""" -import logging - -from chongming_worker.plugin import Plugin - -from .router import * # noqa: F401,F403 触发 @rpc_method 注册 - -# 模块加载时即注册 SQLModel 元数据 —— 基类 _bootstrap 建表(create_all)只创建 -# 已导入的表,若拖到 on_init 才导入会导致全新库缺少 users 表(no such table)。 -# Users 权威定义在 chongming_auth 包,此处 re-export 注册元数据。 -from .models.auth import Users # noqa: F401 - -logger = logging.getLogger(__name__) - - -class AuthPlugin(Plugin): - """Auth 插件 — 统一生命周期。""" - - type = "auth" - db_name = "auth.db" # 自动在 data_dir 下建 SQLite 库(与网关内置登录共用) - auto_bind = True # 首次初始化时自动 auto_bind_models("auth") - - async def on_init(self): - """首次 create_workers 时调用一次。""" - # 写入全局变量供 router 使用(数据库由基类在 cleanup 时统一 dispose) - self.set_var("db", self.db) - - async def on_cleanup(self): - """cleanup 时调用(数据库由基类处理)。""" - logger.info("[auth] 清理完成") - - -# 暴露给 app lifespan 的模块级入口(bound method,签名与旧写法完全一致) -plugin = AuthPlugin() -create_workers = plugin.create_workers -cleanup = plugin.cleanup diff --git a/auth/global_var.py b/auth/global_var.py deleted file mode 100644 index 17f42cd..0000000 --- a/auth/global_var.py +++ /dev/null @@ -1,26 +0,0 @@ -from typing import Any - -from chongming_config import get_config -from chongming_database import Database -from .models import * - -def set_global_var(name: str, value: Any) -> None: - """ - 设置当前模块的全局变量。 - - Args: - name: 变量名(字符串) - value: 任意值 - """ - globals()[name] = value - -config = get_config() - -db: Database | None = None - -# JWT 签发已随登录内置到网关(chongming.app.routes.auth / chongming_auth 包), -# 插件不再持有 jwt 实例 - -__all__ = [ - "db", -] diff --git a/auth/models/__init__.py b/auth/models/__init__.py deleted file mode 100644 index 677de46..0000000 --- a/auth/models/__init__.py +++ /dev/null @@ -1,30 +0,0 @@ -""" -模型导出 — auth -自动生成,请勿手动修改 -(LoginPostRequest/LoginPostResponse 已随内置登录迁移至 chongming_auth 包) -""" - -from .pydantic import AuthResponse -from .pydantic import ChangePasswordPostRequest -from .pydantic import ChangePasswordPostResponse -from .pydantic import MeGetRequest -from .pydantic import MeGetResponse -from .pydantic import RegisterPostRequest -from .pydantic import RegisterPostResponse -from .pydantic import UpdateProfilePostRequest -from .pydantic import UpdateProfilePostResponse - -from .auth import Users - -__all__ = [ - "AuthResponse", - "ChangePasswordPostRequest", - "ChangePasswordPostResponse", - "MeGetRequest", - "MeGetResponse", - "RegisterPostRequest", - "RegisterPostResponse", - "UpdateProfilePostRequest", - "UpdateProfilePostResponse", - "Users", -] diff --git a/auth/models/auth.py b/auth/models/auth.py deleted file mode 100644 index f1c2d7a..0000000 --- a/auth/models/auth.py +++ /dev/null @@ -1,10 +0,0 @@ -"""Users 模型 re-export —— 权威定义已迁移至共享包 chongming_auth。 - -auth 域拆分后:Users 模型 / 建库 / 登录校验 / JWT 签发归 chongming_auth 包 -(网关内置登录与 auth 插件共用,指向同一张 users 表)。本模块保留 re-export -以兼容插件内既有相对导入(from ..models.auth import Users); -re-export 保证全进程单一模型定义,SQLModel 元数据不会重复注册。 -""" -from chongming_auth import Users # noqa: F401 - -__all__ = ["Users"] diff --git a/auth/models/pydantic.py b/auth/models/pydantic.py deleted file mode 100644 index ffad0ff..0000000 --- a/auth/models/pydantic.py +++ /dev/null @@ -1,75 +0,0 @@ -""" -自动生成的 Pydantic 模型 — auth -来源: config/workers/auth/ -""" - -from __future__ import annotations - -from typing import Any, Optional -from pydantic import BaseModel, Field - -class ChangePasswordPostRequest(BaseModel): - """修改密码 — 请求体""" - user_id: str = Field(...) - old_password: str = Field(...) - new_password: str = Field(...) - -class ChangePasswordPostResponse(BaseModel): - """修改密码 — 响应体""" - result: str = Field(...) - message: str = Field(...) - timestamp: Optional[float] = Field(default=None) - -class MeGetRequest(BaseModel): - """获取当前用户信息 — 请求体""" - user_id: str = Field(...) - -class MeGetResponse(BaseModel): - """获取当前用户信息 — 响应体""" - result: str = Field(...) - id: str = Field(...) - username: str = Field(...) - email: str = Field(...) - full_name: Optional[str] = Field(default=None) - avatar_url: Optional[str] = Field(default=None) - phone: Optional[str] = Field(default=None) - is_active: bool = Field(...) - is_verified: bool = Field(...) - is_superuser: bool = Field(...) - created_at: Optional[str] = Field(default=None) - updated_at: Optional[str] = Field(default=None) - last_login_at: Optional[str] = Field(default=None) - -class RegisterPostRequest(BaseModel): - """用户注册 — 请求体""" - username: str = Field(...) - password: str = Field(...) - email: str = Field(...) - full_name: str = Field(...) - phone: str = Field(...) - -class RegisterPostResponse(BaseModel): - """用户注册 — 响应体""" - result: str = Field(...) - message: str = Field(...) - user_id: Optional[str] = Field(default=None) - timestamp: Optional[float] = Field(default=None) - -class UpdateProfilePostRequest(BaseModel): - """更新用户资料 — 请求体""" - user_id: str = Field(...) - full_name: str = Field(...) - email: str = Field(...) - avatar_url: str = Field(...) - phone: str = Field(...) - -class UpdateProfilePostResponse(BaseModel): - """更新用户资料 — 响应体""" - result: str = Field(...) - message: str = Field(...) - timestamp: Optional[float] = Field(default=None) - - -# 所有响应模型的联合类型 -# (登录响应 LoginPostResponse 已随内置登录迁移至 chongming_auth 包) -AuthResponse = ChangePasswordPostResponse | MeGetResponse | UpdateProfilePostResponse | RegisterPostResponse diff --git a/auth/router/__init__.py b/auth/router/__init__.py deleted file mode 100644 index 840f2f0..0000000 --- a/auth/router/__init__.py +++ /dev/null @@ -1,7 +0,0 @@ -from .register import register -from .me import me -from .update_profile import update_profile -from .change_password import change_password - - -__all__ = ["register", "me", "update_profile", "change_password"] diff --git a/auth/router/change_password.py b/auth/router/change_password.py deleted file mode 100644 index 98a6216..0000000 --- a/auth/router/change_password.py +++ /dev/null @@ -1,81 +0,0 @@ -import time -from datetime import datetime - -from sqlalchemy import select - -from chongming_worker import rpc_method -from chongming_password import hash_password, verify_password - -from ..models.pydantic import ( - ChangePasswordPostRequest, - ChangePasswordPostResponse -) -from ..models.auth import Users - - -@rpc_method( - subject="auth.change_password", - method="POST", - summary="修改密码", - response_model=ChangePasswordPostResponse, -) -async def change_password(change_password_post_request: ChangePasswordPostRequest) -> ChangePasswordPostResponse: - from ..global_var import db - - if db is None: - raise RuntimeError("Database not initialized") - - user_id = change_password_post_request.user_id - old_password = change_password_post_request.old_password - new_password = change_password_post_request.new_password - - if not old_password: - return ChangePasswordPostResponse( - result="fail", - message="旧密码不能为空", - timestamp=time.time() - ) - if not new_password: - return ChangePasswordPostResponse( - result="fail", - message="新密码不能为空", - timestamp=time.time() - ) - if old_password == new_password: - return ChangePasswordPostResponse( - result="fail", - message="新密码不能与旧密码相同", - timestamp=time.time() - ) - - async with db.get_session() as session: - stmt = select(Users).filter_by(id=user_id) - result = await session.execute(stmt) - user: Users | None = result.scalar_one_or_none() - - if user is None: - return ChangePasswordPostResponse( - result="fail", - message="用户不存在", - timestamp=time.time() - ) - - # 验证旧密码 - if not verify_password(old_password, user.hashed_password): - return ChangePasswordPostResponse( - result="fail", - message="旧密码错误", - timestamp=time.time() - ) - - # 更新密码 - user.hashed_password = hash_password(new_password) - user.updated_at = datetime.now() - session.add(user) - await session.commit() - - return ChangePasswordPostResponse( - result="success", - message="密码修改成功", - timestamp=time.time() - ) diff --git a/auth/router/me.py b/auth/router/me.py deleted file mode 100644 index 7251a71..0000000 --- a/auth/router/me.py +++ /dev/null @@ -1,56 +0,0 @@ -import time - -from sqlalchemy import select - -from chongming_worker import rpc_method - -from ..models.pydantic import ( - MeGetRequest, - MeGetResponse -) -from ..models.auth import Users - - -@rpc_method( - subject="auth.me", - method="GET", - summary="获取当前用户信息", - response_model=MeGetResponse, -) -async def me(me_get_request: MeGetRequest) -> MeGetResponse: - from ..global_var import db - - if db is None: - raise RuntimeError("Database not initialized") - - async with db.get_session() as session: - stmt = select(Users).filter_by(id=me_get_request.user_id) - result = await session.execute(stmt) - user: Users | None = result.scalar_one_or_none() - - if user is None: - return MeGetResponse( - result="fail", - id="", - username="", - email="", - is_active=False, - is_verified=False, - is_superuser=False, - ) - - return MeGetResponse( - result="success", - id=user.id, - username=user.username, - email=user.email, - full_name=user.full_name, - avatar_url=user.avatar_url, - phone=user.phone, - is_active=user.is_active, - is_verified=user.is_verified, - is_superuser=user.is_superuser, - created_at=user.created_at.isoformat() if user.created_at else None, - updated_at=user.updated_at.isoformat() if user.updated_at else None, - last_login_at=user.last_login_at.isoformat() if user.last_login_at else None, - ) diff --git a/auth/router/register.py b/auth/router/register.py deleted file mode 100644 index 8d2fa20..0000000 --- a/auth/router/register.py +++ /dev/null @@ -1,93 +0,0 @@ -import time -from datetime import datetime - -from sqlalchemy import select - -from chongming_worker import rpc_method -from chongming_password import hash_password - -from ..models.pydantic import ( - RegisterPostRequest, - RegisterPostResponse -) -from ..models.auth import Users - - -@rpc_method( - subject="auth.register", - method="POST", - summary="用户注册", - response_model=RegisterPostResponse, -) -async def register(register_post_request: RegisterPostRequest) -> RegisterPostResponse: - from ..global_var import db - - if db is None: - raise RuntimeError("Database not initialized") - - username = register_post_request.username.strip() - email = register_post_request.email.strip() - full_name = register_post_request.full_name.strip() or None - phone = register_post_request.phone.strip() or None - - # 校验必填字段 - if not username: - return RegisterPostResponse( - result="fail", - message="用户名不能为空", - timestamp=time.time() - ) - if not email: - return RegisterPostResponse( - result="fail", - message="邮箱不能为空", - timestamp=time.time() - ) - if not register_post_request.password: - return RegisterPostResponse( - result="fail", - message="密码不能为空", - timestamp=time.time() - ) - - async with db.get_session() as session: - # 检查用户名是否已存在 - stmt = select(Users).filter_by(username=username) - result = await session.execute(stmt) - if result.scalar_one_or_none() is not None: - return RegisterPostResponse( - result="fail", - message="用户名已存在", - timestamp=time.time() - ) - - # 检查邮箱是否已存在 - stmt = select(Users).filter_by(email=email) - result = await session.execute(stmt) - if result.scalar_one_or_none() is not None: - return RegisterPostResponse( - result="fail", - message="邮箱已被注册", - timestamp=time.time() - ) - - # 创建新用户 - user = Users( - username=username, - hashed_password=hash_password(register_post_request.password), - email=email, - full_name=full_name, - phone=phone, - created_at=datetime.now(), - updated_at=datetime.now(), - ) - session.add(user) - await session.commit() - await session.refresh(user) - - return RegisterPostResponse( - result="success", - message="注册成功", - user_id=user.id, - timestamp=time.time() - ) diff --git a/auth/router/update_profile.py b/auth/router/update_profile.py deleted file mode 100644 index fa4b99a..0000000 --- a/auth/router/update_profile.py +++ /dev/null @@ -1,74 +0,0 @@ -import time -from datetime import datetime - -from sqlalchemy import select - -from chongming_worker import rpc_method - -from ..models.pydantic import ( - UpdateProfilePostRequest, - UpdateProfilePostResponse -) -from ..models.auth import Users - - -@rpc_method( - subject="auth.update_profile", - method="POST", - summary="更新用户资料", - response_model=UpdateProfilePostResponse, -) -async def update_profile(update_profile_post_request: UpdateProfilePostRequest) -> UpdateProfilePostResponse: - from ..global_var import db - - if db is None: - raise RuntimeError("Database not initialized") - - user_id = update_profile_post_request.user_id - - async with db.get_session() as session: - stmt = select(Users).filter_by(id=user_id) - result = await session.execute(stmt) - user: Users | None = result.scalar_one_or_none() - - if user is None: - return UpdateProfilePostResponse( - result="fail", - message="用户不存在", - timestamp=time.time() - ) - - # 更新字段(只更新非空值) - full_name = update_profile_post_request.full_name.strip() - email = update_profile_post_request.email.strip() - avatar_url = update_profile_post_request.avatar_url.strip() - phone = update_profile_post_request.phone.strip() - - if full_name: - user.full_name = full_name - if email: - # 检查邮箱是否被其他用户占用 - if email != user.email: - check_stmt = select(Users).filter_by(email=email) - check_result = await session.execute(check_stmt) - if check_result.scalar_one_or_none() is not None: - return UpdateProfilePostResponse( - result="fail", - message="邮箱已被其他用户使用", - timestamp=time.time() - ) - user.email = email - if avatar_url: - user.avatar_url = avatar_url - if phone: - user.phone = phone - - user.updated_at = datetime.now() - session.add(user) - await session.commit() - - return UpdateProfilePostResponse( - result="success", - message="资料更新成功", - timestamp=time.time() - ) diff --git a/auth_oauth/README.md b/auth_oauth/README.md deleted file mode 100644 index d7bf4b0..0000000 --- a/auth_oauth/README.md +++ /dev/null @@ -1,61 +0,0 @@ -# Auth OAuth 插件 — 第三方授权登录示例(身份源插件) - -基于 `chongming-auth` 共享契约实现**身份源插件**的参考实现:准入判断归插件, -平台凭证签发归 `chongming_auth.issue_token`(claims 契约与网关内置登录完全一致)。 -第三方身份登录后拿到的 token 与内置登录**完全等价**——网关 JWT 验签、 -`user_info` 随信封透传、admin API 判超管全部无缝生效。 - -## 提供的身份源 - -| 身份源 | 模式 | 端点 | -| --- | --- | --- | -| GitHub | OAuth2 授权码(SPA JSON 流程) | `GET /auth/oauth/github_authorize`、`POST /auth/oauth/github_callback` | -| 手机号 | 一次性验证码(演示模式) | `POST /auth/oauth/phone_send_code`、`POST /auth/oauth/phone_login` | - -## 身份源插件的标准形态(写新身份源时照此扩展) - -1. **准入**:校验第三方凭据(OAuth code / OTP / LDAP 密码 / ...); -2. **映射**:在插件自有绑定表 `identity_bindings(provider, provider_uid, user_id)` - 中 find-or-create——老用户换身份源登录仍是同一账号体系; -3. **建号**:无绑定时新建 `users` 账号(密码为随机不可用值)+ 绑定记录, - 与网关内置登录**共用同一张 users 表**(`chongming_auth.open_database`); -4. **签发**:`chongming_auth.issue_token(user, secret, algorithm)`。 - -扩展新身份源(LDAP / 企业微信 / 飞书 / OIDC ...)只需:加一个 provider 对接模块 -+ 一组 TOML 路由 + 一个 `@rpc_method` 端点,调 `store.find_or_create_user` 即可。 - -## 配置 - -`config/workers/auth_oauth/config.toml` 的 `[workers.config]`: - -```toml -return_otp_in_response = true # 演示模式:验证码随响应返回。生产必须 false 并接短信通道 - -[workers.config.github] -client_id / client_secret / redirect_uri / scope -``` - -## 安全说明(务必阅读) - -- **本插件须以 inline 模式运行**(默认):`issue_token` 需要 `[security]` 签名密钥, - external 进程持有该密钥即能铸造超管 token。external 化前需平台提供内部签发 API - 或按 issuer 分离密钥。 -- `return_otp_in_response` 仅限本地演示,生产必须关闭并接真实短信通道。 -- OAuth state 与 OTP 验证码均存共享缓存(`chongming_cache`),一次性消费、300s 过期。 - -## 踩坑记录(写插件必读) - -**插件必须提供 `models/pydantic.py`**(gen_models 产物位置,手写等价物亦可)。 -网关契约校验按 `plugins..models.pydantic` → `src.plugins....` → `workers....` -顺序探测;缺失时 `src.plugins.` 兜底路径会以第二个模块名重新执行整个插件包 -(父包导入链),插件内自定义 SQLModel 表会在全局元数据中重复定义, -网关启动即崩(`Table 'xxx' is already defined for this MetaData instance`)。 - -## 依赖 - -| 工具包 | 用途 | -| --- | --- | -| `chongming-auth` | Users 模型 / 建库 / 登录校验 / issue_token(共享契约) | -| `chongming-cache` | OAuth state 与 OTP 验证码存储 | -| `chongming-worker` | `@rpc_method` 装饰器 + Worker 运行时 | -| `httpx` | GitHub OAuth API 调用 | diff --git a/auth_oauth/__init__.py b/auth_oauth/__init__.py deleted file mode 100644 index 197d544..0000000 --- a/auth_oauth/__init__.py +++ /dev/null @@ -1,65 +0,0 @@ -"""auth_oauth 身份源插件 — 第三方授权登录示例(基于 chongming-auth 共享契约)。 - -演示 auth 域拆分后的身份源插件标准形态: - - 准入判断归插件(GitHub OAuth 授权码模式 / 手机验证码模式) - - 平台凭证签发归 chongming_auth.issue_token(claims 契约与内置登录完全一致) - - 账号落地到共享 users 表,第三方身份与账号的映射存插件自有绑定表 - (identity_bindings,建在同一 auth 库,不改动 chongming-auth 拥有的 users schema) - -演示模式说明: - - 手机验证码默认直接随响应返回 dev_code(无短信通道的本地演示用), - 生产必须接真实短信通道并关闭 return_otp_in_response - - GitHub OAuth 需在 [workers.config.github] 配置 client_id/secret/redirect_uri -""" -import logging - -from chongming_worker.plugin import Plugin - -from .router import * # noqa: F401,F403 触发 @rpc_method 注册 - -# 模块加载时即注册 SQLModel 元数据(identity_bindings 表与 users 同库, -# 由 chongming_auth.open_database 的 create_all 一并创建) -from .models import IdentityBinding # noqa: F401 - -logger = logging.getLogger(__name__) - - -class AuthOAuthPlugin(Plugin): - """AuthOAuth 插件 — 身份源示例,统一生命周期。""" - - type = "auth_oauth" - # 不用基类建库:与网关内置登录共用同一 auth 库(chongming_auth.open_database - # 按 [workers.config] 的 data_dir / database 解析,指向 users + identity_bindings) - db_name = None - auto_bind = False # 请求/响应模型由端点函数的类型注解直接提供,无需 gen_models - - async def on_init(self): - """首次 create_workers 时调用一次:打开共享 auth 库 + 缓存。""" - from chongming_auth import open_database - from chongming_cache import Cache - from chongming_config import get_config - - db = await open_database(self.config) - cache = Cache.from_config(get_config()["cache"]) - - # 写入全局变量供 router 使用(cleanup 时统一释放) - self.set_var("db", db) - self.set_var("cache", cache) - self.set_var("plugin_config", dict(self.config or {})) - logger.info("[auth_oauth] 初始化完成(共享 auth 库 + 缓存就绪)") - - async def on_cleanup(self): - """cleanup 时调用:释放自有资源(基类不再托管本插件的数据库)。""" - db = self.get_var("db") - cache = self.get_var("cache") - if db is not None: - await db.dispose() - if cache is not None: - cache.close() - logger.info("[auth_oauth] 清理完成") - - -# 暴露给 app lifespan 的模块级入口(bound method,签名与旧写法完全一致) -plugin = AuthOAuthPlugin() -create_workers = plugin.create_workers -cleanup = plugin.cleanup diff --git a/auth_oauth/global_var.py b/auth_oauth/global_var.py deleted file mode 100644 index c27ea2d..0000000 --- a/auth_oauth/global_var.py +++ /dev/null @@ -1,27 +0,0 @@ -from typing import Any - -from chongming_config import get_config -from chongming_database import Database -from chongming_cache import Cache - -def set_global_var(name: str, value: Any) -> None: - """ - 设置当前模块的全局变量。 - - Args: - name: 变量名(字符串) - value: 任意值 - """ - globals()[name] = value - -config = get_config() - -db: Database | None = None -cache: Cache | None = None # chongming_cache.Cache(OAuth state / OTP 验证码) -plugin_config: dict = {} # [workers.config] 运行时配置(github 凭据等) - -__all__ = [ - "db", - "cache", - "plugin_config", -] diff --git a/auth_oauth/models/__init__.py b/auth_oauth/models/__init__.py deleted file mode 100644 index d08ef60..0000000 --- a/auth_oauth/models/__init__.py +++ /dev/null @@ -1,21 +0,0 @@ -"""模型导出 — auth_oauth(含插件自有绑定表 + 请求/响应模型)""" - -from .pydantic import ( - GithubAuthorizeGetResponse, - GithubCallbackPostRequest, - PhoneLoginPostRequest, - PhoneSendCodePostRequest, - PhoneSendCodePostResponse, -) -from chongming_auth import LoginPostResponse -from .identity import IdentityBinding - -__all__ = [ - "GithubAuthorizeGetResponse", - "GithubCallbackPostRequest", - "LoginPostResponse", - "PhoneLoginPostRequest", - "PhoneSendCodePostRequest", - "PhoneSendCodePostResponse", - "IdentityBinding", -] diff --git a/auth_oauth/models/identity.py b/auth_oauth/models/identity.py deleted file mode 100644 index 283ae7c..0000000 --- a/auth_oauth/models/identity.py +++ /dev/null @@ -1,34 +0,0 @@ -"""身份源绑定表 — 第三方身份(github open_id / 手机号 / ...)与 users 账号的映射。 - -插件自有表,建在共享 auth 库(与 users 同库不同表): -chongming-auth 拥有的 users schema 保持最小不改动,身份源扩展物归插件。 -""" -from datetime import datetime -from uuid import uuid4 - -from sqlalchemy import UniqueConstraint, func, Column, DateTime -from sqlmodel import Field, SQLModel - - -def _pk() -> str: - return f"{uuid4()}_{int(datetime.now().timestamp())}" - - -class IdentityBinding(SQLModel, table=True): - __tablename__ = "identity_bindings" # type: ignore - __table_args__ = ( - UniqueConstraint("provider", "provider_uid", name="uq_provider_uid"), - ) - - id: str = Field(default_factory=_pk, primary_key=True, description="绑定记录ID") - user_id: str = Field(index=True, max_length=255, description="关联的 users.id") - provider: str = Field(index=True, max_length=32, description="身份源: github / phone / ...") - provider_uid: str = Field(max_length=255, description="身份源侧唯一标识(github user id / 手机号)") - created_at: datetime = Field( - sa_column=Column( - DateTime(timezone=True), - server_default=func.now(), - nullable=False, - comment="创建时间", - ) - ) diff --git a/auth_oauth/models/pydantic.py b/auth_oauth/models/pydantic.py deleted file mode 100644 index 6fefd97..0000000 --- a/auth_oauth/models/pydantic.py +++ /dev/null @@ -1,43 +0,0 @@ -"""auth_oauth 请求/响应模型。 - -框架契约:所有插件须有 models/pydantic.py(gen_models 的产物位置)—— -网关契约校验按 plugins..models.pydantic → src.plugins.... → workers.... -顺序探测,缺失时会以 src.plugins. 前缀二次执行插件包(父包导入链), -插件内自定义 SQLModel 表会在全局元数据中重复定义导致启动失败。 -本插件模型由手写(等价 gen_models 产物),端点函数以类型注解直接绑定。 -登录响应 LoginPostResponse 直接从 chongming_auth 导入(共享契约)。 -""" -from pydantic import BaseModel, Field - - -class GithubAuthorizeGetResponse(BaseModel): - """授权跳转 URL 响应""" - result: str = Field(..., description="success/fail") - authorize_url: str = Field(default="", description="GitHub 授权跳转 URL") - state: str = Field(default="", description="防 CSRF state(一次性,300s)") - expires_in: int = Field(default=300, description="state 有效期(秒)") - - -class GithubCallbackPostRequest(BaseModel): - """GitHub 回调请求""" - code: str = Field(..., description="GitHub 回跳授权码") - state: str = Field(..., description="authorize 返回的 state") - - -class PhoneSendCodePostRequest(BaseModel): - """发送验证码请求""" - phone: str = Field(..., description="手机号") - - -class PhoneSendCodePostResponse(BaseModel): - """发送验证码响应""" - result: str = Field(..., description="success/fail") - message: str = Field(..., description="结果消息") - expires_in: int = Field(default=300, description="验证码有效期(秒)") - dev_code: str | None = Field(default=None, description="验证码(仅演示模式返回)") - - -class PhoneLoginPostRequest(BaseModel): - """验证码登录请求""" - phone: str = Field(..., description="手机号") - code: str = Field(..., description="短信验证码") diff --git a/auth_oauth/providers/__init__.py b/auth_oauth/providers/__init__.py deleted file mode 100644 index d1737d5..0000000 --- a/auth_oauth/providers/__init__.py +++ /dev/null @@ -1 +0,0 @@ -from . import github # noqa: F401 diff --git a/auth_oauth/providers/github.py b/auth_oauth/providers/github.py deleted file mode 100644 index 2d5e06c..0000000 --- a/auth_oauth/providers/github.py +++ /dev/null @@ -1,74 +0,0 @@ -"""GitHub OAuth 提供方对接(授权码模式)。 - -HTTP 调用独立成模块函数,便于测试替身(monkeypatch)注入。 -""" -import logging - -import httpx - -logger = logging.getLogger("plugins.auth_oauth.github") - -_TOKEN_URL = "https://github.com/login/oauth/access_token" -_USER_API = "https://api.github.com/user" -_USER_EMAILS_API = "https://api.github.com/user/emails" - - -async def exchange_code(client_id: str, client_secret: str, code: str, - redirect_uri: str) -> str: - """授权码换 access_token,失败抛 RuntimeError(业务语义,调用方转 fail 响应)。""" - async with httpx.AsyncClient(timeout=10.0) as client: - resp = await client.post( - _TOKEN_URL, - json={ - "client_id": client_id, - "client_secret": client_secret, - "code": code, - "redirect_uri": redirect_uri, - }, - headers={"Accept": "application/json"}, - ) - data = resp.json() - if "access_token" not in data: - detail = data.get("error_description") or data.get("error") or data - raise RuntimeError(f"GitHub 授权码换取失败: {detail}") - return data["access_token"] - - -async def fetch_user(access_token: str) -> dict: - """拉取 GitHub 用户并归一化为建号 profile。 - - Returns: - {uid, username, email, full_name, avatar_url} - Raises: - RuntimeError: API 失败或拿不到邮箱(users.email 非空唯一,必须可得)。 - """ - async with httpx.AsyncClient(timeout=10.0) as client: - headers = { - "Authorization": f"Bearer {access_token}", - "Accept": "application/vnd.github+json", - } - resp = await client.get(_USER_API, headers=headers) - if resp.status_code != 200: - raise RuntimeError(f"GitHub 用户信息获取失败: HTTP {resp.status_code}") - u = resp.json() - - email = u.get("email") - if not email: - # 私密邮箱:走 /user/emails 取主邮箱 - er = await client.get(_USER_EMAILS_API, headers=headers) - if er.status_code == 200 and isinstance(er.json(), list): - emails = er.json() - primary = next( - (e for e in emails if e.get("primary")), emails[0] if emails else None - ) - email = primary.get("email") if primary else None - if not email: - raise RuntimeError("GitHub 账号未提供可用邮箱,无法创建账号") - - return { - "uid": str(u["id"]), - "username": u.get("login") or f"github_{u['id']}", - "email": email, - "full_name": u.get("name"), - "avatar_url": u.get("avatar_url"), - } diff --git a/auth_oauth/router/__init__.py b/auth_oauth/router/__init__.py deleted file mode 100644 index 6f1fa00..0000000 --- a/auth_oauth/router/__init__.py +++ /dev/null @@ -1,2 +0,0 @@ -from .github import github_authorize, github_callback # noqa: F401 -from .phone import phone_send_code, phone_login # noqa: F401 diff --git a/auth_oauth/router/github.py b/auth_oauth/router/github.py deleted file mode 100644 index 0e33fab..0000000 --- a/auth_oauth/router/github.py +++ /dev/null @@ -1,86 +0,0 @@ -"""GitHub OAuth 授权码模式端点(SPA 友好的 JSON 流程)。 - -流程:前端调 authorize 拿 authorize_url → 浏览器跳转 GitHub → -GitHub 回跳前端页(带 code+state)→ 前端调 callback 换平台 token。 -""" -import logging -import secrets -from urllib.parse import urlencode - -from chongming_worker import rpc_method - -from ..models import GithubAuthorizeGetResponse, GithubCallbackPostRequest, LoginPostResponse -from ..providers import github as github_provider -from .. import store - -logger = logging.getLogger("plugins.auth_oauth.github") - -_STATE_TTL = 300 # state 有效期(秒),一次性消费防 CSRF - - -@rpc_method( - subject="auth_oauth.github_authorize", - method="GET", - summary="GitHub 授权登录:生成授权跳转 URL(state 存缓存防 CSRF)", -) -async def github_authorize() -> GithubAuthorizeGetResponse: - from ..global_var import cache, plugin_config - - gh = plugin_config.get("github") or {} - client_id = gh.get("client_id", "") - if not client_id or client_id.startswith("your-"): - return GithubAuthorizeGetResponse( - result="fail", authorize_url="", - state="GitHub OAuth 未配置(config/workers/auth_oauth/config.toml 的 [workers.config.github])", - ) - - state = secrets.token_urlsafe(24) - cache.set(f"auth_oauth:state:{state}", "1", expire=_STATE_TTL) - - params = urlencode({ - "client_id": client_id, - "redirect_uri": gh.get("redirect_uri", ""), - "scope": gh.get("scope", "read:user user:email"), - "state": state, - }) - return GithubAuthorizeGetResponse( - result="success", - authorize_url=f"https://github.com/login/oauth/authorize?{params}", - state=state, - ) - - -@rpc_method( - subject="auth_oauth.github_callback", - method="POST", - summary="GitHub 授权回调:校验 state、换取用户信息,find-or-create 后签发平台 token", -) -async def github_callback(req: GithubCallbackPostRequest) -> LoginPostResponse: - from ..global_var import cache, db, plugin_config - - if db is None or cache is None: - return store.fail_response("插件未初始化") - - # state 一次性消费(防 CSRF / 重放) - state_key = f"auth_oauth:state:{req.state}" - if not cache.get(state_key): - return store.fail_response("state 无效或已过期,请重新发起授权") - cache.delete(state_key) - - gh = plugin_config.get("github") or {} - try: - access_token = await github_provider.exchange_code( - gh.get("client_id", ""), gh.get("client_secret", ""), - req.code, gh.get("redirect_uri", ""), - ) - profile = await github_provider.fetch_user(access_token) - except RuntimeError as e: - logger.warning("GitHub 回调失败: %s", e) - return store.fail_response(str(e)) - - user, message = await store.find_or_create_user( - db, "github", profile["uid"], profile - ) - if user is None: - return store.fail_response(message) - return store.issue_login_response(user) diff --git a/auth_oauth/router/phone.py b/auth_oauth/router/phone.py deleted file mode 100644 index e5fd904..0000000 --- a/auth_oauth/router/phone.py +++ /dev/null @@ -1,82 +0,0 @@ -"""手机验证码登录端点(演示模式:验证码存共享缓存、默认随响应返回)。 - -生产接入真实短信通道后,将 send_code 中的 dev_code 分支关闭 -(config: return_otp_in_response = false),验证码只经短信下发。 -""" -import logging -import re -import secrets - -from chongming_worker import rpc_method - -from ..models import ( - LoginPostResponse, - PhoneLoginPostRequest, - PhoneSendCodePostRequest, - PhoneSendCodePostResponse, -) -from .. import store - -logger = logging.getLogger("plugins.auth_oauth.phone") - -_OTP_TTL = 300 # 验证码有效期(秒) -_PHONE_RE = re.compile(r"^1\d{10}$") # 演示按中国大陆手机号校验 - - -def _otp_key(phone: str) -> str: - return f"auth_oauth:otp:{phone}" - - -@rpc_method( - subject="auth_oauth.phone_send_code", - method="POST", - summary="手机验证码登录:下发验证码(演示模式:存缓存并随响应返回)", -) -async def phone_send_code(req: PhoneSendCodePostRequest) -> PhoneSendCodePostResponse: - from ..global_var import cache, plugin_config - - if cache is None: - return PhoneSendCodePostResponse(result="fail", message="插件未初始化") - if not _PHONE_RE.match(req.phone): - return PhoneSendCodePostResponse(result="fail", message="手机号格式不正确") - - code = f"{secrets.randbelow(1_000_000):06d}" - cache.set(_otp_key(req.phone), code, expire=_OTP_TTL) - - # 生产环境在此接真实短信通道;演示模式打印并将验证码随响应返回 - logger.info("[auth_oauth] 验证码已生成 phone=%s code=%s(演示模式,请勿用于生产)", - req.phone, code) - if plugin_config.get("return_otp_in_response", True): - return PhoneSendCodePostResponse( - result="success", message="验证码已发送(演示模式直接返回)", dev_code=code, - ) - return PhoneSendCodePostResponse(result="success", message="验证码已发送") - - -@rpc_method( - subject="auth_oauth.phone_login", - method="POST", - summary="手机验证码登录:校验验证码,find-or-create 后签发平台 token", -) -async def phone_login(req: PhoneLoginPostRequest) -> LoginPostResponse: - from ..global_var import cache, db - - if db is None or cache is None: - return store.fail_response("插件未初始化") - - key = _otp_key(req.phone) - stored = cache.get(key) - if not stored or stored != req.code: - return store.fail_response("验证码错误或已过期") - cache.delete(key) # 一次性消费 - - user, message = await store.find_or_create_user( - db, "phone", req.phone, - profile={ - "username": req.phone, - "email": f"{req.phone}@phone.local", - }, - ) - if user is None: - return store.fail_response(message) - return store.issue_login_response(user) diff --git a/auth_oauth/store.py b/auth_oauth/store.py deleted file mode 100644 index 589606e..0000000 --- a/auth_oauth/store.py +++ /dev/null @@ -1,114 +0,0 @@ -"""身份源插件共享逻辑:绑定查找、账号 find-or-create、平台 token 签发。 - -账号落点: - - 已有绑定 → 直接映射到既有 users 账号(老用户换身份源登录也是同一账号体系) - - 无绑定 → 新建 users 账号(密码为随机值,本插件身份源不走密码登录)+ 绑定记录 -签发: - - chongming_auth.issue_token —— claims 契约与网关内置登录完全一致, - 网关验签 / user_info 透传 / admin 判超管全部无缝生效 -""" -import logging -import secrets -from datetime import datetime - -from chongming_auth import LoginPostResponse, Users, issue_token -from chongming_config import get_config -from chongming_database import Database -from chongming_password import hash_password -from sqlalchemy import select -from sqlalchemy.exc import IntegrityError -import time - -from .models import IdentityBinding - -logger = logging.getLogger("plugins.auth_oauth") - -# 演示身份源建号使用的占位密码(随机、不可用于密码登录) -_UNUSABLE_PASSWORD_BYTES = 32 - - -async def _lookup_user(db: Database, provider: str, - provider_uid: str) -> Users | None: - """按绑定关系定位既有账号;无绑定或账号已消失返回 None。""" - async with db.get_session() as session: - binding = ( - await session.execute( - select(IdentityBinding).where( - IdentityBinding.provider == provider, # type: ignore - IdentityBinding.provider_uid == provider_uid, # type: ignore - ) - ) - ).scalar_one_or_none() - if binding is None: - return None - user = ( - await session.execute(select(Users).where(Users.id == binding.user_id)) # type: ignore - ).scalar_one_or_none() - return user - - -async def _create_user_with_binding( - db: Database, provider: str, provider_uid: str, profile: dict -) -> None: - """新建 users 账号 + 绑定记录(同一事务)。 - - profile: {username, email, full_name?, avatar_url?} - """ - async with db.get_session() as session: - user = Users( - username=str(profile.get("username", ""))[:50], - email=str(profile["email"]), - hashed_password=hash_password(secrets.token_urlsafe(_UNUSABLE_PASSWORD_BYTES)), - full_name=profile.get("full_name"), - avatar_url=profile.get("avatar_url"), - created_at=datetime.now(), - updated_at=datetime.now(), - is_active=True, - ) - session.add(user) - await session.flush() # 取 user.id 供绑定外键 - session.add(IdentityBinding( - user_id=user.id, provider=provider, provider_uid=provider_uid, - )) # type: ignore - await session.commit() - - -async def find_or_create_user( - db: Database, provider: str, provider_uid: str, profile: dict -) -> tuple[Users | None, str]: - """按身份源 find-or-create 账号,返回 (user, message)。 - - user 非 None 即成功;唯一约束冲突(并发首登 / username·email 被占)时 - 回退到重查绑定,仍无则返回失败原因。 - """ - user = await _lookup_user(db, provider, provider_uid) - if user is not None: - return user, "" - - try: - await _create_user_with_binding(db, provider, provider_uid, profile) - except IntegrityError as e: - logger.warning("[%s] 建号冲突(%s),尝试复用既有绑定: %s", provider, provider_uid, e) - - user = await _lookup_user(db, provider, provider_uid) - if user is None: - return None, "创建用户失败(用户名或邮箱已被占用),请改用其他方式登录" - return user, "" - - -def issue_login_response(user: Users) -> LoginPostResponse: - """按平台契约签发 token 并组装与内置登录同 shape 的响应。""" - security = get_config()["security"] - token = issue_token( - user, security["secret_key"], security["algorithm"], - ) - return LoginPostResponse( - result="success", token=token, message="登录成功", timestamp=time.time() - ) - - -def fail_response(message: str) -> LoginPostResponse: - """业务失败(不抛异常):result=fail + 原文案,保持与内置登录同 shape。""" - return LoginPostResponse( - result="fail", token="", message=message, timestamp=time.time() - ) diff --git a/doubao/README.md b/doubao/README.md deleted file mode 100644 index 492fba0..0000000 --- a/doubao/README.md +++ /dev/null @@ -1,81 +0,0 @@ -# Doubao Worker 插件 - -火山引擎豆包(Doubao)视觉理解模型集成,支持图像分析与描述生成。 - ---- - -## 功能 - -- 接收 **base64 编码**的图片和**提示词**(prompt) -- 调用火山引擎豆包视觉模型 API -- 支持**思考模式**(thinking mode),模型会先推理再输出结果 -- 返回模型生成的图像描述文本 - ---- - -## 配置 - -在 `config/workers/doubao/config.toml` 中设置: - -```toml -[workers.config] -type = "doubao" -api_key = "" # 火山引擎 API Key(也可通过环境变量 VOLC_ACCESSKEY 提供) -base_url = "https://ark.cn-beijing.volces.com/api/v3" # API 端点 -model = "doubao-seed-evolving" # 模型名称 -``` - ---- - -## API 端点 - -### `POST /doubao/analyze` — 图像分析 - -```json -// 请求 -{ - "image_base64": "data:image/png;base64,iVBORw0KGgoAAAANS...", - "prompt": "描述这张图片的内容" -} - -// 响应 -{ - "result": "图片中是一只橘猫坐在米色沙发上,阳光从窗户照进来...", - "error": "" -} -``` - -| 参数 | 类型 | 必填 | 说明 | -| -------------- | ----- | ---- | ----------------------------------------------------------- | -| `image_base64` | `str` | ✅ | Base64 编码的图片数据(可含 `data:image/...;base64,` 前缀) | -| `prompt` | `str` | ✅ | 给模型的提示词,描述期望的分析内容 | - ---- - -## 支持的模型 - -| 模型 | 说明 | -| ---------------------- | ---------------------------- | -| `doubao-seed-evolving` | 豆包视觉理解旗舰模型(默认) | - -可根据火山引擎最新文档更换 `model` 配置项。 - ---- - -## 实现细节 - -- 生命周期基于 `Plugin` 基类:`on_init` 按 `is_aicc` 分支创建 Ark(`AsyncArk`)或 - AICC(`AsyncOpenAI` + `AsyncSecureHttpClient`)客户端并写入 `global_var`; - `on_cleanup` 释放客户端引用(无数据库) -- API Key 优先取配置 `api_key`,其次环境变量 `VOLC_ACCESSKEY`(缺失时插件初始化报错) -- 支持 AICC 安全通信模式(懒加载 SDK,避免非 AICC 环境导入失败) -- 超时时间默认 30 秒,在 `config.toml` 的 `defaults.timeout` 中配置 - ---- - -## 依赖 - -| 工具包 | 用途 | -| ------------------ | ------------------------------------ | -| `httpx` | 异步 HTTP 客户端 | -| `chongming-worker` | `@rpc_method` 装饰器 + Worker 运行时 | diff --git a/doubao/__init__.py b/doubao/__init__.py deleted file mode 100644 index 569d01c..0000000 --- a/doubao/__init__.py +++ /dev/null @@ -1,106 +0,0 @@ -""" -豆包视觉理解 Worker 插件 - -基于 Plugin 基类统一生命周期:on_init 中按配置创建 Ark / AICC 客户端, -on_cleanup 释放客户端引用(无数据库)。 -""" -import os -import logging -from openai import AsyncOpenAI -from volcenginesdkarkruntime import AsyncArk - -from chongming_worker.plugin import Plugin -from .global_var import ( - set_ark_client, set_aicc_client, set_is_aicc, - cleanup_all, -) -from .router import * # noqa: F401,F403 触发 @rpc_method 注册 - -logger = logging.getLogger(__name__) - - -class DoubaoPlugin(Plugin): - """豆包插件 — 统一生命周期。""" - - type = "doubao" - db_name = None # 无数据库 - auto_bind = True # 首次初始化时自动 auto_bind_models("doubao") - - async def on_init(self): - """首次 create_workers 时调用一次:初始化豆包客户端。""" - config = self.config - logger.info("豆包插件初始化开始 (count=%s, aicc=%s)", - config.get("count", 1), config.get("is_aicc", False)) - - is_aicc = config.get("is_aicc", False) - set_is_aicc(is_aicc) - - api_key = config.get("api_key") or os.environ.get("VOLC_ACCESSKEY") - if not api_key: - logger.error("豆包插件初始化失败: 缺少 API_KEY") - raise ValueError("缺少 API_KEY,请在配置中设置 api_key 或环境变量 VOLC_ACCESSKEY") - - model = config.get("model", "doubao-seed-evolving") - os.environ["DOUBAO_MODEL"] = model - logger.info("豆包模型: %s", model) - - if is_aicc: - logger.info("豆包使用 AICC 安全通信模式") - # AICC 模式:懒加载,避免非 AICC 环境导入失败 - try: - from bytedance.volcengine_aicc_sdk.secure_http_client import ( - AICCConfig, AsyncSecureHttpClient, - ) - except ImportError as e: - logger.error("AICC SDK 导入失败: %s,请确认已安装 bytedance_volcengine_aicc_sdk", e) - raise - - base_url = config.get("base_url", "https://ark.cn-beijing.volces.com/api/v3") - # ra_url 为远程证明(安全令牌)端点地址,可能与推理 base_url 不同 - # 若不单独配置则默认使用 base_url 的域名 + /v1/security/token - ra_url = config.get("ra_url", base_url) - ak = config.get("ak", "") - sk = config.get("sk", "") - ra_service_name = config.get("ra_service_name", "AICC.ConfidentialInference") - ra_policy_id = config.get("ra_policy_id", "router_policy") - - os.environ["DOUBAO_BASE_URL"] = base_url - logger.info("豆包 AICC base_url: %s, ra_url: %s, ra_service_name: %s, ra_policy_id: %s", - base_url, ra_url, ra_service_name, ra_policy_id) - - aicc_config = AICCConfig( - ra_url=ra_url, - auth_token=api_key, - ak=ak, - sk=sk, - aicc_llm_service_name=ra_service_name, - policy_id=ra_policy_id, - ) - client = AsyncOpenAI( - api_key=api_key, - base_url=base_url, - http_client=AsyncSecureHttpClient(aicc_config=aicc_config), # type: ignore - ) - set_aicc_client(client) - logger.info("豆包 AICC 客户端已创建") - else: - logger.info("豆包使用标准 Ark SDK 模式") - # Ark 模式:使用 AsyncArk - base_url = config.get("base_url", "https://ark.cn-beijing.volces.com/api/v3") - os.environ["DOUBAO_BASE_URL"] = base_url - client = AsyncArk(base_url=base_url, api_key=api_key) - set_ark_client(client) - logger.info("豆包 Ark 客户端已创建 (base_url=%s)", base_url) - - logger.info("豆包插件初始化完成") - - async def on_cleanup(self): - """cleanup 时调用:释放客户端引用。""" - cleanup_all() - logger.info("[doubao] 清理完成") - - -# 暴露给 app lifespan 的模块级入口(bound method,签名与旧写法完全一致) -plugin = DoubaoPlugin() -create_workers = plugin.create_workers -cleanup = plugin.cleanup diff --git a/doubao/global_var.py b/doubao/global_var.py deleted file mode 100644 index 4a9aadb..0000000 --- a/doubao/global_var.py +++ /dev/null @@ -1,48 +0,0 @@ -from typing import Optional, Union -from volcenginesdkarkruntime import AsyncArk -from openai import AsyncOpenAI - -_ark_client: Optional[AsyncArk] = None -_aicc_client: Optional[AsyncOpenAI] = None -_is_aicc: bool = False - -def set_ark_client(client: AsyncArk): - global _ark_client - _ark_client = client - -def get_ark_client() -> Optional[AsyncArk]: - return _ark_client - -def set_aicc_client(client: AsyncOpenAI): - global _aicc_client - _aicc_client = client - -def get_aicc_client() -> Optional[AsyncOpenAI]: - return _aicc_client - -def set_is_aicc(value: bool): - global _is_aicc - _is_aicc = value - -def get_is_aicc() -> bool: - return _is_aicc - -def get_client() -> Union[AsyncArk, AsyncOpenAI, None]: - """返回当前活跃的客户端(AICC 或 Ark)""" - if _is_aicc: - return _aicc_client - return _ark_client - -def cleanup_ark_client(): - global _ark_client - if _ark_client: - _ark_client = None - -def cleanup_aicc_client(): - global _aicc_client - if _aicc_client: - _aicc_client = None - -def cleanup_all(): - cleanup_ark_client() - cleanup_aicc_client() \ No newline at end of file diff --git a/doubao/models/__init__.py b/doubao/models/__init__.py deleted file mode 100644 index 6955df0..0000000 --- a/doubao/models/__init__.py +++ /dev/null @@ -1,14 +0,0 @@ -""" -模型导出 — doubao -自动生成,请勿手动修改 -""" - -from .pydantic import AnalyzePostRequest -from .pydantic import AnalyzePostResponse -from .pydantic import DoubaoResponse - -__all__ = [ - "AnalyzePostRequest", - "AnalyzePostResponse", - "DoubaoResponse", -] diff --git a/doubao/models/pydantic.py b/doubao/models/pydantic.py deleted file mode 100644 index e3528ae..0000000 --- a/doubao/models/pydantic.py +++ /dev/null @@ -1,23 +0,0 @@ -""" -自动生成的 Pydantic 模型 — doubao -来源: config/workers/doubao/ -""" - -from __future__ import annotations - -from typing import Any, Optional -from pydantic import BaseModel, Field - -class AnalyzePostRequest(BaseModel): - """分析图片内容 — 请求体""" - image_base64: str = Field(...) - prompt: str = Field(...) - -class AnalyzePostResponse(BaseModel): - """分析图片内容 — 响应体""" - result: str = Field(...) - error: Optional[str] = Field(default='') - - -# 所有响应模型的联合类型 -DoubaoResponse = AnalyzePostResponse diff --git a/doubao/router/__init__.py b/doubao/router/__init__.py deleted file mode 100644 index fb843d7..0000000 --- a/doubao/router/__init__.py +++ /dev/null @@ -1,3 +0,0 @@ -from .analyze import analyze - -__all__ = ["analyze"] \ No newline at end of file diff --git a/doubao/router/analyze.py b/doubao/router/analyze.py deleted file mode 100644 index df8f6a3..0000000 --- a/doubao/router/analyze.py +++ /dev/null @@ -1,80 +0,0 @@ -import os -import logging -from chongming_worker import rpc_method -from ..models.pydantic import AnalyzePostRequest, AnalyzePostResponse -from ..global_var import get_client, get_is_aicc - -logger = logging.getLogger("plugins.doubao.analyze") - -@rpc_method( - subject="doubao.analyze", - method="POST", - summary="Analyze image via Doubao vision model", - response_model=AnalyzePostResponse, -) -async def analyze(request: AnalyzePostRequest) -> AnalyzePostResponse: - logger.info("收到分析请求: prompt=%s, image_len=%d", request.prompt, len(request.image_base64)) - client = get_client() - if client is None: - logger.error("豆包客户端未初始化(is_aicc=%s)", get_is_aicc()) - return AnalyzePostResponse(result="", error="客户端未初始化,请检查插件启动日志确认 API_KEY 和 AICC 配置是否正确") - - # 获取图片数据 - image_base64 = request.image_base64 - - if not image_base64: - return AnalyzePostResponse(result="", error="未提供图片数据(image_base64或image_path)") - - if image_base64.startswith('data:image/'): - image_base64 = image_base64.split(',', 1)[1] - - mime_type = "image/png" - - prompt = request.prompt or "" - - try: - # AICC 模式的 AsyncOpenAI 不支持 thinking 参数,需通过 extra_body 传递 - if get_is_aicc(): - create_kwargs = {"extra_body": {"thinking": {"type": "disabled"}}} - else: - create_kwargs = {"thinking": {"type": "disabled"}} - - response = await client.chat.completions.create( - model=os.getenv("DOUBAO_MODEL", "doubao-seed-evolving"), - messages=[ - { - "role": "user", - "content": [ - {"type": "image_url", "image_url": f"data:{mime_type};base64,{image_base64}"}, - {"type": "text", "text": prompt} - ] - } # type: ignore - ], - **create_kwargs # type: ignore - ) - result = response.choices[0].message.content # type: ignore - logger.info("分析完成: 结果长度=%d", len(result) if result else 0) - return AnalyzePostResponse(result=result) # type: ignore - except Exception as e: - error_msg = str(e) - # 针对 AICC 安全通道密钥缺失,给出明确提示 - # KeyMissingError 可能被包装在 APIConnectionError 的 __cause__ 链中 - is_key_missing = ( - "KeyMissingError" in type(e).__name__ - or "Can not get key" in error_msg - or any("KeyMissingError" in type(c).__name__ for c in [getattr(e, '__cause__', None), getattr(e, '__context__', None)] if c is not None) - ) - if is_key_missing: - logger.error( - "AICC 安全通道密钥获取失败,请检查: " - "1) api_key 是否有效 " - "2) ra_url(安全令牌地址)是否正确 " - "3) APIGateway /v1/security/token 端点是否可达" - ) - return AnalyzePostResponse( - result="", - error="AICC 安全通道初始化失败:无法获取加密密钥。" - "请检查 api_key 有效性及安全令牌服务 (/v1/security/token) 连通性。" - ) - logger.error("调用豆包API失败: %s", e, exc_info=True) - return AnalyzePostResponse(result="", error=f"调用豆包API失败: {error_msg}") \ No newline at end of file diff --git a/example/README.md b/example/README.md deleted file mode 100644 index b2e1ca2..0000000 --- a/example/README.md +++ /dev/null @@ -1,72 +0,0 @@ -# Example Worker 插件 - -演示 `@rpc_method` 装饰器与显式模型绑定的完整用法,是学习开发新插件的**入门参考**。 - ---- - -## 实现的功能 - -### 四则运算(GET / POST) - -| 端点 | 方法 | 路径 | 说明 | -| ---- | ---------- | ---------------- | ------------------------ | -| 加法 | GET / POST | `/calc/add` | `a + b` | -| 减法 | GET | `/calc/subtract` | `a - b` | -| 乘法 | GET | `/calc/multiply` | `a * b` | -| 除法 | GET | `/calc/divide` | `a / b`(除数为 0 报错) | - -### 访问计数器 - -| 端点 | 方法 | 路径 | 说明 | -| ---- | ---- | ------------- | ----------------- | -| 计数 | GET | `/calc/count` | 每次调用计数器 +1 | - -基于内存缓存实现,用于演示 GET 端点的无参数调用模式。 - ---- - -## 调用示例 - -```bash -# GET 加法 -curl "http://localhost:8000/api/v1/calc/add?a=3&b=4" -# {"result":7.0,"operation":"add","timestamp":1720000000.123} - -# POST 加法 -curl -X POST "http://localhost:8000/api/v1/calc/add" \ - -H "Content-Type: application/json" \ - -d '{"a": 3.5, "b": 2.5}' -# {"result":6.0,"operation":"add","timestamp":1720000000.456} - -# 除法(除数为零) -curl "http://localhost:8000/api/v1/calc/divide?a=10&b=0" -# {"result":0.0,"operation":"divide","error":"Division by zero"} - -# 计数器 -curl "http://localhost:8000/api/v1/calc/count" -# {"count":1} -``` - ---- - -## 如何工作 - -1. **配置**:`config/workers/example/*.toml` — 定义每个端点的参数和响应结构 -2. **模型**:`models/pydantic.py` — 由 `gen_models` 根据 TOML 自动生成 -3. **路由**:`router/` 下每个文件实现一个 RPC 端点,使用 `@rpc_method` 装饰器 -4. **入口**:`__init__.py` 的 `ExamplePlugin`(Plugin 基类)统一生命周期——`on_init` 初始化 - 数据库并发布到 global_var,`create_workers` / `cleanup` 为其 bound method - ---- - -## 开发参考 - -此插件展示了以下模式: - -- GET 端点的 `params` 定义与 `@rpc_method` 参数绑定 -- POST 端点的 JSON body 自动解析 -- `response_model` 在 TOML 中定义并在装饰器中引用 -- 全局变量 (`global_var.py`) 管理共享资源 -- 错误处理(如除法除零返回错误信息) -- `auth_required` 控制端点是否需要 JWT 鉴权 -- `cacheable` 控制结果是否缓存 diff --git a/example/__init__.py b/example/__init__.py deleted file mode 100644 index 34af96a..0000000 --- a/example/__init__.py +++ /dev/null @@ -1,49 +0,0 @@ -""" -Example Worker 插件 — 演示 @rpc_method + Plugin 基类统一生命周期 - -本插件展示如何使用 chongming_worker 框架定义 RPC 端点,以及如何 -基于 Plugin 基类编写风格统一的插件初始化 / 清理。 - -核心工作流: - 1. 编写 config/workers/example/*.toml 配置文件 - 2. 执行 uv run gen_models 生成 Pydantic 模型 - 3. 在 router/*.py 中用 @rpc_method 定义业务函数 - 4. 在 __init__.py 中用 Plugin 基类统一管理生命周期 - -关键约定: - - 模型命名: {Request|Response} - (如 AddGetRequest, AddGetResponse 由 gen_models 自动生成) - - 装饰器 subject 中的操作名 → 模型前缀(驼峰) - example.add → Add, example.subtract → Subtract -""" -import logging - -from chongming_worker.plugin import Plugin - -from .router import * # noqa: F401,F403 触发 @rpc_method 注册 - -logger = logging.getLogger(__name__) - - -class ExamplePlugin(Plugin): - """Example 插件 — 统一生命周期。""" - - type = "example" - db_name = "example.db" # 自动在 data_dir 下建 SQLite 库 - auto_bind = True # 首次初始化时自动 auto_bind_models("example") - - async def on_init(self): - """首次 create_workers 时调用一次:初始化资源。""" - # 数据库已由基类自动初始化(self.db),写入全局变量供 router 使用 - self.set_var("db", self.db) - logger.info("[example] 初始化完成") - - async def on_cleanup(self): - """cleanup 时调用:关闭自定义客户端等(数据库由基类处理)。""" - logger.info("[example] 清理完成") - - -# 暴露给 app lifespan 的模块级入口(bound method,签名与旧写法完全一致) -plugin = ExamplePlugin() -create_workers = plugin.create_workers -cleanup = plugin.cleanup diff --git a/example/global_var.py b/example/global_var.py deleted file mode 100644 index 22fc6c8..0000000 --- a/example/global_var.py +++ /dev/null @@ -1,57 +0,0 @@ -"""Example 插件全局运行时状态 — 统一容器 - -约定: - - 模块只做"声明默认值",不产生任何导入期副作用 - - 所有值统一在 create_workers 中通过 set_global_var 显式写入 - - set_global_var 仅允许写入 _ALLOWED 白名单,拼写错误会立即抛错 - - require() 用于取"必须已初始化"的值,未初始化时抛清晰错误 -""" -from typing import Any - -from chongming_database import Database - -# 允许写入的全局变量白名单(防止拼写错误静默创建无用的新变量) -_ALLOWED = frozenset({"db"}) - -# 显式声明所有运行时状态(默认 None,运行时由 create_workers 写入) -db: Database | None = None - - -def set_global_var(name: str, value: Any) -> None: - """ - 白名单校验版 setter — 设置模块级运行时状态。 - - 拼错变量名会立即抛 KeyError,而不是静默创建一个无用的新变量。 - - Args: - name: 必须在 _ALLOWED 白名单中的变量名 - value: 任意值 - """ - if name not in _ALLOWED: - raise KeyError( - f"global_var: 未知变量 '{name}'(允许: {sorted(_ALLOWED)})" - ) - globals()[name] = value - - -def require(name: str) -> Any: - """ - 获取必须已初始化的全局变量,未初始化时抛清晰错误。 - - Args: - name: 必须在 _ALLOWED 白名单中的变量名 - - Returns: - 对应全局变量的值 - """ - if name not in _ALLOWED: - raise KeyError(f"global_var: 未知变量 '{name}'(允许: {sorted(_ALLOWED)})") - value = globals().get(name) - if value is None: - raise RuntimeError( - f"global_var: '{name}' 未初始化 —— 请确认插件 create_workers 已执行且配置正确" - ) - return value - - -__all__ = sorted(_ALLOWED) + ["set_global_var", "require"] diff --git a/example/models/__init__.py b/example/models/__init__.py deleted file mode 100644 index 1d8ba98..0000000 --- a/example/models/__init__.py +++ /dev/null @@ -1,38 +0,0 @@ -""" -模型导出 — example -自动生成,请勿手动修改 -""" - -from .pydantic import AddGetRequest -from .pydantic import AddGetResponse -from .pydantic import AddMqttRequest -from .pydantic import AddPostRequest -from .pydantic import AddPostResponse -from .pydantic import CountGetResponse -from .pydantic import DivideGetRequest -from .pydantic import DivideGetResponse -from .pydantic import ExampleResponse -from .pydantic import HeartbeatMqttRequest -from .pydantic import MultiplyGetRequest -from .pydantic import MultiplyGetResponse -from .pydantic import StreamChatPostRequest -from .pydantic import SubtractGetRequest -from .pydantic import SubtractGetResponse - -__all__ = [ - "AddGetRequest", - "AddGetResponse", - "AddMqttRequest", - "AddPostRequest", - "AddPostResponse", - "CountGetResponse", - "DivideGetRequest", - "DivideGetResponse", - "ExampleResponse", - "HeartbeatMqttRequest", - "MultiplyGetRequest", - "MultiplyGetResponse", - "StreamChatPostRequest", - "SubtractGetRequest", - "SubtractGetResponse", -] diff --git a/example/models/counter.py b/example/models/counter.py deleted file mode 100644 index 34fc786..0000000 --- a/example/models/counter.py +++ /dev/null @@ -1,8 +0,0 @@ -from sqlmodel import Field, SQLModel - - -class Counter(SQLModel, table=True): - __tablename__ = "counter" # type: ignore - name: str = Field(primary_key=True, index=True) - value: int = Field(default=0) - \ No newline at end of file diff --git a/example/models/pydantic.py b/example/models/pydantic.py deleted file mode 100644 index 1e5108b..0000000 --- a/example/models/pydantic.py +++ /dev/null @@ -1,88 +0,0 @@ -""" -自动生成的 Pydantic 模型 — example -来源: config/workers/example/ -""" - -from __future__ import annotations - -from typing import Optional -from pydantic import BaseModel, Field - -class AddGetRequest(BaseModel): - """Add two numbers — 请求体""" - a: float = Field(...) - b: float = Field(...) - -class AddGetResponse(BaseModel): - """Add two numbers — 响应体""" - result: float = Field(...) - operation: str = Field(default='add') - timestamp: Optional[float] = Field(default=0.0) - -class AddPostRequest(BaseModel): - """Add two numbers (POST) — 请求体""" - a: float = Field(...) - b: float = Field(...) - -class AddPostResponse(BaseModel): - """Add two numbers (POST) — 响应体""" - result: float = Field(...) - operation: str = Field(default='add') - timestamp: Optional[float] = Field(default=0.0) - -class AddMqttRequest(BaseModel): - """Add two numbers via MQTT — 请求体""" - a: float = Field(...) - b: float = Field(...) - -class CountGetResponse(BaseModel): - """Get visit count and increment — 响应体""" - name: str = Field(...) - value: int = Field(...) - -class DivideGetRequest(BaseModel): - """Divide two numbers — 请求体""" - a: float = Field(...) - b: float = Field(...) - -class DivideGetResponse(BaseModel): - """Divide two numbers — 响应体""" - result: Optional[float] = Field(default=0.0) - operation: str = Field(default='divide') - timestamp: Optional[float] = Field(default=0.0) - error: Optional[str] = Field(default='') - -class HeartbeatMqttRequest(BaseModel): - """Heartbeat ping (no response) — 请求体""" - worker_id: str = Field(...) - -class MultiplyGetRequest(BaseModel): - """Multiply two numbers — 请求体""" - a: float = Field(...) - b: float = Field(...) - -class MultiplyGetResponse(BaseModel): - """Multiply two numbers — 响应体""" - result: float = Field(...) - operation: str = Field(default='multiply') - timestamp: Optional[float] = Field(default=0.0) - -class StreamChatPostRequest(BaseModel): - """模拟 AI 流式对话(SSE,逐 token 输出) — 请求体""" - prompt: str = Field(...) - tokens: int = Field(...) - -class SubtractGetRequest(BaseModel): - """Subtract two numbers — 请求体""" - a: float = Field(...) - b: float = Field(...) - -class SubtractGetResponse(BaseModel): - """Subtract two numbers — 响应体""" - result: float = Field(...) - operation: str = Field(default='subtract') - timestamp: Optional[float] = Field(default=0.0) - - -# 所有响应模型的联合类型 -ExampleResponse = MultiplyGetResponse | AddPostResponse | CountGetResponse | SubtractGetResponse | DivideGetResponse | AddGetResponse diff --git a/example/router/__init__.py b/example/router/__init__.py deleted file mode 100644 index 569d623..0000000 --- a/example/router/__init__.py +++ /dev/null @@ -1,21 +0,0 @@ -from .add import add, add_post -from .add_mqtt import add_mqtt -from .heartbeat import heartbeat -from .subtract import subtract -from .multiply import multiply -from .divide import divide -from .count import count_get -from .stream_chat import stream_chat - - -__all__ = [ - "add", - "add_post", - "add_mqtt", - "heartbeat", - "subtract", - "multiply", - "divide", - "count_get", - "stream_chat" -] \ No newline at end of file diff --git a/example/router/add.py b/example/router/add.py deleted file mode 100644 index cc97e63..0000000 --- a/example/router/add.py +++ /dev/null @@ -1,51 +0,0 @@ -import time - -import zenoh - -from chongming_worker import rpc_method -from chongming_request import Client - -from ..models.pydantic import ( - AddGetRequest, - AddGetResponse, - AddPostRequest, - AddPostResponse -) - -@rpc_method( - subject="example.add", - method="GET", - summary="Add two numbers via GET", - response_model=AddGetResponse, -) -async def add(add_get_request:AddGetRequest) -> AddGetResponse: - """加法运算(GET)— 通过查询参数传入 a、b,返回它们的和""" - return AddGetResponse( - result=add_get_request.a + add_get_request.b, - operation="add", - timestamp=time.time() - ) - -@rpc_method( - subject="example.add", - method="POST", - summary="Add two numbers via POST", - response_model=AddPostResponse, -) -async def add_post(add_post_request: AddPostRequest, session: zenoh.Session, user_info: dict | None = None) -> AddPostResponse: - """加法运算(POST)— 通过 JSON Body 传入 a、b,返回它们的和""" - async with Client(session=session) as client: - await client.call( - worker_type="example", - subject="example.add", - payload={ - "a":add_post_request.a, - "b":add_post_request.b - }, - method="GET" - ) - return AddPostResponse( - result=add_post_request.a + add_post_request.b, - operation="add", - timestamp=time.time(), - ) diff --git a/example/router/add_mqtt.py b/example/router/add_mqtt.py deleted file mode 100644 index a8ef2d4..0000000 --- a/example/router/add_mqtt.py +++ /dev/null @@ -1,24 +0,0 @@ -""" -MQTT 端点处理器 — 通过 MQTT 主题 chongming/example/add 调用 - -与 GET/POST 的 add 共用同一个 subject (example.add), -通过信封 meta.method="MQTT" 区分。 -""" -import time - -from chongming_worker import rpc_method -from ..models.pydantic import AddMqttRequest - - -@rpc_method( - subject="example.add", - method="MQTT", - summary="Add two numbers via MQTT", -) -async def add_mqtt(request: AddMqttRequest) -> dict: - """MQTT 加法 — 通过 MQTT 主题 chongming/example/add 调用,返回 dict""" - return { - "result": request.a + request.b, - "operation": "add_mqtt", - "timestamp": time.time(), - } diff --git a/example/router/count.py b/example/router/count.py deleted file mode 100644 index 11b0fdd..0000000 --- a/example/router/count.py +++ /dev/null @@ -1,42 +0,0 @@ -from sqlalchemy import select, update - -from chongming_worker import rpc_method - -from ..models.pydantic import CountGetResponse -from ..models.counter import Counter - - -@rpc_method( - subject="example.count", - method="GET", - summary="Increment and get counter value", - response_model=CountGetResponse, # 需要定义 CountResponse -) -async def count_get() -> CountGetResponse: - from ..global_var import db - - if db is None: - raise RuntimeError("Database not initialized") - async with db.get_session() as session: - statement = select(Counter).where(Counter.name == "visits") # type: ignore - result = await session.execute(statement) - counter = result.scalar_one_or_none() - - if counter is None: - counter = Counter(name="visits", value=1) - session.add(counter) - await session.commit() - return CountGetResponse(name="visits", value=1) - - else: - new_value = counter.value + 1 - - update_stmt = ( - update(Counter) - .where(Counter.name == "visits") # type: ignore - .values(value=new_value) - ) - await session.execute(update_stmt) - await session.commit() - - return CountGetResponse(name="visits", value=new_value) diff --git a/example/router/divide.py b/example/router/divide.py deleted file mode 100644 index b5bb7b0..0000000 --- a/example/router/divide.py +++ /dev/null @@ -1,29 +0,0 @@ -import time - -from chongming_worker import rpc_method - -from ..models.pydantic import ( - DivideGetResponse, - DivideGetRequest, -) - -@rpc_method( - subject="example.divide", - method="GET", - summary="Divide two numbers", - response_model=DivideGetResponse, -) -async def divide(divide_get_request: DivideGetRequest) -> DivideGetResponse: - """除法运算:a / b(b 不能为 0)""" - if divide_get_request.b == 0: - return DivideGetResponse( - error="除数不能为零", - operation="divide", - result=0.0, - timestamp=time.time(), - ) - return DivideGetResponse( - result=divide_get_request.a / divide_get_request.b, - operation="divide", - timestamp=time.time(), - ) diff --git a/example/router/heartbeat.py b/example/router/heartbeat.py deleted file mode 100644 index eee7c95..0000000 --- a/example/router/heartbeat.py +++ /dev/null @@ -1,20 +0,0 @@ -""" -心跳 MQTT 端点 — 仅上报,无需响应 - -通过 MQTT 主题 chongming/example/heartbeat 发送心跳, -Worker 返回 None,MQTT Bridge 不发布响应。 -""" -import time - -from chongming_worker import rpc_method - - -@rpc_method( - subject="example.heartbeat", - method="MQTT", - summary="Heartbeat ping via MQTT (no response)", -) -async def heartbeat(worker_id: str = "") -> None: - """心跳上报 — 无返回值,MQTT Bridge 跳过响应发布""" - print(f"[Heartbeat] worker={worker_id} time={time.time()}") - return None # 显式返回 None,MQTT Bridge 不发布响应 diff --git a/example/router/multiply.py b/example/router/multiply.py deleted file mode 100644 index eef9494..0000000 --- a/example/router/multiply.py +++ /dev/null @@ -1,22 +0,0 @@ -import time - -from chongming_worker import rpc_method - -from ..models.pydantic import ( - MultiplyGetRequest, - MultiplyGetResponse, -) - -@rpc_method( - subject="example.multiply", - method="GET", - summary="Multiply two numbers", - response_model=MultiplyGetResponse, -) -async def multiply(multiply_get_request=MultiplyGetRequest) -> MultiplyGetResponse: - """乘法运算:a * b""" - return MultiplyGetResponse( - result=multiply_get_request.a * multiply_get_request.b, - operation="multiply", - timestamp=time.time(), - ) diff --git a/example/router/stream_chat.py b/example/router/stream_chat.py deleted file mode 100644 index 896bebc..0000000 --- a/example/router/stream_chat.py +++ /dev/null @@ -1,34 +0,0 @@ -""" -流式端点示例 — @rpc_method 返回异步生成器,框架自动逐块多回复推送 - -网关在 TOML 中声明 stream = true 时,该端点以 SSE(text/event-stream)输出: - POST /api/v1/calc/stream_chat {"prompt": "你好", "tokens": 5} - 响应: data: {"token": "..."} ... data: {"done": true} - -约定: - - 插件侧写法:函数返回生成器(yield),框架逐块 query.reply() 推送 - - {"done": true} 是业务结束标记(插件自定义),不是协议哨兵—— - 流式协议以生成器自然结束为流终止信号 - - 中途异常时框架自动发送 {"error": ...} 收尾块 -""" -import asyncio -import logging - -from chongming_worker import rpc_method - -logger = logging.getLogger("plugins.example.stream_chat") - - -@rpc_method( - subject="example.stream_chat", - method="POST", - summary="模拟 AI 流式对话(SSE,逐 token 输出)", - stream=True, -) -async def stream_chat(prompt: str = "你好", tokens: int = 10): - """逐 token 模拟生成:把 return 改成 yield,框架即自动流式推送""" - logger.info("收到流式请求: prompt=%s, tokens=%d", prompt, tokens) - for i in range(tokens): - await asyncio.sleep(0.05) - yield {"token": f"{prompt}·{i + 1}", "index": i} - yield {"done": True} diff --git a/example/router/subtract.py b/example/router/subtract.py deleted file mode 100644 index b08fe85..0000000 --- a/example/router/subtract.py +++ /dev/null @@ -1,22 +0,0 @@ -import time - -from chongming_worker import rpc_method - -from ..models.pydantic import ( - SubtractGetResponse, - SubtractGetRequest, -) - -@rpc_method( - subject="example.subtract", - method="GET", - summary="Subtract two numbers", - response_model=SubtractGetResponse, -) -async def subtract(subtract_get_request: SubtractGetRequest) -> SubtractGetResponse: - """减法运算:a - b""" - return SubtractGetResponse( - result=subtract_get_request.a - subtract_get_request.b, - operation="subtract", - timestamp=time.time(), - )