118 lines
3.8 KiB
Python
118 lines
3.8 KiB
Python
# coding=utf-8
|
|
"""
|
|
T4_Text_StaticTest.py - 纯文本消息识别与合并测试 (Text Message Static Test)
|
|
|
|
【核心功能】
|
|
- 静态文本分析：专门针对文字消息进行 OCR 识别。
|
|
- 消息合并逻辑：模拟机器人将同一发送者、在短时间内发送的多行文字合并为一条消息的逻辑。
|
|
- 视觉验证：生成标记了识别结果的调试图片 (T4_Debug.jpg)。
|
|
|
|
【使用场景】
|
|
- 验证多行长文字消息是否能被正确合并。
|
|
- 调试文字消息的发送者判定（特别是长文本换行后的判定）。
|
|
- 在不干扰手机交互的情况下，快速验证文本 OCR 的准确性。
|
|
"""
|
|
import os
|
|
import sys
|
|
import logging
|
|
import cv2
|
|
|
|
# Add project root
|
|
project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
|
if project_root not in sys.path:
|
|
sys.path.append(project_root)
|
|
|
|
from WeiXin import WxUtil
|
|
|
|
# Setup Logger
|
|
logger = logging.getLogger("T4_Text")
|
|
|
|
# Configure Root Logger to ensure all logs (including WxUtil) go to file
|
|
root_logger = logging.getLogger()
|
|
root_logger.setLevel(logging.INFO)
|
|
|
|
# Ensure Log Directory exists
|
|
if hasattr(WxUtil, 'LOG_DIR'):
|
|
log_dir = WxUtil.LOG_DIR
|
|
else:
|
|
log_dir = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "Logs")
|
|
|
|
if not os.path.exists(log_dir):
|
|
os.makedirs(log_dir)
|
|
|
|
# File Handler
|
|
log_file = os.path.join(log_dir, "T4_Text.log")
|
|
fh = logging.FileHandler(log_file, mode='w', encoding='utf-8')
|
|
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
|
|
fh.setFormatter(formatter)
|
|
root_logger.addHandler(fh)
|
|
|
|
# Console Handler (if not already added by WxUtil)
|
|
if not any(isinstance(h, logging.StreamHandler) for h in root_logger.handlers):
|
|
ch = logging.StreamHandler()
|
|
ch.setFormatter(formatter)
|
|
root_logger.addHandler(ch)
|
|
|
|
def main():
|
|
logger.info("--- T4: Text Processing Test ---")
|
|
|
|
# 1. Connect
|
|
d = WxUtil.connect_device()
|
|
if not d:
|
|
logger.error("Device connection failed.")
|
|
return
|
|
|
|
# 2. Screenshot
|
|
screenshot_path = os.path.join(WxUtil.OUTPUT_DIR, "T4_Screenshot.jpg")
|
|
logger.info("Taking screenshot...")
|
|
d.screenshot(screenshot_path)
|
|
|
|
# 3. Scan
|
|
logger.info("Scanning image...")
|
|
messages, debug_img, _ = WxUtil._scan_chat_messages(screenshot_path)
|
|
|
|
# 4. Filter and Output
|
|
logger.info("--- Raw Text Messages ---")
|
|
text_msgs = [m for m in messages if m['type'] == 'text']
|
|
|
|
if not text_msgs:
|
|
logger.info("No text messages found.")
|
|
else:
|
|
for i, msg in enumerate(text_msgs):
|
|
logger.info(f"Text [{i}] Sender: {msg['sender']} | y={msg['y']} | Content: {msg['content']}")
|
|
|
|
# 5. Merging Logic (Same as T2)
|
|
logger.info("--- Merged Text Dialogues ---")
|
|
merged_msgs = []
|
|
if text_msgs:
|
|
current_msg = text_msgs[0].copy()
|
|
last_y = current_msg['y']
|
|
|
|
for i in range(1, len(text_msgs)):
|
|
next_msg = text_msgs[i]
|
|
|
|
is_same_sender = (current_msg['sender'] == next_msg['sender'])
|
|
y_diff = next_msg['y'] - last_y
|
|
is_close = (y_diff < 120)
|
|
|
|
if is_same_sender and is_close:
|
|
current_msg['content'] += " " + next_msg['content']
|
|
last_y = next_msg['y']
|
|
else:
|
|
merged_msgs.append(current_msg)
|
|
current_msg = next_msg.copy()
|
|
last_y = current_msg['y']
|
|
|
|
merged_msgs.append(current_msg)
|
|
|
|
for i, msg in enumerate(merged_msgs):
|
|
logger.info(f"[{i}] {msg['sender']}: {msg['content']}")
|
|
|
|
# Save debug
|
|
debug_path = os.path.join(WxUtil.OUTPUT_DIR, "T4_Debug.jpg")
|
|
cv2.imwrite(debug_path, debug_img)
|
|
logger.info(f"Debug image saved to: {debug_path}")
|
|
|
|
if __name__ == "__main__":
|
|
main()
|