From 3f97dd92ef774050d6bc0d602717c88d4b2006ed Mon Sep 17 00:00:00 2001 From: qing <1776177662@qq.com> Date: Thu, 30 Jul 2026 21:57:04 +0800 Subject: [PATCH] update --- .dockerignore | 61 ++++++ .env.example | 11 ++ ai_review/.dockerignore | 18 ++ ai_review/Dockerfile | 43 +++++ backups/.gitkeep | 2 + captcha/captcha_service.py | 46 +---- db/.dockerignore | 20 ++ db/Dockerfile | 53 +++++ db/db_tools.py | 304 +++++++++++++++++++++++++++++ db/dbv2.py | 87 +++++---- db/docker-entrypoint.sh | 58 ++++++ dnfs/.dockerignore | 18 ++ dnfs/Dockerfile | 48 +++++ dnfs/dnfs.py | 33 ++-- dnfs/log_client.py | 69 ++++--- docker-compose.yml | 109 +++++++++++ docker.sh | 386 +++++++++++++++++++++++++++++++++++++ pymsg/README.md | 118 +++++++----- pymsg/README.robot.md | 22 ++- pymsg/pymsg/__init__.py | 6 + pymsg/pymsg/rpc.py | 171 ++++++++++++++++ pymsg/pyproject.toml | 2 +- pymsg/tests/test_rpc.py | 145 ++++++++++++++ smtp/log_client.py | 85 ++++---- smtp/smtp_v2.py | 51 +---- static/log_client.py | 350 +++++++++++++++++---------------- 26 files changed, 1865 insertions(+), 451 deletions(-) create mode 100644 .dockerignore create mode 100644 .env.example create mode 100644 ai_review/.dockerignore create mode 100644 ai_review/Dockerfile create mode 100644 backups/.gitkeep create mode 100644 db/.dockerignore create mode 100644 db/Dockerfile create mode 100644 db/db_tools.py create mode 100644 db/docker-entrypoint.sh create mode 100644 dnfs/.dockerignore create mode 100644 dnfs/Dockerfile create mode 100644 docker-compose.yml create mode 100755 docker.sh create mode 100644 pymsg/pymsg/rpc.py create mode 100644 pymsg/tests/test_rpc.py diff --git a/.dockerignore b/.dockerignore new file mode 100644 index 0000000..37ecf92 --- /dev/null +++ b/.dockerignore @@ -0,0 +1,61 @@ +# Git +.git +.gitignore +.gitattributes + +# Python +__pycache__/ +*.pyc +*.pyo +*.pyd +.Python +*.so +.env +.env/ +.venv/ +venv/ +ENV/ +env/ + +# IDE +.vscode/ +.idea/ +*.swp +*.swo +.DS_Store + +# Build artifacts +*.bin +*.o +*.a +*.out + +# Documentation (exclude READMEs) +*.md +!README.md +docs/ + +# Logs +*.log +logs/ + +# Data files (handled via volumes) +*.db +*.sqlite +*.sqlite3 +data/ +chroma/ + +# Other modules not needed for build +bot4cj/ +core/ +gateway/ +lib*/ +waf/ +captcha/ +smtp/ +static/ +tests/ +3d/ +BotOps/ +pymsg/tests/ diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..6933624 --- /dev/null +++ b/.env.example @@ -0,0 +1,11 @@ +# AI Review 服务配置 +# 复制此文件为 ai_review/.env 并填入你的 API Key + +# DeepSeek/OpenAI API 密钥 +OPENAI_API_KEY=your_api_key_here + +# API Base URL（默认使用 DeepSeek） +OPENAI_BASE_URL=https://api.deepseek.com/v1 + +# 使用的模型名称 +MODEL_NAME=deepseek-chat diff --git a/ai_review/.dockerignore b/ai_review/.dockerignore new file mode 100644 index 0000000..7b284e0 --- /dev/null +++ b/ai_review/.dockerignore @@ -0,0 +1,18 @@ +.env +.env/ +__pycache__/ +*.pyc +*.pyo +*.pyd +.Python +*.so +.git +.gitignore +*.md +!README.md +.envrc +.venv/ +venv/ +ENV/ +chroma/ +.DS_Store diff --git a/ai_review/Dockerfile b/ai_review/Dockerfile new file mode 100644 index 0000000..a31deb5 --- /dev/null +++ b/ai_review/Dockerfile @@ -0,0 +1,43 @@ +FROM python:3.11-slim + +LABEL maintainer="qingsong" +LABEL service="ai_review" +LABEL description="AI内容审核服务 - DeepSeek + RAG" + +WORKDIR /app + +ENV PYTHONDONTWRITEBYTECODE=1 \ + PYTHONUNBUFFERED=1 \ + PIP_NO_CACHE_DIR=1 \ + PIP_DISABLE_PIP_VERSION_CHECK=1 + +RUN apt-get update && apt-get install -y --no-install-recommends \ + && rm -rf /var/lib/apt/lists/* + +# 安装依赖（ai_review不依赖pymsg） +COPY requirements.txt /app/ +RUN pip install --no-cache-dir -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple + +# 复制应用代码 +COPY . /app/ + +# 创建ChromaDB持久化目录 +RUN mkdir -p /app/chroma +VOLUME ["/app/chroma"] + +# 创建非root用户 +RUN useradd -m -u 1000 appuser && chown -R appuser:appuser /app +USER appuser + +EXPOSE 8007 + +# 健康检查 +HEALTHCHECK --interval=30s --timeout=10s --start-period=30s --retries=3 \ + CMD python -c "import urllib.request; urllib.request.urlopen('http://localhost:8007/health', timeout=5)" || exit 1 + +# 默认环境变量 +ENV AI_REVIEW_HOST=0.0.0.0 \ + AI_REVIEW_PORT=8007 \ + CHROMA_PERSIST_DIR=/app/chroma + +CMD ["python", "ai_review.py"] diff --git a/backups/.gitkeep b/backups/.gitkeep new file mode 100644 index 0000000..bbb98fc --- /dev/null +++ b/backups/.gitkeep @@ -0,0 +1,2 @@ +# 数据库备份输出目录 +# 容器内备份会输出到此目录，可直接复制走 diff --git a/captcha/captcha_service.py b/captcha/captcha_service.py index 91e1236..c93331e 100644 --- a/captcha/captcha_service.py +++ b/captcha/captcha_service.py @@ -21,7 +21,6 @@ captcha — 基于 pymsg 协议（MsgServer）的图形验证码服务。 import base64 import io -import json import os import random import secrets @@ -34,7 +33,7 @@ from PIL import Image, ImageDraw, ImageFont # pymsg 导入 import sys as _sys _sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) -from pymsg import MsgServer # noqa: E402 +from pymsg import RpcServer # noqa: E402 # ── 配置 ──────────────────────────────────────────────────────────────── @@ -100,54 +99,21 @@ class CaptchaServer: def __init__(self, addr="127.0.0.1", port=8009, workers=4): self.addr = addr self.port = port - self._server = MsgServer(addr, port, workers=workers) - - # 路由表 - self._routes = { - "image": self._handle_image, - "verify": self._handle_verify, - "health": self._handle_health, - } + self._server = RpcServer(addr, port, workers=workers) + self._server.add_route("image", self._handle_image) + self._server.add_route("verify", self._handle_verify) + self._server.add_route("health", self._handle_health) # ── 启动 / 停止 ────────────────────────────────────────────────── def start(self) -> None: """启动服务（阻塞当前线程）。""" print(f"[captcha] 监听 tcp://{self.addr}:{self.port}") - self._server.start(self._dispatch) + self._server.start() def stop(self) -> None: """从另一线程停止服务。""" self._server.stop() - # ── 消息调度 ------------------------------------------------------- - def _dispatch(self, sock, addr, text): - """解析 JSON 中的 _route，分发给对应的 handler，并回复结果。""" - try: - req = json.loads(text) - except json.JSONDecodeError as e: - self._reply(sock, False, error=f"JSON 解析失败: {e}") - return - - route = req.pop("_route", "") - handler = self._routes.get(route) - if handler is None: - self._reply(sock, False, error=f"未知路由: {route}") - return - - try: - result = handler(req) - self._reply(sock, True, data=result) - except Exception as e: - self._reply(sock, False, error=str(e)) - - def _reply(self, sock, success, data=None, error=None): - resp = {"success": success} - if data is not None: - resp["data"] = data - if error is not None: - resp["error"] = error - self._server.reply(sock, json.dumps(resp, ensure_ascii=False)) - # ═══════════════════════════════════════════════════════════════════ # 路由处理 # ═══════════════════════════════════════════════════════════════════ diff --git a/db/.dockerignore b/db/.dockerignore new file mode 100644 index 0000000..ba73f43 --- /dev/null +++ b/db/.dockerignore @@ -0,0 +1,20 @@ +.env +.env/ +__pycache__/ +*.pyc +*.pyo +*.pyd +.Python +*.so +.git +.gitignore +*.md +!README.md +.envrc +.venv/ +venv/ +ENV/ +*.db +*.sqlite +*.sqlite3 +.DS_Store diff --git a/db/Dockerfile b/db/Dockerfile new file mode 100644 index 0000000..49472cb --- /dev/null +++ b/db/Dockerfile @@ -0,0 +1,53 @@ +FROM python:3.11-slim + +LABEL maintainer="qingsong" +LABEL service="db" +LABEL description="dbv2 - pymsg协议数据库服务" + +WORKDIR /app + +ENV PYTHONDONTWRITEBYTECODE=1 \ + PYTHONUNBUFFERED=1 \ + PIP_NO_CACHE_DIR=1 \ + PIP_DISABLE_PIP_VERSION_CHECK=1 + +RUN apt-get update && apt-get install -y --no-install-recommends \ + && rm -rf /var/lib/apt/lists/* + +# 安装pymsg共享库 +COPY pymsg/ /app/pymsg/ +RUN pip install /app/pymsg/ + +# 安装依赖 +COPY db/requirements.txt /app/ +RUN pip install --no-cache-dir -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple + +# 复制应用代码 +COPY db/ /app/db/ + +# 入口脚本 +RUN chmod +x /app/db/docker-entrypoint.sh + +# 创建数据持久化目录并设置权限 +RUN mkdir -p /app/data && chown -R 1000:1000 /app/data +VOLUME ["/app/data"] + +# 创建非root用户 +RUN useradd -m -u 1000 appuser && chown -R appuser:appuser /app +USER appuser + +WORKDIR /app/db + +EXPOSE 8005 + +# 健康检查：尝试连接TCP端口 +HEALTHCHECK --interval=30s --timeout=5s --start-period=10s --retries=3 \ + CMD python -c "import socket; s=socket.socket(); s.settimeout(2); s.connect(('localhost', 8005)); s.close()" || exit 1 + +# 默认使用SQLite，数据持久化到/app/data/database.db +ENV DATABASE_URL=sqlite:////app/data/database.db \ + LISTEN_ADDR=0.0.0.0 \ + LISTEN_PORT=8005 \ + WORKERS=4 + +ENTRYPOINT ["/app/db/docker-entrypoint.sh"] diff --git a/db/db_tools.py b/db/db_tools.py new file mode 100644 index 0000000..75671f0 --- /dev/null +++ b/db/db_tools.py @@ -0,0 +1,304 @@ +#!/usr/bin/env python3 +""" +db_tools — SQLite 数据库备份、导出、导入工具 + +用法: + python db_tools.py backup # 安全备份数据库（使用 sqlite3 backup API） + python db_tools.py export # 导出所有表数据为 JSON + python db_tools.py import # 从 JSON 导入数据 + python db_tools.py vacuum # 执行 VACUUM 压缩数据库 + python db_tools.py wal-checkpoint # 强制 WAL 检查点（确保数据写入主库） + python db_tools.py info # 显示数据库信息和表统计 +""" + +import json +import os +import shutil +import sys +import sqlite3 +from datetime import datetime, timezone + +DATABASE_URL = os.environ.get("DATABASE_URL", "sqlite:///database.db") + +# 从 sqlite:///path 提取文件路径 +def _get_db_path() -> str: + if DATABASE_URL.startswith("sqlite:////"): + return DATABASE_URL[10:] # 绝对路径 sqlite:////app/data/xxx.db + elif DATABASE_URL.startswith("sqlite:///"): + return DATABASE_URL[9:] # 相对路径 + else: + return "database.db" + + +def _connect() -> sqlite3.Connection: + """创建数据库连接（配置 PRAGMA）。""" + conn = sqlite3.connect(_get_db_path()) + conn.execute("PRAGMA journal_mode=WAL") + conn.execute("PRAGMA synchronous=NORMAL") + conn.row_factory = sqlite3.Row + return conn + + +def backup(output_path: str): + """ + 使用 SQLite 官方 backup API 进行在线热备份（不需要停机）。 + 这是最安全的备份方式，支持读写并发。 + """ + src = _get_db_path() + print(f"[backup] 正在备份数据库: {src} -> {output_path}") + + if not os.path.exists(src): + print(f"[backup] 错误: 数据库文件不存在: {src}", file=sys.stderr) + sys.exit(1) + + start = datetime.now(timezone.utc) + + # 使用 sqlite3 backup API（支持在线备份） + src_conn = sqlite3.connect(src) + dst_conn = sqlite3.connect(output_path) + try: + src_conn.backup(dst_conn) + dst_conn.commit() + finally: + dst_conn.close() + src_conn.close() + + size_mb = os.path.getsize(output_path) / (1024 * 1024) + elapsed = (datetime.now(timezone.utc) - start).total_seconds() + print(f"[backup] 备份完成: {output_path} ({size_mb:.2f} MB, 耗时 {elapsed:.2f}s)") + + +def export_json(output_path: str): + """导出所有表数据为 JSON 格式（便于跨平台传输/迁移）。""" + print(f"[export] 正在导出数据到: {output_path}") + conn = _connect() + + result = { + "_meta": { + "exported_at": datetime.now(timezone.utc).isoformat(), + "db_path": _get_db_path(), + "sqlite_version": sqlite3.sqlite_version, + }, + "tables": {} + } + + try: + # 获取所有用户表 + tables = conn.execute( + "SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%'" + ).fetchall() + + for (table_name,) in tables: + # 获取列信息 + columns = [desc[1] for desc in conn.execute(f"PRAGMA table_info({table_name})").fetchall()] + + # 获取行数 + count = conn.execute(f"SELECT COUNT(*) FROM [{table_name}]").fetchone()[0] + + # 获取所有数据 + rows = conn.execute(f"SELECT * FROM [{table_name}]").fetchall() + data = [dict(row) for row in rows] + + result["tables"][table_name] = { + "columns": columns, + "row_count": count, + "data": data, + } + print(f"[export] 表 {table_name}: {count} 行") + + # 同时导出 schema + schema = [] + for (table_name,) in tables: + ddl = conn.execute( + "SELECT sql FROM sqlite_master WHERE type='table' AND name=?", + (table_name,) + ).fetchone()[0] + if ddl: + schema.append(ddl + ";") + + result["_meta"]["schema"] = schema + + finally: + conn.close() + + with open(output_path, "w", encoding="utf-8") as f: + json.dump(result, f, ensure_ascii=False, indent=2, default=str) + + size_mb = os.path.getsize(output_path) / (1024 * 1024) + print(f"[export] 导出完成: {output_path} ({size_mb:.2f} MB)") + + +def import_json(input_path: str): + """从 JSON 文件导入数据（注意：会清空现有数据！）。""" + print(f"[import] 正在从 {input_path} 导入数据") + + if not os.path.exists(input_path): + print(f"[import] 错误: 文件不存在: {input_path}", file=sys.stderr) + sys.exit(1) + + with open(input_path, "r", encoding="utf-8") as f: + data = json.load(f) + + conn = _connect() + + try: + # 是否清空现有表（需要确认） + tables = data.get("tables", {}) + if tables: + print(f"[import] 警告: 即将导入 {len(tables)} 个表，现有数据将被覆盖!") + # 为安全起见，不自动清空，由用户先备份 + + # 按外键依赖顺序导入（先主表后子表，这里简单按名字排序） + for table_name in sorted(tables.keys()): + table_data = tables[table_name] + rows = table_data.get("data", []) + columns = table_data.get("columns", []) + + if not rows or not columns: + print(f"[import] 跳表 {table_name}: 无数据或列定义") + continue + + # 先清空表 + conn.execute(f"DELETE FROM [{table_name}]") + + # 批量插入 + placeholders = ", ".join(["?"] * len(columns)) + col_names = ", ".join(f"[{c}]" for c in columns) + insert_sql = f"INSERT INTO [{table_name}] ({col_names}) VALUES ({placeholders})" + + conn.executemany(insert_sql, [tuple(row.get(col) for col in columns) for row in rows]) + print(f"[import] 表 {table_name}: 导入 {len(rows)} 行") + + conn.commit() + print("[import] 导入完成") + else: + print("[import] 没有可导入的表数据") + finally: + conn.close() + + +def vacuum(): + """执行 VACUUM 压缩数据库，回收空间、重建索引。""" + print("[vacuum] 正在执行 VACUUM...") + db_path = _get_db_path() + before_size = os.path.getsize(db_path) / (1024 * 1024) + + conn = _connect() + try: + # WAL 检查点先将所有数据写入主库 + conn.execute("PRAGMA wal_checkpoint(TRUNCATE)") + conn.execute("VACUUM") + conn.execute("ANALYZE") + finally: + conn.close() + + after_size = os.path.getsize(db_path) / (1024 * 1024) + saved = before_size - after_size + print(f"[vacuum] 完成: {before_size:.2f} MB -> {after_size:.2f} MB (节省 {saved:.2f} MB)") + + +def wal_checkpoint(): + """强制 WAL 检查点，确保所有写入都持久化到主数据库文件。""" + print("[wal-checkpoint] 执行 WAL 检查点...") + conn = _connect() + try: + result = conn.execute("PRAGMA wal_checkpoint(TRUNCATE)").fetchone() + # busy, log, checkpointed + print(f"[wal-checkpoint] 结果: busy={result[0]}, log_pages={result[1]}, checkpointed={result[2]}") + finally: + conn.close() + print("[wal-checkpoint] 完成") + + +def show_info(): + """显示数据库信息和表统计。""" + db_path = _get_db_path() + print("=" * 50) + print("SQLite 数据库信息") + print("=" * 50) + print(f"文件路径: {os.path.abspath(db_path)}") + + if os.path.exists(db_path): + size_mb = os.path.getsize(db_path) / (1024 * 1024) + mtime = datetime.fromtimestamp(os.path.getmtime(db_path), tz=timezone.utc) + print(f"文件大小: {size_mb:.2f} MB") + print(f"最后修改: {mtime.isoformat()}") + else: + print("文件大小: (不存在)") + + # WAL 文件 + wal_path = db_path + "-wal" + shm_path = db_path + "-shm" + if os.path.exists(wal_path): + wal_size = os.path.getsize(wal_path) / 1024 + print(f"WAL 文件: {wal_size:.2f} KB") + if os.path.exists(shm_path): + shm_size = os.path.getsize(shm_path) / 1024 + print(f"SHM 文件: {shm_size:.2f} KB") + + conn = _connect() + try: + print(f"SQLite 版本: {sqlite3.sqlite_version}") + + # PRAGMA 信息 + journal_mode = conn.execute("PRAGMA journal_mode").fetchone()[0] + synchronous = conn.execute("PRAGMA synchronous").fetchone()[0] + page_size = conn.execute("PRAGMA page_size").fetchone()[0] + page_count = conn.execute("PRAGMA page_count").fetchone()[0] + free_pages = conn.execute("PRAGMA freelist_count").fetchone()[0] + + print(f"Journal 模式: {journal_mode}") + print(f"同步级别: {synchronous}") + print(f"页大小: {page_size} bytes") + + # 表统计 + tables = conn.execute( + "SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%' ORDER BY name" + ).fetchall() + + print(f"\n表统计 ({len(tables)} 个表):") + print("-" * 40) + total_rows = 0 + for (table_name,) in tables: + count = conn.execute(f"SELECT COUNT(*) FROM [{table_name}]").fetchone()[0] + total_rows += count + print(f" {table_name:20s} {count:>8d} 行") + print("-" * 40) + print(f" {'总计':20s} {total_rows:>8d} 行") + + finally: + conn.close() + + +def main(): + if len(sys.argv) < 2: + print(__doc__) + sys.exit(1) + + cmd = sys.argv[1] + + if cmd == "backup": + output = sys.argv[2] if len(sys.argv) > 2 else f"backup_{datetime.now().strftime('%Y%m%d_%H%M%S')}.db" + backup(output) + elif cmd == "export": + output = sys.argv[2] if len(sys.argv) > 2 else f"export_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json" + export_json(output) + elif cmd == "import": + if len(sys.argv) < 3: + print("用法: python db_tools.py import ", file=sys.stderr) + sys.exit(1) + import_json(sys.argv[2]) + elif cmd == "vacuum": + vacuum() + elif cmd == "wal-checkpoint": + wal_checkpoint() + elif cmd == "info": + show_info() + else: + print(f"未知命令: {cmd}", file=sys.stderr) + print(__doc__) + sys.exit(1) + + +if __name__ == "__main__": + main() diff --git a/db/dbv2.py b/db/dbv2.py index 23d25cd..771b206 100644 --- a/db/dbv2.py +++ b/db/dbv2.py @@ -37,6 +37,7 @@ from datetime import datetime, date, time as dt_time, timezone, timedelta from sqlalchemy import ( create_engine, MetaData, Table, select, insert, update, delete, func, and_, + event, ) from sqlalchemy import types as sa_types from sqlalchemy.exc import SQLAlchemyError @@ -44,12 +45,32 @@ from sqlalchemy.exc import SQLAlchemyError # pymsg 导入 import sys as _sys _sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) -from pymsg import MsgServer # noqa: E402 +from pymsg import RpcServer # noqa: E402 # ── 数据库配置 ────────────────────────────────────────────────────────── DATABASE_URL = os.environ.get("DATABASE_URL", "sqlite:///database.db") +# ── SQLite 连接配置（针对 Docker/多线程环境优化）────────────────────── + +def _configure_sqlite(dbapi_connection, connection_record): + """为每个 SQLite 连接设置 PRAGMA，优化并发与持久化性能。 + + - WAL 模式：读写不互斥，并发性能大幅提升，Docker 卷挂载下更稳定 + - NORMAL 同步级别：在断电时可能丢失最近提交的事务（但不会损坏数据库） + 配合 WAL checkpoint 足够安全，性能显著提升 + - foreign_keys=ON：启用外键约束 + - busy_timeout：锁等待超时 5 秒，避免 SQLITE_BUSY 错误 + - mmap_size：使用内存映射 I/O 提升读取性能 + """ + cursor = dbapi_connection.cursor() + cursor.execute("PRAGMA journal_mode=WAL") + cursor.execute("PRAGMA synchronous=NORMAL") + cursor.execute("PRAGMA foreign_keys=ON") + cursor.execute("PRAGMA busy_timeout=5000") + cursor.execute("PRAGMA mmap_size=268435456") # 256MB + cursor.close() + # ── 日志模块常量 ──────────────────────────────────────────────────────── MAX_MESSAGE_LEN = 8192 # message 上限 8KB @@ -80,9 +101,20 @@ class DBServer: def __init__(self, addr="127.0.0.1", port=8005, workers=4): self.addr = addr self.port = port - self.engine = create_engine(DATABASE_URL) + self.engine = create_engine( + DATABASE_URL, + connect_args={"check_same_thread": False} if DATABASE_URL.startswith("sqlite") else {}, + ) + # 为 SQLite 连接应用 PRAGMA 优化 + if DATABASE_URL.startswith("sqlite"): + event.listen(self.engine, "connect", _configure_sqlite) self.metadata = MetaData() - self._server = MsgServer(addr, port, workers=workers) + self._server = RpcServer( + addr, port, workers=workers, + json_default=_json_default, + on_error=lambda route, e: traceback.print_exc(), + ) + self._server.add_route("execute", self._handle_execute) # 日志表缓存（避免重复反射） self._logs_table_cache = {} @@ -98,54 +130,18 @@ class DBServer: # 抛 InvalidRequestError），反射操作需串行化 self._reflect_lock = threading.Lock() - # 路由表 - self._routes = { - "execute": self._handle_execute, - } - # ── 启动 / 停止 ────────────────────────────────────────────────── def start(self) -> None: """启动服务（阻塞当前线程）。""" print(f"[dbv2] 监听 tcp://{self.addr}:{self.port}") - self._ensure_logs_table() # 确保 Logs 表已建好（外部 MsgClient 写日志前） + self._ensure_logs_table() # 确保 Logs 表已建好（外部写日志前） self._start_cleanup_thread() - self._server.start(self._dispatch) + self._server.start() def stop(self) -> None: """从另一线程停止服务。""" self._server.stop() - # ── 消息调度 ------------------------------------------------------- - def _dispatch(self, sock, addr, text): - """解析 JSON 中的 _route，分发给对应的 handler，并回复结果。""" - try: - req = json.loads(text) - except json.JSONDecodeError as e: - self._reply(sock, False, error=f"JSON 解析失败: {e}") - return - - route = req.pop("_route", "") - handler = self._routes.get(route) - if handler is None: - self._reply(sock, False, error=f"未知路由: {route}") - return - - try: - result = handler(req, sock=sock) - self._reply(sock, True, data=result) - except Exception as e: - traceback.print_exc() - self._reply(sock, False, error=str(e)) - - def _reply(self, sock, success, data=None, error=None): - resp = {"success": success} - if data is not None: - resp["data"] = data - if error is not None: - resp["error"] = error - self._server.reply(sock, json.dumps(resp, ensure_ascii=False, - default=_json_default)) - # ── 表反射 --------------------------------------------------------- def _get_table(self, table_name): # double-check：_tables_cache 只在反射完整完成后写入（dict 读写 GIL @@ -245,7 +241,7 @@ class DBServer: # 路由处理 — execute # ═══════════════════════════════════════════════════════════════════ - def _handle_execute(self, data, **kw): + def _handle_execute(self, data): action = data.get("action", "select") table_name = data.get("table") if not table_name: @@ -460,9 +456,12 @@ class DBServer: if __name__ == "__main__": import argparse + default_addr = os.environ.get("LISTEN_ADDR", "0.0.0.0") + default_port = int(os.environ.get("LISTEN_PORT", "8005")) + parser = argparse.ArgumentParser(description="dbv2 — pymsg 数据库服务") - parser.add_argument("--addr", default="127.0.0.1", help="监听地址") - parser.add_argument("--port", type=int, default=8005, help="监听端口") + parser.add_argument("--addr", default=default_addr, help="监听地址") + parser.add_argument("--port", type=int, default=default_port, help="监听端口") parser.add_argument("--workers", type=int, default=4, help="回调并发线程数（0=同步串行模式）") args = parser.parse_args() diff --git a/db/docker-entrypoint.sh b/db/docker-entrypoint.sh new file mode 100644 index 0000000..4eb7f19 --- /dev/null +++ b/db/docker-entrypoint.sh @@ -0,0 +1,58 @@ +#!/bin/bash +set -e + +cd "$(dirname "$0")" + +DATA_DIR="/app/data" +DB_PATH="${DATABASE_URL#sqlite:///}" +if [ "$DB_PATH" = "$DATABASE_URL" ]; then + # URL 不是 sqlite:/// 格式，使用默认路径 + DB_PATH="$DATA_DIR/database.db" +fi + +echo "数据目录: $DATA_DIR" +echo "数据库路径: $DB_PATH" + +# 确保数据目录存在并可写 +mkdir -p "$(dirname "$DB_PATH")" + +# 初始化数据库表（如果不存在） +if [[ "$DATABASE_URL" == sqlite:* ]]; then + echo "初始化数据库 schema..." + python -c " +import os +from sqlalchemy import create_engine, text + +db_url = os.environ.get('DATABASE_URL', 'sqlite:////app/data/database.db') +engine = create_engine( + db_url, + connect_args={'check_same_thread': False} if db_url.startswith('sqlite') else {}, +) +# 应用 SQLite PRAGMA 优化 +from sqlalchemy import event +@event.listens_for(engine, 'connect') +def _configure_sqlite(dbapi_conn, _): + cur = dbapi_conn.cursor() + cur.execute('PRAGMA journal_mode=WAL') + cur.execute('PRAGMA synchronous=NORMAL') + cur.execute('PRAGMA foreign_keys=ON') + cur.execute('PRAGMA busy_timeout=5000') + cur.close() + +with open('create_sqlite3.sql', 'r') as f: + schema_sql = f.read() +with engine.connect() as conn: + for stmt in schema_sql.split(';'): + stmt = stmt.strip() + if stmt: + try: + conn.execute(text(stmt)) + except Exception as e: + pass + conn.commit() +print('Schema 初始化完成.') +" +fi + +echo "启动 dbv2 服务 ${LISTEN_ADDR:-0.0.0.0}:${LISTEN_PORT:-8005}..." +exec python dbv2.py --addr "${LISTEN_ADDR:-0.0.0.0}" --port "${LISTEN_PORT:-8005}" --workers "${WORKERS:-4}" diff --git a/dnfs/.dockerignore b/dnfs/.dockerignore new file mode 100644 index 0000000..38c4359 --- /dev/null +++ b/dnfs/.dockerignore @@ -0,0 +1,18 @@ +.env +.env/ +__pycache__/ +*.pyc +*.pyo +*.pyd +.Python +*.so +.git +.gitignore +*.md +!README.md +.envrc +.venv/ +venv/ +ENV/ +*.bin +.DS_Store diff --git a/dnfs/Dockerfile b/dnfs/Dockerfile new file mode 100644 index 0000000..7ef7505 --- /dev/null +++ b/dnfs/Dockerfile @@ -0,0 +1,48 @@ +FROM python:3.11-slim + +LABEL maintainer="qingsong" +LABEL service="dnfs" +LABEL description="Dendrachy Network File-System - 树状网络文件系统" + +# 设置工作目录 +WORKDIR /app + +# 设置环境变量 +ENV PYTHONDONTWRITEBYTECODE=1 \ + PYTHONUNBUFFERED=1 \ + PIP_NO_CACHE_DIR=1 \ + PIP_DISABLE_PIP_VERSION_CHECK=1 + +# 安装系统依赖（如果有需要） +RUN apt-get update && apt-get install -y --no-install-recommends \ + && rm -rf /var/lib/apt/lists/* + +# 先复制pymsg共享库并安装 +COPY pymsg/ /app/pymsg/ +RUN pip install /app/pymsg/ + +# 复制requirements并安装依赖 +COPY dnfs/requirements.txt /app/ +RUN pip install --no-cache-dir -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple + +# 移除flask_sqlalchemy遗留依赖（已不使用） +RUN pip uninstall -y flask_sqlalchemy 2>/dev/null || true + +# 复制应用代码 +COPY dnfs/ /app/dnfs/ + +# 创建非root用户运行（安全最佳实践） +RUN useradd -m -u 1000 appuser && chown -R appuser:appuser /app +USER appuser + +WORKDIR /app/dnfs + +# 暴露端口 +EXPOSE 8006 + +# 健康检查 +HEALTHCHECK --interval=30s --timeout=5s --start-period=10s --retries=3 \ + CMD python -c "import urllib.request; urllib.request.urlopen('http://localhost:8006/get_notes', data=b'{\"pid\":0}', timeout=3)" || exit 1 + +# 启动服务 +CMD ["python", "dnfs.py"] diff --git a/dnfs/dnfs.py b/dnfs/dnfs.py index c06f7c4..090d00d 100755 --- a/dnfs/dnfs.py +++ b/dnfs/dnfs.py @@ -32,30 +32,36 @@ from log_client import logs # pymsg 导入 import sys as _sys _sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) -from pymsg import MsgClient # noqa: E402 +from pymsg import RpcClient, RpcError # noqa: E402 -_DBV2_ADDR = '127.0.0.1' -_DBV2_PORT = 8005 +_DBV2_ADDR = os.environ.get('DB_HOST', '127.0.0.1') +_DBV2_PORT = int(os.environ.get('DB_PORT', '8005')) # 管理员判定已上移至 gateway：登录时命中 [Admin] users 白名单即在 Token payload # 写入 is_admin=true，本服务通过 info.get('is_admin') 读取，不再本地维护名单。 # 旧 Token 无此字段，按非管理员处理（安全方向）。 +_rpc = RpcClient(_DBV2_ADDR, _DBV2_PORT, "dnfs") + + def _call_dbv2(payload: dict, timeout: int = 10) -> dict: - """通过 pymsg 短连接调用 dbv2 数据库服务。 - 自动注入 ``_route: "execute"``。 + """通过 pymsg RPC 调用 dbv2 数据库服务（自动注入 ``_route: "execute"``）。 + + 返回完整响应 dict（``{"success": ..., "data"/"error": ...}``）， + 与历史行为一致，调用方无需改动。服务端业务错误返回 success=False； + 网络层故障抛 ConnectionError。 """ - payload = {**payload, "_route": "execute"} try: - client = MsgClient(_DBV2_ADDR, _DBV2_PORT, "dnfs") - resp_text = client.call(json.dumps(payload), timeout=timeout) - return json.loads(resp_text) + data = _rpc.call("execute", payload, timeout=timeout) + return {"success": True, "data": data} + except RpcError as e: + return {"success": False, "error": str(e)} except Exception as e: raise ConnectionError( f"dbv2 调用失败 ({_DBV2_ADDR}:{_DBV2_PORT}): {e}" ) from None -REVIEW_SERVER_URL = 'http://127.0.0.1:8007/review' +REVIEW_SERVER_URL = os.environ.get('REVIEW_URL', 'http://127.0.0.1:8007/review') def review_content(content: str, timeout: int = 10) -> Dict[str, Any]: """ 调用 AI 内容审核服务的 /review 接口，对给定文本进行审核。 @@ -545,10 +551,9 @@ def get_note(): # """返回应用主页""" # return render_template('index.html') -ADDR='127.0.0.1' -PORT=8006 +ADDR = os.environ.get('LISTEN_ADDR', '0.0.0.0') +PORT = int(os.environ.get('LISTEN_PORT', '8006')) if __name__ == "__main__": logs.info('dnfs', f'服务启动，监听 http://{ADDR}:{PORT}') print(f'Listen at http://{ADDR}:{PORT}') - #app.run(host="127.0.0.1", port=8006) - serve(app, host='127.0.0.1', port=8006) + serve(app, host=ADDR, port=PORT) diff --git a/dnfs/log_client.py b/dnfs/log_client.py index f3188b6..60a426d 100644 --- a/dnfs/log_client.py +++ b/dnfs/log_client.py @@ -1,8 +1,7 @@ -""" -统一日志客户端 — 各服务共用。 +"""统一日志客户端 — 各服务共用（dnfs / smtp / static 三份副本内容一致，改动请同步）。 设计要点： - - 通过 pymsg TCP 调用 dbv2 execute 路由，逐条 INSERT 到 Logs 表 + - 通过 pymsg RPC 调用 dbv2 execute 路由，逐条 INSERT 到 Logs 表 - 异步非阻塞：内存队列 + 后台线程批量推送，不拖慢业务 - fail-safe：db 服务不可用时静默丢弃，绝不影响业务 - 缓冲批量：队列满 50 条或超时 2 秒时自动 flush @@ -26,12 +25,12 @@ import atexit from datetime import datetime, timezone _sys.path.insert(0, _os.path.join(_os.path.dirname(__file__), "..")) -from pymsg import MsgClient # noqa: E402 +from pymsg import RpcClient, RpcError # noqa: E402 # ── 配置 ────────────────────────────────────────────────────────────── -_DBV2_ADDR = '127.0.0.1' -_DBV2_PORT = 8005 +_DBV2_ADDR = _os.environ.get('DB_HOST', '127.0.0.1') +_DBV2_PORT = int(_os.environ.get('DB_PORT', '8005')) _BATCH_SIZE = 50 # 队列满 50 条时 flush _FLUSH_INTERVAL = 2.0 # 超时 2 秒强制 flush @@ -41,8 +40,10 @@ _TIMEOUT = 5 # socket 超时（秒） class LogClient: """异步批量日志推送客户端。""" - def __init__(self, batch_size=_BATCH_SIZE, + def __init__(self, service_name="log", + batch_size=_BATCH_SIZE, flush_interval=_FLUSH_INTERVAL): + self._service_name = service_name self._batch_size = batch_size self._flush_interval = flush_interval self._queue = queue.Queue() @@ -133,35 +134,33 @@ class LogClient: self._send(batch) def _send(self, batch): - """通过 pymsg 使用 execute 路由逐条 INSERT 到 dbv2 Logs 表。""" + """通过 pymsg RPC（execute 路由）逐条 INSERT 到 dbv2 Logs 表。""" now = datetime.now(timezone.utc).isoformat() - try: - client = MsgClient(_DBV2_ADDR, _DBV2_PORT, "dnfs-log") - for entry in batch: - extra_val = entry.get("extra") - payload = { - "_route": "execute", - "action": "insert", - "table": "Logs", - "values": { - "ts": now, - "service": entry.get("service"), - "level": entry.get("level"), - "message": entry.get("message"), - "trace_id": entry.get("trace_id"), - "extra": json.dumps(extra_val, ensure_ascii=False) - if extra_val is not None else None, - }, - } - resp = client.call(json.dumps(payload), timeout=_TIMEOUT) - resp_obj = json.loads(resp) - if not resp_obj.get("success"): - import sys - print(f"[log_client] INSERT 失败: {resp_obj.get('error')}", - file=sys.stderr) - except Exception as e: - import sys - print(f"[log_client] 发送失败: {e}", file=sys.stderr) + client = RpcClient(_DBV2_ADDR, _DBV2_PORT, self._service_name + "-log") + for entry in batch: + extra_val = entry.get("extra") + payload = { + "action": "insert", + "table": "Logs", + "values": { + "ts": now, + "service": entry.get("service"), + "level": entry.get("level"), + "message": entry.get("message"), + "trace_id": entry.get("trace_id"), + "extra": json.dumps(extra_val, ensure_ascii=False) + if extra_val is not None else None, + }, + } + try: + client.call("execute", payload, timeout=_TIMEOUT) + except RpcError as e: + # 业务错误（如表结构不符）：打印后继续下一条 + print(f"[log_client] INSERT 失败: {e}", file=_sys.stderr) + except Exception as e: + # 网络错误：db 不可用，放弃本批剩余（fail-safe） + print(f"[log_client] 发送失败: {e}", file=_sys.stderr) + break def _shutdown(self): """进程退出时 flush 剩余日志。""" diff --git a/docker-compose.yml b/docker-compose.yml new file mode 100644 index 0000000..b1be30d --- /dev/null +++ b/docker-compose.yml @@ -0,0 +1,109 @@ +version: '3.8' + +# ── 持久化说明 ────────────────────────────────────────────────────── +# 默认使用 Docker named volume（适合生产，性能好）。 +# 开发调试或需要直接访问数据库文件时，可改为 bind mount： +# volumes: +# - ./data/db:/app/data # SQLite 数据库文件 +# - ./backups:/app/backups # 备份输出目录 +# +# 注意：bind mount 在 macOS/Windows 上文件系统性能比 named volume 差， +# 但方便直接复制/查看数据库文件。 + +services: + # ── 数据库服务 (pymsg TCP协议) ────────────────────────────────── + db: + build: + context: . + dockerfile: db/Dockerfile + container_name: micro-db + restart: unless-stopped + ports: + - "127.0.0.1:8005:8005" + volumes: + # 方式1: Docker named volume（推荐生产） + - db_data:/app/data + # 方式2: Bind mount（开发/需要直接访问文件时取消注释） + # - ./data/db:/app/data + # 备份输出目录（bind mount 方便取备份文件） + - ./backups:/app/backups + environment: + - LISTEN_ADDR=0.0.0.0 + - LISTEN_PORT=8005 + - DATABASE_URL=sqlite:////app/data/database.db + - WORKERS=4 + healthcheck: + test: ["CMD", "python", "-c", "import socket; s=socket.socket(); s.settimeout(2); s.connect(('localhost', 8005)); s.close()"] + interval: 15s + timeout: 5s + retries: 3 + start_period: 5s + networks: + - micro_net + + # ── AI内容审核服务 ────────────────────────────────────────────── + ai_review: + build: + context: ./ai_review + dockerfile: Dockerfile + container_name: micro-ai-review + restart: unless-stopped + ports: + - "127.0.0.1:8007:8007" + volumes: + - chroma_data:/app/chroma + # - ./data/chroma:/app/chroma # bind mount 选项 + environment: + - AI_REVIEW_HOST=0.0.0.0 + - AI_REVIEW_PORT=8007 + - CHROMA_PERSIST_DIR=/app/chroma + env_file: + - ./ai_review/.env + healthcheck: + test: ["CMD", "python", "-c", "import urllib.request; urllib.request.urlopen('http://localhost:8007/health', timeout=5)"] + interval: 30s + timeout: 10s + retries: 3 + start_period: 30s + networks: + - micro_net + + # ── DNFS 树状文件系统服务 ────────────────────────────────────── + dnfs: + build: + context: . + dockerfile: dnfs/Dockerfile + container_name: micro-dnfs + restart: unless-stopped + ports: + - "127.0.0.1:8006:8006" + environment: + - LISTEN_ADDR=0.0.0.0 + - LISTEN_PORT=8006 + - DB_HOST=db + - DB_PORT=8005 + - REVIEW_URL=http://ai_review:8007/review + depends_on: + db: + condition: service_healthy + ai_review: + condition: service_healthy + healthcheck: + test: ["CMD", "python", "-c", "import urllib.request; urllib.request.urlopen('http://localhost:8006/get_notes', data=b'{\"pid\":0}', timeout=3)"] + interval: 30s + timeout: 5s + retries: 3 + start_period: 10s + networks: + - micro_net + +networks: + micro_net: + driver: bridge + name: micro-network + +volumes: + db_data: + name: micro-db-data + chroma_data: + name: micro-chroma-data diff --git a/docker.sh b/docker.sh new file mode 100755 index 0000000..281629a --- /dev/null +++ b/docker.sh @@ -0,0 +1,386 @@ +#!/bin/bash +# +# 微服务容器化管理脚本 +# 用法: ./docker.sh [command] +# + +set -e + +cd "$(dirname "$0")" + +# 颜色输出 +RED='\033[0;31m' +GREEN='\033[0;32m' +YELLOW='\033[1;33m' +NC='\033[0m' # No Color + +info() { echo -e "${GREEN}[INFO]${NC} $1"; } +warn() { echo -e "${YELLOW}[WARN]${NC} $1"; } +error() { echo -e "${RED}[ERROR]${NC} $1"; exit 1; } + +check_docker() { + if ! command -v docker &> /dev/null; then + error "Docker 未安装，请先安装 Docker Desktop" + fi + if ! command -v docker-compose &> /dev/null && ! docker compose version &> /dev/null; then + error "Docker Compose 未安装" + fi +} + +# 使用 docker compose 或 docker-compose（兼容旧版本） +docker_compose() { + if docker compose version &> /dev/null; then + docker compose "$@" + else + docker-compose "$@" + fi +} + +cmd_build() { + info "构建所有服务镜像..." + docker_compose build --progress=plain "$@" + info "构建完成" +} + +cmd_up() { + info "启动服务（后台运行）..." + docker_compose up -d "$@" + info "服务已启动" + echo "" + cmd_ps +} + +cmd_down() { + info "停止并移除容器..." + docker_compose down "$@" + info "服务已停止" +} + +cmd_start() { + info "启动已存在的容器..." + docker_compose start "$@" +} + +cmd_stop() { + info "停止运行中的容器..." + docker_compose stop "$@" +} + +cmd_restart() { + info "重启服务..." + docker_compose restart "$@" +} + +cmd_ps() { + info "服务状态:" + docker_compose ps +} + +cmd_logs() { + if [ -z "$1" ]; then + docker_compose logs -f --tail=100 + else + docker_compose logs -f --tail=100 "$@" + fi +} + +cmd_shell() { + local service=${1:-dnfs} + info "进入 $service 容器 shell..." + docker_compose exec "$service" /bin/bash || docker_compose exec "$service" /bin/sh +} + +cmd_rebuild() { + info "重新构建并启动服务..." + docker_compose build --no-cache "$@" + docker_compose up -d "$@" +} + +cmd_clean() { + warn "此操作将停止所有容器并删除数据卷（数据库和ChromaDB数据将丢失！）" + read -p "确认继续? [y/N] " -n 1 -r + echo + if [[ $REPLY =~ ^[Yy]$ ]]; then + docker_compose down -v + info "已清理所有容器和数据卷" + else + info "已取消" + fi +} + +cmd_pull() { + info "拉取基础镜像..." + docker pull python:3.11-slim +} + +cmd_health() { + info "检查服务健康状态..." + local services=("db:8005" "ai_review:8007" "dnfs:8006") + for svc in "${services[@]}"; do + IFS=':' read -r name port <<< "$svc" + if nc -z 127.0.0.1 "$port" 2>/dev/null; then + echo -e " ${GREEN}✓${NC} $name :$port 健康" + else + echo -e " ${RED}✗${NC} $name :$port 不可达" + fi + done +} + +# ── 数据库管理命令 ────────────────────────────────────────────── + +DB_CONTAINER="micro-db" + +_check_db_running() { + if ! docker ps --format '{{.Names}}' | grep -q "^${DB_CONTAINER}$"; then + error "db 容器未运行，请先启动服务: ./docker.sh up" + fi +} + +cmd_db_info() { + _check_db_running + info "数据库信息:" + docker_compose exec db python db_tools.py info +} + +cmd_db_backup() { + _check_db_running + local backup_file="${1:-}" + if [ -z "$backup_file" ]; then + backup_file="backup_$(date +%Y%m%d_%H%M%S).db" + fi + info "执行数据库备份..." + # 先做 WAL checkpoint 确保数据完整 + docker_compose exec db python db_tools.py wal-checkpoint + # 备份到容器内的 /app/backups 目录（映射到宿主机 ./backups） + docker_compose exec db python db_tools.py backup "/app/backups/$backup_file" + echo "" + info "备份文件: ./backups/$backup_file" + ls -lh "./backups/$backup_file" 2>/dev/null || info "(文件已写入容器 /app/backups/) " +} + +cmd_db_export() { + _check_db_running + local export_file="${1:-}" + if [ -z "$export_file" ]; then + export_file="export_$(date +%Y%m%d_%H%M%S).json" + fi + info "导出数据库为 JSON..." + docker_compose exec db python db_tools.py export "/app/backups/$export_file" + echo "" + info "导出文件: ./backups/$export_file" +} + +cmd_db_import() { + local import_file="$1" + if [ -z "$import_file" ]; then + error "用法: ./docker.sh db-import " + fi + if [ ! -f "./backups/$import_file" ] && [ ! -f "$import_file" ]; then + # 检查是否在backups目录 + error "导入文件不存在: $import_file (请放在 ./backups/ 目录下或提供绝对路径)" + fi + _check_db_running + + local container_path="/app/backups/$(basename "$import_file")" + # 如果是宿主机路径且不在backups下，先复制到backups + if [ -f "$import_file" ] && [ ! -f "./backups/$(basename "$import_file")" ]; then + warn "复制文件到 backups 目录..." + cp "$import_file" "./backups/" + fi + + warn "即将导入数据，现有数据将被覆盖！" + read -p "确认继续? [y/N] " -n 1 -r + echo + if [[ $REPLY =~ ^[Yy]$ ]]; then + # 先备份当前数据 + cmd_db_backup "before_import_$(date +%Y%m%d_%H%M%S).db" + info "开始导入..." + docker_compose exec db python db_tools.py import "$container_path" + else + info "已取消" + fi +} + +cmd_db_vacuum() { + _check_db_running + info "执行 VACUUM 压缩数据库..." + docker_compose exec db python db_tools.py vacuum +} + +cmd_db_restore() { + local backup_file="$1" + if [ -z "$backup_file" ]; then + error "用法: ./docker.sh db-restore <备份文件路径> (从 ./backups/ 目录)" + error "示例: ./docker.sh db-restore backup_20260730_120000.db" + fi + + local backup_path="./backups/$backup_file" + if [ ! -f "$backup_path" ]; then + # 尝试绝对路径 + if [ -f "$backup_file" ]; then + backup_path="$backup_file" + else + error "备份文件不存在: $backup_file" + fi + fi + + _check_db_running + + warn "即将从备份恢复数据库，当前数据将被覆盖！" + warn "备份文件: $backup_path" + read -p "确认继续? [y/N] 输入 YES 确认: " -r + echo + if [[ "$REPLY" == "YES" ]]; then + info "停止 dnfs 和 ai_review 服务..." + docker_compose stop dnfs ai_review + + info "复制备份文件到容器..." + docker cp "$backup_path" "$DB_CONTAINER:/app/data/database.db" + + info "重启 db 服务..." + docker_compose restart db + + # 等待db健康 + info "等待 db 服务恢复..." + sleep 5 + + info "启动其他服务..." + docker_compose start dnfs ai_review + + info "恢复完成！" + else + info "已取消" + fi +} + +cmd_db_list_backups() { + info "备份文件列表 (./backups/):" + if [ -d "./backups" ]; then + ls -lhS ./backups/ 2>/dev/null | grep -v "total" | grep -v ".gitkeep" || echo " (无备份文件)" + else + echo " backups 目录不存在" + fi +} + +usage() { + echo "微服务容器化管理脚本" + echo "" + echo "用法: ./docker.sh [options]" + echo "" + echo "── 服务管理 ──" + echo " build 构建所有服务镜像" + echo " up 启动所有服务（后台运行）" + echo " down 停止并移除所有容器" + echo " start 启动已停止的容器" + echo " stop 停止运行中的容器" + echo " restart 重启服务" + echo " ps 查看服务状态" + echo " logs [svc] 查看日志（默认所有服务，可指定服务名）" + echo " shell [svc] 进入容器shell（默认dnfs）" + echo " rebuild 重新构建（无缓存）并启动" + echo " health 检查服务健康状态" + echo " pull 拉取最新基础镜像" + echo " clean 停止并删除所有容器+数据卷（危险！）" + echo "" + echo "── 数据库管理 (SQLite) ──" + echo " db-info 查看数据库信息和表统计" + echo " db-backup [file] 备份数据库（.db 二进制文件，输出到 ./backups/）" + echo " db-export [file] 导出数据为 JSON（跨平台传输，输出到 ./backups/）" + echo " db-import 从 JSON 导入数据（文件放在 ./backups/ 下）" + echo " db-restore 从 .db 备份文件恢复" + echo " db-vacuum VACUUM 压缩数据库" + echo " db-ls 列出备份文件" + echo "" + echo "示例:" + echo " ./docker.sh build dnfs # 仅构建dnfs服务" + echo " ./docker.sh logs dnfs # 查看dnfs日志" + echo " ./docker.sh shell db # 进入db容器" + echo " ./docker.sh db-backup # 立即备份" + echo " ./docker.sh db-export # 导出JSON用于迁移" + echo " ./docker.sh db-ls # 查看现有备份" + echo "" +} + +# 主逻辑 +check_docker + +case "${1:-}" in + build) + shift + cmd_build "$@" + ;; + up) + shift + cmd_up "$@" + ;; + down) + shift + cmd_down "$@" + ;; + start) + shift + cmd_start "$@" + ;; + stop) + shift + cmd_stop "$@" + ;; + restart) + shift + cmd_restart "$@" + ;; + ps|status) + cmd_ps + ;; + logs) + shift + cmd_logs "$@" + ;; + shell) + shift + cmd_shell "$@" + ;; + rebuild) + shift + cmd_rebuild "$@" + ;; + health) + cmd_health + ;; + pull) + cmd_pull + ;; + clean) + cmd_clean + ;; + # ── 数据库管理命令 ── + db-info|dbinfo) + cmd_db_info + ;; + db-backup|dbbackup) + shift + cmd_db_backup "$@" + ;; + db-export|dbexport) + shift + cmd_db_export "$@" + ;; + db-import|dbimport) + shift + cmd_db_import "$@" + ;; + db-restore|dbrestore) + shift + cmd_db_restore "$@" + ;; + db-vacuum|dbvacuum) + cmd_db_vacuum + ;; + db-ls|dbls) + cmd_db_list_backups + ;; + *) + usage + exit 1 + ;; +esac diff --git a/pymsg/README.md b/pymsg/README.md index 3c77548..c5694a8 100644 --- a/pymsg/README.md +++ b/pymsg/README.md @@ -179,61 +179,85 @@ server = MsgServer("0.0.0.0", 8005, workers=4) - `MsgClient.stop()` 可从另一线程调用以中断 `listen()`。 - `send()` 和 `recv()` 内部不做同步 —— 如果多线程共享同一 socket，调用方需自行串行化。 +## JSON RPC 约定层（v0.3.0） + +项目内所有 pymsg 通信都遵循同一 JSON RPC 约定（与 C++/Go 客户端字节级兼容）： + +``` +请求: {"_route": "<路由名>", ...业务字段} +响应: {"success": true, "data": ...} | {"success": false, "error": "..."} +``` + +`RpcServer` / `RpcClient` 把这套约定的编解码、路由分发、错误包装收敛为开箱即用的类（`pymsg/rpc.py`），各服务无需再手写 `_dispatch` / `_reply`。 + +### `RpcServer` — JSON RPC 服务端 + +```python +from pymsg import RpcServer + +rpc = RpcServer("127.0.0.1", 8005, workers=4) + +@rpc.route("execute") +def _execute(req): # handler(req: dict) -> data + return {"rows": []} # 返回值自动包成 {"success": true, "data": ...} + # 抛异常自动包成 {"success": false, "error": str(e)} +rpc.start() # 阻塞当前线程 +``` + +| 构造参数 | 含义 | +|----------|------| +| `addr` / `port` | 监听地址与端口 | +| `workers` | 回调并发线程数（透传 `MsgServer`，默认 0） | +| `json_default` | 响应 JSON 序列化兜底函数（如日期转 ISO 8601） | +| `on_error` | 可选钩子 `on_error(route, exc)`，handler 异常时调用（记日志用） | + +### `RpcClient` — JSON RPC 客户端 + +```python +from pymsg import RpcClient, RpcError + +rpc = RpcClient("127.0.0.1", 8005, "dnfs", timeout=10.0) +try: + data = rpc.call("execute", {"action": "select", "table": "Users"}) +except RpcError as e: + ... # 服务端业务错误（success=false） +``` + +`call(route, payload=None, *, timeout=None)`：自动注入 `_route`、JSON 编解码、检查 `success`，返回 `data` 字段；业务错误抛 `RpcError`，网络故障仍抛 `ConnectionRefusedError` / `TimeoutError`（来自 `MsgClient`）。 + ## 实际用例 ### 服务端 — 数据库服务（dbv2） -db 服务通过 pymsg 暴露通用 CRUD 接口。每条请求是带 `_route` 字段的 JSON 对象，服务端按路由分发到对应处理函数，通过同一 socket 回复结果。 +db 服务基于 `RpcServer` 暴露通用 CRUD 接口（路由 `execute`）： ```python -import json -from pymsg import MsgServer +from pymsg import RpcServer class DBServer: - def __init__(self): - self._server = MsgServer("127.0.0.1", 8005) - self._routes = {"execute": self._handle_execute} + def __init__(self, addr="127.0.0.1", port=8005, workers=4): + self._server = RpcServer(addr, port, workers=workers, + json_default=_json_default) + self._server.add_route("execute", self._handle_execute) def start(self): - self._server.start(self._dispatch) - - def _dispatch(self, sock, addr, text): - req = json.loads(text) - route = req.pop("_route", "") - handler = self._routes.get(route) - if handler is None: - self._server.reply(sock, - json.dumps({"success": False, "error": "未知路由"})) - return - try: - result = handler(req) - self._server.reply(sock, - json.dumps({"success": True, "data": result})) - except Exception as e: - self._server.reply(sock, - json.dumps({"success": False, "error": str(e)})) + self._server.start() def _handle_execute(self, data): - # ... 通过 SQLAlchemy 执行 SQL，返回 dict 列表 + # ... 通过 SQLAlchemy 执行 SQL，返回 dict 列表；异常自动包装为错误响应 pass ``` ### 客户端 — RPC 调用（dnfs → db） -dnfs 笔记服务自身没有数据库，所有 CRUD 操作均通过 pymsg 调用 db。 +dnfs 笔记服务自身没有数据库，所有 CRUD 操作均通过 `RpcClient` 调用 db： ```python -import json -from pymsg import MsgClient +from pymsg import RpcClient -def _call_dbv2(payload): - client = MsgClient("127.0.0.1", 8005, "dnfs") - resp = client.call(json.dumps(payload), timeout=10.0) - return json.loads(resp) +_rpc = RpcClient("127.0.0.1", 8005, "dnfs") -# 创建笔记 -result = _call_dbv2({ - "_route": "execute", +data = _rpc.call("execute", { "action": "insert", "table": "Notes", "values": {"title": "会议记录", "content": "..."} @@ -242,24 +266,19 @@ result = _call_dbv2({ ### 客户端 — 异步日志推送（log_client） -各服务通过 pymsg 将日志异步推送到 db。使用 `MsgClient.call()` 逐批插入。 +各服务通过统一 `log_client.py` 将日志异步推送到 db，底层为 `RpcClient`： ```python -import json -from pymsg import MsgClient +from pymsg import RpcClient -def _send(batch): - client = MsgClient("127.0.0.1", 8005, "gateway-log") - for entry in batch: - client.call(json.dumps({ - "_route": "execute", - "action": "insert", - "table": "Logs", - "values": { - "ts": now, "service": entry["service"], - "level": entry["level"], "message": entry["message"], - } - }), timeout=5) +client = RpcClient("127.0.0.1", 8005, "log-client") +for entry in batch: + client.call("execute", { + "action": "insert", + "table": "Logs", + "values": {"ts": now, "service": entry["service"], + "level": entry["level"], "message": entry["message"]}, + }, timeout=5) ``` ## 跨语言互操作 @@ -289,6 +308,7 @@ pymsg 的线格式有意设计得极其简单，使其他语言的服务也能 | `ConnectionError`（内置） | 完整消息接收前 socket 断开 | | `ConnectionRefusedError`（内置） | `MsgClient.call()` 或 `send()` 无法连接目标 | | `RuntimeError` | `MsgClient.listen()` 在 `connect()` 之前调用 | +| `pymsg.RpcError` | `RpcClient.call()` 收到 `success=false` 业务错误或非法 RPC 响应 | `MsgServer` 内部会吞掉客户端断连错误（移除该客户端，继续服务）。`broadcast()` 静默跳过已断开的 socket。 diff --git a/pymsg/README.robot.md b/pymsg/README.robot.md index 8a8d79b..0c64f94 100644 --- a/pymsg/README.robot.md +++ b/pymsg/README.robot.md @@ -1,7 +1,7 @@ # pymsg 模块技术手册 > 文档版本：2026-07-30 -> 代码版本：0.2.0 +> 代码版本：0.3.0 > 语言：Python 3.9+（核心）、C++（gateway 集成）、Go（waf 集成） > 定位：跨语言 TCP 消息帧协议库 @@ -144,9 +144,11 @@ log_client.push(entry) ``` pymsg/ ├── pymsg/ -│ └── __init__.py # 核心实现（~460 行） +│ ├── __init__.py # 核心协议与 MsgClient/MsgServer（~470 行） +│ └── rpc.py # JSON RPC 约定层 RpcClient/RpcServer（~170 行，v0.3.0） ├── tests/ -│ └── test_threadpool.py # 线程池/同步模式单元测试（纯 stdlib） +│ ├── test_threadpool.py # 线程池/同步模式单元测试（纯 stdlib） +│ └── test_rpc.py # RPC 约定层单元测试（纯 stdlib） ├── README.md # 用户文档 ├── pyproject.toml # 项目配置 └── .gitignore @@ -189,6 +191,19 @@ pymsg/ | `MsgServer._run_task(...)` | L413 | 工作线程：执行回调、失败打标记 | | `MsgServer._reap_failed(socketlist)` | L434 | 主循环统一移除失败连接 | +#### RPC 约定层（rpc.py，v0.3.0） + +| 函数/类 | 行号 | 功能 | +|---------|------|------| +| `RpcError` | L28 | 业务错误异常（success=false / 非法响应） | +| `RpcClient` | L32 | JSON RPC 客户端 | +| `RpcClient.call(route, payload, *, timeout)` | L46 | 注入 _route、编解码、检查 success，返回 data | +| `RpcServer` | L80 | JSON RPC 服务端（基于 MsgServer） | +| `RpcServer.route(name)` / `add_route` | L119/L126 | 注册路由 handler（`fn(req) -> data`） | +| `RpcServer.start()` / `stop()` | L131/L135 | 启动（阻塞）/ 停止 | +| `RpcServer._dispatch` | L140 | 路由分发与异常包装 | +| `RpcServer._reply` | L164 | 响应包装（支持 json_default） | + #### C++ 实现（Msg.hpp） | 函数/类 | 行号 | 功能 | @@ -736,6 +751,7 @@ python test_runner.py |------|------|---------| | 0.1.0 | — | 初始版本：核心协议、MsgClient、MsgServer、跨语言实现 | | 0.2.0 | 2026-07-30 | `MsgServer` 新增 `workers` 参数：回调可提交线程池并发执行；同连接串行（busy 标记）、每连接发送锁防帧交错、回调异常故障隔离；`tests/test_threadpool.py` 单元测试 | +| 0.3.0 | 2026-07-30 | 新增 `rpc.py` JSON RPC 约定层：`RpcClient`（_route 注入 + 编解码 + success 检查 + `RpcError`）、`RpcServer`（路由分发 + 响应包装 + `json_default`/`on_error` 钩子）；db/smtp/captcha 服务端与 dnfs/log_client 客户端完成迁移；`tests/test_rpc.py` 单元测试 | --- diff --git a/pymsg/pymsg/__init__.py b/pymsg/pymsg/__init__.py index 4f0a706..5efe085 100644 --- a/pymsg/pymsg/__init__.py +++ b/pymsg/pymsg/__init__.py @@ -456,3 +456,9 @@ class MsgServer: pass self._clients.clear() self._send_locks.clear() + + +# ── JSON RPC 约定层（依赖上面的 MsgClient / MsgServer，故放在末尾导入）── +from .rpc import RpcClient, RpcError, RpcServer # noqa: E402 + +__all__ += ["RpcClient", "RpcServer", "RpcError"] diff --git a/pymsg/pymsg/rpc.py b/pymsg/pymsg/rpc.py new file mode 100644 index 0000000..74b2977 --- /dev/null +++ b/pymsg/pymsg/rpc.py @@ -0,0 +1,171 @@ +"""pymsg.rpc — pymsg 线协议之上的 JSON RPC 约定层。 + +约定（与 C++/Go 客户端字节级兼容）:: + + 请求: {"_route": "<路由名>", ...业务字段} + 响应: {"success": true, "data": ...} + {"success": false, "error": "..."} + +把各服务（db / smtp / captcha）重复的 ``_dispatch`` / ``_reply`` 服务端 +分发逻辑，以及各调用方（dnfs / log_client）重复的 ``_route`` 注入 + +JSON 编解码客户端逻辑，收敛为 ``RpcServer`` / ``RpcClient`` 两个类。 + +异常分层:: + + RpcError — 远端返回的业务错误（success=false）或非法响应 + pymsg.TimeoutError — 网络层超时（来自 MsgClient） + ConnectionRefusedError — 目标不可达（来自 MsgClient） + ConnectionError — 连接中断（来自 MsgClient） +""" + +import json + +from . import MsgClient, MsgServer + +__all__ = ["RpcClient", "RpcServer", "RpcError"] + + +class RpcError(Exception): + """远端返回 ``success=false`` 的业务错误，或响应不符合 RPC 约定。""" + + +class RpcClient: + """JSON RPC 客户端（短连接，每次调用独立 TCP 连接）。 + + Usage:: + + rpc = RpcClient("127.0.0.1", 8005, "dnfs") + data = rpc.call("execute", {"action": "select", "table": "Users"}) + """ + + def __init__(self, addr: str, port: int, name: str = "rpc", + timeout: float = 10.0) -> None: + self._client = MsgClient(addr, port, name) + self._timeout = timeout + + def call(self, route: str, payload: "dict | None" = None, *, + timeout: "float | None" = None): + """调用指定路由，返回响应中的 ``data`` 字段。 + + Args: + route: 路由名（注入为 ``_route`` 字段）。 + payload: 业务字段 dict（``_route`` 键会被覆盖）。 + timeout: 本次调用的超时秒数，缺省用构造时的值。 + + Returns: + 服务端返回的 ``data``（可能为 None）。 + + Raises: + RpcError: 服务端返回 success=false，或响应非法。 + pymsg.TimeoutError: 等待响应超时。 + ConnectionRefusedError: 目标服务不可达。 + """ + req = dict(payload or {}) + req["_route"] = route + resp = self._client.call( + json.dumps(req, ensure_ascii=False), + timeout=self._timeout if timeout is None else timeout, + ) + try: + obj = json.loads(resp) + except json.JSONDecodeError as e: + raise RpcError(f"响应不是合法 JSON: {e}") from None + if not isinstance(obj, dict) or "success" not in obj: + raise RpcError(f"非法 RPC 响应: {resp[:200]!r}") + if not obj["success"]: + raise RpcError(obj.get("error", "未知错误")) + return obj.get("data") + + +class RpcServer: + """JSON RPC 服务端：基于 MsgServer，自带路由分发与响应包装。 + + handler 签名为 ``handler(req: dict) -> data``： + + - 返回值放入 ``{"success": true, "data": ...}`` 回复； + - 抛出的异常自动包装为 ``{"success": false, "error": str(e)}``， + 服务不会因此崩溃。 + + Usage:: + + rpc = RpcServer("127.0.0.1", 8005, workers=4) + + @rpc.route("execute") + def _execute(req): + return {"rows": []} + + rpc.start() # 阻塞当前线程 + """ + + def __init__(self, addr: str, port: int, *, workers: int = 0, + json_default: "callable | None" = None, + on_error: "callable | None" = None) -> None: + """ + Args: + addr: 监听地址。 + port: 监听端口。 + workers: 回调并发线程数（透传 MsgServer，0=同步串行）。 + json_default: 响应 JSON 序列化的兜底函数（如日期转 ISO）， + 透传 ``json.dumps(default=...)``。 + on_error: 可选钩子 ``on_error(route, exc)``，handler 抛异常 + 时调用（用于记日志）；None 表示静默。 + """ + self._server = MsgServer(addr, port, workers=workers) + self._routes: "dict" = {} + self._json_default = json_default + self._on_error = on_error + + # ── 路由注册 ------------------------------------------------------- + def route(self, name: str): + """装饰器：注册路由处理函数。""" + def deco(fn): + self._routes[name] = fn + return fn + return deco + + def add_route(self, name: str, fn: "callable") -> None: + """直接注册路由处理函数（等价于 route 装饰器）。""" + self._routes[name] = fn + + # ── 启动 / 停止 ---------------------------------------------------- + def start(self) -> None: + """启动服务（阻塞当前线程）。""" + self._server.start(self._dispatch) + + def stop(self) -> None: + """从另一线程停止服务。""" + self._server.stop() + + # ── 分发与响应包装 -------------------------------------------------- + def _dispatch(self, sock, addr, text): + try: + req = json.loads(text) + except json.JSONDecodeError as e: + self._reply(sock, False, error=f"JSON 解析失败: {e}") + return + + route = req.pop("_route", "") + handler = self._routes.get(route) + if handler is None: + self._reply(sock, False, error=f"未知路由: {route}") + return + + try: + result = handler(req) + self._reply(sock, True, data=result) + except Exception as e: + if self._on_error is not None: + try: + self._on_error(route, e) + except Exception: + pass + self._reply(sock, False, error=str(e)) + + def _reply(self, sock, success, data=None, error=None): + resp = {"success": success} + if data is not None: + resp["data"] = data + if error is not None: + resp["error"] = error + self._server.reply(sock, json.dumps( + resp, ensure_ascii=False, default=self._json_default)) diff --git a/pymsg/pyproject.toml b/pymsg/pyproject.toml index 3bbfeb0..ddb2d78 100644 --- a/pymsg/pyproject.toml +++ b/pymsg/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "pymsg" -version = "0.2.0" +version = "0.3.0" description = "Length-prefixed variable-length socket message framing" requires-python = ">=3.9" license = "MIT" diff --git a/pymsg/tests/test_rpc.py b/pymsg/tests/test_rpc.py new file mode 100644 index 0000000..6a919d4 --- /dev/null +++ b/pymsg/tests/test_rpc.py @@ -0,0 +1,145 @@ +"""RpcClient / RpcServer（JSON RPC 约定层）单元测试。 + +纯标准库，直接运行： + + cd pymsg + python -m unittest discover -s tests -v +""" + +import json +import socket +import threading +import time +import unittest +from datetime import datetime, timezone + +from pymsg import RpcClient, RpcError, RpcServer, send + + +def _free_port() -> int: + with socket.socket() as s: + s.bind(("127.0.0.1", 0)) + return s.getsockname()[1] + + +class _RpcFixture: + """后台线程运行 RpcServer；routes: {name: fn}。""" + + def __init__(self, routes, workers=0, json_default=None): + self.port = _free_port() + self.server = RpcServer("127.0.0.1", self.port, workers=workers, + json_default=json_default) + for name, fn in routes.items(): + self.server.add_route(name, fn) + self.thread = threading.Thread(target=self.server.start, daemon=True) + + def __enter__(self): + self.thread.start() + deadline = time.time() + 5 + while time.time() < deadline: + try: + with socket.create_connection(("127.0.0.1", self.port), timeout=0.2): + return self + except OSError: + time.sleep(0.02) + raise RuntimeError("server did not start in time") + + def __exit__(self, *exc): + self.server.stop() + self.thread.join(timeout=5) + return False + + +class TestRpc(unittest.TestCase): + + def test_call_returns_data(self): + with _RpcFixture({"echo": lambda req: {"got": req["msg"]}}) as fx: + rpc = RpcClient("127.0.0.1", fx.port, "t") + self.assertEqual(rpc.call("echo", {"msg": "hi"}), {"got": "hi"}) + + def test_route_field_is_injected(self): + seen = {} + + def handler(req): + seen.update(req) + return None + + with _RpcFixture({"r": handler}) as fx: + RpcClient("127.0.0.1", fx.port).call("r", {"a": 1}) + # _route 已被服务端 pop，handler 只看到业务字段 + self.assertEqual(seen, {"a": 1}) + + def test_business_error_raises_rpcerror(self): + def handler(req): + raise ValueError("bad input") + + with _RpcFixture({"x": handler}) as fx: + with self.assertRaises(RpcError) as cm: + RpcClient("127.0.0.1", fx.port).call("x") + self.assertIn("bad input", str(cm.exception)) + + def test_unknown_route_raises_rpcerror(self): + with _RpcFixture({}) as fx: + with self.assertRaises(RpcError) as cm: + RpcClient("127.0.0.1", fx.port).call("nope") + self.assertIn("未知路由", str(cm.exception)) + + def test_malformed_json_gets_error_response(self): + """服务端收到非法 JSON：返回 success=false 且不崩溃，后续请求正常。""" + with _RpcFixture({"ok": lambda req: "fine"}) as fx: + with socket.create_connection(("127.0.0.1", fx.port), timeout=5) as s: + send(s, "not-json{{{") + from pymsg import recv + resp = json.loads(recv(s, timeout=5)) + self.assertFalse(resp["success"]) + self.assertIn("JSON 解析失败", resp["error"]) + # 服务仍然存活 + self.assertEqual(RpcClient("127.0.0.1", fx.port).call("ok"), "fine") + + def test_data_none_when_absent(self): + with _RpcFixture({"n": lambda req: None}) as fx: + self.assertIsNone(RpcClient("127.0.0.1", fx.port).call("n")) + + def test_json_default_serialization(self): + def default(o): + if isinstance(o, datetime): + return o.isoformat() + return str(o) + + now = datetime(2026, 7, 30, 12, 0, tzinfo=timezone.utc) + with _RpcFixture({"t": lambda req: {"ts": now}}, json_default=default) as fx: + data = RpcClient("127.0.0.1", fx.port).call("t") + self.assertEqual(data["ts"], now.isoformat()) + + def test_concurrent_rpc_with_workers(self): + def handler(req): + time.sleep(0.3) + return req["i"] * 2 + + with _RpcFixture({"dbl": handler}, workers=4) as fx: + results = [None] * 6 + + def w(i): + results[i] = RpcClient("127.0.0.1", fx.port).call("dbl", {"i": i}) + + t0 = time.monotonic() + ts = [threading.Thread(target=w, args=(i,)) for i in range(6)] + for t in ts: + t.start() + for t in ts: + t.join(10) + elapsed = time.monotonic() - t0 + + self.assertEqual(results, [i * 2 for i in range(6)]) + self.assertLess(elapsed, 1.2, f"workers=4 应并发，实际 {elapsed:.2f}s") + + def test_per_call_timeout_override(self): + with _RpcFixture({"slow": lambda req: time.sleep(2)}) as fx: + rpc = RpcClient("127.0.0.1", fx.port, timeout=10.0) + from pymsg import TimeoutError + with self.assertRaises(TimeoutError): + rpc.call("slow", timeout=0.3) + + +if __name__ == "__main__": + unittest.main() diff --git a/smtp/log_client.py b/smtp/log_client.py index e2358a5..60a426d 100644 --- a/smtp/log_client.py +++ b/smtp/log_client.py @@ -1,8 +1,7 @@ -""" -统一日志客户端 — 通过 pymsg TCP 调用 dbv2 execute 路由，逐条 INSERT 到 Logs 表。 +"""统一日志客户端 — 各服务共用（dnfs / smtp / static 三份副本内容一致，改动请同步）。 设计要点： - - 通过 pymsg TCP 调用 dbv2 execute 路由，逐条 INSERT 到 Logs 表 + - 通过 pymsg RPC 调用 dbv2 execute 路由，逐条 INSERT 到 Logs 表 - 异步非阻塞：内存队列 + 后台线程批量推送，不拖慢业务 - fail-safe：db 服务不可用时静默丢弃，绝不影响业务 - 缓冲批量：队列满 50 条或超时 2 秒时自动 flush @@ -10,9 +9,9 @@ 用法： from log_client import logs - logs.info("smtp", "验证码发送成功") - logs.error("smtp", "验证码发送失败", extra={"errno": 2002}) - logs.push("smtp", "WARNING", "登录失败次数超限", + logs.info("dnfs", "节点新增成功 id=42") + logs.error("dnfs", "数据库连接失败", extra={"errno": 2002}) + logs.push("gateway", "WARNING", "登录失败次数超限", trace_id="req-abc", extra={"ip": "1.2.3.4"}) """ @@ -26,12 +25,12 @@ import atexit from datetime import datetime, timezone _sys.path.insert(0, _os.path.join(_os.path.dirname(__file__), "..")) -from pymsg import MsgClient # noqa: E402 +from pymsg import RpcClient, RpcError # noqa: E402 # ── 配置 ────────────────────────────────────────────────────────────── -_DBV2_ADDR = '127.0.0.1' -_DBV2_PORT = 8005 +_DBV2_ADDR = _os.environ.get('DB_HOST', '127.0.0.1') +_DBV2_PORT = int(_os.environ.get('DB_PORT', '8005')) _BATCH_SIZE = 50 # 队列满 50 条时 flush _FLUSH_INTERVAL = 2.0 # 超时 2 秒强制 flush @@ -39,12 +38,9 @@ _TIMEOUT = 5 # socket 超时（秒） class LogClient: - """异步批量日志推送客户端。 + """异步批量日志推送客户端。""" - 通过 MsgClient（pymsg TCP 协议）逐条 INSERT 到 dbv2 的 Logs 表。 - """ - - def __init__(self, service_name="smtp", + def __init__(self, service_name="log", batch_size=_BATCH_SIZE, flush_interval=_FLUSH_INTERVAL): self._service_name = service_name @@ -88,9 +84,7 @@ class LogClient: def flush(self): """手动 flush 队列中剩余的日志（含后台线程持有的局部 batch）。""" self._flush_signal.set() # 通知后台线程立即 flush 其局部 batch - # 等待后台线程响应（queue.get 超时最长 0.5s） - time.sleep(0.6) - # 排空主队列中尚未被线程取走的条目 + time.sleep(0.6) # 等待后台线程响应 batch = [] while True: try: @@ -113,7 +107,6 @@ class LogClient: while not self._stop.is_set(): timeout = self._flush_interval - (time.monotonic() - last_flush) - # 限制最长阻塞时间，以便及时响应 flush() 信号 chk_timeout = max(0.1, min(timeout, 0.5)) try: item = self._queue.get(timeout=chk_timeout) @@ -141,35 +134,33 @@ class LogClient: self._send(batch) def _send(self, batch): - """通过 pymsg 使用 execute 路由逐条 INSERT 到 dbv2 Logs 表。""" + """通过 pymsg RPC（execute 路由）逐条 INSERT 到 dbv2 Logs 表。""" now = datetime.now(timezone.utc).isoformat() - try: - client = MsgClient(_DBV2_ADDR, _DBV2_PORT, self._service_name + "-log") - for entry in batch: - extra_val = entry.get("extra") - payload = { - "_route": "execute", - "action": "insert", - "table": "Logs", - "values": { - "ts": now, - "service": entry.get("service"), - "level": entry.get("level"), - "message": entry.get("message"), - "trace_id": entry.get("trace_id"), - "extra": json.dumps(extra_val, ensure_ascii=False) - if extra_val is not None else None, - }, - } - resp = client.call(json.dumps(payload), timeout=_TIMEOUT) - resp_obj = json.loads(resp) - if not resp_obj.get("success"): - import sys - print(f"[log_client] INSERT 失败: {resp_obj.get('error')}", - file=sys.stderr) - except Exception as e: - import sys - print(f"[log_client] 发送失败: {e}", file=sys.stderr) + client = RpcClient(_DBV2_ADDR, _DBV2_PORT, self._service_name + "-log") + for entry in batch: + extra_val = entry.get("extra") + payload = { + "action": "insert", + "table": "Logs", + "values": { + "ts": now, + "service": entry.get("service"), + "level": entry.get("level"), + "message": entry.get("message"), + "trace_id": entry.get("trace_id"), + "extra": json.dumps(extra_val, ensure_ascii=False) + if extra_val is not None else None, + }, + } + try: + client.call("execute", payload, timeout=_TIMEOUT) + except RpcError as e: + # 业务错误（如表结构不符）：打印后继续下一条 + print(f"[log_client] INSERT 失败: {e}", file=_sys.stderr) + except Exception as e: + # 网络错误：db 不可用，放弃本批剩余（fail-safe） + print(f"[log_client] 发送失败: {e}", file=_sys.stderr) + break def _shutdown(self): """进程退出时 flush 剩余日志。""" @@ -179,4 +170,4 @@ class LogClient: # ── 全局单例 ────────────────────────────────────────────────────────── -logs = LogClient(service_name="smtp") +logs = LogClient() diff --git a/smtp/smtp_v2.py b/smtp/smtp_v2.py index 463d84a..8255075 100644 --- a/smtp/smtp_v2.py +++ b/smtp/smtp_v2.py @@ -19,7 +19,6 @@ SMTP 账号经 .env 配置：SMTP_SERVER / SMTP_PORT / SMTP_USER / __PASSWORD （__PASSWORD 兼容别名 SMTP_PASSWORD）。 """ -import json import os import random import smtplib @@ -60,7 +59,7 @@ from log_client import logs # pymsg 导入 import sys as _sys _sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) -from pymsg import MsgServer # noqa: E402 +from pymsg import RpcServer # noqa: E402 # ── 配置 ──────────────────────────────────────────────────────────────── @@ -96,14 +95,14 @@ class SmtpServer: def __init__(self, addr="127.0.0.1", port=8008, workers=4): self.addr = addr self.port = port - self._server = MsgServer(addr, port, workers=workers) - - # 路由表 - self._routes = { - "send_email": self._handle_send_email, - "verify_email": self._handle_verify_email, - "health": self._handle_health, - } + self._server = RpcServer( + addr, port, workers=workers, + on_error=lambda route, e: logs.error( + "smtp", f"路由 {route} 处理异常: {e}"), + ) + self._server.add_route("send_email", self._handle_send_email) + self._server.add_route("verify_email", self._handle_verify_email) + self._server.add_route("health", self._handle_health) # ── 启动 / 停止 ────────────────────────────────────────────────── def start(self) -> None: @@ -118,42 +117,12 @@ class SmtpServer: print(f"[smtp] 警告: SMTP 配置缺失: {','.join(missing)}（.env 加载 {_env_loaded} 项）") print(f"[smtp] 监听 tcp://{self.addr}:{self.port}") logs.info("smtp", f"开始监听 tcp://{self.addr}:{self.port}") - self._server.start(self._dispatch) + self._server.start() def stop(self) -> None: """从另一线程停止服务。""" self._server.stop() - # ── 消息调度 ------------------------------------------------------- - def _dispatch(self, sock, addr, text): - """解析 JSON 中的 _route，分发给对应的 handler，并回复结果。""" - try: - req = json.loads(text) - except json.JSONDecodeError as e: - self._reply(sock, False, error=f"JSON 解析失败: {e}") - return - - route = req.pop("_route", "") - handler = self._routes.get(route) - if handler is None: - self._reply(sock, False, error=f"未知路由: {route}") - return - - try: - result = handler(req) - self._reply(sock, True, data=result) - except Exception as e: - logs.error("smtp", f"路由 {route} 处理异常: {e}") - self._reply(sock, False, error=str(e)) - - def _reply(self, sock, success, data=None, error=None): - resp = {"success": success} - if data is not None: - resp["data"] = data - if error is not None: - resp["error"] = error - self._server.reply(sock, json.dumps(resp, ensure_ascii=False)) - # ═══════════════════════════════════════════════════════════════════ # 路由处理 # ═══════════════════════════════════════════════════════════════════ diff --git a/static/log_client.py b/static/log_client.py index f70e750..60a426d 100644 --- a/static/log_client.py +++ b/static/log_client.py @@ -1,177 +1,173 @@ -""" -统一日志客户端 — 通过 pymsg TCP 调用 dbv2 execute 路由，逐条 INSERT 到 Logs 表。 - -设计要点： - - 通过 pymsg TCP 调用 dbv2 execute 路由，逐条 INSERT 到 Logs 表 - - 异步非阻塞：内存队列 + 后台线程批量推送，不拖慢业务 - - fail-safe：db 服务不可用时静默丢弃，绝不影响业务 - - 缓冲批量：队列满 50 条或超时 2 秒时自动 flush - -用法： - from log_client import logs - - logs.info("static", "服务启动成功") - logs.error("static", "文件系统未启用", extra={"path": "/data"}) -""" - -import json -import os as _os -import sys as _sys -import threading -import queue -import time -import atexit -from datetime import datetime, timezone - -_sys.path.insert(0, _os.path.join(_os.path.dirname(__file__), "..")) -from pymsg import MsgClient # noqa: E402 - -# ── 配置 ────────────────────────────────────────────────────────────── - -_DBV2_ADDR = '127.0.0.1' -_DBV2_PORT = 8005 - -_BATCH_SIZE = 50 # 队列满 50 条时 flush -_FLUSH_INTERVAL = 2.0 # 超时 2 秒强制 flush -_TIMEOUT = 5 # socket 超时（秒） - - -class LogClient: - """异步批量日志推送客户端。 - - 通过 MsgClient（pymsg TCP 协议）逐条 INSERT 到 dbv2 的 Logs 表。 - """ - - def __init__(self, service_name="static", - batch_size=_BATCH_SIZE, - flush_interval=_FLUSH_INTERVAL): - self._service_name = service_name - self._batch_size = batch_size - self._flush_interval = flush_interval - self._queue = queue.Queue() - self._stop = threading.Event() - self._flush_signal = threading.Event() - self._thread = threading.Thread(target=self._flush_loop, - daemon=True, name='log-client') - self._thread.start() - atexit.register(self._shutdown) - - # ── 公开接口 ──────────────────────────────────────────────────────── - - def push(self, service, level, message, trace_id=None, extra=None): - """推送一条日志（非阻塞，立即返回）。""" - self._queue.put({ - 'service': service, - 'level': level, - 'message': message, - 'trace_id': trace_id, - 'extra': extra, - }) - - def debug(self, service, message, **kw): - self.push(service, 'DEBUG', message, **kw) - - def info(self, service, message, **kw): - self.push(service, 'INFO', message, **kw) - - def warning(self, service, message, **kw): - self.push(service, 'WARNING', message, **kw) - - def error(self, service, message, **kw): - self.push(service, 'ERROR', message, **kw) - - def critical(self, service, message, **kw): - self.push(service, 'CRITICAL', message, **kw) - - def flush(self): - """手动 flush 队列中剩余的日志（含后台线程持有的局部 batch）。""" - self._flush_signal.set() # 通知后台线程立即 flush 其局部 batch - time.sleep(0.6) # 等待后台线程响应 - batch = [] - while True: - try: - batch.append(self._queue.get_nowait()) - if len(batch) >= self._batch_size: - self._send(batch) - batch = [] - except queue.Empty: - break - if batch: - self._send(batch) - self._flush_signal.clear() - - # ── 内部实现 ──────────────────────────────────────────────────────── - - def _flush_loop(self): - """后台线程：积攒到 batch_size 或超时 flush_interval 时发送。""" - batch = [] - last_flush = time.monotonic() - - while not self._stop.is_set(): - timeout = self._flush_interval - (time.monotonic() - last_flush) - chk_timeout = max(0.1, min(timeout, 0.5)) - try: - item = self._queue.get(timeout=chk_timeout) - batch.append(item) - while len(batch) < self._batch_size: - try: - batch.append(self._queue.get_nowait()) - except queue.Empty: - break - except queue.Empty: - pass - - should_flush = ( - len(batch) >= self._batch_size or - self._flush_signal.is_set() or - (batch and time.monotonic() - last_flush >= self._flush_interval) - ) - if should_flush and batch: - self._send(batch) - batch = [] - last_flush = time.monotonic() - - # 退出前 flush 剩余 - if batch: - self._send(batch) - - def _send(self, batch): - """通过 pymsg 使用 execute 路由逐条 INSERT 到 dbv2 Logs 表。""" - now = datetime.now(timezone.utc).isoformat() - try: - client = MsgClient(_DBV2_ADDR, _DBV2_PORT, self._service_name + "-log") - for entry in batch: - extra_val = entry.get("extra") - payload = { - "_route": "execute", - "action": "insert", - "table": "Logs", - "values": { - "ts": now, - "service": entry.get("service"), - "level": entry.get("level"), - "message": entry.get("message"), - "trace_id": entry.get("trace_id"), - "extra": json.dumps(extra_val, ensure_ascii=False) - if extra_val is not None else None, - }, - } - resp = client.call(json.dumps(payload), timeout=_TIMEOUT) - resp_obj = json.loads(resp) - if not resp_obj.get("success"): - import sys - print(f"[log_client] INSERT 失败: {resp_obj.get('error')}", - file=sys.stderr) - except Exception as e: - import sys - print(f"[log_client] 发送失败: {e}", file=sys.stderr) - - def _shutdown(self): - """进程退出时 flush 剩余日志。""" - self._stop.set() - self.flush() - - -# ── 全局单例 ────────────────────────────────────────────────────────── - -logs = LogClient(service_name="static") +"""统一日志客户端 — 各服务共用（dnfs / smtp / static 三份副本内容一致，改动请同步）。 + +设计要点： + - 通过 pymsg RPC 调用 dbv2 execute 路由，逐条 INSERT 到 Logs 表 + - 异步非阻塞：内存队列 + 后台线程批量推送，不拖慢业务 + - fail-safe：db 服务不可用时静默丢弃，绝不影响业务 + - 缓冲批量：队列满 50 条或超时 2 秒时自动 flush + +用法： + from log_client import logs + + logs.info("dnfs", "节点新增成功 id=42") + logs.error("dnfs", "数据库连接失败", extra={"errno": 2002}) + logs.push("gateway", "WARNING", "登录失败次数超限", + trace_id="req-abc", extra={"ip": "1.2.3.4"}) +""" + +import json +import os as _os +import sys as _sys +import threading +import queue +import time +import atexit +from datetime import datetime, timezone + +_sys.path.insert(0, _os.path.join(_os.path.dirname(__file__), "..")) +from pymsg import RpcClient, RpcError # noqa: E402 + +# ── 配置 ────────────────────────────────────────────────────────────── + +_DBV2_ADDR = _os.environ.get('DB_HOST', '127.0.0.1') +_DBV2_PORT = int(_os.environ.get('DB_PORT', '8005')) + +_BATCH_SIZE = 50 # 队列满 50 条时 flush +_FLUSH_INTERVAL = 2.0 # 超时 2 秒强制 flush +_TIMEOUT = 5 # socket 超时（秒） + + +class LogClient: + """异步批量日志推送客户端。""" + + def __init__(self, service_name="log", + batch_size=_BATCH_SIZE, + flush_interval=_FLUSH_INTERVAL): + self._service_name = service_name + self._batch_size = batch_size + self._flush_interval = flush_interval + self._queue = queue.Queue() + self._stop = threading.Event() + self._flush_signal = threading.Event() + self._thread = threading.Thread(target=self._flush_loop, + daemon=True, name='log-client') + self._thread.start() + atexit.register(self._shutdown) + + # ── 公开接口 ──────────────────────────────────────────────────────── + + def push(self, service, level, message, trace_id=None, extra=None): + """推送一条日志（非阻塞，立即返回）。""" + self._queue.put({ + 'service': service, + 'level': level, + 'message': message, + 'trace_id': trace_id, + 'extra': extra, + }) + + def debug(self, service, message, **kw): + self.push(service, 'DEBUG', message, **kw) + + def info(self, service, message, **kw): + self.push(service, 'INFO', message, **kw) + + def warning(self, service, message, **kw): + self.push(service, 'WARNING', message, **kw) + + def error(self, service, message, **kw): + self.push(service, 'ERROR', message, **kw) + + def critical(self, service, message, **kw): + self.push(service, 'CRITICAL', message, **kw) + + def flush(self): + """手动 flush 队列中剩余的日志（含后台线程持有的局部 batch）。""" + self._flush_signal.set() # 通知后台线程立即 flush 其局部 batch + time.sleep(0.6) # 等待后台线程响应 + batch = [] + while True: + try: + batch.append(self._queue.get_nowait()) + if len(batch) >= self._batch_size: + self._send(batch) + batch = [] + except queue.Empty: + break + if batch: + self._send(batch) + self._flush_signal.clear() + + # ── 内部实现 ──────────────────────────────────────────────────────── + + def _flush_loop(self): + """后台线程：积攒到 batch_size 或超时 flush_interval 时发送。""" + batch = [] + last_flush = time.monotonic() + + while not self._stop.is_set(): + timeout = self._flush_interval - (time.monotonic() - last_flush) + chk_timeout = max(0.1, min(timeout, 0.5)) + try: + item = self._queue.get(timeout=chk_timeout) + batch.append(item) + while len(batch) < self._batch_size: + try: + batch.append(self._queue.get_nowait()) + except queue.Empty: + break + except queue.Empty: + pass + + should_flush = ( + len(batch) >= self._batch_size or + self._flush_signal.is_set() or + (batch and time.monotonic() - last_flush >= self._flush_interval) + ) + if should_flush and batch: + self._send(batch) + batch = [] + last_flush = time.monotonic() + + # 退出前 flush 剩余 + if batch: + self._send(batch) + + def _send(self, batch): + """通过 pymsg RPC（execute 路由）逐条 INSERT 到 dbv2 Logs 表。""" + now = datetime.now(timezone.utc).isoformat() + client = RpcClient(_DBV2_ADDR, _DBV2_PORT, self._service_name + "-log") + for entry in batch: + extra_val = entry.get("extra") + payload = { + "action": "insert", + "table": "Logs", + "values": { + "ts": now, + "service": entry.get("service"), + "level": entry.get("level"), + "message": entry.get("message"), + "trace_id": entry.get("trace_id"), + "extra": json.dumps(extra_val, ensure_ascii=False) + if extra_val is not None else None, + }, + } + try: + client.call("execute", payload, timeout=_TIMEOUT) + except RpcError as e: + # 业务错误（如表结构不符）：打印后继续下一条 + print(f"[log_client] INSERT 失败: {e}", file=_sys.stderr) + except Exception as e: + # 网络错误：db 不可用，放弃本批剩余（fail-safe） + print(f"[log_client] 发送失败: {e}", file=_sys.stderr) + break + + def _shutdown(self): + """进程退出时 flush 剩余日志。""" + self._stop.set() + self.flush() + + +# ── 全局单例 ────────────────────────────────────────────────────────── + +logs = LogClient()