192 lines
5.8 KiB
Python
192 lines
5.8 KiB
Python
#!/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() |