修复短句碎片和模糊重复跳过的误报：允许包含括号的短消息，问句不视为模糊重复
This commit is contained in:
youpaishidifu
+192
@@ -0,0 +1,192 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
实时监控msgbot日志 - 诊断两个问题：
|
||||
1. 知识库匹配错误（牛头不对马嘴）
|
||||
2. OCR连续性问题（需要表情包/图片清屏才能触发OCR）
|
||||
|
||||
使用方法：
|
||||
1. 先启动msgbot
|
||||
2. 运行此脚本开始监控
|
||||
3. 在QQ/微信中发送消息测试
|
||||
4. 实时查看分析结果
|
||||
"""
|
||||
|
||||
import re
|
||||
import time
|
||||
import sys
|
||||
from collections import defaultdict
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
|
||||
class LogAnalyzer:
|
||||
def __init__(self):
|
||||
self.kb_matches = []
|
||||
self.ocr_events = []
|
||||
self.pixel_stats = []
|
||||
self.replies = []
|
||||
|
||||
def analyze_line(self, line: str):
|
||||
"""分析单行日志"""
|
||||
line = line.strip()
|
||||
if not line:
|
||||
return
|
||||
|
||||
# 分析KB匹配
|
||||
if "KB搜索输入" in line:
|
||||
self.kb_matches.append({
|
||||
'time': datetime.now(),
|
||||
'type': 'search_input',
|
||||
'content': line
|
||||
})
|
||||
|
||||
# 分析KB结果
|
||||
elif "KB结果" in line and "标题" in line:
|
||||
self.kb_matches.append({
|
||||
'time': datetime.now(),
|
||||
'type': 'search_result',
|
||||
'content': line
|
||||
})
|
||||
|
||||
# 分析内容相关性
|
||||
elif "内容相关性检查" in line:
|
||||
self.kb_matches.append({
|
||||
'time': datetime.now(),
|
||||
'type': 'relevance_check',
|
||||
'content': line
|
||||
})
|
||||
|
||||
# 分析像素检测
|
||||
elif "像素统计" in line or "像素检测" in line:
|
||||
self.pixel_stats.append({
|
||||
'time': datetime.now(),
|
||||
'content': line
|
||||
})
|
||||
|
||||
# 分析OCR事件
|
||||
elif "OCR" in line and ("跳过" in line or "触发" in line or "输出" in line):
|
||||
self.ocr_events.append({
|
||||
'time': datetime.now(),
|
||||
'content': line
|
||||
})
|
||||
|
||||
# 分析回复
|
||||
elif "新消息" in line and ("回复" in line or "发送" in line):
|
||||
self.replies.append({
|
||||
'time': datetime.now(),
|
||||
'content': line
|
||||
})
|
||||
|
||||
def print_summary(self):
|
||||
"""打印分析摘要"""
|
||||
print("\n" + "="*80)
|
||||
print(f"分析摘要 - {datetime.now().strftime('%H:%M:%S')}")
|
||||
print("="*80)
|
||||
|
||||
# KB匹配分析
|
||||
print(f"\n【知识库匹配】共 {len(self.kb_matches)} 条记录")
|
||||
if self.kb_matches:
|
||||
print(" 最近5条:")
|
||||
for item in self.kb_matches[-5:]:
|
||||
print(f" [{item['time'].strftime('%H:%M:%S')}] {item['content'][:80]}")
|
||||
|
||||
# 像素检测分析
|
||||
print(f"\n【像素检测】共 {len(self.pixel_stats)} 条记录")
|
||||
if self.pixel_stats:
|
||||
print(" 最近5条:")
|
||||
for item in self.pixel_stats[-5:]:
|
||||
print(f" [{item['time'].strftime('%H:%M:%S')}] {item['content'][:80]}")
|
||||
|
||||
# OCR事件分析
|
||||
print(f"\n【OCR事件】共 {len(self.ocr_events)} 条记录")
|
||||
if self.ocr_events:
|
||||
print(" 最近5条:")
|
||||
for item in self.ocr_events[-5:]:
|
||||
print(f" [{item['time'].strftime('%H:%M:%S')}] {item['content'][:80]}")
|
||||
|
||||
# 回复分析
|
||||
print(f"\n【回复记录】共 {len(self.replies)} 条记录")
|
||||
if self.replies:
|
||||
print(" 最近5条:")
|
||||
for item in self.replies[-5:]:
|
||||
print(f" [{item['time'].strftime('%H:%M:%S')}] {item['content'][:80]}")
|
||||
|
||||
print("="*80)
|
||||
|
||||
def monitor_log_file(log_file: str):
|
||||
"""监控日志文件"""
|
||||
print(f"开始监控日志文件: {log_file}")
|
||||
print("按 Ctrl+C 停止监控")
|
||||
print("-"*80)
|
||||
|
||||
analyzer = LogAnalyzer()
|
||||
|
||||
try:
|
||||
with open(log_file, 'r', encoding='utf-8', errors='ignore') as f:
|
||||
# 跳到文件末尾
|
||||
f.seek(0, 2)
|
||||
|
||||
while True:
|
||||
line = f.readline()
|
||||
if not line:
|
||||
time.sleep(0.1)
|
||||
continue
|
||||
|
||||
analyzer.analyze_line(line)
|
||||
print(line.rstrip())
|
||||
|
||||
# 每30秒打印一次摘要
|
||||
if int(time.time()) % 30 == 0:
|
||||
analyzer.print_summary()
|
||||
|
||||
except KeyboardInterrupt:
|
||||
print("\n监控已停止")
|
||||
analyzer.print_summary()
|
||||
except FileNotFoundError:
|
||||
print(f"日志文件不存在: {log_file}")
|
||||
print("请先启动msgbot，然后运行此脚本")
|
||||
|
||||
def main():
|
||||
"""主函数"""
|
||||
# 查找最新的日志文件
|
||||
log_patterns = [
|
||||
"msgbot.log",
|
||||
"monitor.log",
|
||||
"debug.log",
|
||||
"*.log"
|
||||
]
|
||||
|
||||
log_files = []
|
||||
for pattern in log_patterns:
|
||||
log_files.extend(list(Path(".").glob(pattern)))
|
||||
|
||||
if not log_files:
|
||||
print("未找到日志文件，请先启动msgbot")
|
||||
print("然后运行此脚本开始监控")
|
||||
return
|
||||
|
||||
print("找到以下日志文件:")
|
||||
for i, f in enumerate(log_files, 1):
|
||||
print(f" {i}. {f.name}")
|
||||
|
||||
print("\n监控功能说明:")
|
||||
print("1. 实时显示日志内容")
|
||||
print("2. 自动分析KB匹配过程")
|
||||
print("3. 跟踪像素检测结果")
|
||||
print("4. 记录OCR事件")
|
||||
print("5. 监控回复发送")
|
||||
|
||||
# 选择要监控的文件
|
||||
if len(log_files) == 1:
|
||||
monitor_log_file(str(log_files[0]))
|
||||
else:
|
||||
try:
|
||||
choice = int(input("\n请选择要监控的日志文件编号: ")) - 1
|
||||
if 0 <= choice < len(log_files):
|
||||
monitor_log_file(str(log_files[choice]))
|
||||
else:
|
||||
print("无效的选择")
|
||||
except ValueError:
|
||||
print("请输入有效的数字")
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user