huanghai/aiData
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

'commit'

Browse Source
This commit is contained in:
HuangHai
2026-01-26 16:56:09 +08:00
parent 7514d3119f
commit cbc59c6628
4 changed files with 360 additions and 311 deletions
Download Patch File Download Diff File Expand all files Collapse all files

218
WeiXin/WxUtil.py
View File

@@ -14,6 +14,7 @@ if project_root not in sys.path:
sys.path.append(project_root)
import json
from datetime import datetime
from Util.EasyOcrKit import EasyOcrKit
# 初始化 EasyOcrKit
@@ -29,14 +30,30 @@ LOG_DIR = os.path.join(BASE_DATA_DIR, "Logs")
OUTPUT_DIR = os.path.join(BASE_DATA_DIR, "Output")
TEMPLATE_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "Templates")
def clear_directory(dir_path):
"""清理指定目录下的所有文件"""
# 全局调试图片计数器
_debug_counter = 0
def get_next_debug_path(desc="step"):
"""获取下一个顺序命名的调试图片路径 (debug_N_desc.jpg)"""
global _debug_counter
_debug_counter += 1
filename = f"debug_{_debug_counter}_{desc}.jpg"
return os.path.join(OUTPUT_DIR, filename)
def clear_directory(dir_path, exclude_files=None):
"""清理指定目录下的所有文件，支持排除特定文件"""
if not os.path.exists(dir_path):
os.makedirs(dir_path)
return
if exclude_files is None:
exclude_files = []
import shutil
for filename in os.listdir(dir_path):
if filename in exclude_files:
continue
file_path = os.path.join(dir_path, filename)
try:
if os.path.isfile(file_path) or os.path.islink(file_path):
@@ -49,7 +66,11 @@ def clear_directory(dir_path):
def setup_script_environment():
"""运行前清理日志和输出目录"""
logger.info("清理运行环境: Logs 和 Output 目录...")
clear_directory(LOG_DIR)
# 重置调试计数器
global _debug_counter
_debug_counter = 0
# 排除当前正在使用的日志文件
clear_directory(LOG_DIR, exclude_files=["T2_ChatMonitor.log", "WxUtil.log"])
clear_directory(OUTPUT_DIR)
def connect_device():
@@ -93,10 +114,11 @@ def safe_device_click(d, x, y):
logger.error(f"重试点击操作依然失败: {e2}")
return False
async def analyze_chat_image(image_path, output_path, device=None, target_name="对方"):
async def analyze_chat_image(image_path, output_path, device=None, target_name="对方", only_process_last_voice=False):
"""
全面采用 CV + OCR 识别微信聊天截图中的最后一条消息
不再使用 VLM
:param only_process_last_voice: 如果为 True，仅处理（转文字）屏幕上最后一条未转换的语音消息
"""
try:
# 1. 初始化
@@ -122,14 +144,31 @@ async def analyze_chat_image(image_path, output_path, device=None, target_name="
logger.info("正在执行 OCR 识别...")
ocr_results = ocr_kit.read_text(image_path)
# 微信菜单关键字（用于排除干扰）
MENU_KEYWORDS = ["听筒播放", "收藏", "背景播放", "删除", "多选", "取消转文字", "转文字", "引用", "提醒"]
# 5. 整合所有消息
messages = []
debug_img = img.copy() # 初始化调试图
# 绘制过滤区域边界 (可视化)
cv2.line(debug_img, (0, 150), (w, 150), (255, 0, 255), 2) # 顶部线
cv2.line(debug_img, (0, h - 100), (w, h - 100), (255, 0, 255), 2) # 底部线 (从 180 改为 100)
cv2.putText(debug_img, "TOP_FILTER", (10, 140), cv2.FONT_HERSHEY_SIMPLEX, 0.6, (255, 0, 255), 1)
cv2.putText(debug_img, "BOTTOM_FILTER", (10, h - 110), cv2.FONT_HERSHEY_SIMPLEX, 0.6, (255, 0, 255), 1)
# A. 添加语音消息
for ax, ay in audio_matches:
# 过滤掉顶部和底部的非聊天区域 (经验值: 顶部150, 底部250)
if ay < 150 or ay > h - 250:
# 标记所有找到的语音图标 (用于调试)
cv2.circle(debug_img, (ax, ay), 10, (255, 255, 0), -1) # 青色实心圆表示原始匹配点
# 过滤掉顶部和底部的非聊天区域
# 顶部标题栏通常在 150 像素以内
# 底部输入栏通常在 100 像素以内 (捕捉最底部的文字)
if ay < 150 or ay > h - 100:
logger.info(f"忽略区域外语音图标: ({ax}, {ay})")
cv2.rectangle(debug_img, (ax-35, ay-35), (ax+35, ay+35), (128, 128, 128), 1) # 灰色框表示被过滤
cv2.putText(debug_img, "FILTERED", (ax - 40, ay - 45), cv2.FONT_HERSHEY_SIMPLEX, 0.5, (128, 128, 128), 1)
continue
sender = "对方" if ax < w / 2 else "我"
@@ -144,15 +183,58 @@ async def analyze_chat_image(image_path, output_path, device=None, target_name="
color = (0, 0, 255) if is_unread else (0, 255, 0)
cv2.rectangle(debug_img, (ax-30, ay-30), (ax+30, ay+30), color, 2)
# --- 新增：判断是否已转文字 ---
# --- 改进：判断是否已转文字 ---
is_converted = False
converted_trigger_text = ""
for bbox, text, conf in ocr_results:
c_x = int((bbox[0][0] + bbox[2][0]) / 2)
c_y = int((bbox[0][1] + bbox[2][1]) / 2)
# 转换后的文字通常在语音图标下方 30-300 像素内，且水平位置相近
if 30 < c_y - ay < 300 and abs(c_x - ax) < 200:
is_converted = True
break
# 判定逻辑：
# 1. 转换后的文字通常在语音图标下方 30-600 像素 (增加到 600 以适配长文本)
# 2. 水平位置偏移在 600 像素内 (增加到 600 以适配宽文本泡)
# 3. 关键：确保这两个坐标之间没有其他的语音图标（防止第一个语音“偷”了第二个语音的文字）
if 30 < c_y - ay < 600 and abs(c_x - ax) < 600:
# 检查中间是否有其他语音图标
has_intermediate_audio = False
for other_ax, other_ay in audio_matches:
# 增加 20 像素缓冲区，防止判定到自身或极近的干扰点
if ay + 20 < other_ay < c_y - 10:
has_intermediate_audio = True
logger.info(f"语音({ax},{ay}) 被中间语音图标({other_ax},{other_ay}) 阻断，无法关联文本 '{text[:10]}...'")
break
if has_intermediate_audio:
continue
clean_text = text.strip()
# 判定是否为时间戳 (如 13:49, 09:26)
# 增强：同时支持 "昨天 14:15" 这种格式
is_timestamp = re.search(r'(\d{1,2}:\d{2})', clean_text) and (len(clean_text) < 15)
# 判定是否为纯数字或时长 (如 5", 3", 少3")
# 增强：允许前面有少量杂讯字符，只要结尾是数字或 "
is_duration = re.search(r'\d{1,2}"?$', clean_text) and len(clean_text) < 6
# 排除掉语音时长、时间戳和菜单关键字的干扰
if not is_duration and not is_timestamp and clean_text not in MENU_KEYWORDS:
is_converted = True
converted_trigger_text = clean_text
logger.info(f"语音({ax},{ay}) 判定为已转换，关联到有效文本: '{clean_text}'")
break
else:
if is_timestamp:
logger.info(f"语音({ax},{ay}) 忽略下方时间戳文本: '{clean_text}'")
elif is_duration:
logger.info(f"语音({ax},{ay}) 忽略时长文本: '{clean_text}'")
if is_converted:
logger.info(f"语音消息 ({ax}, {ay}) 已有转换文字: '{converted_trigger_text}'，跳过")
# --- 恢复绘图反馈 ---
# 根据已读/未读画框：未读红框，已读绿框
color = (0, 0, 255) if is_unread else (0, 255, 0)
cv2.rectangle(debug_img, (ax-30, ay-30), (ax+30, ay+30), color, 2)
label = "YES" if is_converted else "NO"
# 在框的右侧标注 YES 或 NO
@@ -165,7 +247,8 @@ async def analyze_chat_image(image_path, output_path, device=None, target_name="
"center": (ax, ay),
"y": ay,
"is_unread": is_unread,
"is_converted": is_converted
"is_converted": is_converted,
"content": None
})
# B. 添加文本消息
@@ -174,30 +257,24 @@ async def analyze_chat_image(image_path, output_path, device=None, target_name="
c_y = int((bbox[0][1] + bbox[2][1]) / 2)
# 过滤区域 (顶部标题栏和底部输入栏)
# 底部输入栏通常在最后 150 像素左右
if c_y < 150 or c_y > h - 150:
# 底部输入栏通常在 100 像素以内 (捕捉最底部的文字)
if c_y < 150 or c_y > h - 100:
continue
# 过滤掉明显的系统词 (通常是日期或时间)
# 匹配如: "2025年12月28日 11:18", "11:18", "昨天 09:26" 等
# 增加对 OCR 误识别的容错 (如 28811:18)
time_pattern = r'(\d{4}年|\d{1,2}月|\d{1,2}日|\d{1,2}:\d{2}|昨天|今天|星期|上午|下午|晚上)'
# 如果文本包含这些关键词且长度较短，或者是纯数字/标点组合
if len(text) < 20 and (re.search(time_pattern, text) or re.match(r'^[0-9:\s日年月\-]+$', text)):
continue
# 过滤掉语音时长标识 (如 "5\"", "10\"", "小8\"")
if re.match(r'^.?[0-9]{1,2}"?$', text.strip()):
continue
# 过滤掉“撤回了一条消息”等系统提示
if "撤回了一条消息" in text or "打招呼的消息" in text:
continue
# 排除干扰：语音时长、菜单关键字、系统提示
clean_text = text.strip()
if re.match(r'^.?[0-9]{1,2}"?$', clean_text): continue
if clean_text in MENU_KEYWORDS: continue
if "撤回了一条消息" in text or "打招呼的消息" in text: continue
# 改进发送者判定：查看文本块的左边界
# 对方的消息靠左，我的消息靠右
left_x = bbox[0][0]
sender = "对方" if left_x < w * 0.3 else "我"
sender = "对方" if left_x < w * 0.5 else "我"
messages.append({
"type": "text",
@@ -229,8 +306,15 @@ async def analyze_chat_image(image_path, output_path, device=None, target_name="
# 获取所有语音消息（不论已读未读，只要没转换成文字就处理）
unconverted_voices = [m for m in messages if m['type'] == 'voice' and not m.get('is_converted')]
# 按 Y 坐标排序，确保从上到下顺序
unconverted_voices.sort(key=lambda x: x['y'])
if unconverted_voices:
logger.info(f"发现 {len(unconverted_voices)} 条未转换的语音，开始处理...")
if only_process_last_voice:
logger.info(f"策略限制：仅处理最后一条未转换语音 (共发现 {len(unconverted_voices)} 条)")
unconverted_voices = [unconverted_voices[-1]]
else:
logger.info(f"发现 {len(unconverted_voices)} 条未转换的语音，开始全部处理...")
for v_msg in unconverted_voices:
vx, vy = int(v_msg['center'][0]), int(v_msg['center'][1])
@@ -242,7 +326,7 @@ async def analyze_chat_image(image_path, output_path, device=None, target_name="
time.sleep(1.5)
# B. 截图寻找“转文字”按钮
menu_shot = os.path.join(OUTPUT_DIR, f"voice_menu_{vy}.jpg")
menu_shot = get_next_debug_path("step_long_press")
d.screenshot(menu_shot)
zhuan_template = os.path.join(TEMPLATE_DIR, "zhun_wen_zi.jpg")
@@ -258,7 +342,7 @@ async def analyze_chat_image(image_path, output_path, device=None, target_name="
time.sleep(5.0)
# C. 再次截图 OCR 获取转换后的文字
after_convert_shot = os.path.join(OUTPUT_DIR, f"after_auto_{vy}.jpg")
after_convert_shot = get_next_debug_path("step_convert_result")
try:
d.screenshot(after_convert_shot)
convert_ocr = ocr_kit.read_text(after_convert_shot)
@@ -266,48 +350,12 @@ async def analyze_chat_image(image_path, output_path, device=None, target_name="
logger.error(f"截图或 OCR 失败: {e}")
convert_ocr = []
# 提取转换文字（合并多行结果）
text_blocks = []
for c_bbox, c_text, c_conf in convert_ocr:
cc_x = (c_bbox[0][0] + c_bbox[2][0]) / 2
cc_y = (c_bbox[0][1] + c_bbox[2][1]) / 2
# 扩大搜索范围，适应更长的转换结果
# 增加 sender 判断 (通过水平位置判定)
c_left_x = c_bbox[0][0]
c_sender = "对方" if c_left_x < w * 0.3 else "我"
if 30 < cc_y - vy < 600 and abs(cc_x - vx) < 400 and c_sender == v_msg['sender']:
text_blocks.append((cc_y, c_text))
# ... (中间提取文字逻辑不变) ...
# 按 Y 坐标排序并合并
text_blocks.sort(key=lambda x: x[0])
converted_text = "".join([t[1] for t in text_blocks])
if converted_text:
logger.info(f"✨ OCR 识别成功: {converted_text}")
v_msg['content'] = converted_text
v_msg['is_converted'] = True
else:
logger.warning("❌ OCR 未能提取到转换后的文字内容")
# D. 长按并点击“取消转文字”恢复界面
try:
logger.info("正在恢复界面状态 (点击'取消转文字')...")
d.long_click(vx, vy, 1.5)
time.sleep(1.0)
cancel_shot = os.path.join(OUTPUT_DIR, f"cancel_menu_{vy}.jpg")
d.screenshot(cancel_shot)
cancel_template = os.path.join(TEMPLATE_DIR, "cancel_zhuan_wen_zi.jpg")
cancel_btn = find_template_match(cancel_shot, cancel_template, threshold=0.7)
if cancel_btn:
c_btn_x, c_btn_y = int(cancel_btn[0]), int(cancel_btn[1])
safe_device_click(d, c_btn_x, c_btn_y)
logger.info(f"✅ 已点击'取消转文字' ({c_btn_x}, {c_btn_y})，界面已恢复")
else:
logger.warning("⚠️ 未找到'取消转文字'按钮，尝试点击空白处关闭菜单")
safe_device_click(d, vx + 300, vy)
except Exception as e:
logger.error(f"恢复界面状态时发生错误: {e}")
# D. (已移除) 不再执行“取消转文字”操作，保留文字以避免重复识别
# 之前此处会执行 long_click -> cancel_template -> safe_device_click
# 为了解决“反复打开”的问题，现在改为保留转出来的文字
logger.info("保留语音转换后的文字，不进行恢复界面操作")
else:
logger.warning("❌ 未能找到'转文字'按钮，点击空白处退出")
safe_device_click(d, vx + 300, vy)
@@ -326,13 +374,26 @@ async def analyze_chat_image(image_path, output_path, device=None, target_name="
if msg['type'] == 'text':
cx, cy = msg['center']
# 1. 垂直距离在合理范围内 (30 到 600 像素)
# 2. 发送者一致 (确保归属正确)
# 3. 水平偏移在合理范围内 (对于对方，cx 应该在左侧；对于我，cx 应该在右侧)
if 30 < cy - vy < 600 and msg['sender'] == v_msg['sender']:
# 进一步检查水平位置，确保文字在语音图标的大致垂直线上或稍有偏移
if abs(cx - vx) < 400:
# 2. 水平偏移在合理范围内 (增加到 600 像素以适配宽文本泡)
# 3. 关键：确保这两个坐标之间没有其他的语音图标（防止第一个语音“偷”了第二个语音的文字）
v_dist = cy - vy
h_dist = abs(cx - vx)
if 30 < v_dist < 600 and h_dist < 600:
# 检查中间是否有其他语音图标
has_intermediate_audio = False
for other_ax, other_ay in audio_matches:
if vy < other_ay < cy:
has_intermediate_audio = True
break
if has_intermediate_audio:
continue
# 发送者判定
if msg['sender'] == v_msg['sender']:
v_content_blocks.append(msg)
msg['is_voice_part'] = True
logger.info(f"关联成功: 语音({vx}, {vy}) -> 文本('{msg['content']}') [h_dist={h_dist:.1f}, v_dist={v_dist:.1f}]")
# 如果有内容块，按 Y 排序并合并
if v_content_blocks:
@@ -340,6 +401,9 @@ async def analyze_chat_image(image_path, output_path, device=None, target_name="
combined_content = "".join([m['content'] for m in v_content_blocks])
v_msg['content'] = combined_content
v_msg['is_converted'] = True
else:
if not v_msg.get('content'):
logger.warning(f"语音({vx}, {vy}) 未能关联到任何文本块")
# 2. 收集最终要显示的消息（排除被标记为语音部分的文本）
for msg in messages:
@@ -485,6 +549,10 @@ def find_all_template_matches(screen_path, template_path, threshold=0.8):
h, w = template.shape[:2]
res = cv2.matchTemplate(img, template, cv2.TM_CCOEFF_NORMED)
# 记录最大匹配度，方便调试阈值
min_val, max_val, min_loc, max_loc = cv2.minMaxLoc(res)
logger.info(f"模板匹配 {os.path.basename(template_path)}: 最大相似度 = {max_val:.4f} (阈值={threshold})")
# 找到所有大于阈值的点
loc = np.where(res >= threshold)