Improve anti-injector status handling and statistics

Moved statistics update to only occur when the anti-injector system is enabled. Enhanced statistics reporting to handle disabled state and improved uptime calculation to use session start time. Updated status command to provide clearer feedback when the anti-injector system is disabled or when errors occur. Also fixed a docstring in prompt_builder.py for clarity.
This commit is contained in:
雅诺狐
2025-08-20 23:18:09 +08:00
committed by Windpicker-owo
parent dfcb4d5628
commit 90f5e2357e
4 changed files with 35 additions and 7 deletions

View File

@@ -60,11 +60,12 @@ class AntiPromptInjector:
start_time = time.time()
try:
# 统计更新
await self.statistics.update_stats(total_messages=1)
# 1. 检查系统是否启用
if not self.config.enabled:
return ProcessResult.ALLOWED, None, "反注入系统未启用"
# 统计更新 - 只有在系统启用时才进行统计
await self.statistics.update_stats(total_messages=1)
logger.debug(f"开始处理消息: {message.processed_plain_text}")
# 2. 检查用户是否被封禁

View File

@@ -10,6 +10,7 @@ from typing import Dict, Any, Optional
from src.common.logger import get_logger
from src.common.database.sqlalchemy_models import AntiInjectionStats, get_db_session
from src.config.config import global_config
logger = get_logger("anti_injector.statistics")
@@ -19,7 +20,8 @@ class AntiInjectionStatistics:
def __init__(self):
"""初始化统计管理器"""
pass
self.session_start_time = datetime.datetime.now()
"""当前会话开始时间"""
async def get_or_create_stats(self):
"""获取或创建统计记录"""
@@ -78,6 +80,22 @@ class AntiInjectionStatistics:
async def get_stats(self) -> Dict[str, Any]:
"""获取统计信息"""
try:
# 检查反注入系统是否启用
if not global_config.anti_prompt_injection.enabled:
return {
"status": "disabled",
"message": "反注入系统未启用",
"uptime": "N/A",
"total_messages": 0,
"detected_injections": 0,
"blocked_messages": 0,
"shielded_messages": 0,
"detection_rate": "N/A",
"average_processing_time": "N/A",
"last_processing_time": "N/A",
"error_count": 0
}
stats = await self.get_or_create_stats()
# 计算派生统计信息 - 处理None值
@@ -88,10 +106,13 @@ class AntiInjectionStatistics:
detection_rate = (detected_injections / total_messages * 100) if total_messages > 0 else 0
avg_processing_time = (processing_time_total / total_messages) if total_messages > 0 else 0
# 使用当前会话开始时间计算运行时间而不是数据库中的start_time
# 这样可以避免重启后显示错误的运行时间
current_time = datetime.datetime.now()
uptime = current_time - stats.start_time
uptime = current_time - self.session_start_time
return {
"status": "enabled",
"uptime": str(uptime),
"total_messages": total_messages,
"detected_injections": detected_injections,

View File

@@ -151,7 +151,7 @@ class Prompt(str):
@staticmethod
def _process_escaped_braces(template) -> str:
"""处理模板中的转义花括号,将 \{\} 替换为临时标记""" # type: ignore
"""处理模板中的转义花括号,将 \\{\\} 替换为临时标记""" # type: ignore
# 如果传入的是列表,将其转换为字符串
if isinstance(template, list):
template = "\n".join(str(item) for item in template)

View File

@@ -32,8 +32,14 @@ class AntiInjectorStatusCommand(BaseCommand):
anti_injector = get_anti_injector()
stats = await anti_injector.get_stats()
if stats.get("stats_disabled"):
return True, "反注入系统统计功能已禁用", True
# 检查反注入系统是否禁用
if stats.get("status") == "disabled":
await self.send_text("❌ 反注入系统未启用\n\n💡 请在配置文件中启用反注入功能后重试")
return True, "反注入系统未启用", True
if stats.get("error"):
await self.send_text(f"❌ 获取状态失败: {stats['error']}")
return False, f"获取状态失败: {stats['error']}", True
status_text = f"""🛡️ 反注入系统状态报告