From 03d9dd05431acf32e79547fcb96efef6d398193d Mon Sep 17 00:00:00 2001 From: Kris <2893855659@qq.com> Date: Wed, 13 May 2026 16:50:28 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E6=96=B0=E5=A2=9E=E5=8F=8D=E9=A6=88?= =?UTF-8?q?=E6=B8=A0=E9=81=93=E5=AD=97=E6=AE=B5=E3=80=81=E6=8F=92=E4=BB=B6?= =?UTF-8?q?=E7=8A=B6=E6=80=81=E8=A1=A8=E4=B8=8E=E6=97=A5=E5=BF=97=E6=95=8F?= =?UTF-8?q?=E6=84=9F=E4=BF=A1=E6=81=AF=E8=84=B1=E6=95=8F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 1. 为消息反馈表新增channel字段并完善相关接口返回 2. 创建渠道插件状态存储表与对应索引 3. 调整路由规则agent_config_id字段类型为字符串 4. 添加日志敏感信息脱敏功能 5. 优化消息记录仓库方法，新增响应时间统计和查询接口 --- .../channel_message_record_repository.py | 55 +++++++++++++++---- .../package/yuxi/services/feedback_service.py | 4 ++ .../package/yuxi/storage/postgres/manager.py | 18 ++++++ .../yuxi/storage/postgres/models_business.py | 4 ++ .../yuxi/storage/postgres/models_channels.py | 38 ++++++++++++- backend/package/yuxi/utils/logging_config.py | 22 ++++++++ 6 files changed, 127 insertions(+), 14 deletions(-) diff --git a/backend/package/yuxi/repositories/channel_message_record_repository.py b/backend/package/yuxi/repositories/channel_message_record_repository.py index 96b216ce..f5139c58 100644 --- a/backend/package/yuxi/repositories/channel_message_record_repository.py +++ b/backend/package/yuxi/repositories/channel_message_record_repository.py @@ -2,7 +2,7 @@ from __future__ import annotations from datetime import timedelta -from sqlalchemy import select, update +from sqlalchemy import func, select, update from sqlalchemy.ext.asyncio import AsyncSession from yuxi.channels.models import ChannelMessage, ChannelResponse @@ -43,17 +43,21 @@ class ChannelMessageRecordRepository: await self.db.refresh(record) return record - async def mark_success(self, record_id: int, response: ChannelResponse) -> None: - await self.db.execute( - update(ChannelMsgRecord) - .where(ChannelMsgRecord.id == record_id) - .values( - status="success", - reply_message_id=response.identity.channel_message_id, - reply_content_preview=self._truncate(response.content, MAX_REPLY_PREVIEW), - replied_at=utc_now_naive(), - ) - ) + async def mark_success( + self, + record_id: int, + response: ChannelResponse, + response_time_ms: int | None = None, + ) -> None: + values: dict = { + "status": "success", + "reply_message_id": response.identity.channel_message_id, + "reply_content_preview": self._truncate(response.content, MAX_REPLY_PREVIEW), + "replied_at": utc_now_naive(), + } + if response_time_ms is not None: + values["response_time_ms"] = response_time_ms + await self.db.execute(update(ChannelMsgRecord).where(ChannelMsgRecord.id == record_id).values(**values)) await self.db.commit() async def mark_error(self, record_id: int, error_message: str) -> None: @@ -91,6 +95,33 @@ class ChannelMessageRecordRepository: cutoff = utc_now_naive() - timedelta(seconds=timeout_seconds) return [r for r in records if r.created_at and r.created_at < cutoff] + async def get_recent_records( + self, channel_id: str, chat_id: str, limit: int = 10 + ) -> list[ChannelMsgRecord]: + result = await self.db.execute( + select(ChannelMsgRecord) + .where( + ChannelMsgRecord.channel_id == channel_id, + ChannelMsgRecord.chat_id == chat_id, + ) + .order_by(ChannelMsgRecord.created_at.desc()) + .limit(limit) + ) + return list(result.scalars().all()) + + async def get_chat_message_count( + self, channel_id: str, chat_id: str, since_hours: int = 24 + ) -> int: + cutoff = utc_now_naive() - timedelta(hours=since_hours) + result = await self.db.execute( + select(func.count()).select_from(ChannelMsgRecord).where( + ChannelMsgRecord.channel_id == channel_id, + ChannelMsgRecord.chat_id == chat_id, + ChannelMsgRecord.created_at >= cutoff, + ) + ) + return result.scalar() or 0 + @staticmethod def _truncate(text: str, max_length: int) -> str: if len(text) <= max_length: diff --git a/backend/package/yuxi/services/feedback_service.py b/backend/package/yuxi/services/feedback_service.py index 1b5a8de8..46478a25 100644 --- a/backend/package/yuxi/services/feedback_service.py +++ b/backend/package/yuxi/services/feedback_service.py @@ -12,6 +12,7 @@ async def submit_message_feedback_view( message_id: int, rating: str, reason: str | None, + channel: str = "webchat", db: AsyncSession, current_user_id: str, ) -> dict: @@ -41,6 +42,7 @@ async def submit_message_feedback_view( user_id=str(current_user_id), rating=rating, reason=reason, + channel=channel, ) db.add(new_feedback) @@ -54,6 +56,7 @@ async def submit_message_feedback_view( "message_id": new_feedback.message_id, "rating": new_feedback.rating, "reason": new_feedback.reason, + "channel": new_feedback.channel, "created_at": new_feedback.created_at.isoformat(), } @@ -86,6 +89,7 @@ async def get_message_feedback_view( "id": feedback.id, "rating": feedback.rating, "reason": feedback.reason, + "channel": feedback.channel, "created_at": feedback.created_at.isoformat(), }, } diff --git a/backend/package/yuxi/storage/postgres/manager.py b/backend/package/yuxi/storage/postgres/manager.py index 82e9d1b2..94ac97b9 100644 --- a/backend/package/yuxi/storage/postgres/manager.py +++ b/backend/package/yuxi/storage/postgres/manager.py @@ -364,6 +364,24 @@ class PostgresManager(metaclass=SingletonMeta): "CREATE INDEX IF NOT EXISTS idx_msg_records_stuck" " ON channels_msg_records(channel_id, created_at) WHERE status = 'processing'", "CREATE INDEX IF NOT EXISTS idx_msg_records_chat ON channels_msg_records(channel_id, chat_id)", + """ + CREATE TABLE IF NOT EXISTS channel_plugin_states ( + id BIGSERIAL PRIMARY KEY, + channel_id VARCHAR(32) NOT NULL, + namespace VARCHAR(64) NOT NULL DEFAULT 'default', + entry_key VARCHAR(128) NOT NULL, + value_json JSONB NOT NULL DEFAULT '{}', + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + expires_at TIMESTAMPTZ, + CONSTRAINT uq_channel_state UNIQUE (channel_id, namespace, entry_key) + ) + """, + "CREATE INDEX IF NOT EXISTS idx_plugin_states_expiry" + " ON channel_plugin_states(expires_at) WHERE expires_at IS NOT NULL", + "CREATE INDEX IF NOT EXISTS idx_plugin_states_channel ON channel_plugin_states(channel_id, namespace)", + "ALTER TABLE IF EXISTS channels_routing_rules" + " ALTER COLUMN agent_config_id TYPE VARCHAR(64) USING agent_config_id::VARCHAR", ] async with self.async_engine.begin() as conn: for stmt in stmts: diff --git a/backend/package/yuxi/storage/postgres/models_business.py b/backend/package/yuxi/storage/postgres/models_business.py index 171b4734..bf567d15 100644 --- a/backend/package/yuxi/storage/postgres/models_business.py +++ b/backend/package/yuxi/storage/postgres/models_business.py @@ -412,6 +412,9 @@ class MessageFeedback(Base): user_id = Column(String(64), nullable=False, index=True, comment="User ID who provided feedback") rating = Column(String(10), nullable=False, comment="Feedback rating: like or dislike") reason = Column(Text, nullable=True, comment="Optional reason for dislike feedback") + channel = Column( + String(32), nullable=False, default="webchat", comment="Feedback source channel: webchat/wechat/telegram/..." + ) created_at = Column(DateTime, default=utc_now_naive, comment="Feedback creation time") # Relationships @@ -424,6 +427,7 @@ class MessageFeedback(Base): "user_id": self.user_id, "rating": self.rating, "reason": self.reason, + "channel": self.channel, "created_at": format_utc_datetime(self.created_at), } diff --git a/backend/package/yuxi/storage/postgres/models_channels.py b/backend/package/yuxi/storage/postgres/models_channels.py index 92afb370..470752ab 100644 --- a/backend/package/yuxi/storage/postgres/models_channels.py +++ b/backend/package/yuxi/storage/postgres/models_channels.py @@ -1,4 +1,4 @@ -"""多渠道网关 ORM 模型 - 渠道用户映射、会话映射、消息记录""" +"""多渠道网关 ORM 模型 - 渠道用户映射、会话映射、消息记录、插件状态""" from typing import Any @@ -10,6 +10,7 @@ from sqlalchemy import ( Integer, String, UniqueConstraint, + text, ) from sqlalchemy.dialects.postgresql import JSONB from yuxi.storage.postgres.models_business import Base @@ -181,7 +182,7 @@ class ChannelRoutingRule(Base): id = Column(BigInteger, primary_key=True, autoincrement=True) channel_id = Column(String(32), nullable=False) command = Column(String(64), nullable=False) - agent_config_id = Column(Integer, nullable=False) + agent_config_id = Column(String(64), nullable=False) priority = Column(Integer, nullable=False, default=0) created_at = Column(DateTime, nullable=False, default=utc_now_naive) updated_at = Column(DateTime, nullable=False, default=utc_now_naive, onupdate=utc_now_naive) @@ -201,3 +202,36 @@ class ChannelRoutingRule(Base): "created_at": format_utc_datetime(self.created_at), "updated_at": format_utc_datetime(self.updated_at), } + + +class ChannelPluginState(Base): + """渠道插件运行时状态统一存储""" + + __tablename__ = "channel_plugin_states" + + id = Column(BigInteger, primary_key=True, autoincrement=True) + channel_id = Column(String(32), nullable=False) + namespace = Column(String(64), nullable=False, default="default") + entry_key = Column(String(128), nullable=False) + value_json = Column(JSONB, nullable=False, default=dict) + created_at = Column(DateTime, nullable=False, default=utc_now_naive) + updated_at = Column(DateTime, nullable=False, default=utc_now_naive, onupdate=utc_now_naive) + expires_at = Column(DateTime, nullable=True) + + __table_args__ = ( + UniqueConstraint("channel_id", "namespace", "entry_key", name="uq_channel_state"), + Index("idx_plugin_states_expiry", "expires_at", postgresql_where=(text("expires_at IS NOT NULL"))), + Index("idx_plugin_states_channel", "channel_id", "namespace"), + ) + + def to_dict(self) -> dict[str, Any]: + return { + "id": self.id, + "channel_id": self.channel_id, + "namespace": self.namespace, + "entry_key": self.entry_key, + "value_json": self.value_json or {}, + "created_at": format_utc_datetime(self.created_at), + "updated_at": format_utc_datetime(self.updated_at), + "expires_at": format_utc_datetime(self.expires_at) if self.expires_at else None, + } diff --git a/backend/package/yuxi/utils/logging_config.py b/backend/package/yuxi/utils/logging_config.py index d76f9c99..88cdc76b 100644 --- a/backend/package/yuxi/utils/logging_config.py +++ b/backend/package/yuxi/utils/logging_config.py @@ -1,5 +1,6 @@ import logging import os +import re import sys from loguru import logger as loguru_logger @@ -10,6 +11,25 @@ SAVE_DIR = os.getenv("SAVE_DIR") or "saves" DATETIME = shanghai_now().strftime("%Y-%m-%d") LOG_FILE = f"{SAVE_DIR}/logs/yuxi-{DATETIME}.log" +_SENSITIVE_RE_PATTERNS = [ + (re.compile(r'(token[=:]\s*["\']?)([^\s"\'&,]+)', re.IGNORECASE), r"\1***"), + (re.compile(r'(secret[=:]\s*["\']?)([^\s"\'&,]+)', re.IGNORECASE), r"\1***"), + (re.compile(r'(password[=:]\s*["\']?)([^\s"\'&,]+)', re.IGNORECASE), r"\1***"), + (re.compile(r'(api_key[=:]\s*["\']?)([^\s"\'&,]+)', re.IGNORECASE), r"\1***"), + (re.compile(r"(Authorization:\s*)(Bearer\s+\S+)", re.IGNORECASE), r"\1***"), + (re.compile(r"(X-API-Key:\s*)(\S+)", re.IGNORECASE), r"\1***"), + (re.compile(r"'(app_secret|app_key|bot_token|access_token)':\s*'([^']+)'"), r"'\1': '***'"), + (re.compile(r'"(app_secret|app_key|bot_token|access_token)":\s*"([^"]+)"'), r'"\1": "***"'), +] + + +def _redact_sensitive_in_message(record: dict) -> None: + message = record.get("message", "") + if isinstance(message, str): + for pattern, replacement in _SENSITIVE_RE_PATTERNS: + message = pattern.sub(replacement, message) + record["message"] = message + class LoguruHandler(logging.Handler): """将 Python logging 桥接到 loguru 的 handler""" @@ -69,6 +89,7 @@ def setup_logger(name, level="DEBUG", console=True): retention="30 days", compression="zip", enqueue=True, + filter=_redact_sensitive_in_message, ) # 添加控制台日志（有颜色） @@ -84,6 +105,7 @@ def setup_logger(name, level="DEBUG", console=True): ), colorize=True, enqueue=True, + filter=_redact_sensitive_in_message, ) return loguru_logger