微信自动重连监控系统
任务描述
一个运行在本地电脑上的 Python 自动化脚本，用于监控远程服务器上的微信登录状态，并在微信掉线时自动完成重新登录的全流程。
系统架构
本地电脑（Python 脚本 + ADB）
↕ SSH/SFTP
远程服务器（微信 + 检测程序 + 常驻 AHK 脚本）
↕ USB
手机（微信确认登录）
核心流程
常规监控：
通过 HTTP 接口检测微信登录状态（检测程序提供）
接口不可用时，通过 SSH 执行
tasklist检查微信和检测程序进程是否存活检测程序挂了 → 认为微信掉线，触发重连
重连流程：SSH 到服务器，通过
schtasks启动微信进程（解决 SSH session 无法启动 GUI 程序的问题）通过 SFTP 写信号文件，触发服务器上常驻的 AHK 脚本
AHK 脚本发送 Enter 键点击微信登录按钮，然后截图 + 调用 Tesseract OCR 识别界面状态
根据 OCR 结果判断：
等待手机确认 → 进入 ADB 流程
需要扫码 → 记录日志并终止程序
点击失败 → 重试（最多 3 次）
ADB 唤醒手机屏幕、解锁、截图 + OCR 检测确认登录界面
检测到确认界面后点击登录按钮
启动检测程序，通过接口验证登录成功
技术难点与解决方案
文件清单
本地：
wechat_monitor.py— 主监控脚本test_adb_ocr.py— ADB OCR 测试脚本upload_and_test.py— 上传 AHK 脚本到服务器
服务器：click_login.ahk— 常驻 AHK 脚本（信号监听 + 点击登录 + 截图 + OCR）Tesseract OCR（含中文语言包）
依赖
本地 Python 环境：
paramiko（SSH/SFTP）
requests（HTTP 接口检测）
pytesseract + Pillow（手机端 OCR）
ADB 工具（手机控制）
服务器：AutoHotkey v2
Tesseract OCR + chi_sim 语言包
Windows OpenSSH 服务
运行前提
服务器上双击运行
click_login.ahk（常驻后台）手机通过 USB 连接本地电脑，
adb devices能识别本地运行
python wechat_monitor.py
知识点整理
一、SSH/SFTP 远程控制（paramiko）
1. paramiko 持久连接管理
import paramiko
class SSHManager:
def __init__(self):
self.client = None
def connect(self):
self.client = paramiko.SSHClient()
self.client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
self.client.connect(host, port=22, username=user, password=pwd, timeout=10)
def is_connected(self):
if self.client is None:
return False
transport = self.client.get_transport()
return transport is not None and transport.is_active()
def ensure_connected(self):
if not self.is_connected():
self.close()
self.connect()
def execute(self, command):
self.ensure_connected()
stdin, stdout, stderr = self.client.exec_command(command, timeout=20)
out = stdout.read().decode("gbk", errors="ignore").strip()
err = stderr.read().decode("gbk", errors="ignore").strip()
return out, err
def close(self):
if self.client:
try:
self.client.close()
except Exception:
pass
self.client = None
要点：
用
get_transport().is_active()判断连接是否存活每次执行命令前
ensure_connected()，断了自动重连Windows 服务器输出用
gbk解码exec_command设置timeout防止命令卡死
2. SFTP 文件操作
# 写文件（比 echo > 更可靠）
ssh.ensure_connected()
sftp = ssh.client.open_sftp()
with sftp.open(remote_path, 'w') as f:
f.write("content")
sftp.close()
# 读文件
sftp = ssh.client.open_sftp()
with sftp.open(remote_path, 'r') as f:
content = f.read().decode('utf-8').strip()
sftp.close()
# 上传/下载
sftp.put(local_path, remote_path)
sftp.get(remote_path, local_path)
# 列目录
sftp.listdir_attr(remote_path)
要点：
SFTP 写文件比
echo >重定向可靠（echo通过 SSH 有时会失败）写完文件后句柄可能未立即释放，其他程序删除该文件会报"文件被占用"
读文件时加
try/except FileNotFoundError处理文件不存在的情况读文件加重试逻辑应对文件刚创建还没写完的情况
3. SSH Session 隔离（核心坑）
SSH 登录的 session（Session 0）与交互式桌面 session 完全隔离：
4. 通过 SSH 启动 GUI 程序（schtasks 方案）
# 创建一次性计划任务并立即执行
# TN=TaskName TR=TashRun
ssh.execute(
f'schtasks /Create /TN "TaskName" /TR "\'程序路径\'" '
f'/SC ONCE /ST 00:00 /F /RL HIGHEST'
)
ssh.execute('schtasks /Run /TN "TaskName"')
ssh.execute('schtasks /Delete /TN "TaskName" /F')
要点：
/SC ONCE /ST 00:00设置为一次性任务（时间已过不影响手动 Run）/F强制覆盖已存在的同名任务/RL HIGHEST最高权限运行任务在交互式桌面 session 执行，GUI 程序能正常显示
执行后删除任务保持干净
5. 通过 SSH 操作桌面（信号文件方案）
问题： SSH 无法操作桌面窗口 解决： 在桌面 session 运行常驻脚本，SSH 通过文件通信触发操作
# Python 端：写信号文件触发
sftp = ssh.client.open_sftp()
with sftp.open(signal_file, 'w') as f:
f.write("click")
sftp.close()
# 等待执行完成后读取结果
time.sleep(18)
sftp = ssh.client.open_sftp()
with sftp.open(result_file, 'r') as f:
result = f.read().decode('utf-8').strip()
sftp.close()
要点：
常驻脚本在交互式桌面运行（手动双击启动一次）
SSH 端只负责写信号文件和读结果文件
信号文件应在操作完成后再删除（避免句柄竞争）
等待时间要足够长，覆盖：信号检测间隔 + 操作执行时间
6. SSH 远程进程检测
# FI = Filter（过滤条件）
# NH = 不显示表头（让输出更干净，方便脚本解析）
# 检查进程是否存在
out, _ = ssh.execute('tasklist /FI "IMAGENAME eq WeChat.exe" /NH')
is_running = "WeChat.exe" in out
# IM = Image Name（进程名）
# 杀进程
ssh.execute("taskkill /F /IM WeChat.exe")
7. SSH 认证问题
Windows OpenSSH 默认先尝试公钥认证，失败后尝试密码
用户名注意大小写（
administratorvsAdministrator）AutoAddPolicy()自动接受未知主机密钥（生产环境应改用RejectPolicy+ 已知主机文件）
8. 编码问题
Windows CMD 输出：GBK 编码 →
.decode("gbk", errors="ignore")文件内容（Tesseract 输出等）：UTF-8 →
.decode("utf-8")写入文件时注意编码一致性
二、Windows 远程启动 GUI 程序
schtasks 方案
schtasks /Create /TN "名称" /TR "命令" /SC ONCE /ST 00:00 /F /RL HIGHESTschtasks /Run /TN "名称"立即执行schtasks /Delete /TN "名称" /F清理计划任务在交互式桌面 session 运行，能启动 GUI 程序
但计划任务执行的程序也受 session 限制，不一定能操作桌面窗口（取决于用户配置）
信号文件 + 常驻脚本方案在交互式桌面运行一个常驻脚本（AHK/bat）
远程通过 SSH/SFTP 创建信号文件触发操作
常驻脚本检测到信号文件后执行操作，完成后删除信号文件
解决了 SSH session 隔离的所有问题
三、AutoHotkey v2
1. 脚本基础结构
#Requires AutoHotkey v2.0 ; 声明需要 v2 版本
#SingleInstance Force ; 防止重复运行（新实例替换旧实例）
Persistent ; 脚本常驻不退出（没有热键/GUI时必须加）
A_IconTip — 设置托盘图标的鼠标悬停提示文字：
A_IconTip := "My Script - Running"
2. 变量与字符串
; 赋值
myVar := "hello"
myNum := 42
; 字符串拼接用 .（点号）
fullPath := "C:\folder\" . filename . ".txt"
; 多行拼接
content := "line1`n"
. "line2`n"
. "line3`n"
; 在函数内访问全局变量
DoSomething() {
global RESULT_FILE, SCREENSHOT_PATH
; ...
}
注意： n 是换行符（AHK 中反引号 是转义字符，不是反斜杠）
3. 窗口操作
; 等待窗口出现（超时30秒，返回0表示超时）
if !WinWait("ahk_class WeChatLoginWndForPC", , 30) {
; 超时处理
}
; 检查窗口是否存在
if WinExist("ahk_class WeChatLoginWndForPC") {
; 存在
}
; 激活窗口（带到前台）
WinActivate "ahk_class WeChatLoginWndForPC"
; 获取窗口位置和大小
WinGetPos(&x, &y, &w, &h, "ahk_class WeChatLoginWndForPC")
; 置顶/取消置顶
WinSetAlwaysOnTop 1, "ahk_class WeChatLoginWndForPC"
WinSetAlwaysOnTop 0, "ahk_class WeChatLoginWndForPC"
窗口标识方式：
ahk_class ClassName— 按窗口类名匹配（最可靠）ahk_exe process.exe— 按进程名匹配"窗口标题"— 按标题匹配
4. 发送按键和点击（不依赖鼠标位置）
; 给指定窗口发送按键（窗口不需要在前台）
ControlSend("{Enter}", , "ahk_class WeChatLoginWndForPC");两个, ,占位符
; 给指定窗口发送鼠标点击（窗口内相对坐标）
SetControlDelay(-1)
ControlClick("x175 y400", "ahk_class WeChatLoginWndForPC", , , , "NA")
ControlSend 要点：
第一个参数：按键（
{Enter},{Tab},{Space},a,{Ctrl down}c{Ctrl up}等）第二个参数：控件名（留空表示发给窗口本身）
第三个参数：窗口标识
不需要窗口在前台，不移动鼠标
ControlClick要点："NA"参数表示不激活窗口（后台点击）坐标是窗口内的相对坐标
SetControlDelay(-1)取消点击间延迟
5. 文件操作
; 检查文件是否存在
if FileExist("C:\path\to\file.txt") {
; 存在
}
; 删除文件
FileDelete("C:\path\to\file.txt")
; 写入文件（追加模式）
FileAppend("内容", "C:\path\to\file.txt")
; 读取文件（指定编码）
content := FileRead("C:\path\to\file.txt", "UTF-8")
; 字符串拼接路径
ocrFile := OCR_OUTPUT ".txt" ; 结果: "C:\...\ocr_output.txt"
6. 运行外部程序
; 运行并等待完成（隐藏窗口）
RunWait('powershell -ExecutionPolicy Bypass -File "script.ps1"', , "Hide")
; 运行带引号的路径
RunWait('"C:\Program Files\app.exe" "参数"', , "Hide")
; 只运行不等待
Run("notepad.exe")
引号嵌套技巧：
AHK v2 字符串用单引号或双引号
外层用单引号，内层用双引号（或反之）
路径有空格必须加引号
7. 错误处理
try {
FileDelete(SIGNAL_FILE)
} catch {
; 删除失败，忽略
}
注意： 不加 try/catch 时，AHK 遇到错误会弹对话框并暂停脚本执行。对于常驻脚本必须加错误处理。
8. 流程控制
; 无限循环
Loop {
; ...
Sleep 2000 ; 毫秒
}
; 条件判断
if (ww > 0 and wh > 0) {
; ...
} else {
; ...
}
; 字符串包含检查
if InStr(ocrText, "关键词") {
; 找到了
}
; 多条件用 or
if InStr(text, "词1") or InStr(text, "词2") {
; 任一匹配
}
; 函数定义
DoClick() {
global VAR1, VAR2 ; 声明要用的全局变量
; ...
return ; 提前返回
}
9. 像素搜索与找色（PixelSearch）
; 在指定区域搜索颜色（允许色差）
if PixelSearch(&Px, &Py, x1, y1, x2, y2, 0x07C160, 3) {
; 找到了，坐标在 Px, Py
} else {
; 没找到
}
限制：
需要窗口在前台且可见
通过 SSH/schtasks 执行时无法使用（看不到桌面）
只有在交互式桌面 session 的常驻脚本里才能用
10. 调用 PowerShell 截图（动态生成脚本）
; 避免命令行引号嵌套问题：动态生成 .ps1 文件再调用
psScript := "C:\path\do_screenshot.ps1"
if FileExist(psScript)
FileDelete(psScript)
scriptContent := "Add-Type -AssemblyName System.Drawing`n"
. "$bmp = New-Object System.Drawing.Bitmap(" ww "," wh ")`n"
. "$g = [System.Drawing.Graphics]::FromImage($bmp)`n"
. "$g.CopyFromScreen(" wx "," wy ", 0, 0, ...)`n"
. "..."
FileAppend(scriptContent, psScript)
RunWait('powershell -ExecutionPolicy Bypass -File "' psScript '"', , "Hide")
为什么不直接内联 PowerShell 命令：
AHK 字符串 + PowerShell 命令 + 引号嵌套极易出错
动态生成 .ps1 文件再调用，引号问题完全消失
11. 信号文件模式（与外部程序通信）
; 主循环：检测信号文件 → 执行操作 → 删除信号文件
Loop {
if FileExist(SIGNAL_FILE) {
; 清理旧结果
if FileExist(RESULT_FILE)
FileDelete(RESULT_FILE)
; 执行操作
DoClick()
; 操作完成后再删信号文件（避免文件锁冲突）
Sleep 1000
try {
FileDelete(SIGNAL_FILE)
} catch {
}
}
Sleep 2000
}
设计要点：
信号文件在操作完成后再删除（不是检测到就删）
避免与写入方的文件句柄竞争
删除失败用
try/catch忽略，不会导致重复触发（因为操作已执行完）结果通过另一个文件返回给调用方
12. 常见坑
四、Windows 窗口类名识别
微信窗口类名
WeChatLoginWndForPC— 登录界面WeChatMainWndForPC— 主界面（已登录）
用途判断微信登录状态（不依赖接口）
定位窗口进行截图
注意：只有在交互式桌面 session 里才能通过 FindWindow/WinExist 找到
五、OCR（Tesseract）
安装
Windows 安装包：
tesseract-ocr-w64-setup-*.exe中文语言包：
chi_sim.traineddata放到tessdata/目录
命令行调用tesseract input.png output -l chi_sim输出文件自动加
.txt后缀输出编码为 UTF-8
Python 调用（pytesseract）pytesseract.image_to_string(img, lang="chi_sim")配合 Pillow 的
Image.open().convert("L")转灰度提高识别率
识别技巧彩色文字（绿色）在白底上识别率低，转灰度后改善
用完整短语匹配而非单个关键词，避免误匹配
截图范围越精确，干扰越少，识别越准
ColorMatrix 灰度转换在 PowerShell 中可能导致全黑图片，直接用 Pillow 的
.convert("L")更可靠
六、ADB 手机控制
1. 基础连接与设备管理
# 检查设备连接
adb devices
# 输出示例（已连接）
# List of devices attached
# nzojcyus8xamqsjr device
# 获取屏幕分辨率
adb shell wm size
# 输出: Physical size: 1080x2400
2. 触摸操作
# 点击指定坐标
adb shell input tap 540 1530
# 滑动（从 x1,y1 滑到 x2,y2）
adb shell input swipe 540 2000 540 1000
# 长按（滑动距离为0，持续时间长）
adb shell input swipe 540 1530 540 1530 1000
坐标系：
X：从左到右递增，左上角为 0
Y：从上到下递增，左上角为 0
坐标基于屏幕物理分辨率（如 1080x2400）
3. 按键事件
# 唤醒屏幕
adb shell input keyevent KEYCODE_WAKEUP
# 电源键
adb shell input keyevent KEYCODE_POWER
# 返回键
adb shell input keyevent KEYCODE_BACK
# Home键
adb shell input keyevent KEYCODE_HOME
# 回车
adb shell input keyevent KEYCODE_ENTER
4. 截图与文件传输
# 手机端截图保存
adb shell screencap -p /sdcard/screen.png
# 拉取到本地
adb pull /sdcard/screen.png D:\local\screen.png
# 推送文件到手机
adb push local_file.txt /sdcard/
5. 坐标定位方法
# 开启触摸坐标显示（屏幕顶部实时显示触摸坐标）
adb shell settings put system pointer_location 1
# 手指点击目标位置，记录屏幕顶部显示的 X, Y 值
# 关闭坐标显示
adb shell settings put system pointer_location 0
6. Python 中调用 ADB
import subprocess
# 唤醒屏幕
subprocess.run(["adb", "shell", "input", "keyevent", "KEYCODE_WAKEUP"], timeout=5)
# 上滑解锁
subprocess.run(["adb", "shell", "input", "swipe", "540", "2000", "540", "1000"], timeout=5)
# 点击坐标
subprocess.run(["adb", "shell", "input", "tap", "540", "1530"], timeout=5)
# 截图
subprocess.run(["adb", "shell", "screencap", "-p", "/sdcard/screen.png"], timeout=10)
# 拉取截图到本地
subprocess.run(["adb", "pull", "/sdcard/screen.png", "local_screenshot.png"], timeout=10)
要点：
所有参数必须是字符串（坐标也要
str()）设置
timeout防止命令卡死subprocess.run会等待命令完成
7. ADB + OCR 识别手机界面
import subprocess
import os
from PIL import Image
import pytesseract
def adb_screenshot_and_ocr():
"""截图 + OCR 识别手机屏幕文字"""
remote_path = "/sdcard/screen.png"
local_path = "phone_screenshot.png"
# 截图并拉到本地
subprocess.run(["adb", "shell", "screencap", "-p", remote_path], timeout=10)
subprocess.run(["adb", "pull", remote_path, local_path], timeout=10)
if not os.path.exists(local_path):
return ""
# OCR 识别（转灰度提高识别率）
img = Image.open(local_path).convert("L")
text = pytesseract.image_to_string(img, lang="chi_sim")
return text.strip()
OCR 技巧：
.convert("L")转灰度，提高彩色文字识别率lang="chi_sim"指定中文简体用完整短语匹配而非单个关键词，避免误判
8. 基于 OCR 的状态检测与操作
def adb_check_phone_state():
"""根据OCR结果判断手机当前界面状态"""
text = adb_screenshot_and_ocr()
if not text:
return "other"
if "微信已登录" in text or "手机通知已关闭" in text:
return "logged_in"
elif "登录 Windows 微信" in text or "登录确认" in text or "同步最近的消息" in text:
return "waiting_confirm"
else:
return "other"
设计模式：截图 → OCR → 关键词匹配 → 决定操作
9. 完整的 ADB 自动化流程
def adb_confirm_login():
"""唤醒手机 → 检测界面 → 点击确认"""
# 1. 唤醒 + 解锁
subprocess.run(["adb", "shell", "input", "keyevent", "KEYCODE_WAKEUP"], timeout=5)
time.sleep(1)
subprocess.run(["adb", "shell", "input", "swipe", "540", "2000", "540", "1000"], timeout=5)
time.sleep(3)
# 2. 轮询检测界面状态
for attempt in range(5):
state = adb_check_phone_state()
if state == "waiting_confirm":
# 3. 点击确认按钮
subprocess.run(["adb", "shell", "input", "tap", "540", "1530"], timeout=5)
time.sleep(3)
return True
elif state == "logged_in":
return True
else:
time.sleep(3) # 等待界面加载
return False # 超时
流程要点：
先唤醒解锁，等几秒让界面加载
轮询检测而非盲目点击（避免点错位置）
设置最大重试次数防止死循环
检测到已登录直接返回成功
10. 解锁屏幕的常见方式
# 无密码：上滑解锁
subprocess.run(["adb", "shell", "input", "swipe", "540", "2000", "540", "1000"], timeout=5)
# PIN 码解锁：上滑 + 输入PIN
subprocess.run(["adb", "shell", "input", "swipe", "540", "2000", "540", "1000"], timeout=5)
time.sleep(1)
subprocess.run(["adb", "shell", "input", "text", "1234"], timeout=5)
subprocess.run(["adb", "shell", "input", "keyevent", "KEYCODE_ENTER"], timeout=5)
# 检查屏幕是否亮着
# dumpsys 输出中 mScreenOn=true 或 Display Power: state=ON
11. 常见坑
七、进程检测与管理
tasklist 检查进程
tasklist /FI "IMAGENAME eq WeChat.exe" /NH输出包含进程名则说明在运行
taskkill 杀进程taskkill /F /IM WeChat.exe强制杀掉
进程存活作为状态指标检测程序（pywxcode.exe）在微信未登录时会闪退
检测程序活着 = 微信已登录
检测程序挂了 = 微信可能掉线
八、文件竞争与时序问题
信号文件竞争
写入方（SFTP）和读取/删除方（AHK）操作同一文件可能冲突
解决：先执行操作，最后再删信号文件（此时写入方已释放句柄）
文件系统延迟Windows 上
del删除文件后，文件系统可能有短暂延迟才完全释放删除后等 2-3 秒再创建同名文件更可靠
读取时机文件刚创建可能还没写完，读取时加重试逻辑
或者等足够长的时间再读取
九、编码问题
Windows CMD 输出
中文 Windows 默认 GBK 编码
SSH 执行命令的输出用
.decode("gbk")解码
Tesseract 输出输出文件是 UTF-8 编码
AHK 读取时需指定
FileRead(path, "UTF-8")
PowerShell 脚本含中文的 .ps1 文件传输时可能乱码
用纯英文输出避免编码问题
十、远程桌面（RDP）相关
分辨率
远程桌面连接时，服务器分辨率跟随客户端窗口大小。
断开后分辨率可能变化，绝对坐标会失效。
session远程桌面断开（不注销）后，交互式 session 仍然存在。
常驻脚本继续运行，schtasks 任务也能在该 session 执行。
鼠标操作远程桌面连着时，脚本和用户操作的是同一个鼠标光标。
测试时不要手动移动鼠标，避免冲突。
完整代码
wechat_monitor.py
"""
微信登录状态监控脚本
功能：定期检查微信登录状态，掉线后自动重启微信并通过手机确认登录
流程：
1. 通过SSH窗口检测判断微信状态
2. 未登录/未运行 → 启动微信
3. ADB 控制手机确认登录
4. 轮询窗口检测等待登录成功
5. 登录成功后启动检测程序
6. 之后用接口做常规监控
"""
import requests
import time
import logging
import subprocess
import os
import paramiko
# 配置
CHECK_URL = "your_url"
CHECK_INTERVAL = 10 # 常规监控检查间隔（秒）
# SSH 配置
SSH_HOST = "your_ip" # 服务器IP
SSH_PORT = 22 # SSH端口
SSH_USER = "your_user_name" # 用户名，根据实际修改
SSH_PASSWORD = "your_passward" # 密码，根据实际修改
# 服务器上的路径/命令配置
WECHAT_EXE_PATH = r"C:\Users\Administrator\Desktop\app\WeChat.exe"
DETECTOR_EXE_PATH = r"C:\Users\Administrator\Desktop\WXService\pywxcode.exe"
DETECTOR_PROCESS_NAME = "pywxcode.exe"
AHK_EXE_PATH = r"C:\Program Files\AutoHotkey\v2\AutoHotkey64.exe"
CLICK_LOGIN_SCRIPT = r"C:\Users\Administrator\Desktop\click_login.ahk"
# 日志配置
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(message)s",
handlers=[
logging.StreamHandler(),
logging.FileHandler("wechat_monitor.log", encoding="utf-8"),
],
)
logger = logging.getLogger(__name__)
class SSHManager:
"""持久SSH连接管理，断线自动重连"""
def __init__(self):
self.client = None
def connect(self):
self.client = paramiko.SSHClient()
self.client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
self.client.connect(SSH_HOST, port=SSH_PORT, username=SSH_USER, password=SSH_PASSWORD, timeout=10)
logger.info("SSH连接已建立: %s@%s:%s", SSH_USER, SSH_HOST, SSH_PORT)
def is_connected(self):
if self.client is None:
return False
transport = self.client.get_transport()
return transport is not None and transport.is_active()
def ensure_connected(self):
if not self.is_connected():
logger.info("SSH连接不可用，正在重连...")
self.close()
self.connect()
def execute(self, command):
self.ensure_connected()
logger.info("SSH执行命令: %s", command)
stdin, stdout, stderr = self.client.exec_command(command, timeout=20)
out = stdout.read().decode("gbk", errors="ignore").strip()
err = stderr.read().decode("gbk", errors="ignore").strip()
if out:
logger.info("输出: %s", out)
if err:
logger.warning("错误输出: %s", err)
return out, err
def close(self):
if self.client:
try:
self.client.close()
except Exception:
pass
self.client = None
# 全局SSH管理器
ssh = SSHManager()
# ==================== 状态检测 ====================
def check_wechat_by_api():
"""
通过检测程序接口判断微信状态（检测程序必须已运行）
"""
try:
resp = requests.get(CHECK_URL, timeout=5)
data = resp.json()
if "message" in data:
logger.warning("[接口检测] 微信进程未启动: %s", data["message"])
return "not_running"
if data.get("logined") == 1:
logger.info("[接口检测] 微信已登录, nickname=%s", data.get("nickname"))
return "logged_in"
logger.warning("[接口检测] 微信未登录, logined=%s", data.get("logined"))
return "not_logged_in"
except requests.exceptions.RequestException:
return "error"
except (ValueError, KeyError):
return "error"
def check_wechat_by_process():
"""
通过SSH检查进程存活状态判断微信是否掉线
- WeChat.exe 不存在 → not_running
- WeChat.exe 存在但 pywxcode.exe 不存在 → 检测程序挂了，可能掉线
- 两个都在 → 可能正常（但无法确认登录状态）
"""
try:
out, _ = ssh.execute('tasklist /FI "IMAGENAME eq WeChat.exe" /NH')
wechat_running = "WeChat.exe" in out
if not wechat_running:
logger.warning("[进程检测] 微信进程不存在")
return "not_running"
out2, _ = ssh.execute(f'tasklist /FI "IMAGENAME eq {DETECTOR_PROCESS_NAME}" /NH')
detector_running = DETECTOR_PROCESS_NAME in out2
if not detector_running:
logger.warning("[进程检测] 微信在运行但检测程序不在，可能掉线")
return "not_logged_in"
logger.info("[进程检测] 微信和检测程序都在运行")
return "logged_in"
except Exception as e:
logger.error("[进程检测] SSH执行失败: %s", e)
return "error"
# ==================== 服务器操作 ====================
def start_wechat():
"""启动微信进程"""
logger.info(">>> 启动微信...")
try:
ssh.execute("taskkill /F /IM WeChat.exe")
ssh.execute(f"taskkill /F /IM {DETECTOR_PROCESS_NAME}")
time.sleep(3)
ssh.execute(
f'schtasks /Create /TN "StartWeChat" /TR "\'{WECHAT_EXE_PATH}\'" '
f'/SC ONCE /ST 00:00 /F /RL HIGHEST'
)
ssh.execute('schtasks /Run /TN "StartWeChat"')
ssh.execute('schtasks /Delete /TN "StartWeChat" /F')
logger.info("微信进程已启动")
time.sleep(10)
return True
except Exception as e:
logger.error("启动微信失败: %s", e)
return False
def click_login_button():
"""
通过SFTP写信号文件触发AHK点击登录按钮，并读取OCR检测结果
返回值：
"waiting_confirm" - 点击成功，等待手机确认
"need_scan_qr" - 需要扫码，长时间未登录
"retry" - 点击没成功，需要重试
"already_logged_in" - 已经登录了
"error" - 出错
"""
logger.info(">>> 触发点击微信登录按钮...")
try:
signal_file = r"C:\Users\Administrator\Desktop\do_click.txt"
result_file = r"C:\Users\Administrator\Desktop\click_result.txt"
# 用SFTP写信号文件
ssh.ensure_connected()
sftp = ssh.client.open_sftp()
with sftp.open(signal_file, 'w') as f:
f.write("click")
sftp.close()
logger.info("信号文件已写入，等待AHK执行+OCR...")
time.sleep(18) # AHK需要：2s检测 + 1s等待 + 0.5s激活 + 3s点击后等待 + 截图+OCR
# 读取结果文件
ssh.ensure_connected()
sftp = ssh.client.open_sftp()
try:
with sftp.open(result_file, 'r') as f:
result = f.read().decode('utf-8').strip()
sftp.close()
except FileNotFoundError:
sftp.close()
logger.warning("结果文件不存在，AHK可能未执行")
return "error"
logger.info("AHK返回结果: %s", result)
if "WAITING_PHONE_CONFIRM" in result:
logger.info("点击成功，等待手机确认登录")
return "waiting_confirm"
elif "NEED_SCAN_QR" in result:
logger.warning("需要扫码登录（长时间未登录）")
return "need_scan_qr"
elif "LOGIN_BUTTON_STILL_THERE" in result:
logger.warning("登录按钮仍在，点击可能未成功")
return "retry"
elif "ALREADY_LOGGED_IN" in result:
logger.info("微信已经登录")
return "already_logged_in"
else:
logger.error("OCR识别失败，无法判断状态: %s", result)
return "ocr_failed"
except Exception as e:
logger.error("触发点击失败: %s", e)
return "error"
def start_detector():
"""启动检测程序（必须在微信登录成功后调用）"""
logger.info(">>> 启动检测程序...")
try:
detector_dir = r"C:\Users\Administrator\Desktop\WXService"
ssh.execute(
f'schtasks /Create /TN "StartDetector" '
f'/TR "cmd /c cd /d {detector_dir} && {DETECTOR_PROCESS_NAME}" '
f'/SC ONCE /ST 00:00 /F /RL HIGHEST'
)
ssh.execute('schtasks /Run /TN "StartDetector"')
ssh.execute('schtasks /Delete /TN "StartDetector" /F')
logger.info("检测程序已启动")
time.sleep(5)
return True
except Exception as e:
logger.error("启动检测程序失败: %s", e)
return False
# ==================== ADB 操作 ====================
# 手机确认登录按钮坐标（1080x2400 分辨率）
ADB_CONFIRM_X = 540
ADB_CONFIRM_Y = 1530
# 本地截图路径
ADB_SCREENSHOT_LOCAL = r"d:\Desktop\restart\phone_screenshot.png"
ADB_SCREENSHOT_REMOTE = "/sdcard/screen.png"
# 最大重试次数
ADB_MAX_RETRY = 3
def adb_screenshot_and_ocr():
"""ADB截图并OCR识别"""
try:
from PIL import Image
import pytesseract
subprocess.run(["adb", "shell", "screencap", "-p", ADB_SCREENSHOT_REMOTE], timeout=10)
subprocess.run(["adb", "pull", ADB_SCREENSHOT_REMOTE, ADB_SCREENSHOT_LOCAL], timeout=10)
if not os.path.exists(ADB_SCREENSHOT_LOCAL):
return ""
img = Image.open(ADB_SCREENSHOT_LOCAL).convert("L")
text = pytesseract.image_to_string(img, lang="chi_sim")
return text.strip()
except Exception as e:
logger.error("[ADB] 截图/OCR失败: %s", e)
return ""
def adb_check_phone_state():
"""
检测手机当前状态
返回值：
"waiting_confirm" - 确认登录界面
"logged_in" - 已登录成功
"other" - 其他界面
"""
text = adb_screenshot_and_ocr()
if not text:
return "other"
if "微信已登录" in text or "手机通知已关闭" in text:
return "logged_in"
elif "登录 Windows 微信" in text or "登录确认" in text or "同步最近的消息" in text:
return "waiting_confirm"
else:
return "other"
def adb_confirm_login():
"""
ADB 完整流程：
1. 唤醒屏幕 + 解锁
2. 等待确认登录界面出现
3. 点击登录按钮
4. 验证是否点击成功
"""
logger.info("[ADB] 开始手机确认登录流程...")
try:
# 1. 唤醒屏幕
subprocess.run(["adb", "shell", "input", "keyevent", "KEYCODE_WAKEUP"], timeout=5)
time.sleep(1)
# 解锁屏幕（上滑）
subprocess.run(["adb", "shell", "input", "swipe", "540", "2000", "540", "1000"], timeout=5)
time.sleep(3)
# 2. 等待确认登录界面出现（重试几次）
for attempt in range(ADB_MAX_RETRY):
state = adb_check_phone_state()
if state == "waiting_confirm":
logger.info("[ADB] 检测到确认登录界面，点击登录按钮...")
# 3. 点击确认登录
subprocess.run(["adb", "shell", "input", "tap", str(ADB_CONFIRM_X), str(ADB_CONFIRM_Y)], timeout=5)
time.sleep(3)
logger.info("[ADB] 确认登录已点击")
return True
elif state == "logged_in":
logger.info("[ADB] 手机显示已登录成功")
return True
else:
logger.info("[ADB] 未检测到确认界面，等待重试 (%d/%d)...", attempt + 1, ADB_MAX_RETRY)
time.sleep(3)
logger.warning("[ADB] 等待确认界面超时")
return False
except Exception as e:
logger.error("[ADB] 操作失败: %s", e)
return False
# ==================== 主流程 ====================
def reconnect_flow():
"""
完整重连流程：
1. 启动微信
2. 点击登录按钮 + OCR检测结果
3. 根据结果决定下一步
返回值：
True - 重连成功
False - 重连失败但可继续监控
"exit" - 需要终止程序
"""
MAX_CLICK_RETRY = 3
# 1. 启动微信
if not start_wechat():
return False
# 2. 点击登录按钮并检测结果
for attempt in range(MAX_CLICK_RETRY):
result = click_login_button()
if result == "waiting_confirm":
# 点击成功，进入ADB流程
break
elif result == "need_scan_qr":
# 需要扫码，记录日志并终止程序
logger.error("=" * 50)
logger.error("需要扫码登录，自动登录流程无法继续")
logger.error("可能原因：长时间未登录，微信要求重新扫码验证")
logger.error("请手动扫码登录后重启监控脚本")
logger.error("=" * 50)
return "exit"
elif result == "retry":
logger.info("点击未成功，重试第%d/%d次...", attempt + 1, MAX_CLICK_RETRY)
continue
elif result == "already_logged_in":
logger.info("微信已登录，跳过ADB确认")
if not start_detector():
return False
return True
elif result == "ocr_failed":
logger.error("=" * 50)
logger.error("OCR识别失败，无法判断微信状态，终止程序")
logger.error("请检查服务器上Tesseract和AHK脚本是否正常")
logger.error("=" * 50)
return "exit"
else:
logger.warning("检测异常，重试第%d/%d次...", attempt + 1, MAX_CLICK_RETRY)
continue
else:
# 达到最大重试次数
logger.error("=" * 50)
logger.error("点击登录按钮%d次均未成功，终止程序", MAX_CLICK_RETRY)
logger.error("可能原因：AHK常驻脚本未运行，或微信界面异常")
logger.error("请检查服务器状态后重启监控脚本")
logger.error("=" * 50)
return "exit"
# 3. ADB 确认登录
if not adb_confirm_login():
# ADB没找到确认界面，可能是服务器端登录按钮没点成功，重新尝试点击
logger.warning("ADB未检测到确认界面，重新尝试服务器端点击登录...")
click_login_button()
time.sleep(3)
# 再试一次ADB
if not adb_confirm_login():
logger.error("ADB二次确认仍失败，重连流程终止")
return False
# 4. 启动检测程序（ADB已通过OCR确认登录状态，不需要额外等待）
if not start_detector():
return False
# 5. 验证
time.sleep(5)
status = check_wechat_by_api()
if status == "logged_in":
logger.info(">>> 重连流程全部完成！微信已登录")
else:
logger.warning(">>> 重连流程完成，接口状态: %s", status)
return True
def main():
logger.info("=== 微信状态监控启动 ===")
# 建立SSH连接
try:
ssh.connect()
except Exception as e:
logger.error("初始SSH连接失败: %s（后续会自动重连）", e)
try:
while True:
# 先尝试接口检测（检测程序运行时）
status = check_wechat_by_api()
# 接口不可用时用进程检测
if status == "error":
status = check_wechat_by_process()
if status == "logged_in":
pass # 正常，继续监控
elif status in ("not_logged_in", "not_running"):
logger.info("检测到掉线，启动重连流程...")
result = reconnect_flow()
if result == "exit":
logger.error("程序终止")
break
# 重连后等待一段时间再检查，避免立刻又触发重连
time.sleep(20)
elif status == "error":
logger.info("检测异常，等待下次重试...")
time.sleep(CHECK_INTERVAL)
finally:
ssh.close()
logger.info("=== 监控脚本退出，SSH连接已关闭 ===")
if __name__ == "__main__":
main()
click_login.ahk
#Requires AutoHotkey v2.0
#SingleInstance Force
Persistent
; 常驻AHK脚本：监听信号文件，点击登录并OCR检测结果
SIGNAL_FILE := "C:\Users\Administrator\Desktop\do_click.txt"
RESULT_FILE := "C:\Users\Administrator\Desktop\click_result.txt"
SCREENSHOT_PATH := "C:\Users\Administrator\Desktop\wechat_screenshot.png"
TESSERACT_EXE := "C:\Program Files\Tesseract-OCR\tesseract.exe"
OCR_OUTPUT := "C:\Users\Administrator\Desktop\ocr_output"
A_IconTip := "WeChat Login Clicker - Waiting"
Loop {
if FileExist(SIGNAL_FILE) {
; 清理旧结果文件
if FileExist(RESULT_FILE)
FileDelete(RESULT_FILE)
if FileExist(SCREENSHOT_PATH)
FileDelete(SCREENSHOT_PATH)
if FileExist(OCR_OUTPUT ".txt")
FileDelete(OCR_OUTPUT ".txt")
; 先执行点击流程
DoClick()
; 执行完后再删信号文件（此时Python端已释放句柄）
Sleep 1000
try {
FileDelete(SIGNAL_FILE)
} catch {
}
}
Sleep 2000
}
DoClick() {
global RESULT_FILE, SCREENSHOT_PATH, TESSERACT_EXE, OCR_OUTPUT
; 等待微信登录窗口出现（最多等30秒）
if !WinWait("ahk_class WeChatLoginWndForPC", , 30) {
if WinExist("ahk_class WeChatMainWndForPC") {
FileAppend("ALREADY_LOGGED_IN", RESULT_FILE)
} else {
FileAppend("ERROR:LOGIN_WINDOW_NOT_FOUND", RESULT_FILE)
}
return
}
Sleep 1000
WinActivate "ahk_class WeChatLoginWndForPC"
Sleep 500
; 发送Enter点击登录
ControlSend("{Enter}", , "ahk_class WeChatLoginWndForPC")
Sleep 10000
; 截图
TakeScreenshot()
Sleep 500
; 调用Tesseract OCR识别
if !FileExist(SCREENSHOT_PATH) {
FileAppend("ERROR:SCREENSHOT_FAILED", RESULT_FILE)
return
}
RunWait('"' TESSERACT_EXE '" "' SCREENSHOT_PATH '" "' OCR_OUTPUT '" -l chi_sim', , "Hide")
Sleep 500
; 读取OCR结果
ocrFile := OCR_OUTPUT ".txt"
if !FileExist(ocrFile) {
FileAppend("ERROR:OCR_FAILED", RESULT_FILE)
return
}
ocrText := FileRead(ocrFile, "UTF-8")
; 优先匹配点击成功状态
if InStr(ocrText, "需在手机上完成登录") or InStr(ocrText, "在手机上完成登录") {
FileAppend("WAITING_PHONE_CONFIRM", RESULT_FILE)
} else if InStr(ocrText, "切换账号") {
FileAppend("LOGIN_BUTTON_STILL_THERE", RESULT_FILE)
} else if InStr(ocrText, "扫码登录") or InStr(ocrText, "仅传输文件") {
FileAppend("NEED_SCAN_QR", RESULT_FILE)
} else {
FileAppend("UNKNOWN:" ocrText, RESULT_FILE)
}
}
TakeScreenshot() {
global SCREENSHOT_PATH
wx := 0
wy := 0
ww := 0
wh := 0
if WinExist("ahk_class WeChatLoginWndForPC") {
WinGetPos(&wx, &wy, &ww, &wh, "ahk_class WeChatLoginWndForPC")
} else if WinExist("ahk_class WeChatMainWndForPC") {
WinGetPos(&wx, &wy, &ww, &wh, "ahk_class WeChatMainWndForPC")
}
; 生成截图PowerShell脚本（带灰度转换，提高OCR识别率）
psScript := "C:\Users\Administrator\Desktop\do_screenshot.ps1"
if FileExist(psScript)
FileDelete(psScript)
if (ww > 0 and wh > 0) {
scriptContent := "Add-Type -AssemblyName System.Drawing`n"
. "$bmp = New-Object System.Drawing.Bitmap(" ww "," wh ")`n"
. "$g = [System.Drawing.Graphics]::FromImage($bmp)`n"
. "$g.CopyFromScreen(" wx "," wy ", 0, 0, (New-Object System.Drawing.Size(" ww "," wh ")))`n"
. "$g.Dispose()`n"
. "$gray = New-Object System.Drawing.Bitmap(" ww "," wh ")`n"
. "$g2 = [System.Drawing.Graphics]::FromImage($gray)`n"
. "$cm = New-Object System.Drawing.Imaging.ColorMatrix`n"
. "$cm.Matrix00 = 0.3; $cm.Matrix01 = 0.3; $cm.Matrix02 = 0.3`n"
. "$cm.Matrix10 = 0.59; $cm.Matrix11 = 0.59; $cm.Matrix12 = 0.59`n"
. "$cm.Matrix20 = 0.11; $cm.Matrix21 = 0.11; $cm.Matrix22 = 0.11`n"
. "$cm.Matrix33 = 1; $cm.Matrix44 = 1`n"
. "$attr = New-Object System.Drawing.Imaging.ImageAttributes`n"
. "$attr.SetColorMatrix($cm)`n"
. "$rect = New-Object System.Drawing.Rectangle(0, 0, " ww ", " wh ")`n"
. "$g2.DrawImage($bmp, $rect, 0, 0, " ww ", " wh ", [System.Drawing.GraphicsUnit]::Pixel, $attr)`n"
. "$g2.Dispose()`n"
. "$gray.Save('" SCREENSHOT_PATH "')`n"
. "$bmp.Dispose(); $gray.Dispose()`n"
} else {
scriptContent := "Add-Type -AssemblyName System.Windows.Forms`n"
. "Add-Type -AssemblyName System.Drawing`n"
. "$s = [System.Windows.Forms.Screen]::PrimaryScreen.Bounds`n"
. "$bmp = New-Object System.Drawing.Bitmap($s.Width, $s.Height)`n"
. "$g = [System.Drawing.Graphics]::FromImage($bmp)`n"
. "$g.CopyFromScreen(0, 0, 0, 0, (New-Object System.Drawing.Size($s.Width, $s.Height)))`n"
. "$g.Dispose()`n"
. "$gray = New-Object System.Drawing.Bitmap($s.Width, $s.Height)`n"
. "$g2 = [System.Drawing.Graphics]::FromImage($gray)`n"
. "$cm = New-Object System.Drawing.Imaging.ColorMatrix`n"
. "$cm.Matrix00 = 0.3; $cm.Matrix01 = 0.3; $cm.Matrix02 = 0.3`n"
. "$cm.Matrix10 = 0.59; $cm.Matrix11 = 0.59; $cm.Matrix12 = 0.59`n"
. "$cm.Matrix20 = 0.11; $cm.Matrix21 = 0.11; $cm.Matrix22 = 0.11`n"
. "$cm.Matrix33 = 1; $cm.Matrix44 = 1`n"
. "$attr = New-Object System.Drawing.Imaging.ImageAttributes`n"
. "$attr.SetColorMatrix($cm)`n"
. "$rect = New-Object System.Drawing.Rectangle(0, 0, $s.Width, $s.Height)`n"
. "$g2.DrawImage($bmp, $rect, 0, 0, $s.Width, $s.Height, [System.Drawing.GraphicsUnit]::Pixel, $attr)`n"
. "$g2.Dispose()`n"
. "$gray.Save('" SCREENSHOT_PATH "')`n"
. "$bmp.Dispose(); $gray.Dispose()`n"
}
FileAppend(scriptContent, psScript)
RunWait('powershell -ExecutionPolicy Bypass -File "' psScript '"', , "Hide")
}
评论
匿名评论
隐私政策
你无需删除空行，直接评论以获取最佳展示效果
