diff --git a/output.md b/output.md deleted file mode 100644 index 9163522..0000000 --- a/output.md +++ /dev/null @@ -1,5235 +0,0 @@ -`./pyproject.toml` -``` -[build-system] -build-backend = "hatchling.build" -requires = ["hatchling"] - -[project] -dependencies = [ - "apscheduler", - "boto3>=1.43.48", - "chongming-worker", - "diskcache", - "eclipse-zenoh", - "fastapi[uvicorn]", - "gunicorn", - "modelvault>=0.16.0", - "module-bank", - "nanio>=0.1.5", - "orjson>=3.11.9", - "python-multipart>=0.0.32", - "sqlite-vfs", - "tomli", - "typing-extensions", - "uvicorn", -] -description = "基于 modules-bank 的 fastapi 打包脚手架" -name = "chongming-zenoh" -readme = "README.md" -requires-python = ">=3.12" -version = "0.1.0" - -[dependency-groups] -dev = [ - "nuitka", - "pyarmor", - "pyinstaller", - "ruff", -] - -[tool.hatch.build.targets.wheel] -packages = ["src/chongming"] - -[[tool.uv.index]] -url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/simple" - -[tool.uv.workspace] -members = [ - "utils/worker", - "utils/request", -] - -[tool.uv.sources] -chongming-worker = { workspace = true } - -[project.scripts] -build = "chongming.scripts.build:build" -gen_models = "chongming.scripts.gen_models:gen_models" -gunicorn = "chongming:gunicorn_serve" -serve = "chongming:serve" - -``` - -`./config.toml` -``` -[default] -app.debug = false -app.description = "基于 FastAPI 的 chongming 打包脚手架" -app.name = "chongming" -app.version = "chongming" -env = "development" -prefix = "/api/v1" -upload_path = "uploads" - -[database] -database_path = "./data/database.db" -type = "sqlite" -url = "sqlite+aiosqlite:///./data/database.db" - -[database.sqlite] -connect_args.check_same_thread = false -connect_args.timeout = 15 -echo = true -echo_pool = true -max_overflow = 10 -pool_pre_ping = true -pool_recycle = 300 -pool_size = 20 -pool_timeout = 30 - -[scheduler] -job_store_path = "scheduler_jobs.db" - -[cache] -cache_store_path = "./cache_data" - -[security] -access_token_expire_minutes = 1440 -algorithm = "HS256" -max_sessions_per_user = 5 -refresh_token_expire_days = 7 -secret_key = "your-secret-key-change-in-production" - -[development] -debug = true -env = "development" - -[development.server] -access_log = false -host = "0.0.0.0" -port = 8000 -reload = true - -[development.logging] -file = "logs" -format = "%(asctime)s - %(name)s - %(levelname)s - %(message)s" -level = "DEBUG" - -[development.cors] -allow_credentials = true -allow_headers = ["*"] -allow_methods = ["*"] -allow_origins = ["*"] - -[plugins_worker] -config_path = "./config/workers/config.toml" - -[pyinstaller_build_import] -base = [ - "utils", # public 骨架 - "utils.launch", - "utils.config", - "zenoh", - "chongming_worker", - "server", - "module_bank", # 模块化打包工具 - "sqlite_vfs", - "sqlite_vfs.core", - "fastapi", # web 应用依赖 - "fastapi.staticfiles", - "fastapi.middleware.cors", - "uvicorn", - "apscheduler", - "diskcache", - "orjson", - "python_multipart", - "nanio", - "boto3", -] -linux = [ - "gunicorn.glogging", - "gunicorn.app", - "gunicorn.app.wsgiapp", -] -windows = [] - -[production.module_system] -path = [ - ["resources/app.mbank", "src/chongming/app", "app"], - ["resources/plugins.mbank", "src/plugins", "plugins"], -] -type = "module-bank" -``` - -`./README.md` -``` - -``` - -`./build_nuitka/config.toml` -``` -[default] -app.debug = false -app.description = "基于 FastAPI 的 chongming 打包脚手架" -app.name = "chongming" -app.version = "0.1.0" -env = "production" -prefix = "/api/v1" -upload_path = "uploads" - -[database] -generate_schemas = true # 生产环境使用迁移工具 -type = "sqlite" -url = "sqlite+aiosqlite:///./data/database.db" - -[database.sqlite] -connect_args.check_same_thread = false -connect_args.timeout = 15 -echo = false -echo_pool = false -max_overflow = 10 -pool_pre_ping = true -pool_recycle = 300 -pool_size = 20 -pool_timeout = 30 - -[scheduler] -job_store_path = "scheduler_jobs.db" - -[cache] -cache_store_path = "./" - -[security] -access_token_expire_minutes = 30 -algorithm = "HS256" -max_sessions_per_user = 5 -refresh_token_expire_days = 7 -secret_key = "LClT3DaMKZzvT1L4bgsxvCSM5q7uPeX3R2_FZ8xzwu8=" - -[development] -debug = true -env = "development" - -[development.server] -access_log = false -host = "127.0.0.1" -port = 8001 -reload = true - -[development.logging] -file = "logs" -format = "%(asctime)s - %(name)s - %(levelname)s - %(message)s" -level = "DEBUG" - -[development.cors] -allow_credentials = true -allow_headers = ["*"] -allow_methods = ["*"] -allow_origins = ["*"] - -[production] -debug = false -env = "production" - -[production.server] -access_log = false -host = "0.0.0.0" -port = 8000 -reload = false -timeout_keep_alive = 65 -workers = 4 - -[production.logging] -backup_count = 5 # 最多保存 10 个日志文件 -file = "logs" -format = "%(asctime)s - %(name)s - %(levelname)s - %(message)s" -level = "INFO" -max_size = 10485760 # 10MB - -[production.cors] -allow_credentials = true -allow_headers = ["*"] -allow_methods = ["*"] -allow_origins = ["http://192.168.137.5:8000", "http://localhost:8000", "http://127.0.0.1:8000"] - -[production.module_system] -path = [ - "resources/app.mbank", - "resources/plugins.mbank", -] -type = "module-bank" - -[production.file_system] -path = "static.svfs" -type = "svfs" - -[[workers]] -type = "example" -module = "plugins.example" -factory = "create_workers" -count = 3 -config = { type = "example" } - -``` - -`./build_nuitka/config/workers/example/multiply.toml` -``` -# multiply 接口 - -[GET] -summary = "Multiply two numbers" -docstring = "Multiply two numbers via GET query params (a * b)" -params = ["a: float", "b: float"] -response_model = { result = { type = "float", required = true }, operation = { type = "str", default = "multiply" }, timestamp = { type = "float", required = false, default = 0.0 } } - -``` - -`./build_nuitka/config/workers/example/divide.toml` -``` -# divide 接口 - -[GET] -summary = "Divide two numbers" -docstring = "Divide two numbers via GET query params (a / b). Returns error if b == 0." -params = ["a: float", "b: float"] -response_model = { result = { type = "float", required = false, default = 0.0 }, operation = { type = "str", default = "divide" }, timestamp = { type = "float", required = false, default = 0.0 }, error = { type = "str", required = false, default = "" } } - -``` - -`./build_nuitka/config/workers/example/subtract.toml` -``` -# subtract 接口 - -[GET] -summary = "Subtract two numbers" -docstring = "Subtract two numbers via GET query params (a - b)" -params = ["a: float", "b: float"] -response_model = { result = { type = "float", required = true }, operation = { type = "str", default = "subtract" }, timestamp = { type = "float", required = false, default = 0.0 } } - -``` - -`./build_nuitka/config/workers/example/add.toml` -``` -# add 接口 — 按请求类型分节 -# subject 会带上方法后缀: example.add.GET / example.add.POST - -[GET] -cacheable = true -summary = "Add two numbers" -docstring = "Add two numbers via GET query params" -params = ["a: float", "b: float"] -response_model = { result = { type = "float", required = true }, operation = { type = "str", default = "add" }, timestamp = { type = "float", required = false, default = 0.0 } } - -[POST] -cacheable = true -summary = "Add two numbers (POST)" -docstring = "Add two numbers via POST JSON body" -params = ["a: float", "b: float"] -response_model = { result = { type = "float", required = true }, operation = { type = "str", default = "add" }, timestamp = { type = "float", required = false, default = 0.0 } } - -``` - -`./build_nuitka/config/workers/example/config.toml` -``` -[worker] -name = "example" -version = "0.1.0" -description = "chongming worker example — demonstrates new config features" -config_version = "1.0" -defaults = { method = "GET", ttl = 30, timeout = 2.0, auth_required = true, cacheable = false } - - [worker.registration] - type = "register" - service = "example" - worker_type = "calc-workers" - router_prefix = "/calc" - tags = ["example"] - -``` - -`./utils/worker/pyproject.toml` -``` -[build-system] -build-backend = "hatchling.build" -requires = ["hatchling"] - -[project] -dependencies = [ - "eclipse-zenoh>=1.9.0", - "tomli>=2.4.1", -] -description = "chongming worker" -name = "chongming-worker" -readme = "README.md" -requires-python = ">=3.12" -version = "0.1.0" - -[tool.hatch.build.targets.wheel] -packages = ["src/chongming_worker"] - -``` - -`./utils/worker/README.md` -``` - -``` - -`./utils/worker/src/chongming_worker/decorators.py` -``` -""" -通用 Working 装饰器框架 - -提供 @rpc_method 装饰器，用于将普通函数注册为 Zenoh RPC 端点。 -装饰器自动处理： -- 访问类型（GET/POST/PUT/DELETE） -- 接口名称（subject） -- 参数解析与类型转换 -- 响应序列化 - -兼容 gen_models 命令生成的 Pydantic 模型： -- 通过 request_model / response_model 参数绑定已生成的模型 -- 自动从插件 models 目录发现匹配的模型 -- 请求参数自动校验与响应格式自动序列化 -""" -import asyncio -import json -import time -import inspect -import importlib -from pathlib import Path -from typing import Any, Callable, Dict, List, Optional, Tuple, Type, get_type_hints, Union - -from pydantic import BaseModel, Field, create_model, ValidationError - - -# ============================================================ -# 类型系统 -# ============================================================ - -TYPE_MAP: Dict[str, Any] = { - "float": float, - "int": int, - "str": str, - "bool": bool, - "any": Any, -} - - -def _resolve_type(annotation: Any) -> Any: - """将类型注解解析为可用的 Python 类型""" - if annotation is inspect.Parameter.empty: - return str - origin = getattr(annotation, "__origin__", None) - if origin is list: - args = getattr(annotation, "__args__", (Any,)) - return List[args[0]] if args else List[Any] # type: ignore - if origin is dict: - args = getattr(annotation, "__args__", (str, Any)) - return Dict[args[0], args[1]] if len(args) == 2 else Dict[str, Any] # type: ignore - if origin is tuple: - args = getattr(annotation, "__args__", (Any,)) - return Tuple[tuple(args)] # type: ignore - if origin is Optional: - args = getattr(annotation, "__args__", (Any,)) - return Optional[args[0]] # type: ignore - return annotation - - -def _type_to_config_str(py_type: Any) -> str: - """将 Python 类型转换为配置字符串（用于 gen_models 兼容）""" - origin = getattr(py_type, "__origin__", None) - # 处理 Union 类型（包括 Optional[X]） - if origin is Union: - args = getattr(py_type, "__args__", (Any,)) - non_none_args = [a for a in args if a is not type(None)] - if len(non_none_args) == 1: - return _type_to_config_str(non_none_args[0]) - return "str" - if origin is list: - args = getattr(py_type, "__args__", (Any,)) - return f"list[{_type_to_config_str(args[0])}]" - if origin is dict: - args = getattr(py_type, "__args__", (str, Any)) - return f"dict[{_type_to_config_str(args[0])}, {_type_to_config_str(args[1])}]" - if py_type is float: - return "float" - if py_type is int: - return "int" - if py_type is bool: - return "bool" - if py_type is str: - return "str" - return "str" - - -# ============================================================ -# 装饰器定义 -# ============================================================ - -class RPCMethod: - """ - 装饰后的 RPC 方法元数据 - - 存储原始函数、访问类型、接口名称、参数信息等， - 供 Worker 在运行时发现和调用。 - - 支持绑定 gen_models 生成的 Pydantic 模型： - - request_model: 用于请求参数的自动校验和转换 - - response_model: 用于响应数据的格式化和序列化 - """ - - def __init__( - self, - func: Callable, - subject: str, - method: str = "GET", - summary: str = "", - timeout: float = 3.0, - request_model: Optional[Type[BaseModel]] = None, - response_model: Optional[Type[BaseModel]] = None, - ): - self.func = func - self.subject = subject - self.method = method.upper() - self.summary = summary or func.__doc__ or "" - self.timeout = timeout - self.name = func.__name__ - - # 绑定的 Pydantic 模型（由 gen_models 生成，或手动传入） - self.request_model: Optional[Type[BaseModel]] = request_model - self.response_model: Optional[Type[BaseModel]] = response_model - - # 解析函数签名 - sig = inspect.signature(func) - hints = get_type_hints(func) - self.params: Dict[str, Any] = {} - self.model_param: Optional[str] = None # 如果有参数是 BaseModel，记录参数名 - self.return_type: Any = str - - for param_name, param in sig.parameters.items(): - if param_name == "self": - continue - py_type = _resolve_type(hints.get(param_name, str)) - # 检查参数类型是否是 BaseModel 子类 - if isinstance(py_type, type) and issubclass(py_type, BaseModel): - self.model_param = param_name - # 如果是 BaseModel 参数，直接用它的字段作为 params（用于配置和文档） - for field_name, field in py_type.model_fields.items(): - self.params[field_name] = _resolve_type(field.annotation) - else: - self.params[param_name] = py_type - - # 返回值类型 - self.return_type = _resolve_type(hints.get("return", Any)) - - def build_pydantic_request_model(self, model_name: str = "AutoRequest") -> Type[BaseModel]: - """根据参数注解生成 Pydantic 请求模型""" - fields: Dict[str, Any] = {} - for param_name, py_type in self.params.items(): - fields[param_name] = (py_type, Field(...)) - return create_model(model_name, **fields) - - def build_pydantic_response_model(self, model_name: str = "AutoResponse") -> Type[BaseModel]: - """根据返回值注解生成 Pydantic 响应模型""" - fields: Dict[str, Any] = {} - # 如果返回类型是 BaseModel 子类，直接使用 - if isinstance(self.return_type, type) and issubclass(self.return_type, BaseModel): - return self.return_type - # 否则包装为 {result: return_type} - fields["result"] = (self.return_type, Field(...)) - return create_model(model_name, **fields) - - def get_request_model(self) -> Optional[Type[BaseModel]]: - """ - 获取有效的请求模型，优先级： - 1. 绑定的 request_model（由用户或 gen_models 提供） - 2. 如果函数参数本身就是 Pydantic 模型（model_param），从中提取字段生成模型 - 3. 自动生成的模型（根据函数参数注解，仅 POST/PUT/PATCH） - """ - if self.request_model is not None: - return self.request_model - # 如果函数接受了 Pydantic 模型参数，用它的字段生成请求模型 - if self.model_param is not None: - sig = inspect.signature(self.func) - hints = get_type_hints(self.func) - model_type = hints.get(self.model_param) - if isinstance(model_type, type) and issubclass(model_type, BaseModel): - return model_type - # 对于 POST/PUT/PATCH，根据 params 自动生成 - if self.params and self.method in ("POST", "PUT", "PATCH"): - return self.build_pydantic_request_model(f"{self.name.title()}_Request") - return None - - def get_response_model(self) -> Optional[Type[BaseModel]]: - """ - 获取有效的响应模型，优先级： - 1. 绑定的 response_model（由用户或 gen_models 提供） - 2. 自动生成的模型（根据函数返回值注解） - """ - if self.response_model is not None: - return self.response_model - return self.build_pydantic_response_model(f"{self.name.title()}_Response") - - def validate_request(self, payload: Dict[str, Any]) -> Dict[str, Any]: - """ - 使用 Pydantic 模型校验请求参数 - - Args: - payload: 原始请求参数字典 - - Returns: - 校验通过后的参数字典（包含类型转换） - """ - req_model = self.get_request_model() - if req_model is None: - return payload - try: - validated = req_model(**payload) - return validated.model_dump() - except ValidationError as e: - raise ValueError(f"请求参数校验失败: {e.errors()}") - - def serialize_response(self, raw_result: Any) -> Any: - """ - 使用 Pydantic 模型序列化响应数据 - - Args: - raw_result: 原始函数返回值 - - Returns: - 序列化后的响应数据（dict 或列表等可 JSON 序列化的类型） - """ - resp_model = self.get_response_model() - if resp_model is None: - return raw_result - if isinstance(raw_result, dict): - try: - validated = resp_model(**raw_result) - return validated.model_dump() - except ValidationError: - # 如果无法用响应模型校验，返回原始结果 - return raw_result - if isinstance(raw_result, resp_model): - return raw_result.model_dump() - return raw_result - - async def call(self, payload: Dict[str, Any]) -> Any: - """ - 调用原始函数，自动从 payload 提取参数 - - 流程： - 1. 校验请求参数（使用绑定的 request_model 或自动生成） - 2. 提取函数需要的参数 - 3. 调用函数 - 4. 序列化响应（使用绑定的 response_model 或自动生成） - - 支持两种函数签名： - - 展开参数: def add(a: float, b: float) -> ... # 传统方式 - - Pydantic 模型: def add(req: AddGetRequest) -> ... # 工程化方式 - - Args: - payload: 请求参数字典 - - Returns: - 序列化后的响应数据 - """ - # 1. 校验请求参数 - validated_payload = self.validate_request(payload) - - # 2. 提取函数需要的参数 - kwargs = {} - - if self.model_param is not None: - # 2a. 函数参数是 Pydantic 模型 — 将整个校验后的 payload 转为模型实例传入 - req_model = self.get_request_model() - if req_model is not None: - kwargs[self.model_param] = req_model(**validated_payload) - else: - # 如果没有 request_model，从函数签名获取模型类型 - sig = inspect.signature(self.func) - hints = get_type_hints(self.func) - model_type = hints.get(self.model_param) - if isinstance(model_type, type) and issubclass(model_type, BaseModel): - kwargs[self.model_param] = model_type(**validated_payload) - else: - kwargs[self.model_param] = validated_payload - else: - # 2b. 传统方式：逐字段提取参数 - for param_name in self.params: - if param_name in validated_payload: - kwargs[param_name] = validated_payload[param_name] - - # 3. 调用函数 - if asyncio.iscoroutinefunction(self.func): - result = await self.func(**kwargs) - else: - result = self.func(**kwargs) - - # 4. 序列化响应 - return self.serialize_response(result) - - -# 全局注册表：worker_type -> {subject -> RPCMethod} -_RPC_REGISTRY: Dict[str, Dict[str, RPCMethod]] = {} - - -def rpc_method( - subject: str, - method: str = "GET", - summary: str = "", - timeout: float = 3.0, - request_model: Optional[Type[BaseModel]] = None, - response_model: Optional[Type[BaseModel]] = None, -): - """ - 装饰器：将函数注册为 RPC 端点 - - 用法: - # 基本用法（自动生成模型） - @rpc_method(subject="calc.add", method="GET", summary="Add two numbers") - def add(a: float, b: float) -> float: - return a + b - - # 绑定 gen_models 生成的 Pydantic 模型 - @rpc_method( - subject="calc.add", - method="GET", - summary="Add two numbers", - response_model=AddGetResponse, - ) - def add(a: float, b: float): - return a + b - - 参数: - subject: 接口名称，格式为 "worker_type.operation" - method: HTTP 方法 (GET/POST/PUT/DELETE) - summary: 接口描述 - timeout: 超时时间（秒） - request_model: 可选的 Pydantic 请求模型（用于验证请求参数） - response_model: 可选的 Pydantic 响应模型（用于格式化响应） - """ - def decorator(func: Callable) -> Callable: - rpc = RPCMethod( - func=func, - subject=subject, - method=method, - summary=summary, - timeout=timeout, - request_model=request_model, - response_model=response_model, - ) - # 提取 worker_type（subject 的第一部分） - worker_type = subject.split(".")[0] if "." in subject else "default" - if worker_type not in _RPC_REGISTRY: - _RPC_REGISTRY[worker_type] = {} - # 使用 subject + "#" + method 作为唯一 key，允许多个 HTTP 方法共享同一 subject - registry_key = f"{subject}#{method.upper()}" - _RPC_REGISTRY[worker_type][registry_key] = rpc - - # 给函数附加元数据，方便外部访问 - func._rpc_meta = rpc # type: ignore - return func - - return decorator - - -def bind_model( - subject: str = "", - method: str = "", - request_model: Optional[Type[BaseModel]] = None, - response_model: Optional[Type[BaseModel]] = None, -): - """ - 装饰器：为已注册的 RPC 端点绑定 Pydantic 模型 - - 用于 gen_models 生成模型后，将模型绑定到用 @rpc_method 定义的端点。 - 无需修改原有的 @rpc_method 调用。 - - 用法: - from .models.pydantic import AddGetResponse - - @bind_model(subject="example.add", method="GET", response_model=AddGetResponse) - def add(): - pass # 函数体不重要，仅用于匹配已注册的端点 - - 参数: - subject: 要绑定的接口名称（与 @rpc_method 中的 subject 一致） - method: 要绑定的 HTTP 方法 - request_model: Pydantic 请求模型 - response_model: Pydantic 响应模型 - """ - def decorator(func: Callable) -> Callable: - nonlocal subject, method - if not subject and hasattr(func, '_rpc_meta'): - subject = func._rpc_meta.subject # type: ignore[attr-defined] - method = func._rpc_meta.method # type: ignore[attr-defined] - - worker_type = subject.split(".")[0] if "." in subject else "default" - method_upper = method.upper() - registry = _RPC_REGISTRY.get(worker_type, {}) - # 用新 key 格式查找: subject#METHOD - rpc = registry.get(f"{subject}#{method_upper}") - - if rpc is None: - raise ValueError( - f"未找到已注册的 RPC 端点: subject='{subject}', method='{method}'. " - f"请确保 @rpc_method 装饰器已先于 @bind_model 执行。" - ) - - # 如果指定了 method，进一步筛选 - if method and rpc.method != method.upper(): - return func - - if request_model is not None: - rpc.request_model = request_model - if response_model is not None: - rpc.response_model = response_model - - func._rpc_meta = rpc # type: ignore[attr-defined] - return func - - return decorator - - -def auto_bind_models(plugin_name: str): - """ - 自动从插件 models 目录发现并绑定 Pydantic 模型 - - 扫描 src/plugins//models/pydantic.py 中定义的所有模型， - 尝试按命名约定匹配已注册的 RPC 端点。 - - 命名约定： - - 请求模型: Request (e.g., AddPostRequest) - - 响应模型: Response (e.g., AddGetResponse) - - 用法: - # 在插件 __init__.py 中调用 - from ...decorators import auto_bind_models - auto_bind_models("example") - """ - try: - models_module = importlib.import_module(f"plugins.{plugin_name}.models.pydantic") - except ImportError: - try: - models_module = importlib.import_module(f"src.plugins.{plugin_name}.models.pydantic") - except ImportError: - print(f"⚠️ auto_bind_models: 未找到 models 模块 (plugins.{plugin_name}.models.pydantic)") - return - - # 获取模块中所有 BaseModel 子类 - model_classes = {} - for name in dir(models_module): - obj = getattr(models_module, name) - if isinstance(obj, type) and issubclass(obj, BaseModel) and obj is not BaseModel: - model_classes[name] = obj - - if not model_classes: - print(f"⚠️ auto_bind_models: 未找到任何 Pydantic 模型") - return - - registry = _RPC_REGISTRY.get(plugin_name, {}) - bound_count = 0 - - for _key, rpc in registry.items(): - op_name = rpc.subject.split(".")[-1] if "." in rpc.subject else rpc.subject - op_camel = op_name.title().replace("-", "").replace("_", "") - method_tag = rpc.method.title() - - # 尝试匹配请求模型: Request - req_model_name = f"{op_camel}{method_tag}Request" - if req_model_name in model_classes: - rpc.request_model = model_classes[req_model_name] - bound_count += 1 - print(f" ✓ 绑定请求模型: {req_model_name} -> {rpc.subject} [{rpc.method}]") - - # 尝试匹配响应模型: Response - resp_model_name = f"{op_camel}{method_tag}Response" - if resp_model_name in model_classes: - rpc.response_model = model_classes[resp_model_name] - bound_count += 1 - print(f" ✓ 绑定响应模型: {resp_model_name} -> {rpc.subject} [{rpc.method}]") - - if bound_count > 0: - print(f"✅ auto_bind_models: 已绑定 {bound_count} 个模型到 {plugin_name} 的 RPC 端点") - else: - print(f"⚠️ auto_bind_models: 未找到匹配的模型绑定") - - -def get_rpc_methods(worker_type: str) -> Dict[str, RPCMethod]: - """获取指定 worker 类型的所有 RPC 方法""" - return _RPC_REGISTRY.get(worker_type, {}) - - -def get_all_rpc_methods() -> Dict[str, Dict[str, RPCMethod]]: - """获取所有已注册的 RPC 方法""" - return _RPC_REGISTRY - - -def clear_registry(): - """清空注册表（主要用于测试）""" - _RPC_REGISTRY.clear() - - -# ============================================================ -# 路由配置生成器（从装饰器元数据生成 TOML 兼容的配置） -# ============================================================ - -def generate_route_configs(worker_type: str) -> List[Dict[str, Any]]: - """ - 从装饰器注册表生成路由配置列表 - - 返回的配置与 config/workers//*.toml 格式兼容， - 可用于动态路由注册或生成配置文件。 - """ - methods = get_rpc_methods(worker_type) - configs: List[Dict[str, Any]] = [] - for _key, rpc in methods.items(): - # 从 subject 提取操作名（最后一部分） - op_name = rpc.subject.split(".")[-1] if "." in rpc.subject else rpc.subject - configs.append({ - "subject": rpc.subject, - "path": f"/{op_name}", - "method": rpc.method, - "params": [f"{name}: {_type_to_config_str(pt)}" for name, pt in rpc.params.items()], - "response_model": _build_response_spec(rpc), - "summary": rpc.summary, - "timeout": rpc.timeout, - }) - return configs - - -def _build_response_spec(rpc: RPCMethod) -> Dict[str, Any]: - """从 RPC 方法构建 response_model 配置""" - spec: Dict[str, Any] = {} - - # 优先使用绑定的 response_model - resp_model = rpc.response_model - if resp_model is None: - resp_model = rpc.build_pydantic_response_model() - - if isinstance(resp_model, type) and issubclass(resp_model, BaseModel): - # 遍历 Pydantic 模型字段 - for field_name, field in resp_model.model_fields.items(): - field_type = _type_to_config_str(field.annotation) - required = field.is_required() - default = field.default if not field.is_required() else None - spec[field_name] = { - "type": field_type, - "required": required, - } - if default is not None: - spec[field_name]["default"] = default - else: - spec["result"] = {"type": _type_to_config_str(rpc.return_type), "required": True} - return spec - -``` - -`./utils/worker/src/chongming_worker/runtime.py` -``` -""" -Worker 运行时 — 通用的生命周期管理 - -提供 create_worker / create_workers 工厂函数，供所有 Worker 插件共享。 -""" -import os -import asyncio -import json -import orjson -from typing import Any, Dict, List, Optional - -import zenoh - -from chongming_worker.decorators import ( - RPCMethod, - get_rpc_methods, -) - -async def create_worker( - session: zenoh.Session, - worker_type: str, - worker_id: str, - config: Optional[Dict[str, Any]] = None, - worker_manager=None, -): - """ - 创建一个 Worker 协程 - - 自动发现通过 @rpc_method 注册的端点，为每个 subject 声明 Zenoh Queryable， - 收到请求后按 _method 字段分派到具体方法。 - - Args: - session: Zenoh 会话 - worker_type: Worker 类型标识（匹配 subject 的第一部分） - worker_id: Worker 唯一 ID - config: 自定义配置参数 - worker_manager: WorkerManager 实例（用于本地注册） - """ - print(f"[Worker] {worker_id} 启动 (type={worker_type})") - - # 1. 注册存活令牌 - token = session.liveliness().declare_token(f"worker/{worker_type}/{worker_id}") - - # 2. 注册到本地 WorkerManager（若有） - if worker_manager is not None: - worker_manager.register_worker(worker_type, worker_id) - - # 3. 获取该 worker 类型的所有 RPC 方法 - rpc_methods = get_rpc_methods(worker_type) - if not rpc_methods: - print(f"[Worker] {worker_id} 警告: 未找到任何 RPC 方法") - else: - print(f"[Worker] {worker_id} 发现 {len(rpc_methods)} 个端点:") - for _key, rpc in rpc_methods.items(): - print(f" {rpc.method} {rpc.subject} -> {rpc.name}({', '.join(rpc.params.keys())})") - - # 4. 按 subject 分组（多个方法可共享同一 subject，如 GET + POST） - subject_methods: Dict[str, List[RPCMethod]] = {} - for registry_key, rpc in rpc_methods.items(): - subject_methods.setdefault(rpc.subject, []).append(rpc) - - # 5. 为每个 subject 创建回调函数并声明 Queryable - stop_event = asyncio.Event() - - def make_callback(subject: str, methods: List[RPCMethod]): - """创建处理特定 subject 的回调函数""" - def callback(query: zenoh.Query): - try: - # 解析请求负载 - # payload_str = str(query.payload) if query.payload else "{}" - payload_bytes = bytes(query.payload) if query.payload else b"" - if not payload_bytes: - data = {} - else: - data = orjson.loads(payload_bytes) - http_method = data.get("_method", "GET").upper() - - # 匹配方法（按 HTTP 方法） - matched = next((m for m in methods if m.method.upper() == http_method), None) - if not matched and methods: - matched = methods[0] # 降级到第一个 - if not matched: - query.reply( - key_expr=query.key_expr, - payload=json.dumps({"error": f"No method for {subject}"}).encode("utf-8") - ) - return - - # 调用 RPC 方法（支持同步和异步） - result = asyncio.run(matched.call(data)) - - # 回复 - query.reply( - key_expr=query.key_expr, - payload=orjson.dumps(result) - ) - print(f"[Worker] {worker_id} 回复 {matched.method} {subject}: {result}") - except Exception as e: - error_response = { - "result": 0.0, - "operation": "error", - "error": str(e), - "worker": worker_id - } - query.reply( - key_expr=query.key_expr, - payload=json.dumps(error_response).encode("utf-8") - ) - print(f"[Worker] {worker_id} 异常 {subject}: {e}") - return callback - - # 声明所有 Queryable - queryables: List[zenoh.Queryable] = [] - for subject, methods in subject_methods.items(): - key = f"worker/{worker_type}/{worker_id}/{subject}" - # 使用回调方式声明 - q = session.declare_queryable(key, make_callback(subject, methods)) - queryables.append(q) - print(f"[Worker] {worker_id} Queryable: '{key}' [{', '.join(m.method for m in methods)}]") - - # 6. 保持 worker 存活直到被取消 - try: - await stop_event.wait() - except asyncio.CancelledError: - print(f"[Worker] {worker_id} 正在关闭...") - except Exception as e: - print(f"[Worker] {worker_id} 发生异常: {e}") - finally: - # 清理 - if worker_manager is not None: - worker_manager.unregister_worker(worker_type, worker_id) - token.undeclare() - for q in queryables: - q.undeclare() - print(f"[Worker] {worker_id} 已关闭") - -def create_workers( - session: zenoh.Session, - config: dict, - count: int = 1, - worker_manager=None, -) -> list[asyncio.Task]: - """ - 创建一组 Worker 协程任务 - - 工厂函数，供 WorkerManager 动态加载使用。 - - Args: - session: Zenoh 会话 - config: Worker 配置字典（至少需要 "type" 字段） - count: 启动的 Worker 副本数 - worker_manager: WorkerManager 实例（可选） - - Returns: - list[asyncio.Task]: Worker 协程任务列表 - """ - worker_type = config.get("type", "default") - pid = os.getpid() - tasks = [] - for i in range(count): - worker_id = f"worker-{i}-{pid}" - task = asyncio.create_task( - create_worker( - session=session, - worker_type=worker_type, - worker_id=worker_id, - config=config, - worker_manager=worker_manager, - ) - ) - tasks.append(task) - print(f"[Worker] 已创建任务: {worker_id} (type={worker_type})") - return tasks - - -__all__ = [ - "create_worker", - "create_workers", -] - -``` - -`./utils/worker/src/chongming_worker/config.py` -``` -import sys -from pathlib import Path -from typing import TypedDict, Literal, Optional, cast, List, Dict, Any - -# 根据Python版本选择合适的TOML解析库和Required类型 -if sys.version_info >= (3, 11): - import tomllib - from typing import Required -else: - import tomli as tomllib - from typing_extensions import Required - - -# ============================================================ -# Worker 配置类型 -# ============================================================ - -class WorkerParams(TypedDict, total=False): - """Worker 操作参数定义""" - type: str - required: bool - default: Any - description: str - - -class WorkerResponse(TypedDict, total=False): - """Worker 操作响应定义""" - type: str - required: bool - default: Any - - -class OperationConfig(TypedDict, total=False): - """ - Worker 操作配置结构 - - 对应 config/workers//*.toml 中的每个操作定义。 - """ - name: Required[str] - method: Required[Literal["GET", "POST", "PUT", "DELETE"]] - path: str - subject: Required[str] - summary: str - timeout: float - params: Dict[str, WorkerParams] - response_model: Dict[str, WorkerResponse] - - -class WorkerConfig(TypedDict, total=False): - """ - Worker 配置结构 - - 对应 config/workers//config.toml 中的顶级配置。 - """ - name: Required[str] - type: Required[str] - description: str - version: str - enabled: bool - operations: List[OperationConfig] - - -def load_worker_config(config_dir: Path) -> Dict[str, Any]: - """ - 加载 Worker 配置目录中的所有 TOML 配置文件 - - Args: - config_dir: Worker 配置目录路径，如 Path("config/workers/example") - - Returns: - 解析后的配置字典，包含所有操作的配置 - - Raises: - FileNotFoundError: 当配置目录不存在时抛出异常 - """ - if not config_dir.exists(): - raise FileNotFoundError(f"Worker 配置目录不存在: {config_dir}") - - config: Dict[str, Any] = {"operations": []} - - # 首先加载 config.toml（如果存在） - config_file = config_dir / "config.toml" - if config_file.exists(): - with open(config_file, "rb") as f: - base_config = tomllib.load(f) - config.update(base_config) - - # 然后加载所有 *.toml 中的操作定义 - for toml_file in sorted(config_dir.glob("*.toml")): - if toml_file.name == "config.toml": - continue - with open(toml_file, "rb") as f: - op_data = tomllib.load(f) - # 将文件名（不含扩展名）作为默认操作名 - op_name = toml_file.stem - if "name" not in op_data: - op_data["name"] = op_name - config["operations"].append(op_data) - - return config - - -# ============================================================ -# 旧的配置系统（与主应用 config.py 兼容） -# ============================================================ - -# 设置 env 为 development 或 production - - -class DefaultApp(TypedDict): - """应用默认配置结构""" - - name: str - version: str - description: str - debug: bool - - -class Default(TypedDict): - """系统默认配置结构，包含应用信息、环境、路径等基础配置""" - - app: DefaultApp - env: Literal["production", "development"] - prefix: str - upload_path: str - - -class SqliteConnectionArgs(TypedDict, total=False): - """SQLite数据库连接参数配置""" - - timeout: int - check_same_thread: bool - - -class DatabaseSqlite(TypedDict, total=False): - """SQLite数据库连接池及行为配置""" - - echo: bool - pool_size: int - max_overflow: int - pool_timeout: int - pool_recycle: int - pool_pre_ping: bool - connect_args: SqliteConnectionArgs - - -class Database(TypedDict, total=False): - """数据库配置结构，支持SQLite、MySQL、PostgreSQL三种数据库类型""" - - type: Required[Literal["sqlite", "mysql", "postgresql"]] - url: Required[str] - generate_schemas: bool - sqlite: DatabaseSqlite - mysql: dict - postgresql: dict - database_path: str - - -class Security(TypedDict): - """安全相关配置，包括JWT令牌和安全策略设置""" - - secret_key: str - algorithm: Literal["HS256"] - access_token_expire_minutes: int - refresh_token_expire_days: int - max_sessions_per_user: int - - -class Server(TypedDict, total=False): - """服务器运行配置，包含主机、端口、工作进程等设置""" - - host: Required[str] - port: Required[int] - reload: bool - workers: int - access_log: bool - - -class Logging(TypedDict): - """日志系统配置，定义日志级别、输出格式和文件管理策略""" - - level: str - file: str - format: str - max_size: int - backup_count: int - - -class Cors(TypedDict, total=False): - """跨域资源共享(CORS)配置""" - - allow_origins: List[str] - allow_credentials: bool - allow_methods: List[str] - allow_headers: List[str] - - -class ModuleSystem(TypedDict, total=False): - """模块系统配置，定义模块加载类型和路径""" - - type: str - path: List[str] - - -class FileSystem(TypedDict, total=False): - """文件系统配置，定义存储类型和路径""" - - type: str - path: str - - -class Env(TypedDict, total=False): - """环境特定配置结构，用于开发和生产环境的差异化配置""" - - debug: Required[bool] - server: Required[Server] - logging: Required[Logging] - cors: Cors - module_system: ModuleSystem - file_system: FileSystem - - -class Scheduler(TypedDict): - """任务调度器配置""" - - job_store_path: str - - -class Cache(TypedDict): - """缓存系统配置""" - - cache_store_path: str - - -class Config(TypedDict): - """完整配置结构，整合所有配置项，包含默认配置、数据库、调度器、缓存、安全和环境配置""" - - default: Default - database: Database - scheduler: Scheduler - cache: Cache - security: Security - development: Env - production: Env - - -def load_config(config_path: Path = Path("config.toml")) -> Config: - """ - 从TOML配置文件加载配置数据 - - Args: - config_path: 配置文件路径，默认为当前目录下的config.toml - - Returns: - Config: 解析后的配置字典对象 - - Raises: - FileNotFoundError: 当配置文件不存在时抛出异常 - """ - if not config_path.exists(): - raise FileNotFoundError(f"配置文件不存在: {config_path}") - with open(config_path, "rb") as f: - return cast(Config, tomllib.load(f)) - -``` - -`./utils/worker/src/chongming_worker/__init__.py` -``` -""" -chongming_worker - 通用 Worker 框架 - -提供 Zenoh RPC 端点的装饰器定义、生命周期管理和配置管理。 -""" -from chongming_worker.decorators import ( - RPCMethod, - rpc_method, - bind_model, - auto_bind_models, - get_rpc_methods, - get_all_rpc_methods, - clear_registry, - generate_route_configs, -) -from chongming_worker.config import ( - load_worker_config, - WorkerConfig, - OperationConfig, -) -from chongming_worker.runtime import ( - create_worker, - create_workers, -) - -__all__ = [ - "RPCMethod", - "rpc_method", - "bind_model", - "auto_bind_models", - "get_rpc_methods", - "get_all_rpc_methods", - "clear_registry", - "generate_route_configs", - "load_worker_config", - "WorkerConfig", - "OperationConfig", - "create_worker", - "create_workers", -] - -``` - -`./utils/request/pyproject.toml` -``` -[build-system] -build-backend = "hatchling.build" -requires = ["hatchling"] - -[project] -name = "chongming-request" -version = "0.1.0" -description = "worker direct communication" -readme = "README.md" -requires-python = ">=3.12" -dependencies = [] - -[tool.hatch.build.targets.wheel] -packages = ["src/chongming_request"] - - -``` - -`./utils/request/README.md` -``` - -``` - -`./utils/request/src/chongming_request/__init__.py` -``` - -``` - -`./public/config.toml` -``` -[default] -app.debug = false -app.description = "基于 FastAPI 的 chongming 打包脚手架" -app.name = "chongming" -app.version = "0.1.0" -env = "production" -prefix = "/api/v1" -upload_path = "uploads" - -[database] -generate_schemas = true # 生产环境使用迁移工具 -type = "sqlite" -url = "sqlite+aiosqlite:///./data/database.db" - -[database.sqlite] -connect_args.check_same_thread = false -connect_args.timeout = 15 -echo = false -echo_pool = false -max_overflow = 10 -pool_pre_ping = true -pool_recycle = 300 -pool_size = 20 -pool_timeout = 30 - -[scheduler] -job_store_path = "scheduler_jobs.db" - -[cache] -cache_store_path = "./" - -[security] -access_token_expire_minutes = 30 -algorithm = "HS256" -max_sessions_per_user = 5 -refresh_token_expire_days = 7 -secret_key = "your-secret-key-change-in-production" - -[production] -debug = false -env = "production" - -[production.server] -access_log = false -host = "0.0.0.0" -port = 8000 -reload = false -timeout_keep_alive = 65 -workers = 4 - -[production.logging] -backup_count = 5 # 最多保存 10 个日志文件 -file = "logs" -format = "%(asctime)s - %(name)s - %(levelname)s - %(message)s" -level = "INFO" -max_size = 10485760 # 10MB - -[production.cors] -allow_credentials = true -allow_headers = ["*"] -allow_methods = ["*"] -allow_origins = ["http://192.168.137.5:8000", "http://localhost:8000", "http://127.0.0.1:8000"] - -[production.module_system] -path = [ - "resources/app.mbank", - "resources/plugins.mbank", -] -type = "module-bank" - -[production.file_system] -path = "static.svfs" -type = "svfs" - -[[workers]] -type = "example" -module = "plugins.example" -factory = "create_workers" -count = 3 -config = { type = "example" } - -``` - -`./public/server.py` -``` -# server.py -import sys -from utils.launch import init_module_bank -from utils.config import ConfigLoader - -# 加载配置文件并初始化模块银行 -config = ConfigLoader("config.toml") -init_module_bank(config) - -# 从数据库动态导入 app 模块 -# 该模块存储在 SQLite 数据库中，通过自定义导入机制加载 -try: - import app # type: ignore - - app_instance = app.app -except ImportError as e: - print(f"无法从数据库导入 app 模块: {e}") - sys.exit(1) - -# 暴露 ASGI 应用实例，供 uvicorn 服务器使用 -app = app_instance - -``` - -`./public/main.py` -``` -# main.py - 修复 reload 问题 -#!/usr/bin/env python3 -""" -FastAPI chongming 示例 -主入口文件，适配 chongming 打包系统 -""" -import os -import sys -import platform -from utils.launch import launch -from multiprocessing import freeze_support -import server - -use_shell = platform.system() == "Windows" - - -def app_run(app_config: dict, default_config: dict): - """ - 应用运行函数 - 适配模块银行环境 - - Args: - app_config: 应用配置字典，包含服务器配置、环境设置等 - default_config: 默认配置字典，包含应用名称、版本等基础信息 - - Returns: - None: 此函数不返回值，直接启动服务器并阻塞运行 - """ - from app.core.logger import get_logger # type: ignore - - logger = get_logger("app") - - # 获取服务器配置 - server_config = app_config.get("server", {}) - host = server_config.get("host", "0.0.0.0") - port = server_config.get("port", 8000) - timeout_keep_alive = server_config.get("timeout_keep_alive", 5) - - # 打包环境强制禁用重载 - is_frozen = getattr(sys, "frozen", False) - reload = False if is_frozen else server_config.get("reload", False) - # 支持通过环境变量 APP_WORKERS 覆盖 worker 数量（方便嵌入式设备调优） - default_workers = server_config.get("workers", 4) - workers = int(os.environ.get("APP_WORKERS", str(default_workers))) - - # 显示启动信息 - env = app_config.get("env", "development") - logger.info("=" * 50) - logger.info(f"🚀 启动 {default_config.get('app.name', 'FastAPI 应用')}") - logger.info(f"📦 版本: {default_config.get('app.version', '1.0.0')}") - logger.info(f"🌍 环境: {env}") - logger.info(f"📍 地址: http://{host}:{port}") - logger.info(f"📚 文档: http://{host}:{port}/docs") - logger.info(f"🔧 调试: {'启用' if app_config.get('debug', False) else '禁用'}") - logger.info(f"🔄 重载: {'启用' if reload else '禁用'}") - logger.info(f"👥 Workers: {workers}") - logger.info("=" * 50) - - # 根据操作系统选择不同的服务器启动方式 - if use_shell: - import uvicorn - - uvicorn.run( - "server:app", - host=host, - port=port, - reload=reload, - workers=workers, - log_level="debug" if app_config.get("debug", False) else "info", - access_log=False, - timeout_keep_alive=timeout_keep_alive, - ) - else: - from gunicorn.app.wsgiapp import run - - sys.argv = [ - "gunicorn", - "server:app", - "-k", - "uvicorn.workers.UvicornWorker", - "-w", - str(workers), - "-b", - f"{host}:{port}", - ] - run() - - -def main(): - """ - 主函数入口 - 初始化多进程支持并启动应用 - - Returns: - None: 此函数不返回值，调用launch函数启动应用 - """ - freeze_support() - launch(app_run, "production") - - -if __name__ == "__main__": - main() - -``` - -`./public/utils/config.py` -``` -# config_loader.py - 读取config.toml的示例代码 - - -import os -import sys -from pathlib import Path -from typing import Any, Dict, Optional - -# 根据Python版本选择合适的TOML解析库 -if sys.version_info >= (3, 11): - import tomllib -else: - import tomli as tomllib - - -class ConfigLoader: - """TOML配置文件加载器 - - 提供配置文件的加载、访问和管理功能。支持自动查找配置文件、 - 点号分隔的嵌套键访问、配置节获取以及配置重载等功能。 - - Attributes: - config_path: 配置文件的路径 - _config: 存储解析后的配置数据 - """ - - def __init__(self, config_path: Optional[str] = None): - """ - 初始化配置加载器 - - Args: - config_path: 配置文件路径，如果为None则自动查找 - - Raises: - FileNotFoundError: 当无法找到配置文件时抛出异常 - """ - if config_path is None: - # 自动查找配置文件 - config_path = self._find_config_file() - - self.config_path = Path(config_path) - self._config: Dict[str, Any] = {} - self.load() - - def _find_config_file(self) -> str: - """ - 自动查找配置文件 - - 按照预定义的优先级顺序搜索配置文件： - 1. 当前目录下的config.toml - 2. config子目录下的config.toml - 3. 用户配置目录~/.config/myapp/config.toml - 4. 系统配置目录/etc/myapp/config.toml - - Returns: - str: 找到的第一个配置文件路径 - - Raises: - FileNotFoundError: 当所有可能的位置都未找到配置文件时抛出异常 - """ - possible_paths = [ - "config.toml", - "config/config.toml", - "~/.config/myapp/config.toml", - "/etc/myapp/config.toml", - ] - - for path in possible_paths: - expanded_path = os.path.expanduser(path) - if os.path.exists(expanded_path): - return expanded_path - - raise FileNotFoundError("未找到配置文件") - - def load(self) -> None: - """ - 加载配置文件 - - 读取并解析TOML格式的配置文件，将配置数据存储到内部字典中。 - - Raises: - FileNotFoundError: 当配置文件不存在时抛出异常 - tomllib.TOMLDecodeError: 当配置文件格式不正确时抛出异常 - """ - if not self.config_path.exists(): - raise FileNotFoundError(f"配置文件不存在: {self.config_path}") - - with open(self.config_path, "r", encoding="utf-8") as f: - self._config = tomllib.loads(f.read()) - - def get(self, key: str, default: Any = None) -> Any: - """ - 获取配置值，支持点号分隔的路径 - - 通过点号分隔的键路径访问嵌套的配置值。例如： - - "database.host" 访问 {"database": {"host": "localhost"}} - - "server.port" 访问 {"server": {"port": 8080}} - - Args: - key: 配置键，支持点号分隔的嵌套路径 - default: 当键不存在时返回的默认值，默认为None - - Returns: - Any: 配置值，如果键不存在则返回默认值 - """ - keys = key.split(".") - value = self._config - - try: - for k in keys: - value = value[k] - return value - except (KeyError, TypeError): - return default - - def get_section(self, section: str) -> Dict[str, Any]: - """ - 获取整个配置节 - - 返回指定配置节的完整字典内容，适用于需要访问某个配置节下 - 所有配置项的场景。 - - Args: - section: 配置节名称 - - Returns: - Dict[str, Any]: 配置节的内容字典，如果节不存在则返回空字典 - """ - return self._config.get(section, {}) - - def reload(self) -> None: - """ - 重新加载配置文件 - - 从磁盘重新读取配置文件并更新内部配置数据。 - 适用于配置文件在运行时被修改的场景。 - - Raises: - FileNotFoundError: 当配置文件不存在时抛出异常 - """ - self.load() - - def __getitem__(self, key: str) -> Any: - """ - 支持字典式访问 - - 允许使用方括号语法访问配置值，如：config["database.host"] - - Args: - key: 配置键 - - Returns: - Any: 配置值 - - Raises: - KeyError: 当键不存在且无默认值时抛出异常（通过get方法处理） - """ - return self.get(key) - - def __contains__(self, key: str) -> bool: - """ - 检查配置是否存在 - - 支持使用in操作符检查配置键是否存在，如："database.host" in config - - Args: - key: 要检查的配置键 - - Returns: - bool: 如果配置键存在且值不为None则返回True，否则返回False - """ - return self.get(key) is not None - - @property - def all(self) -> Dict[str, Any]: - """ - 获取所有配置 - - 返回完整的配置字典，适用于需要遍历或导出所有配置的场景。 - - Returns: - Dict[str, Any]: 包含所有配置的字典 - """ - return self._config - -``` - -`./public/utils/__init__.py` -``` - -``` - -`./public/utils/launch.py` -``` -from .config import ConfigLoader -import os -from module_bank import PythonToSQLite -from typing import Callable - - - -def init_module_bank(config): - """ - 初始化模块银行系统，将指定路径下的Python模块打包并安装导入器。 - - 该函数会验证所有模块路径的有效性，创建PythonToSQLite打包器实例， - 列出所有可用模块，并为每个打包器安装导入器以支持模块的动态加载。 - - Args: - config (dict): 配置字典，必须包含 'production' -> 'module_system' -> 'path' 键， - 其值为模块路径列表 - - Raises: - ValueError: 当配置的模块路径中有任何一个不存在时抛出异常 - """ - MODULE_BANK_PATH_LIST = config["production"]["module_system"]["path"] - for module_path in MODULE_BANK_PATH_LIST: - if not os.path.exists(module_path): - raise ValueError(f"模块路径不存在: {module_path}") - - # 为每个模块路径创建PythonToSQLite打包器实例 - packer_list = [ - PythonToSQLite(module_path) for module_path in MODULE_BANK_PATH_LIST - ] - # 遍历所有打包器，列出模块信息并安装导入器 - for packer in packer_list: - key = packer.get_key() - modules = packer.list_modules() - for module in modules: - print(f"{module['module_name']} {'[包]' if module['is_package'] else ''}") - packer.install_importer(key=key) - - -def launch(run_fun: Callable[[dict, dict], None], run_mode: str): - """ - 启动应用程序，加载配置并执行指定的运行函数。 - - 该函数从config.toml文件中加载配置，根据运行模式获取对应的应用配置和默认配置， - 然后调用传入的运行函数执行应用程序逻辑。 - - Args: - run_fun (Callable[[dict, dict], None]): 运行函数，接收两个参数： - - app_config (dict): 当前运行模式的配置 - - default_config (dict): 默认配置 - run_mode (str): 运行模式，用于从配置文件中获取对应模式的配置项 - - Returns: - None - """ - config = ConfigLoader("config.toml") - - # 获取对应运行模式的配置 - default_config = config.get("default", {}) - app_config = config.get(run_mode, {}) - - run_fun(app_config, default_config) - -``` - -`./data/storage/chongming/.nanio-meta/string/95e51f17-ad95-48d9-a349-e0fe5deba291/k8s部署文档.md.json` -``` -{"etag":"\"99bb1226df2b46c13a8b2f099ecb4778\"","size":5756,"last_modified":"2026-07-16T09:55:58.795584+00:00","content_type":"text/markdown","user_metadata":{},"storage_class":"STANDARD","content_encoding":null,"content_disposition":null,"cache_control":null} -``` - -`./data/storage/chongming/.nanio-meta/test/1f7f0e0e-c347-489c-bfb9-1f3e8083303e/k8s部署文档.md.json` -``` -{"etag":"\"99bb1226df2b46c13a8b2f099ecb4778\"","size":5756,"last_modified":"2026-07-16T09:51:00.518358+00:00","content_type":"text/markdown","user_metadata":{},"storage_class":"STANDARD","content_encoding":null,"content_disposition":null,"cache_control":null} -``` - -`./data/storage/chongming/.nanio-meta/test/c6b0c0db-c681-4d44-b56a-350366a03866/test.mp4.json` -``` -{"etag":"\"cdbfb2788ef3025530e985b2c9ddb850\"","size":918742,"last_modified":"2026-07-16T01:36:13.643254+00:00","content_type":"application/octet-stream","user_metadata":{},"storage_class":"STANDARD","content_encoding":null,"content_disposition":null,"cache_control":null} -``` - -`./data/storage/chongming/.nanio-meta/test/e4269914-e2d1-48c1-875e-b114295943c3/test.mp4.json` -``` -{"etag":"\"cdbfb2788ef3025530e985b2c9ddb850\"","size":918742,"last_modified":"2026-07-16T01:34:26.862901+00:00","content_type":"application/octet-stream","user_metadata":{},"storage_class":"STANDARD","content_encoding":null,"content_disposition":null,"cache_control":null} -``` - -`./data/storage/chongming/string/95e51f17-ad95-48d9-a349-e0fe5deba291/k8s部署文档.md` -``` -# Kubernetes 部署 - -## 使用 containerd 配置 Kubernetes - -自Kubernetes 1.24 版本起，不再支持 docker 作为容器运行时，因此 1.24 及以后版本默认使 用containerd 作为容器。 - -推荐使用1.28 及以上Kubernetes，本例程中以 1.28 版本为例。 - -## 安装 containerd - -本节操作需要在所有节点上进行，包括Master节点与Node节点。 - -1. 卸载设备上原有containerd - - ```shell - sudo apt-get remove containerd - sudo apt-get remove containerd.io - sudo apt-get remove runc - ``` - -2. 更新 apt 包索引 - - ```shell - sudo apt-get update - sudo apt-get install ca-certificates curl gnupg lsb-release - ``` - - 添加官方GPG密钥： - - ``` - sudo mkdir -p /etc/apt/keyrings - curl-fsSL https://download.docker.com/linux/ubuntu/gpg | sudo gpg--dearmor-o /etc/apt/ - →keyrings/docker.gpg - ``` - - 置存储库： - - ``` - echo "deb [arch=$(dpkg--print-architecture) signedby=/etc/apt/keyrings/docker.gpg] https://download.docker.com/linux/ubuntu $(lsb_release-cs) stable" | sudo tee /etc/apt/sources.list.d/docker.list > /dev/null - ``` - -3. 安装 containerd - - 安装最新版本 containerd - - ```shell - sudo apt-get update - sudo apt-get install-y containerd.io - ``` - -## 安装 Kubernetes - -本节操作需要在所有节点上进行，包括Master节点与Node节点 - -1. 卸载设备上已安装Kubernetes - - ```shell - sudo apt-get autoremove kube* - ``` - -2. 更新 apt 包索引并安装 Kubernetes 依赖： - - ```shell - sudo apt-get update - sudo apt-get install -y apt-transport-https ca-certificates curl - ``` - -3. 下载密钥 - - ```shell - sudo curl-fsSL https://pkgs.k8s.io/core:/stable:/v1.28/deb/Release.key | sudo gpg--dearmor-o /etc/apt/keyrings/kubernetes-apt-keyring.gpg - ``` - -4. 添加 Kubernetes 仓库 - - ```shell - sudo echo "deb [signed-by=/etc/apt/keyrings/kubernetes-apt-keyring.gpg] https://pkgs.k8s.io/core:/stable:/v1.28/deb/ /" | sudo tee /etc/apt/sources.list.d/kubernetes.list - ``` - -5. 更新 apt 包索引，并安装 Kubernetes 管理工具 - - ```shell - sudo apt-get update - ``` - - 使用以下指令查看当前可用kubectl 版本 - - ```shell - sudo apt-cache madison kubectl - ``` - - 安装特定版本的 kubectl，kubeadm，kubectl - - ```shell - sudo apt-get install-y kubelet=1.28.0-1.1 kubeadm=1.28.0-1.1 kubectl=1.28.0-1.1 - ``` - -6. 锁定 Kubernetes 版本 - - ```shell - sudo apt-mark hold kubelet kubeadm kubectl - ``` - -## 主节点 - -主节点初始化 - -1. 关闭 swap 分区 - - ```shell - sudo su - swapoff -a - free -m - ``` - -2. 关闭 SELinux - - ```shell - setenforce 0 - ``` - -3. 加载必要的内核模块并设置 sysctl 参数 - - ```shell - modprobe overlay - modprobe br_netfilter - ``` - - 确保模块开机自动加载： - - ```shell - cat < /dev/null - ``` - -3. 安装 containerd - - 安装最新版本 containerd - - ```shell - sudo apt-get update - sudo apt-get install-y containerd.io - ``` - -## 安装 Kubernetes - -本节操作需要在所有节点上进行，包括Master节点与Node节点 - -1. 卸载设备上已安装Kubernetes - - ```shell - sudo apt-get autoremove kube* - ``` - -2. 更新 apt 包索引并安装 Kubernetes 依赖： - - ```shell - sudo apt-get update - sudo apt-get install -y apt-transport-https ca-certificates curl - ``` - -3. 下载密钥 - - ```shell - sudo curl-fsSL https://pkgs.k8s.io/core:/stable:/v1.28/deb/Release.key | sudo gpg--dearmor-o /etc/apt/keyrings/kubernetes-apt-keyring.gpg - ``` - -4. 添加 Kubernetes 仓库 - - ```shell - sudo echo "deb [signed-by=/etc/apt/keyrings/kubernetes-apt-keyring.gpg] https://pkgs.k8s.io/core:/stable:/v1.28/deb/ /" | sudo tee /etc/apt/sources.list.d/kubernetes.list - ``` - -5. 更新 apt 包索引，并安装 Kubernetes 管理工具 - - ```shell - sudo apt-get update - ``` - - 使用以下指令查看当前可用kubectl 版本 - - ```shell - sudo apt-cache madison kubectl - ``` - - 安装特定版本的 kubectl，kubeadm，kubectl - - ```shell - sudo apt-get install-y kubelet=1.28.0-1.1 kubeadm=1.28.0-1.1 kubectl=1.28.0-1.1 - ``` - -6. 锁定 Kubernetes 版本 - - ```shell - sudo apt-mark hold kubelet kubeadm kubectl - ``` - -## 主节点 - -主节点初始化 - -1. 关闭 swap 分区 - - ```shell - sudo su - swapoff -a - free -m - ``` - -2. 关闭 SELinux - - ```shell - setenforce 0 - ``` - -3. 加载必要的内核模块并设置 sysctl 参数 - - ```shell - modprobe overlay - modprobe br_netfilter - ``` - - 确保模块开机自动加载： - - ```shell - cat <{Request|Response} - (如 AddGetRequest, AddGetResponse 由 gen_models 自动生成) - - 装饰器 subject 中的操作名 → 模型前缀（驼峰） - example.add → Add, example.subtract → Subtract -""" -import time - -from chongming_worker import rpc_method, auto_bind_models -from chongming_worker.runtime import create_workers as _create_workers - -# ============================================================ -# 显式导入 Pydantic 模型（避免 PyArmor 打包后 importlib 动态导入异常） -# ============================================================ -from .models.pydantic import ( - AddGetRequest, - AddGetResponse, - AddPostRequest, - AddPostResponse, - SubtractGetRequest, - SubtractGetResponse, - MultiplyGetRequest, - MultiplyGetResponse, - DivideGetRequest, - DivideGetResponse, -) - - -# ============================================================ -# 注册 RPC 端点 -# ============================================================ -# 直接在 @rpc_method 中指定 response_model（和可选的 request_model）， -# 不再依赖 auto_bind_models 进行动态导入绑定。 -# 这样 PyArmor 打包后不会因 importlib.import_module 路径解析失败而出问题。 - - -@rpc_method( - subject="example.add", - method="GET", - summary="Add two numbers via GET", - response_model=AddGetResponse, -) -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, -) -def add_post(add_post_request: AddPostRequest) -> AddPostResponse: - """加法运算（POST）— 通过 JSON Body 传入 a、b，返回它们的和""" - return AddPostResponse( - result=add_post_request.a + add_post_request.b, - operation="add", - timestamp=time.time(), - ) - - -@rpc_method( - subject="example.subtract", - method="GET", - summary="Subtract two numbers", - response_model=SubtractGetResponse, -) -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(), - ) - - -@rpc_method( - subject="example.multiply", - method="GET", - summary="Multiply two numbers", - response_model=MultiplyGetResponse, -) -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(), - ) - -@rpc_method( - subject="example.divide", - method="GET", - summary="Divide two numbers", - response_model=DivideGetResponse, -) -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(), - ) - -# ============================================================ -# 自动绑定 Pydantic 模型（作为备用，但在显式绑定后实际不再需要） -# ============================================================ -# 保留 auto_bind_models 调用作为 fallback，确保任何未显式指定模型的端点 -# 也能自动匹配到合适的模型。由于大部分端点已在 @rpc_method 中显式指定， -# auto_bind_models 会跳过已绑定的端点。 -auto_bind_models("example") - - -# ============================================================ -# 工厂函数 — 供 app lifespan 动态加载 -# ============================================================ -# 框架通过 __import__ + getattr 获取此函数， -# 然后调用 create_workers(session, config, count, worker_manager=worker_manager)。 -# 详见 src/chongming/app/__init__.py 的 lifespan。 - - -def create_workers(session, config, count=1, worker_manager=None): - """插件入口 — 委托给 chongming_worker 运行时创建 Worker 协程""" - return _create_workers(session, config, count, worker_manager) - -``` - -`./src/plugins/example/models/__init__.py` -``` -""" -模型导出 — example -自动生成，请勿手动修改 -""" - -from .pydantic import AddGetRequest -from .pydantic import AddGetResponse -from .pydantic import AddPostRequest -from .pydantic import AddPostResponse -from .pydantic import DivideGetRequest -from .pydantic import DivideGetResponse -from .pydantic import ExampleResponse -from .pydantic import MultiplyGetRequest -from .pydantic import MultiplyGetResponse -from .pydantic import SubtractGetRequest -from .pydantic import SubtractGetResponse - -__all__ = [ - "AddGetRequest", - "AddGetResponse", - "AddPostRequest", - "AddPostResponse", - "DivideGetRequest", - "DivideGetResponse", - "ExampleResponse", - "MultiplyGetRequest", - "MultiplyGetResponse", - "SubtractGetRequest", - "SubtractGetResponse", -] - -``` - -`./src/plugins/example/models/pydantic.py` -``` -""" -自动生成的 Pydantic 模型 — example -来源: config/workers/example/ -""" - -from __future__ import annotations - -from typing import Any, 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 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 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 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 | AddGetResponse | SubtractGetResponse | AddPostResponse | DivideGetResponse - -``` - -`./src/chongming/__init__.py` -``` -import uvicorn -from .app.core.config import load_config -from .app.core.logger import get_logger - - -def serve(): - """ - 使用 uvicorn 启动服务 - - 从配置文件中加载服务器配置和日志配置，然后使用 uvicorn 启动应用服务。 - 该函数适用于开发环境或单进程部署场景。 - - Returns: - None: 该函数不返回值，会阻塞运行直到服务停止 - """ - config = load_config() - env = config["default"]["env"] - server_config = config[env]["server"] - log_config = config[env]["logging"] - logger = get_logger("app") - logger.info(f"启动服务: {server_config}") - logger.info(f"启动日志: {log_config}") - uvicorn.run("chongming.app:app", **server_config) - - -def gunicorn_serve(): - """ - 使用 gunicorn 启动服务 - - 通过 gunicorn 作为进程管理器启动应用，使用 UvicornWorker 作为 worker 类型， - 支持多进程部署，适用于生产环境。 - - 从配置文件中读取服务器配置，动态设置 gunicorn 的 worker 数量、绑定地址和端口等参数。 - - Returns: - None: 该函数不返回值，会阻塞运行直到服务停止 - """ - import sys - from gunicorn.app.wsgiapp import run - - config = load_config() - env = config["default"]["env"] - server_config = config[env]["server"] - - # 构建 gunicorn 命令行参数并启动服务 - sys.argv = [ - "gunicorn", - "chongming.app:app", - "-k", - "uvicorn.workers.UvicornWorker", - "-w", - f"{server_config.get('workers', 4)}", - "-b", - f"{server_config.get('host', '0.0.0.0')}:{server_config.get('port', 8000)}", - ] - run() - - -if __name__ == "__main__": - serve() - -``` - -`./src/chongming/scripts/code_content.py` -``` -import os -import sys - - -def generate_markdown_from_py_files(directory, output_file): - with open(output_file, "w", encoding="utf-8") as md_file: - for root, dirs, files in os.walk(directory): - # 排除 venv 目录 - dirs[:] = [d for d in dirs if d != ".venv"] - dirs[:] = [d for d in dirs if d != ".vscode"] - dirs[:] = [d for d in dirs if d != "node_modules"] - dirs[:] = [d for d in dirs if d != "build"] - dirs[:] = [d for d in dirs if d != "dist"] - dirs[:] = [d for d in dirs if d != "chongming-web"] - for file in files: - if ( - file.endswith(".py") - or file.endswith(".rs") - or file.endswith(".toml") - or file.endswith(".md") - or file.endswith(".css") - or file.endswith(".vue") - or file.endswith(".ts") - or file.endswith(".json") - or file.endswith(".html") - or file.endswith(".sh") - or file == "Dockerfile" - ): - file_path = os.path.join(root, file) - if file == "output.md" or file == "package-lock.json": - continue - if "generated" in file_path: - continue - md_file.write(f"`{file_path}`\n") - md_file.write("```\n") - with open(file_path, "r", encoding="utf-8") as py_file: - md_file.write(py_file.read()) - md_file.write("\n```\n\n") - - -if __name__ == "__main__": - # 指定目录和输出文件名 - target_directory = sys.argv[1] # 替换为你的目标目录 - output_markdown_file = "output.md" # 输出的 Markdown 文件名 - - generate_markdown_from_py_files(target_directory, output_markdown_file) - print(f"Markdown 文件已生成：{output_markdown_file}") - -``` - -`./src/chongming/scripts/gen_models.py` -``` -""" -模型生成脚本 — 从 config/workers//*.toml 生成 Pydantic 模型 - -用法: - uv run gen_models config/workers/example src/plugins/example - -工作流: - 1. 编写 config/workers//*.toml 配置文件 - 2. 执行 gen_models 生成 Pydantic 模型 - 3. 在业务代码中 import 生成的模型 - -功能: - 1. 扫描配置目录下的所有 .toml 文件 - 2. 解析每个路由的 params 和 response_model - 3. 生成 Pydantic 请求/响应模型 - 4. 写入 src/plugins//models/__init__.py 和 models/pydantic.py -""" - -import sys -import json -from pathlib import Path -from typing import Any - - -def parse_type_str(type_str: str) -> str: - """将 TOML 中的类型字符串映射为 Python 类型注解""" - mapping = { - "float": "float", - "int": "int", - "str": "str", - "bool": "bool", - "string": "str", - "integer": "int", - "number": "float", - "boolean": "bool", - "any": "Any", - } - - if type_str.startswith("list[") and type_str.endswith("]"): - inner = type_str[5:-1] - inner_py = mapping.get(inner, inner) - return f"list[{inner_py}]" - - if type_str == "dict": - return "dict[str, Any]" - - if type_str.startswith("dict[") and type_str.endswith("]"): - inner = type_str[5:-1] - if "," in inner: - parts = [p.strip() for p in inner.split(",", 1)] - inner_py = mapping.get(parts[-1], parts[-1]) - else: - inner_py = mapping.get(inner, inner) - return f"dict[str, {inner_py}]" - - if type_str.startswith("Optional[") and type_str.endswith("]"): - inner = type_str[9:-1] - inner_py = mapping.get(inner, inner) - return f"Optional[{inner_py}]" - - return mapping.get(type_str, type_str) - - -def parse_params(params: list[str]) -> list[dict[str, str]]: - """解析 params 列表，如 ['a: float', 'b: float']""" - result = [] - for p in params: - if ":" in p: - name, type_str = p.split(":", 1) - result.append({"name": name.strip(), "type": parse_type_str(type_str.strip())}) - else: - result.append({"name": p.strip(), "type": "str"}) - return result - - -def parse_response_model(response_model: dict[str, Any]) -> list[dict[str, Any]]: - """解析 response_model 字典，返回字段列表""" - fields: list[dict[str, Any]] = [] - for field_name, field_spec in response_model.items(): - entry: dict[str, Any] = {"name": field_name} - if isinstance(field_spec, dict): - entry["type"] = parse_type_str(field_spec.get("type", "str")) - entry["required"] = field_spec.get("required", True) - if "default" in field_spec: - entry["default"] = field_spec["default"] - else: - entry["type"] = parse_type_str(str(field_spec)) - entry["required"] = True - fields.append(entry) - return fields - - -def _model_name(base: str, method: str, suffix: str) -> str: - """生成模型名: e.g. add + GET + Request -> AddGetRequest""" - return f"{base.title().replace('-', '').replace('_', '')}{method.title()}{suffix}" - - -def generate_pydantic_code( - plugin_name: str, - routes: list[dict[str, Any]], -) -> str: - """生成 pydantic.py 文件内容""" - lines = [] - lines.append('"""') - lines.append(f"自动生成的 Pydantic 模型 — {plugin_name}") - lines.append(f"来源: config/workers/{plugin_name}/") - lines.append('"""') - lines.append("") - lines.append("from __future__ import annotations") - lines.append("") - lines.append("from typing import Any, Optional") - lines.append("from pydantic import BaseModel, Field") - lines.append("") - - seen_models: set[str] = set() - - for route in routes: - route_name = route["name"] - method = route["method"] - summary = route.get("summary", "") - - # --- Request Model --- - params = route.get("parsed_params", []) - if params: - model_name = _model_name(route_name, method, "Request") - if model_name not in seen_models: - seen_models.add(model_name) - lines.append(f"class {model_name}(BaseModel):") - lines.append(f' """{summary} — 请求体"""') - for p in params: - lines.append(f" {p['name']}: {p['type']} = Field(...)") - lines.append("") - - # --- Response Model --- - response_fields = route.get("parsed_response", []) - if response_fields: - model_name = _model_name(route_name, method, "Response") - if model_name not in seen_models: - seen_models.add(model_name) - lines.append(f"class {model_name}(BaseModel):") - lines.append(f' """{summary} — 响应体"""') - for f in response_fields: - field_type = f["type"] - if f.get("required", True): - if "default" in f: - default_repr = ( - json.dumps(f["default"]) - if not isinstance(f["default"], str) - else f"'{f['default']}'" - ) - lines.append(f" {f['name']}: {field_type} = Field(default={default_repr})") - else: - lines.append(f" {f['name']}: {field_type} = Field(...)") - else: - if "default" in f: - default_repr = ( - json.dumps(f["default"]) - if not isinstance(f["default"], str) - else f"'{f['default']}'" - ) - lines.append(f" {f['name']}: Optional[{field_type}] = Field(default={default_repr})") - else: - lines.append(f" {f['name']}: Optional[{field_type}] = Field(default=None)") - lines.append("") - - # --- Union type for all responses --- - response_models = [m for m in seen_models if m.endswith("Response")] - if response_models: - plugin_title = plugin_name.title().replace("-", "").replace("_", "") - lines.append("") - lines.append("# 所有响应模型的联合类型") - lines.append(f"{plugin_title}Response = " + " | ".join(response_models)) - lines.append("") - - return "\n".join(lines) - - -def generate_init_code( - plugin_name: str, - routes: list[dict[str, Any]], -) -> str: - """生成 __init__.py 文件内容""" - lines = [] - lines.append('"""') - lines.append(f"模型导出 — {plugin_name}") - lines.append("自动生成，请勿手动修改") - lines.append('"""') - lines.append("") - - seen_models: set[str] = set() - for route in routes: - rname = route["name"] - method = route["method"] - if route.get("parsed_params"): - seen_models.add(_model_name(rname, method, "Request")) - if route.get("parsed_response"): - seen_models.add(_model_name(rname, method, "Response")) - - response_models = [m for m in seen_models if m.endswith("Response")] - if response_models: - plugin_title = plugin_name.title().replace("-", "").replace("_", "") - seen_models.add(f"{plugin_title}Response") - - for model_name in sorted(seen_models): - lines.append(f"from .pydantic import {model_name}") - - lines.append("") - lines.append("__all__ = [") - for model_name in sorted(seen_models): - lines.append(f' "{model_name}",') - lines.append("]") - lines.append("") - - return "\n".join(lines) - - -def scan_config_dir(config_dir: Path) -> list[dict[str, Any]]: - """扫描配置目录，解析所有 .toml 文件中的路由定义""" - import tomli - - routes = [] - toml_files = sorted(config_dir.glob("*.toml")) - - group_config = {} - config_file = config_dir / "config.toml" - if config_file.exists(): - with open(config_file, "rb") as f: - group_config = tomli.load(f) - - defaults = group_config.get("worker", {}).get("defaults", {}) - router_prefix = ( - group_config.get("worker", {}) - .get("registration", {}) - .get("router_prefix", "") - ) - - for toml_file in toml_files: - if toml_file.name == "config.toml": - continue - - with open(toml_file, "rb") as f: - data = tomli.load(f) - - route_name = toml_file.stem - - for method in ("GET", "POST", "PUT", "DELETE", "PATCH"): - if method not in data: - continue - - entry = data[method] - params_raw = entry.get("params", []) - response_raw = entry.get("response_model", {}) - - route = { - "name": route_name, - "method": method, - "summary": entry.get("summary", ""), - "docstring": entry.get("docstring", ""), - "params": params_raw, - "response_model": response_raw, - "parsed_params": parse_params(params_raw), - "parsed_response": parse_response_model(response_raw), - "path": entry.get("path", f"{router_prefix}/{route_name}"), - "timeout": entry.get("timeout", defaults.get("timeout", 3.0)), - "auth_required": entry.get("auth_required", defaults.get("auth_required", False)), - } - routes.append(route) - - return routes - - -def main(): - if len(sys.argv) < 3: - print("用法: uv run gen_models ") - print("示例:") - print(" uv run gen_models config/workers/example src/plugins/example") - sys.exit(1) - - config_dir = Path(sys.argv[1]) - plugin_dir = Path(sys.argv[2]) - plugin_name = plugin_dir.name - - if not config_dir.exists(): - print(f"❌ 配置目录不存在: {config_dir}") - sys.exit(1) - if not plugin_dir.exists(): - print(f"❌ 插件目录不存在: {plugin_dir}") - sys.exit(1) - - print(f"🔍 扫描配置目录: {config_dir}") - routes = scan_config_dir(config_dir) - print(f" 发现 {len(routes)} 个路由定义") - - if not routes: - print("⚠️ 未找到路由定义，跳过模型生成") - return - - models_dir = plugin_dir / "models" - models_dir.mkdir(parents=True, exist_ok=True) - - pydantic_code = generate_pydantic_code(plugin_name, routes) - pydantic_file = models_dir / "pydantic.py" - pydantic_file.write_text(pydantic_code, encoding="utf-8") - print(f"✅ 生成: {pydantic_file}") - - init_code = generate_init_code(plugin_name, routes) - init_file = models_dir / "__init__.py" - init_file.write_text(init_code, encoding="utf-8") - print(f"✅ 生成: {init_file}") - - print(f"\n📋 生成的模型:") - for route in routes: - method = route["method"] - name = route["name"] - params = route.get("parsed_params", []) - response = route.get("parsed_response", []) - if params: - req_model = _model_name(name, method, "Request") - param_strs = [f"{p['name']}: {p['type']}" for p in params] - print(f" 📥 {req_model}: {', '.join(param_strs)}") - if response: - resp_model = _model_name(name, method, "Response") - field_strs = [f"{f['name']}: {f['type']}" for f in response] - print(f" 📤 {resp_model}: {', '.join(field_strs)}") - - -def gen_models(): - """CLI 入口""" - main() - - -if __name__ == "__main__": - main() - -``` - -`./src/chongming/scripts/build.py` -``` -""" -构建脚本 - 将 chongming 应用打包为可执行文件 - -功能阶段: - 1. 模块打包(module-bank): 加密打包 Python 模块 - 2. 资源准备: 创建 build 目录, 复制必要文件 - 3. 混淆与打包: pyarmor 混淆 + PyInstaller/Nuitka 打包 - 4. 数据库初始化: 创建表结构和默认数据 - 5. 前端构建: 编译 Vue 项目并打包为 VFS - -用法: - python -m chongming.scripts.build [pyinstaller|nuitka] -""" - -import json -import ast -import platform -import shutil -import subprocess -import sys -import asyncio -import time -from dataclasses import dataclass, field -from pathlib import Path - -from module_bank import PythonToSQLite -from module_bank.encryption import Encryption -from sqlite_vfs.core import SQLiteVFS -from sqlite_vfs.folder_packer import FolderPacker -from ..app.core.config import get_config, Config - -# ── 常量 ──────────────────────────────────────────────────────────────────── - -PYINSTALLER = "pyinstaller" -NUITKA = "nuitka" -MODES = (PYINSTALLER, NUITKA) - -is_windows = platform.system() == "Windows" - -# ── 构建结果数据结构 ───────────────────────────────────────────────────────── - - -@dataclass -class BuildStepResult: - """单个构建步骤的执行结果""" - - name: str # 步骤名称 - success: bool # 是否成功 - duration: float # 耗时(秒) - message: str = "" # 附加信息(如包大小) - - -@dataclass -class BuildReport: - """构建总结报告""" - - mode: str - start_time: float - end_time: float = 0.0 - steps: list[BuildStepResult] = field(default_factory=list) - - @property - def total_duration(self) -> float: - return self.end_time - self.start_time if self.end_time else 0.0 - - @property - def all_success(self) -> bool: - return all(s.success for s in self.steps) - - def add_step(self, name: str, success: bool, duration: float, message: str = ""): - self.steps.append(BuildStepResult(name, success, duration, message)) - - def print_summary(self): - """打印结构化的构建总结""" - sep = "=" * 60 - print(f"\n{sep}") - print(f" 🏗️ CHONGMING 构建报告") - print(f"{sep}") - print(f" 构建模式 : {self.mode}") - print(f" 总耗时 : {self.total_duration:.2f}s") - print(f" 构建状态 : {'✅ 成功' if self.all_success else '❌ 失败'}") - print(f"{sep}") - print(f" 步骤明细:") - for i, step in enumerate(self.steps, 1): - icon = "✅" if step.success else "❌" - msg = f" ({step.message})" if step.message else "" - print( - f" {icon} [{i}/{len(self.steps)}] {step.name:<20s} {step.duration:>7.2f}s{msg}" - ) - print(f"{sep}\n") - - -# ── 工具函数 ──────────────────────────────────────────────────────────────── - - -def _run_cmd( - cmd: list[str], - cwd: str | None = None, - capture: bool = True, - shell: bool | None = None, -) -> subprocess.CompletedProcess: - """统一执行外部命令""" - if shell is None: - shell = is_windows - return subprocess.run( - cmd, - cwd=cwd, - capture_output=capture, - text=True, - shell=shell, - encoding="utf-8", - errors="replace", - ) - - -def _format_size(path: str | Path) -> str: - """获取文件可读大小""" - size = Path(path).stat().st_size - for unit in ("B", "KB", "MB", "GB"): - if size < 1024: - return f"{size:.2f} {unit}" - size /= 1024.0 - return f"{size:.2f} TB" - - -def _ensure_dir(path: str | Path): - """确保目录存在""" - Path(path).mkdir(parents=True, exist_ok=True) - - -def _resolve_mode() -> str: - """解析构建模式""" - mode = PYINSTALLER - if len(sys.argv) >= 2: - mode = sys.argv[1].lower() - if mode not in MODES: - print(f" ❌ 不支持的构建模式: {mode}，可选: {', '.join(MODES)}") - sys.exit(1) - return mode - - -# ── 阶段 1: 模块打包 (module-bank) ────────────────────────────────────────── - - -def _stage_pack_modules(config: Config) -> str: - """ - 使用 module-bank 加密打包 Python 模块为 .mbank 文件。 - 返回打包过程中的提示信息。 - """ - _ensure_dir("resources") - mbank_list = config.get("production", {}).get("module_system", {}).get("path", []) - if not mbank_list: - return "⚠️ 未配置 module_system.path，跳过模块打包" - - messages = [] - for output_path, source_dir, package_name in mbank_list: - key = Encryption.generate_key() - packer = PythonToSQLite( - output_path, - key, - pyarmor_obfuscate=True, - ) - packer.pack_directory(source_dir, package_name) - packer.verify_package_structure() - packer.packer_key(key) - packer.delete_source_code(None) - size = _format_size(output_path) - messages.append(f"{Path(output_path).name} ({size})") - return " - " + "\n - ".join(messages) - - -def _stage_prepare_build_dir(config: Config, mode: str) -> str: - """准备 build 目录并复制必要文件""" - build_path = Path("build") - if build_path.exists(): - shutil.rmtree(str(build_path)) - _ensure_dir("build") - - # 复制 utils - utils_src = Path("public/utils") - if utils_src.exists(): - shutil.copytree(utils_src, build_path / "utils") - - # 复制 resources - resources_src = Path("resources") - if resources_src.exists(): - shutil.copytree(resources_src, build_path / "resources") - - resources_src = Path("config") - if resources_src.exists(): - shutil.copytree(resources_src, build_path / "config") - - # 复制核心文件 - for f in ("main.py", "server.py", "config.toml"): - shutil.copy(f"public/{f}", f"build/{f}") - - # 注入 JWT 密钥 - jwt_key = Encryption.generate_key() - cfg_path = build_path / "config.toml" - content = cfg_path.read_text(encoding="utf-8") - content = content.replace("your-secret-key-change-in-production", jwt_key) - cfg_path.write_text(content, encoding="utf-8") - - return "资源准备完成" - - -# ── 阶段 3: PyArmor 混淆 ─────────────────────────────────────────────────── - - -def _stage_pyarmor_obfuscate() -> str: - """对关键脚本执行 PyArmor 混淆""" - files = [ - ("main.py", "./"), - ("server.py", "./"), - ("utils/launch.py", "./utils"), - ("utils/config.py", "./utils"), - ] - failed = [] - for filename, output in files: - result = _run_cmd( - ["pyarmor", "gen", f"--output={output}", filename], - cwd="build", - ) - if result.returncode != 0: - failed.append(filename) - if failed: - return f"❌ 混淆失败: {', '.join(failed)}" - return "✅ 所有文件混淆成功" - - -# ── 阶段 4: PyInstaller 打包 ──────────────────────────────────────────────── - - -def _stage_pyinstaller(config: Config) -> tuple[str, str]: - """执行 PyInstaller 打包，返回 (状态消息, 可执行文件路径)""" - base_imports = config.get("pyinstaller_build_import", {}) - build_imports = list(base_imports.get("base", [])) - build_imports.extend( - base_imports.get("windows", []) if is_windows else base_imports.get("linux", []) - ) - - icon = Path("../../public/chongming.ico") - cmd = [ - sys.executable, - "-m", - "PyInstaller", - "--onefile", - "--noconfirm", - "--optimize", - "2", - "--name", - "chongming", - "--distpath", - ".", - "--workpath", - "./build_temp", - "--specpath", - "./specs", - "--clean", - ] - if not is_windows: - cmd.append("--strip") - if icon.exists(): - cmd.extend(["--icon", str(icon)]) - for imp in build_imports: - cmd.extend(["--hidden-import", imp]) - - cmd.append("main.py") - - result = _run_cmd(cmd, cwd="build", capture=True) - if result.returncode != 0: - return f"❌ PyInstaller 失败", "" - - exe_path = Path("build/chongming") - if is_windows: - exe_path = exe_path.with_suffix(".exe") - - suffix = ".exe" if is_windows else "" - exe = Path("build") / f"chongming{suffix}" - if exe.exists(): - size = _format_size(exe) - return f"✅ PyInstaller 打包成功 ({size})", str(exe) - return "✅ PyInstaller 打包完成 (但未找到可执行文件)", "" - - -# ── 阶段 5: Nuitka 打包 ──────────────────────────────────────────────────── - - -def _stage_nuitka(config: Config) -> tuple[str, str]: - """执行 Nuitka 打包，返回 (状态消息, 可执行文件路径)""" - base_imports = config.get("pyinstaller_build_import", {}).get("base", []) - extra = config.get("pyinstaller_build_import", {}).get( - "windows" if is_windows else "linux", [] - ) - - packages = [ - "module_bank", - "sqlite_vfs", - "fastapi", - "uvicorn", - "gunicorn", - "diskcache", - "apscheduler", - "tomli", - "typing_extensions", - ] - - cmd = [ - sys.executable, - "-m", - "nuitka", - "--standalone", - "--onefile", - "--follow-imports", - "--static-libpython=no", - "--remove-output", - "--assume-yes-for-downloads", - "--lto=yes", - "--disable-plugin=tk-inter", - "--disable-plugin=numpy", - "--output-dir=.", - "--output-filename=chongming", - "main.py", - "--jobs=2", - ] - if shutil.which("upx"): - cmd.append("--enable-plugin=upx") - for pkg in packages: - cmd.append(f"--include-package={pkg}") - for mod in list(base_imports) + extra: - cmd.append(f"--include-module={mod}") - - if is_windows: - icon = Path("../../public/chongming.ico") - if icon.exists(): - cmd.append(f"--windows-icon-from-ico={icon}") - - try: - _run_cmd(cmd, cwd="build", capture=False) - except subprocess.CalledProcessError: - # 降级重试（移除可能不兼容的参数） - fallback = [ - arg for arg in cmd if arg not in ("--lto=yes", "--enable-plugin=upx") - ] - _run_cmd(fallback, cwd="build", capture=False) - - exe = Path("build/chongming") - if exe.exists(): - # 后处理瘦身 - if not is_windows: - subprocess.run(["strip", str(exe)], check=False) - if shutil.which("upx"): - subprocess.run(["upx", "--best", str(exe)], check=False) - size = _format_size(exe) - return f"✅ Nuitka 编译完成 ({size})", str(exe) - - return "❌ Nuitka 编译失败 (未生成可执行文件)", "" - - -# ── 阶段 6: 清理临时文件 ──────────────────────────────────────────────────── - - -def _stage_cleanup(mode: str) -> str: - """清理构建过程中的临时文件""" - build = Path("build") - dirs_to_remove = [ - "build_temp", - "specs", - "utils", - "pyarmor_runtime_000000", - "__pycache__", - ] - files_to_remove = ["main.py", "server.py"] - - for d in dirs_to_remove: - target = build / d - if target.exists(): - shutil.rmtree(str(target), ignore_errors=True) - - for f in files_to_remove: - target = build / f - if target.exists(): - target.unlink(missing_ok=True) - - return "临时文件已清理" - -# ── 阶段 8: Vue 前端构建 ──────────────────────────────────────────────────── - - -def _stage_build_vue() -> str: - """构建 Vue 前端并打包为 SQLite VFS""" - dist_dir = "./src/chongming-web/dist" - output_db = "./build/static.svfs" - fs_name = "frontend" - - # 1. npm build - result = _run_cmd( - ["npm", "run", "build"], - cwd="./src/chongming-web", - capture=True, - ) - if result.returncode != 0: - err = (result.stderr or "").strip()[:200] - return f"❌ Vue 构建失败: {err}" - - # 2. 检查产物 - if not Path(dist_dir).exists(): - return "❌ Vue 构建成功但未找到 dist 目录" - - # 3. 打包为 VFS - vfs = SQLiteVFS(output_db, compress=False) - packer = FolderPacker(sqlite_vfs=vfs, exclude_patterns=[], fs_name=fs_name) - try: - packer.pack_folder(dist_dir) - finally: - vfs.close() - - size = _format_size(output_db) - return f"✅ Vue 打包完成 ({size})" - - -# ── 主流程 ────────────────────────────────────────────────────────────────── - - -async def main(): - """规范化构建主流程""" - mode = _resolve_mode() - report = BuildReport(mode=mode, start_time=time.time()) - config = get_config() - - async def run_step(name: str, fn, *args, **kwargs) -> str: - """执行一个构建步骤并记录结果（支持同步和异步函数）""" - start = time.time() - try: - result = fn(*args, **kwargs) - if asyncio.iscoroutine(result): - msg = await result - else: - msg = result - elapsed = time.time() - start - report.add_step(name, success=True, duration=elapsed, message=msg or "") - print(f" ✅ {name:<25s} {elapsed:>7.2f}s {msg}") - return msg - except Exception as e: - elapsed = time.time() - start - err_msg = str(e).strip()[:200] - report.add_step(name, success=False, duration=elapsed, message=err_msg) - print(f" ❌ {name:<25s} {elapsed:>7.2f}s {err_msg}") - return "" - - print("\n" + "=" * 60) - print(" 🏗️ CHONGMING 构建开始") - print(f" 模式: {mode} | 平台: {platform.system()}") - print("=" * 60 + "\n") - - # 阶段 1: 模块打包 - await run_step("模块打包(module-bank)", _stage_pack_modules, config) - - # 阶段 2: 准备构建目录 - await run_step("准备构建目录", _stage_prepare_build_dir, config, mode) - - # 阶段 3+4: 混淆 + 打包 - if mode == PYINSTALLER: - await run_step("PyArmor 混淆", _stage_pyarmor_obfuscate) - start = time.time() - try: - msg, exe_path = _stage_pyinstaller(config) - elapsed = time.time() - start - success = "❌" not in msg - report.add_step( - "PyInstaller 打包", success, elapsed, msg.replace("✅ PyInstaller ", "") - ) - print( - f" {'✅' if success else '❌'} PyInstaller 打包{' ' * 12} {elapsed:>7.2f}s {msg}" - ) - except Exception as e: - elapsed = time.time() - start - report.add_step("PyInstaller 打包", False, elapsed, str(e)[:200]) - print(f" ❌ PyInstaller 打包 {elapsed:>7.2f}s {str(e)[:200]}") - else: - await run_step("Nuitka 编译", lambda: _stage_nuitka(config)[0]) - - # 阶段 5: 清理 - await run_step("清理临时文件", _stage_cleanup, mode) - - # 阶段 7: Vue 前端构建 - await run_step("Vue 前端构建", _stage_build_vue) - - # 总结 - report.end_time = time.time() - report.print_summary() - - if not report.all_success: - sys.exit(1) - - -def build(): - """构建入口""" - asyncio.run(main()) - - -if __name__ == "__main__": - build() - -``` - -`./src/chongming/app/__init__.py` -``` -""" -chongming FastAPI 应用入口 - -重构为配置驱动的微服务网关： -- WorkerManager: 统一管理 Zenoh worker 存活检测、负载均衡、超时重试、缓存 -- RouteGenerator: 根据配置自动生成 FastAPI 端点 -- 插件工厂函数动态加载 -""" -import os - -import time -import subprocess -import asyncio -from contextlib import asynccontextmanager -import boto3 -from botocore.exceptions import ClientError -from botocore.config import Config -from fastapi import FastAPI, Request -from fastapi.responses import JSONResponse -from fastapi.middleware.cors import CORSMiddleware - -from .core.config import get_config, get_workers_config -from .core.logger import get_logger -from .core.static_files import SVFSStaticFiles -from .routes import storage_router - -config = get_config() -logger = get_logger("app") - -try: - env = config["default"]["env"] -except KeyError: # pragma: no cover - raise ValueError("配置不存在") - - -@asynccontextmanager -async def lifespan(app: FastAPI): - import zenoh - from diskcache import Cache - from .core.worker_manager import WorkerManager - from .core.route_generator import register_dynamic_routes - - env = os.environ.copy() - env["NANIO_ACCESS_KEY"] = "minioadmin" - env["NANIO_SECRET_KEY"] = "minioadmin" - - nanio_process = subprocess.Popen( - ["nanio", "serve", "--data-dir", "./data/storage"], - env=env, - stdout=subprocess.DEVNULL, - stderr=subprocess.DEVNULL - ) - app.state.nanio_process = nanio_process - - await asyncio.sleep(1) # 等待 nanio 启动完成 - - s3_client_for_init = boto3.client( - "s3", - endpoint_url="http://127.0.0.1:9000", - aws_access_key_id="minioadmin", - aws_secret_access_key="minioadmin", - config=Config(signature_version="s3v4"), - region_name="us-east-1" - ) - try: - s3_client_for_init.head_bucket(Bucket="chongming") - except ClientError as e: - if e.response["Error"]["Code"] == "404": - s3_client_for_init.create_bucket(Bucket="chongming") - logger.info("Bucket 'chongming' 已创建") - else: - raise - app.state.s3_client = s3_client_for_init - app.state.storage_bucket = "chongming" - app.state.storage_presign_expire = 300 - app.state.storage_max_size = 100 * 1024 * 1024 # 100MB - - # 1. 建立 Zenoh 会话 - zconfig = zenoh.Config() - zconfig.from_file("config/zenoh_config.json5") - - try: - session = zenoh.open(zconfig) - logger.info("Zenoh 会话已建立 (TCP 自连接模式)") - except Exception as e: - logger.error(f"Zenoh 会话建立失败: {e}") - raise - - app.state.zenoh_session = session - logger.info("Zenoh 会话已建立") - - # 2. 创建 WorkerManager - cache = Cache(config["cache"]["cache_store_path"]) - worker_manager = WorkerManager(session, cache) - app.state.worker_manager = worker_manager - - # 3. 启动 Worker 存活刷新任务 - refresh_task = asyncio.create_task(worker_manager.refresh_workers()) - app.state.refresh_task = refresh_task - - # 4. 动态加载 worker 插件（根据配置） - workers_config_path = config.get("plugins_worker", {}).get("config_path", "config/workers/config.toml") - workers_conf = get_workers_config(workers_config_path).get("workers", []) - print(workers_conf) - worker_tasks = [] - for wc in workers_conf: - module_name = wc.get("module", "") - factory_name = wc.get("factory", "") - count = wc.get("count", 1) - worker_config = wc.get("config", {}) - - if not module_name or not factory_name: - logger.warning(f"Worker 配置不完整: {wc}") - continue - - try: - # 动态导入模块 - module = __import__(module_name, fromlist=[factory_name]) - factory = getattr(module, factory_name) - print(f"[factory]({factory_name}):{factory}") - # 调用工厂函数，传入 session、自定义配置、副本数和 worker_manager - # worker_manager 会传递给每个 worker，让 worker 启动后直接注册到本地管理器 - tasks = factory(session, worker_config, count, worker_manager=worker_manager) - worker_tasks.extend(tasks) - logger.info(f"已加载 Worker 插件: {module_name}.{factory_name} (副本数: {count})") - except Exception as e: - logger.error(f"加载 Worker 插件失败 {module_name}: {e}") - - app.state.worker_tasks = worker_tasks - logger.info(f"已启动 {len(worker_tasks)} 个 Worker") - - # 5. 注册动态路由 - register_dynamic_routes(app, worker_manager) - - # 短暂等待确保 worker 状态已同步 - await asyncio.sleep(0.5) - - # 初始同步完成后打印 worker 状态 - worker_status = { - wtype: len(workers) - for wtype, workers in app.state.worker_manager._workers_cache.items() - } - logger.info(f"Worker 就绪状态: {worker_status}") - - yield # 应用运行中... - - # 6. 清理 - logger.info("正在关闭应用...") - - # 停止 Worker 刷新任务 - refresh_task.cancel() - try: - await refresh_task - except asyncio.CancelledError: - pass - - # 取消所有 Worker 任务 - for task in worker_tasks: - task.cancel() - if worker_tasks: - await asyncio.gather(*worker_tasks, return_exceptions=True) - - # 关闭 Zenoh 会话 - session.close() - logger.info("应用已关闭") - - nanio_process.terminate() - nanio_process.wait() - logger.info("Nanio 进程已终止") - - -cors = config[env].get("cors", {}) -app = FastAPI( - title=config["default"]["app"]["name"], - version=config["default"]["app"]["version"], - description=config["default"]["app"]["description"], - docs_url="/docs" if config[env]["debug"] else None, - redoc_url="/redoc" if config[env]["debug"] else None, - openapi_url="/openapi.json" if config[env]["debug"] else None, - lifespan=lifespan, -) - -app.add_middleware(CORSMiddleware, **cors) - - -@app.middleware("http") -async def access_log_middleware(request: Request, call_next): - """ - HTTP访问日志中间件，记录每个请求的详细信息。 - - 记录的信息包括： - - 请求方法和路径 - - 响应状态码 - - 请求处理时长（毫秒） - - 客户端IP地址 - - 参数: - request: FastAPI请求对象 - call_next: 下一个中间件或路由处理函数 - - 返回: - Response: 处理后的响应对象 - """ - start = time.perf_counter() - response = await call_next(request) - duration = (time.perf_counter() - start) * 1000 - client = request.client.host if request.client else "-" - - logger.info( - f"{request.method} {request.url.path} status={response.status_code} " - f"duration={duration:.2f}ms client={client}", - extra={"status_code": response.status_code}, - ) - return response - - -@app.get("/", summary="根路径") -async def root(): - """ - 根路径端点，返回应用的基本信息。 - - 返回: - dict: 包含应用名称、版本、文档地址和健康检查地址的字典 - """ - return { - "message": f"欢迎使用 {config['default']['app']['name']}", - "version": config["default"]["app"]["version"], - "docs": "/docs" if config[env]["debug"] else None, - "health": f"{config['default']['prefix']}/health", - } - - -@app.exception_handler(Exception) -async def global_exception_handler(_: Request, exc: Exception): - """ - 全局异常处理器，捕获并处理应用中未捕获的异常。 - - 在调试模式下返回详细的异常信息，在生产模式下返回通用错误提示。 - - 参数: - _: FastAPI请求对象（未使用） - exc: 捕获到的异常对象 - - 返回: - JSONResponse: 包含错误信息的JSON响应，状态码为500 - """ - logger.error(f"全局异常: {exc}") - return JSONResponse( - status_code=500, - content={ - "error": "内部服务器错误", - "detail": str(exc) if config[env]["debug"] else "请查看服务器日志", - }, - ) - - -# 静态文件服务（可选） -if vfs_db_path := config[env].get("file_system", {}).get("path", None): - logger.info("静态文件已存在，将使用静态文件服务") - static_files_handler = SVFSStaticFiles(directory="/", vfs_db_path=vfs_db_path) - app.mount("/static", static_files_handler, name="static") - -app.include_router(storage_router) - -``` - -`./src/chongming/app/core/route_generator.py` -``` -""" -动态路由生成器 - -根据配置文件中的 worker 路由定义，动态生成 FastAPI 端点。 -为每个端点生成带有正确 Pydantic 类型注解的处理函数，确保 OpenAPI 文档 -正确反映参数类型、响应结构，并注册到 components/schemas。 -""" -import asyncio -import json -from typing import Any, Dict, List, Optional, get_type_hints -from inspect import Parameter, Signature - -from fastapi import FastAPI, HTTPException, APIRouter, Query, Path as FastAPIPath, Depends -from pydantic import create_model, Field, BaseModel - -from .worker_manager import WorkerManager -from .logger import get_logger - -logger = get_logger("route_generator") - -# 简单类型映射 -TYPE_MAP: Dict[str, Any] = { - "float": float, - "int": int, - "str": str, - "bool": bool, - "any": Any, -} - - -def _parse_python_type(type_str: str) -> Any: - """解析配置中的类型字符串为 Python 类型对象""" - type_str = type_str.strip() - if type_str.startswith("list[") or type_str.startswith("List["): - inner = type_str[5:-1].strip() - inner_type = _parse_python_type(inner) - return List[inner_type] # type: ignore - if type_str.startswith("dict[") or type_str.startswith("Dict["): - inner = type_str[5:-1].strip() - parts = [p.strip() for p in inner.split(",", 1)] - if len(parts) == 2: - key_type = _parse_python_type(parts[0]) - val_type = _parse_python_type(parts[1]) - return Dict[key_type, val_type] # type: ignore - return Dict[str, Any] # type: ignore - return TYPE_MAP.get(type_str, str) - - -def generate_request_model( - param_spec: Dict[str, str], model_name: str = "AutoRequest" -) -> type[BaseModel]: - """生成 Pydantic 请求体模型""" - fields: Dict[str, Any] = {} - for param_name, param_type_str in param_spec.items(): - py_type = _parse_python_type(param_type_str) - fields[param_name] = (py_type, Field(...)) - return create_model(model_name, **fields) - - -def generate_response_model( - response_spec: Dict[str, Any], model_name: str = "AutoResponse" -) -> Optional[type[BaseModel]]: - """生成 Pydantic 响应模型""" - if not response_spec: - return None - fields: Dict[str, Any] = {} - for field_name, field_spec in response_spec.items(): - if isinstance(field_spec, dict): - py_type = _parse_python_type(field_spec.get("type", "str")) - required = field_spec.get("required", True) - default_val = field_spec.get("default", ...) - if required: - fields[field_name] = (py_type, Field(...)) - else: - fields[field_name] = (py_type, Field(default=default_val if default_val is not ... else None)) - else: - py_type = _parse_python_type(str(field_spec)) - fields[field_name] = (py_type, Field(...)) - return create_model(model_name, **fields) - - -class RouteConfig: - """单个路由配置模型""" - - def __init__(self, data: Dict[str, Any]): - self.subject: str = data.get("subject", "") - self.path: str = data.get("path", "") - self.method: str = data.get("method", "GET").upper() - self.params: List[str] = data.get("params", []) - self.response_model: Dict[str, Any] = data.get("response_model", {}) - self.summary: str = data.get("summary", "") - self.docstring: str = data.get("docstring", "") - self.timeout: float = data.get("timeout", 3.0) - self.cacheable: bool = data.get("cacheable", False) - - def parse_params(self) -> Dict[str, str]: - """解析 params 列表为 {name: type} 字典""" - result: Dict[str, str] = {} - for param in self.params: - if ":" in param: - name, type_str = param.split(":", 1) - result[name.strip()] = type_str.strip() - else: - result[param.strip()] = "str" - return result - - def extract_path_params(self) -> List[str]: - """从路径中提取路径参数，如 /add/{a}/{b} -> ['a', 'b']""" - return [ - seg[1:-1] for seg in self.path.split("/") - if seg.startswith("{") and seg.endswith("}") - ] - - -# 默认的 worker 配置目录 -WORKER_CONFIG_DIR = "config/workers" - - -def load_worker_routes_from_dir(worker_type: str) -> List[Dict[str, Any]]: - """ - 从 config/workers/{worker_type}/ 目录加载路由配置。 - - 目录结构: - config/workers/{worker_type}/ - config.toml # [worker] 分组信息 + registration + defaults - add.toml # 接口名 = 文件名（不含扩展名） - [GET] - summary = ... - params = [...] - response_model = { ... } - [POST] - ... # 同一个接口支持多个请求方法 - - 返回: List[Dict], 每个 dict 包含 flatten 后的路由配置, - 每个方法节(如 GET) 生成一个独立的路由条目。 - """ - from pathlib import Path - import tomllib - - base_dir = Path(WORKER_CONFIG_DIR) / worker_type - if not base_dir.is_dir(): - logger.warning(f"Worker 配置目录不存在: {base_dir}") - return [] - - # 1. 读取分组配置 - group_cfg_path = base_dir / "config.toml" - group_defaults: Dict[str, Any] = {} - registration: Dict[str, Any] = {} - router_prefix: str = "" - if group_cfg_path.exists(): - with open(group_cfg_path, "rb") as f: - cfg = tomllib.load(f) - w = cfg.get("worker", {}) - group_defaults = w.get("defaults", {}) - registration = w.get("registration", {}) - router_prefix = registration.get("router_prefix", "") - - # 2. 遍历所有接口 toml 文件 - routes: List[Dict[str, Any]] = [] - for route_file in sorted(base_dir.glob("*.toml")): - if route_file.name == "config.toml": - continue - - # 文件名（不含 .toml）作为路径名 - route_name = route_file.stem # 例如 "add" - base_path = router_prefix + "/" + route_name # 例如 "/calc/add" - - with open(route_file, "rb") as f: - route_data = tomllib.load(f) - - # 每个方法节（GET/POST/PUT/DELETE）生成一个独立路由 - for method, spec in route_data.items(): - if not isinstance(spec, dict): - logger.warning(f"忽略非 dict 的方法节: {route_file.name}[{method}]") - continue - - # 合并: spec 显式字段优先，然后 group_defaults 兜底 - route = dict(spec) - for key, default_val in group_defaults.items(): - if key not in route: - route[key] = default_val - - # 覆盖关键字段 - method 始终从 TOML 节键获取，不被 defaults 覆盖 - route["method"] = method.upper() - route["path"] = route.get("path", base_path) - if "subject" not in route: - # 自动生成 subject: worker_type.route_name (如 "calc.add") - route["subject"] = f"{worker_type}.{route_name}" - - routes.append(route) - logger.debug( - f" 已加载: {route['method']} {route['path']} -> {route['subject']}" - ) - - logger.info(f"从 {base_dir} 加载了 {len(routes)} 个路由") - return routes - - -def register_dynamic_routes(app: FastAPI, worker_manager: WorkerManager): - """根据 config/workers/ 目录结构注册所有动态路由""" - from pathlib import Path - - workers_base = Path(WORKER_CONFIG_DIR) - if not workers_base.is_dir(): - logger.info(f"Worker 配置目录不存在: {WORKER_CONFIG_DIR}") - return - - router = APIRouter() - extra_schemas: Dict[str, type[BaseModel]] = {} - - # 扫描所有 worker 子目录（每个子目录名 = worker_type） - for worker_dir in sorted(workers_base.iterdir()): - if not worker_dir.is_dir(): - continue - worker_type = worker_dir.name - routes = load_worker_routes_from_dir(worker_type) - if not routes: - continue - - for route_conf in routes: - _register_single_route(router, worker_manager, worker_type, route_conf, extra_schemas) - - app.include_router(router) - - # 将额外 schema 注册到 OpenAPI 的 components/schemas - if extra_schemas: - _inject_schemas_to_openapi(app, extra_schemas) - - logger.info(f"动态路由注册完成，worker 配置目录: {WORKER_CONFIG_DIR}") - - -def _inject_schemas_to_openapi(app: FastAPI, schemas: Dict[str, type[BaseModel]]): - """ - 将自定义 Pydantic 模型注册到 FastAPI 的 OpenAPI components/schemas 中。 - - 通过重写 app.openapi() 方法注入，因为 FastAPI 在生命周期内只生成一次 - OpenAPI schema，之后会被缓存。 - """ - original_openapi = app.openapi - - def patched_openapi() -> dict[str, Any]: - if app.openapi_schema is not None: - return app.openapi_schema - schema = original_openapi() - if schema is None: - return {"openapi": "3.1.0", "info": {"title": "Chongming", "version": "1.0.0"}, "paths": {}} - # 注入我们的 schema - schemas_dict = schema.setdefault("components", {}).setdefault("schemas", {}) - for name, model in schemas.items(): - if name not in schemas_dict: - schemas_dict[name] = model.model_json_schema() - app.openapi_schema = schema - return schema - - app.openapi = patched_openapi - - -def _make_body_endpoint(worker_manager: WorkerManager, rc: RouteConfig, worker_type: str, RequestModel: type[BaseModel]): - """创建用于 Body 请求（POST/PUT/PATCH）的端点函数""" - path = rc.path - subject = rc.subject - timeout = rc.timeout - http_method = rc.method - cacheable = rc.cacheable - - async def endpoint( - body: RequestModel, # type: ignore - ) -> Any: - try: - payload = body.model_dump() - payload.setdefault("_method", http_method) - result = await worker_manager.call( - worker_type=worker_type, - subject=subject, - payload=payload, - timeout=timeout, - cacheable=cacheable, - ) - return result - except RuntimeError as e: - raise HTTPException(status_code=503, detail=str(e)) - except Exception as e: - logger.error(f"路由 {path} 调用异常: {e}") - raise HTTPException(status_code=500, detail="内部服务器错误") - - return endpoint - - -def _make_query_endpoint(worker_manager: WorkerManager, rc: RouteConfig, worker_type: str, path_params: List[str], query_params: Dict[str, str]): - """ - 创建用于 Query 请求（GET/DELETE）的端点函数。 - - 通过设置函数 __signature__ 和 __annotations__ 为每个参数赋予正确的 Pydantic 类型， - 使 FastAPI 能在 OpenAPI schema 中生成正确的 type / format 信息。 - 请求模型 schema 通过 extra_schemas 机制单独注入到 components/schemas。 - """ - path = rc.path - subject = rc.subject - timeout = rc.timeout - http_method = rc.method - cacheable = rc.cacheable - - # 合并路径参数和查询参数 - param_spec_combined: Dict[str, str] = {} - param_spec_combined.update({k: query_params.get(k, "str") for k in path_params if k in query_params}) - param_spec_combined.update(query_params) - - # 无参数情况 - if not param_spec_combined: - async def _simple_endpoint() -> Any: - try: - result = await worker_manager.call( - worker_type=worker_type, - subject=subject, - payload={"_method": http_method}, - timeout=timeout, - cacheable=cacheable, - ) - return result - except RuntimeError as e: - raise HTTPException(status_code=503, detail=str(e)) - except Exception as e: - logger.error(f"路由 {path} 调用异常: {e}") - raise HTTPException(status_code=500, detail="内部服务器错误") - return _simple_endpoint - - # 构造参数名列表（用于从 kwargs 中提取） - param_names = list(param_spec_combined.keys()) - - async def endpoint(**kwargs: Any) -> Any: - payload = {k: kwargs[k] for k in param_names} - payload.setdefault("_method", http_method) - try: - result = await worker_manager.call( - worker_type=worker_type, - subject=subject, - payload=payload, - timeout=timeout, - cacheable=cacheable, - ) - return result - except RuntimeError as e: - raise HTTPException(status_code=503, detail=str(e)) - except Exception as e: - logger.error(f"路由 {path} 调用异常: {e}") - raise HTTPException(status_code=500, detail="内部服务器错误") - - # 构建带类型注解的 Parameter 列表 - params: List[Parameter] = [] - annotations: Dict[str, Any] = {} - for name, type_str in param_spec_combined.items(): - py_type = _parse_python_type(type_str) - annotations[name] = py_type - if name in path_params: - params.append(Parameter(name, Parameter.KEYWORD_ONLY, default=FastAPIPath(...), annotation=py_type)) - else: - params.append(Parameter(name, Parameter.KEYWORD_ONLY, default=Query(...), annotation=py_type)) - - # 设置 __annotations__（FastAPI 通过 get_type_hints 读取） - endpoint.__annotations__ = annotations - # 设置 __signature__（FastAPI 优先于此获取参数列表） - endpoint.__signature__ = Signature(params, return_annotation=Dict[str, Any]) # type: ignore - - return endpoint - - -def _register_single_route( - router: APIRouter, - worker_manager: WorkerManager, - worker_type: str, - route_conf: Dict[str, Any], - extra_schemas: Optional[Dict[str, type[BaseModel]]] = None, -): - """注册单个路由，生成正确的函数签名以便 OpenAPI 正确生成文档""" - rc = RouteConfig(route_conf) - param_spec = rc.parse_params() - path_params = rc.extract_path_params() - query_params = {k: v for k, v in param_spec.items() if k not in path_params} - - # 生成干净的模型名称（替换路径中的 / 和 {} 为 _） - # 模型名称包含 HTTP method 以区分同名路径的不同调用类型 - clean_path = rc.path.replace("/", "_").replace("{", "").replace("}", "").strip("_") - if not clean_path: - clean_path = "root" - method_tag = rc.method.upper() - response_model_name = f"{worker_type}_{clean_path}_{method_tag}_Response" - request_model_name = f"{worker_type}_{clean_path}_{method_tag}_Request" - - # 生成响应模型 - ResponseModel = generate_response_model( - rc.response_model, response_model_name - ) - - if rc.method in ("POST", "PUT", "PATCH"): - RequestModel = generate_request_model( - param_spec, request_model_name - ) - endpoint = _make_body_endpoint(worker_manager, rc, worker_type, RequestModel) - else: - # 对 GET 端点也生成 Request 模型并注册到 extra_schemas - if param_spec: - RequestModel = generate_request_model( - param_spec, request_model_name - ) - if extra_schemas is not None and request_model_name not in extra_schemas: - extra_schemas[request_model_name] = RequestModel - endpoint = _make_query_endpoint(worker_manager, rc, worker_type, path_params, query_params) - - # 注册路由 - 使用 worker_type 作为 OpenAPI 分组标签 - # 每个 worker 类型自动成为一个独立的分组，显示在 Swagger UI 中 - router.add_api_route( - path=rc.path, - endpoint=endpoint, - methods=[rc.method], - summary=rc.summary, - description=rc.docstring, - response_model=ResponseModel, - tags=[worker_type], - include_in_schema=True, - ) - logger.info(f"已注册动态路由: {rc.method} {rc.path} -> worker {worker_type}.{rc.subject} | tags=[{worker_type}]") - -``` - -`./src/chongming/app/core/worker_manager.py` -``` -import asyncio -import random -import json -import orjson -import time -from typing import Any, Dict, List, Optional, Set -import zenoh -from diskcache import Cache -from .logger import get_logger - -logger = get_logger("worker_manager") - -_SENTINEL = object() # 哨兵值，用于标记"No reply received" - - -class WorkerManager: - """ - 统一 Worker 管理器 - - 封装 Zenoh 会话，维护每个 worker 类型的存活节点列表， - 提供负载均衡、超时重试、缓存等能力的统一 RPC 调用入口。 - - 支持两种 worker 注册方式： - 1. 分布式：通过 Zenoh liveliness 自动发现 - 2. 本地：通过 register_worker() 直接注册（用于同进程 worker） - """ - - def __init__(self, session: zenoh.Session, cache: Optional[Cache] = None): - self.session = session - self.cache = cache - self._workers_cache: Dict[str, List[str]] = {} # type -> list of worker key_exprs - self._local_workers: Dict[str, List[str]] = {} # type -> list of locally registered workers - self._lock = asyncio.Lock() - self._refresh_interval = 15 # seconds - - def register_worker(self, worker_type: str, worker_id: str): - """ - 直接注册一个本地 worker（同进程内使用） - 这样无需等待 liveliness 发现，立即可用。 - - 同步执行（仅修改内存字典），确保调用后立即可被 get_worker 发现。 - """ - key_expr = f"worker/{worker_type}/{worker_id}" - # 同步直接写入 _local_workers，无需异步锁 - # 因为 _local_workers 和 _workers_cache 在调用 call() 时通过 await get_worker() 读取 - self._local_workers.setdefault(worker_type, []).append(key_expr) - logger.debug(f"本地 worker 已注册: {key_expr}") - - def unregister_worker(self, worker_type: str, worker_id: str): - """取消注册一个本地 worker""" - key_expr = f"worker/{worker_type}/{worker_id}" - if worker_type in self._local_workers: - try: - self._local_workers[worker_type].remove(key_expr) - except ValueError: - pass - - async def refresh_workers(self): - """定期刷新存活 worker 列表（合并 Zenoh liveliness + 本地注册）""" - while True: - try: - # 1. 从 Zenoh liveliness 获取分布式 workers - replies = self.session.liveliness().get("worker/*/*") - new_cache: Dict[str, List[str]] = {} - for reply in replies: - if reply.ok: - key = str(reply.ok.key_expr) # e.g. "worker/example/worker-0" - parts = key.split("/") - if len(parts) >= 3: - worker_type = parts[1] - new_cache.setdefault(worker_type, []).append(key) - - # 2. 合并本地注册的 workers - for wtype, workers in self._local_workers.items(): - existing = set(new_cache.get(wtype, [])) - for w in workers: - if w not in existing: - new_cache.setdefault(wtype, []).append(w) - - async with self._lock: - self._workers_cache = new_cache - logger.debug(f"Worker 存活列表更新: {self._workers_cache}") - except Exception as e: - logger.error(f"刷新 worker 列表失败: {e}") - await asyncio.sleep(self._refresh_interval) - - async def get_worker(self, worker_type: str) -> Optional[str]: - """获取一个可用的 worker 地址（简单随机负载均衡）""" - async with self._lock: - workers = self._workers_cache.get(worker_type, []) - # 如果缓存为空，回退到本地注册表（适用于启动初期 refresh_workers 尚未执行时） - if not workers: - local = self._local_workers.get(worker_type, []) - if local: - return random.choice(local) - return None - return random.choice(workers) - - def is_success_response(self, result: dict) -> bool: - """根据业务约定判断响应是否成功。 - 假设成功响应没有 'error' 字段，或 'code' 为 0。 - """ - if "error" in result and result["error"]: - return False - if "code" in result and result["code"] != 0: - return False - return True - - async def call( - self, - worker_type: str, - subject: str, - payload: Dict[str, Any], - timeout: float = 3.0, - retries: int = 1, - cacheable: bool = False, - ) -> Any: - """ - 统一调用入口：缓存 > 负载均衡 > 超时重试 - - Args: - worker_type: Worker 类型标识（如 "example"） - subject: 调用的主题（如 "calc.add"） - payload: 请求参数 - timeout: 单次调用超时（秒） - retries: 失败重试次数 - cacheable: 是否允许缓存结果（若为 False 则跳过缓存读写） - - Returns: - Dict[str, Any]: 调用结果 - - Raises: - RuntimeError: 无可用 worker 或所有重试均失败 - """ - # 1. 缓存检查（仅在 cacheable=True 时检查） - cache_key = f"{worker_type}:{subject}:{json.dumps(payload, sort_keys=True)}" - if cacheable and self.cache is not None: - if cache_key in self.cache: - logger.debug(f"命中缓存: {cache_key}") - return self.cache[cache_key] - - # 2. 获取可用 worker - worker_expr = await self.get_worker(worker_type) - if not worker_expr: - raise RuntimeError(f"No available worker for type {worker_type}") - - full_key = f"{worker_expr}/{subject}" - payload_bytes = orjson.dumps(payload) - - for attempt in range(retries + 1): - try: - reply_iter = self.session.get( - full_key, - payload=payload_bytes, - consolidation=zenoh.ConsolidationMode.NONE, - ) - # 使用辅助函数获取第一个回复，避免 StopIteration 在 asyncio.to_thread 中传播的问题 - def _get_first(iterator): - """线程安全地获取迭代器的第一个元素""" - for item in iterator: - return item - return _SENTINEL - - first_reply = await asyncio.wait_for( - asyncio.to_thread(_get_first, reply_iter), - timeout=timeout, - ) - if first_reply is _SENTINEL: - logger.warning( - f"Worker {worker_expr} 无回复，尝试重试 {attempt+1}/{retries+1}" - ) - continue - - reply = first_reply - if reply.ok: # type: ignore - # result_str = str(reply.ok.payload) - payload_bytes = bytes(reply.ok.payload) if reply.ok.payload else b"" # type: ignore - if not payload_bytes: - result = {} # 或根据业务返回默认值 - else: - result = orjson.loads(payload_bytes) - # 缓存结果（仅在 cacheable=True 时写入） - if cacheable and self.cache is not None and self.is_success_response(result): - self.cache[cache_key] = result - return result - else: - error_payload = reply.err.payload # type: ignore - error_msg = str(error_payload) if error_payload else "unknown error" - logger.warning(f"Worker 返回错误: {error_msg} (尝试 {attempt+1}/{retries+1})") - continue # 重试 - except asyncio.TimeoutError: - logger.warning( - f"Worker {worker_expr} 超时，尝试重试 {attempt+1}/{retries+1}" - ) - continue - except Exception as e: - logger.error(f"调用 worker 异常: {e}") - raise - - raise RuntimeError( - f"Failed to call worker {worker_type} after {retries} retries" - ) - -``` - -`./src/chongming/app/core/config.py` -``` -import sys -from pathlib import Path -from typing import TypedDict, Literal, Optional, cast, List - - -# 根据Python版本选择合适的TOML解析库和Required类型 -if sys.version_info >= (3, 11): - import tomllib - from typing import Required -else: - import tomli as tomllib - from typing_extensions import Required - -# 设置 env 为 development 或 production - - -class DefaultApp(TypedDict): - """应用默认配置结构""" - - name: str - version: str - description: str - debug: bool - - -class Default(TypedDict): - """系统默认配置结构，包含应用信息、环境、路径等基础配置""" - - app: DefaultApp - env: Literal["production", "development"] - prefix: str - upload_path: str - - -class SqliteConnectionArgs(TypedDict, total=False): - """SQLite数据库连接参数配置""" - - timeout: int - check_same_thread: bool - - -class DatabaseSqlite(TypedDict, total=False): - """SQLite数据库连接池及行为配置""" - - echo: bool - pool_size: int - max_overflow: int - pool_timeout: int - pool_recycle: int - pool_pre_ping: bool - connect_args: SqliteConnectionArgs - - -class Database(TypedDict, total=False): - """数据库配置结构，支持SQLite、MySQL、PostgreSQL三种数据库类型""" - - type: Required[Literal["sqlite", "mysql", "postgresql"]] - url: Required[str] - generate_schemas: bool - sqlite: DatabaseSqlite - mysql: dict - postgresql: dict - database_path: str - - -class Security(TypedDict): - """安全相关配置，包括JWT令牌和安全策略设置""" - - secret_key: str - algorithm: Literal["HS256"] - access_token_expire_minutes: int - refresh_token_expire_days: int - max_sessions_per_user: int - - -class Server(TypedDict, total=False): - """服务器运行配置，包含主机、端口、工作进程等设置""" - - host: Required[str] - port: Required[int] - reload: bool - workers: int - access_log: bool - - -class Logging(TypedDict): - """日志系统配置，定义日志级别、输出格式和文件管理策略""" - - level: str - file: str - format: str - max_size: int - backup_count: int - - -class Cors(TypedDict, total=False): - """跨域资源共享(CORS)配置""" - - allow_origins: List[str] - allow_credentials: bool - allow_methods: List[str] - allow_headers: List[str] - - -class ModuleSystem(TypedDict, total=False): - """模块系统配置，定义模块加载类型和路径""" - - type: str - path: List[str] - - -class FileSystem(TypedDict, total=False): - """文件系统配置，定义存储类型和路径""" - - type: str - path: str - - -class Env(TypedDict, total=False): - """环境特定配置结构，用于开发和生产环境的差异化配置""" - - debug: Required[bool] - server: Required[Server] - logging: Required[Logging] - cors: Cors - module_system: ModuleSystem - file_system: FileSystem - - -class Scheduler(TypedDict): - """任务调度器配置""" - - job_store_path: str - - -class Cache(TypedDict): - """缓存系统配置""" - - cache_store_path: str - - -class Config(TypedDict): - """完整配置结构，整合所有配置项，包含默认配置、数据库、调度器、缓存、安全和环境配置""" - - default: Default - database: Database - scheduler: Scheduler - cache: Cache - security: Security - development: Env - production: Env - - -def load_config(config_path: Path = Path("config.toml")) -> Config: - """ - 从TOML配置文件加载配置数据 - - Args: - config_path: 配置文件路径，默认为当前目录下的config.toml - - Returns: - Config: 解析后的配置字典对象 - - Raises: - FileNotFoundError: 当配置文件不存在时抛出异常 - """ - if not config_path.exists(): - raise FileNotFoundError(f"配置文件不存在: {config_path}") - with open(config_path, "rb") as f: - return cast(Config, tomllib.load(f)) - - -# 全局配置缓存变量，避免重复加载配置文件 -_config: Optional[Config] = None - - -def get_config(config_path: str = "config.toml") -> Config: - """ - 获取应用程序配置（带缓存机制） - - 首次调用时加载配置文件并缓存，后续调用直接返回缓存的配置对象 - - Args: - config_path: 配置文件路径，默认为config.toml - - Returns: - Config: 配置字典对象 - """ - global _config - if _config is None: - _config = load_config(Path(config_path)) - return _config - -def get_workers_config(config_path: str = "config/workers/config.toml") -> dict: - """ - 获取Worker插件配置列表 - - Args: - config_path: Worker配置文件路径，默认为config/workers/config.toml - - Returns: - list: Worker配置字典列表 - """ - if not Path(config_path).exists(): - raise FileNotFoundError(f"Worker配置文件不存在: {config_path}") - with open(config_path, "rb") as f: - return cast(dict, tomllib.load(f)) - -``` - -`./src/chongming/app/core/__init__.py` -``` - -``` - -`./src/chongming/app/core/logger.py` -``` -import re -import logging -import datetime -from pathlib import Path -from logging.handlers import RotatingFileHandler - -from ..core.config import get_config - - -config = get_config() -env = config["default"]["env"] -server_config = config[env]["server"] -log_config = config[env]["logging"] - - -class ColoredFormatter(logging.Formatter): - """ - 为 HTTP 状态码添加颜色的日志格式化器 - - 根据 HTTP 状态码的首位数字为日志消息中的状态码添加不同颜色： - - 2xx: 绿色（成功） - - 3xx: 青色（重定向） - - 4xx: 黄色（客户端错误） - - 5xx: 红色（服务器错误） - - Attributes: - COLORS: 状态码首位数字到 ANSI 颜色代码的映射字典 - RESET: ANSI 重置颜色代码 - """ - - COLORS = { - "2": "\033[32m", # 绿色 2xx - "3": "\033[36m", # 青色 3xx - "4": "\033[33m", # 黄色 4xx - "5": "\033[31m", # 红色 5xx - } - RESET = "\033[0m" - - def format(self, record): - """ - 格式化日志记录，为状态码添加颜色 - - 如果日志记录包含 status_code 属性，则查找消息中的 "status=数字" 模式 - 并为其添加对应的颜色代码。格式化完成后恢复原始消息以避免影响其他 handler。 - - Args: - record: 日志记录对象，应包含 msg 属性和可选的 status_code 属性 - - Returns: - str: 格式化后的日志消息字符串，状态码部分可能包含 ANSI 颜色代码 - """ - # 保存原始消息 - original_msg = record.msg - # 如果日志记录包含 status_code 属性，则为其添加颜色 - status_code = getattr(record, "status_code", None) - if status_code and isinstance(status_code, int): - first_digit = str(status_code)[0] - color = self.COLORS.get(first_digit, "") - colored_status = f"{color}{status_code}{self.RESET}" - # 替换消息中的 "status=数字" 为带颜色的版本 - record.msg = re.sub( - rf"status={status_code}\b", f"status={colored_status}", original_msg - ) - result = super().format(record) - # 恢复原始消息，避免影响其他 handler - record.msg = original_msg - return result - - -class ExcludeWatchFilesFilter(logging.Filter): - """ - 日志过滤器，用于排除第三方库的日志输出 - - 过滤掉来自 watchfiles、aiosqlite、apscheduler、sqlalchemy 等库的日志记录， - 避免这些库的详细日志污染应用日志输出。 - """ - - def __init__(self): - """ - 初始化过滤器，设置需要过滤的日志名称前缀列表 - """ - super().__init__() - self._filter_starts_with_tulpe = ( - "watchfiles", - "aiosqlite", - "apscheduler", - "sqlalchemy", - "sqlitedict", - "sqlalchemy.engine.Engine", - ) - - def filter(self, record): - """ - 判断是否应该保留该日志记录 - - Args: - record: 日志记录对象，包含 name 属性表示日志来源 - - Returns: - bool: 如果日志名称不以过滤列表中的任何前缀开头则返回 True（保留）， - 否则返回 False（过滤掉） - """ - return not record.name.startswith(self._filter_starts_with_tulpe) - - -_configured_loggers = set() - - -def setup_logging(logger: logging.Logger, log_name: str = "app"): - """ - 配置日志系统，包括文件和控制台输出 - - 为指定的 logger 配置以下功能： - 1. 设置日志级别和格式 - 2. 创建按日期命名的轮转文件处理器（支持文件大小限制和备份数量控制） - 3. 创建带颜色格式化的控制台处理器 - 4. 应用过滤器排除第三方库日志 - - 该函数具有防重复配置机制，同一个 log_name 只会配置一次。 - - Args: - logger: 需要配置的 logging.Logger 实例 - log_name: 日志名称标识，用于区分不同的日志配置和生成日志文件名，默认为 "app" - - Returns: - None - - Raises: - 无显式抛出异常，但可能因文件系统权限问题导致目录创建或文件写入失败 - """ - if log_name in _configured_loggers: - return - _configured_loggers.add(log_name) - log_level = getattr(logging, log_config["level"], logging.INFO) - log_format = log_config["format"] - # 创建文件夹 - log_file_path = Path(log_config["file"]) - if log_file_path != Path("."): - log_file_path.mkdir(parents=True, exist_ok=True) - - # 配置根日志记录器 - logger.setLevel(log_level) - - # 创建格式化器 - plain_formatter = logging.Formatter(log_format) - color_formatter = ColoredFormatter(log_format) - - # 创建轮转文件处理器 - 限制文件大小为 10MB，保留 5 个备份文件 - max_bytes = log_config.get("max_size", 10 * 1024 * 1024) # 默认 10MB - backup_count = log_config.get("backup_count", 5) # 默认保留 5 个备份 - - file_handler = RotatingFileHandler( - f"{log_config['file']}/{log_name}-{datetime.date.today()}.log", - encoding="utf-8", - maxBytes=max_bytes, - backupCount=backup_count, - ) - file_handler.setFormatter(plain_formatter) - file_handler.addFilter(ExcludeWatchFilesFilter()) - - # 创建控制台处理器并添加过滤器 - console_handler = logging.StreamHandler() - console_handler.setFormatter(color_formatter) - console_handler.addFilter(ExcludeWatchFilesFilter()) - - # 添加处理器到根日志记录器 - logger.addHandler(file_handler) - logger.addHandler(console_handler) - - print( - f"日志系统已配置 - 级别: {logging.getLevelName(log_level)}, 文件: {log_config['file']} " - f"(最大：{max_bytes / 1024 / 1024:.1f}MB, 备份：{backup_count})" - ) - - -def get_logger(name: str) -> logging.Logger: - """ - 获取配置好的日志记录器实例 - - 根据名称获取或创建 logger 实例，并自动调用 setup_logging 进行配置。 - 如果该名称的 logger 已经配置过，则直接返回已配置的实例。 - - Args: - name: 日志记录器名称，通常使用模块名如 __name__ - - Returns: - logging.Logger: 已配置完成的日志记录器实例，可直接用于记录日志 - """ - logger = logging.getLogger(name) - setup_logging(logger, name) - return logger - -``` - -`./src/chongming/app/core/static_files.py` -``` -from fastapi import HTTPException -from fastapi.responses import Response -from fastapi.staticfiles import StaticFiles -from sqlite_vfs.core import SQLiteVFS - - -class SVFSStaticFiles(StaticFiles): - """ - 基于 SQLite VFS 的静态文件服务类 - - 该类继承自 FastAPI 的 StaticFiles，使用 SQLite 虚拟文件系统来存储和提供静态文件。 - 支持单页应用（SPA）的路由回退机制，当请求的文件不存在时，会自动返回 index.html。 - """ - - def __init__(self, directory, vfs_db_path="static.svfs"): - """ - 初始化 SVFSStaticFiles 实例 - - 参数: - directory (str): 静态文件的根目录路径 - vfs_db_path (str): SQLite VFS 数据库文件路径，默认为 "static.svfs" - - 返回值: - 无 - """ - vfs_db_path = vfs_db_path - self._vfs = SQLiteVFS(vfs_db_path, compress=False) - super().__init__(directory=directory) - - async def get_response(self, path: str, scope): - """ - 根据请求路径获取对应的静态文件响应 - - 该方法处理静态文件的请求逻辑，包括： - 1. 路径标准化处理 - 2. 文件存在性检查 - 3. SPA 路由回退支持 - 4. 目录索引处理 - 5. MIME 类型自动识别 - - 参数: - path (str): 请求的文件路径 - scope: ASGI 作用域对象，包含请求上下文信息 - - 返回值: - Response: FastAPI 响应对象，包含文件内容和正确的 Content-Type - - 异常: - HTTPException: 当文件不存在且无法回退到 index.html 时，抛出 404 错误 - """ - # 统一路径格式：以 / 开头，并将 Windows 反斜杠转为正斜杠 - normalized_path = "/" + path.replace("\\", "/") - - # 获取文件信息 - file_info = self._vfs.get_file_info(normalized_path) - - # 处理文件不存在的情况，尝试 SPA 回退机制 - if file_info is None: - # SPA 回退：如果不是静态资源扩展名，尝试返回 index.html - if not any( - normalized_path.endswith(ext) - for ext in [ - ".js", - ".css", - ".png", - ".jpg", - ".ico", - ".svg", - ".webp", - ".json", - ] - ): - index_info = self._vfs.get_file_info("/index.html") - if index_info: - content = self._vfs.read_file("/index.html") - return Response(content=content, media_type="text/html") - raise HTTPException(status_code=404, detail="File not found") - - # 处理目录请求，返回目录下的 index.html - if file_info["is_directory"]: - index_info = self._vfs.get_file_info(f"{normalized_path}/index.html") - if index_info: - content = self._vfs.read_file(f"{normalized_path}/index.html") - return Response(content=content, media_type="text/html") - raise HTTPException(status_code=404, detail="Directory index not found") - - # 读取文件内容并确定 MIME 类型 - content = self._vfs.read_file(normalized_path) - content_type = file_info.get("content_type") or "application/octet-stream" - # 简单根据扩展名补充 MIME 类型（可选） - if normalized_path.endswith(".js"): - content_type = "application/javascript" - elif normalized_path.endswith(".css"): - content_type = "text/css" - elif normalized_path.endswith(".html"): - content_type = "text/html" - elif normalized_path.endswith(".ico"): - content_type = "image/x-icon" - - return Response(content=content, media_type=content_type) - -``` - -`./src/chongming/app/routes/storage.py` -``` -# src/chongming/app/routes/storage.py -import uuid -import urllib.parse -from fastapi import APIRouter, UploadFile, File, Form, HTTPException, Request -from fastapi.responses import StreamingResponse -from botocore.exceptions import ClientError - -router = APIRouter(prefix="/api/v1/storage", tags=["Storage"]) - - -@router.post("/upload", summary="上传文件") -async def upload_file( - request: Request, - file: UploadFile = File(...), - worker_type: str = Form(...), -): - """后端代理上传，直接写入 nanio""" - s3_client = request.app.state.s3_client - bucket = request.app.state.storage_bucket - object_key = f"{worker_type}/{uuid.uuid4()}/{file.filename}" - - try: - s3_client.upload_fileobj( - file.file, - bucket, - object_key, - ExtraArgs={"ContentType": file.content_type or "application/octet-stream"} - ) - except ClientError as e: - raise HTTPException(status_code=500, detail=f"S3 上传失败: {str(e)}") - - return { - "file_key": object_key, - "bucket": bucket, - "filename": file.filename, - "content_type": file.content_type, - "size": file.size or 0, # 注意：UploadFile.size 可能为 None，需读取后计算 - } - - -@router.get("/download/{file_key:path}") -async def download_file(request: Request, file_key: str): - s3_client = request.app.state.s3_client - bucket = request.app.state.storage_bucket - - try: - response = s3_client.get_object(Bucket=bucket, Key=file_key) - content_type = response.get('ContentType', 'application/octet-stream') - raw_filename = file_key.split('/')[-1] - # 百分号编码，确保仅 ASCII 字符 - encoded = urllib.parse.quote(raw_filename, safe='') - # 仅使用 filename*（现代浏览器都支持） - content_disposition = f"attachment; filename*=UTF-8''{encoded}" - - return StreamingResponse( - response['Body'].iter_chunks(), - media_type=content_type, - headers={'Content-Disposition': content_disposition} - ) - except ClientError as e: - if e.response['Error']['Code'] == 'NoSuchKey': - raise HTTPException(404, "文件不存在") - raise HTTPException(500, f"下载失败: {e}") -``` - -`./src/chongming/app/routes/__init__.py` -``` -from .storage import router as storage_router - -__all__ = ["storage_router"] -``` -