From: 小龙 Date: Wed, 15 Jul 2026 09:31:05 +0000 (+0800) Subject: v3.0 - 微信监控初始版本(2026-07-15 测试通过) X-Git-Url: https://szjxzxh.cn/gitweb/avatars/avatar_6.jpg?a=commitdiff_plain;h=a3644170088572230574567b9644a85a887d1167;p=%E5%BE%AE%E4%BF%A1%E7%9B%91%E6%8E%A7 v3.0 - 微信监控初始版本(2026-07-15 测试通过) --- a3644170088572230574567b9644a85a887d1167 diff --git a/web/requirements.txt b/web/requirements.txt new file mode 100644 index 0000000..355ecbb --- /dev/null +++ b/web/requirements.txt @@ -0,0 +1,6 @@ +psutil +pygetwindow +pyautogui +pywin32 +Pillow +requests diff --git a/web/start_wechat_monitor.bat b/web/start_wechat_monitor.bat new file mode 100644 index 0000000..e88b8a9 --- /dev/null +++ b/web/start_wechat_monitor.bat @@ -0,0 +1,4 @@ +@echo off +chcp 936 >nul +start "" "%~dp0wechat_auto_login_v3.py" +exit diff --git a/web/test_wechat_monitor.bat b/web/test_wechat_monitor.bat new file mode 100644 index 0000000..636a3c6 --- /dev/null +++ b/web/test_wechat_monitor.bat @@ -0,0 +1,11 @@ +@echo off +chcp 936 >nul +echo =============================== +echo 微信自动登录监控 - 测试模式 +echo 按 Ctrl+C 退出 +echo =============================== +echo. +python "%~dp0wechat_auto_login_v3.py" +echo. +echo 程序已退出。 +pause diff --git a/web/wechat_auto_login_v3.py b/web/wechat_auto_login_v3.py new file mode 100644 index 0000000..d043aaa --- /dev/null +++ b/web/wechat_auto_login_v3.py @@ -0,0 +1,737 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +"""微信自动登录监控程序""" + +import sys, os, ctypes, traceback + +try: + # 切换到脚本所在目录,保证相对路径正确 + os.chdir(os.path.dirname(os.path.abspath(__file__))) +except Exception: + pass # 切换目录失败不影响后续 + +try: + # ===== 正常的模块初始化从这里开始 ===== + + import time + import hashlib + import random + import logging + import configparser + import atexit + import tempfile + import base64 + + # ── logging (先初始化,再导入第三方库,便于排查导入失败) ───────────────── + log_format = "%(asctime)s.%(msecs)03d [%(levelname)s] %(message)s" + logging.basicConfig( + level=logging.INFO, + format=log_format, + datefmt="%Y-%m-%d %H:%M:%S", + handlers=[ + logging.StreamHandler(), + logging.FileHandler("wechat_monitor.log", encoding="utf-8"), + ], + ) + log = logging.getLogger(__name__) + + # ── 第三方库导入 ───────────────────────────────────────────────────────── + try: + import psutil + import pygetwindow as gw + import pyautogui + import win32gui + import win32con + import win32api + from PIL import ImageGrab + import tkinter as tk + from tkinter import font as tkfont + import requests + import smtplib + from email.utils import formataddr + from email.mime.image import MIMEImage + from email.mime.multipart import MIMEMultipart + from email.mime.text import MIMEText + except ImportError as e: + log.critical("导入第三方库失败: %s", e) + sys.exit(1) + + # ── single instance lock ───────────────────────────────────────────────── + _lock_file = os.path.join(tempfile.gettempdir(), "wechat_auto_login.lock") + _lock_fd = None + + + def _acquire_lock(): + global _lock_fd + try: + _lock_fd = os.open(_lock_file, os.O_CREAT | os.O_EXCL | os.O_RDWR) + return True + except FileExistsError: + return False + + + def _release_lock(): + global _lock_fd + if _lock_fd is not None: + os.close(_lock_fd) + try: + os.unlink(_lock_file) + except OSError: + pass + + + # ── 模块初始化(用 try 捕获所有意外错误) ─────────────────────────────── + try: + if not _acquire_lock(): + log.error("已有实例在运行,退出") + sys.exit(1) + atexit.register(_release_lock) + + # ── config ───────────────────────────────────────────────────────────────── + config = configparser.ConfigParser() + config.read( + os.path.join(os.path.dirname(os.path.abspath(__file__)), "wechat_auto_login.ini"), + encoding="utf-8", + ) + EMAIL_ENABLED = config.getboolean("Email", "email_enabled", fallback=True) + SMTP_SERVER = config.get("Email", "smtp_server", fallback="smtp.qq.com") + SMTP_PORT = config.getint("Email", "smtp_port", fallback=465) + USE_SSL = config.getboolean("Email", "use_ssl", fallback=True) + EMAIL_ACCOUNT = config.get("Email", "email_account", fallback="") + EMAIL_PASSWORD = config.get("Email", "email_password", fallback="") + EMAIL_TO = config.get("Email", "email_to", fallback="") + MAX_EMAILS = config.getint("Email", "max_emails", fallback=5) + SMS_PHONE = config.get("Email", "sms_phone", fallback="") + + # ── constants ───────────────────────────────────────────────────────────────── + SCREEN_W = win32api.GetSystemMetrics(0) + SCREEN_H = win32api.GetSystemMetrics(1) + CROP_W = 696 + CROP_H = 905 + CENTER_X, CENTER_Y = SCREEN_W // 2, SCREEN_H // 2 + PAUSE = 5 + except Exception: + tb = traceback.format_exc() + with open("wechat_fatal.log", "w", encoding="utf-8") as f: + f.write(f"模块初始化失败\n{tb}\n") + print(f"FATAL: 模块初始化失败\n{tb}", file=sys.stderr) + sys.exit(1) + + + + def _img_to_b64(path): + with open(path, "rb") as f: + return base64.b64encode(f.read()).decode("utf-8") + + + def _md5_of_image(img): + return hashlib.md5(img.tobytes()).hexdigest() + + + # ── helpers ────────────────────────────────────────────────────────────── + + + def send_sms(code, phone): + """腾讯云短信发送(完整示例)""" + sdkappid = "1401148277" + appkey = "1fa7ae8213f97cb5da5882b207dfa959" + sign = "深圳市健行者户外运动协会" + template_id = 2673158 + rand = random.randint(100000, 999999) + t = int(time.time()) + sig_str = f"appkey={appkey}&random={rand}&time={t}&mobile={phone}" + sig = hashlib.sha256(sig_str.encode()).hexdigest() + url = f"https://yun.tim.qq.com/v5/tlssmssvr/sendsms?sdkappid={sdkappid}&random={rand}" + body = { + "ext": "", + "extend": "", + "params": [code, "5"], + "sig": sig, + "sign": sign, + "tel": {"mobile": phone, "nationcode": "86"}, + "time": t, + "tpl_id": template_id, + } + try: + resp = requests.post(url, json=body, timeout=10) + return resp.status_code, resp.text + except Exception as e: + log.error("短信发送失败: %s", e) + return -1, str(e) + + + def send_qr_email(qr_img_path): + """发送带二维码附件的邮件""" + if not EMAIL_ENABLED or not EMAIL_ACCOUNT: + log.warning("邮件未启用或账号未配置,跳过邮件发送") + return + try: + with open(qr_img_path, "rb") as f: + img_data = f.read() + + msg = MIMEMultipart("related") + msg["From"] = formataddr(("微信助手", EMAIL_ACCOUNT)) + msg["To"] = EMAIL_TO + msg["Subject"] = "微信登录二维码" + + # HTML 正文嵌入二维码(cid) + html = '' + msg.attach(MIMEText(html, "html")) + + # 附件 + img_attach = MIMEImage(img_data, _subtype="png") + img_attach.add_header("Content-Disposition", "attachment", filename="qrcode.png") + msg.attach(img_attach) + + # 正文内嵌图片 + img_inline = MIMEImage(img_data, _subtype="png") + img_inline.add_header("Content-ID", "") + msg.attach(img_inline) + + if USE_SSL: + s = smtplib.SMTP_SSL(SMTP_SERVER, SMTP_PORT, timeout=10) + else: + s = smtplib.SMTP(SMTP_SERVER, SMTP_PORT, timeout=10) + s.starttls() + s.login(EMAIL_ACCOUNT, EMAIL_PASSWORD) + s.send_message(msg) + s.quit() + log.info("二维码邮件已发送至 %s", EMAIL_TO) + except Exception as e: + log.error("邮件发送失败: %s", e) + + + def _find_wechat_windows(): + """找到标题严格等于'微信'的可见窗口列表(排除包含'微信'的其他窗口)""" + return [w for w in gw.getAllWindows() if w.title == "微信"] + + + def _minimize_all_except_wechat(): + """最小化除微信外的所有可见窗口""" + # 获取微信的窗口句柄 + wechat_hwnds = set() + try: + w = _find_wechat_windows() + for win in w: + wechat_hwnds.add(win._hWnd) + except Exception: + pass + + def enum_callback(hwnd, _): + if not win32gui.IsWindowVisible(hwnd): + return + # 用句柄排除微信窗口 + if hwnd in wechat_hwnds: + return + title = win32gui.GetWindowText(hwnd) + # 排除监控程序自身 + if title == "微信监控": + return + if not title: + return + if title in ("任务管理器", "Task Manager"): + return + try: + win32gui.ShowWindow(hwnd, win32con.SW_SHOWMINIMIZED) + except Exception: + pass + + win32gui.EnumWindows(enum_callback, None) + + + def _capture_center(width=CROP_W, height=CROP_H): + """截取屏幕中心 width×height 并返回 PIL Image""" + log.debug("截取屏幕中心 %dx%d", width, height) + left = CENTER_X - width // 2 + top = CENTER_Y - height // 2 + img = ImageGrab.grab(bbox=(left, top, left + width, top + height)) + log.debug("截图完成: %s", img.size) + return img + + + def _get_ai_result(img): + """ + 使用通义千问视觉模型分析截图。 + 返回: "A" | "B" | "未知" | "ERROR" + AI返回"未知"表示正常工作但没看到A/B。 + AI返回"ERROR"表示AI代码出异常。 + """ + temp_path = os.path.join(tempfile.gettempdir(), "wechat_screen_temp.png") + img.save(temp_path, "PNG") + + prompt = ( + '分析以下微信登录界面截图,请从以下两个状态中返回一个字母:\n' + 'A = 截图中有绿色的"进入微信"按钮\n' + 'B = 截图中有二维码图案并带有"扫码登录"文字\n' + '如果以上都不是,返回"未知"\n' + '请只返回一个字母或"未知"' + ) + + try: + b64 = _img_to_b64(temp_path) + resp = requests.post( + "https://dashscope.aliyuncs.com/compatible-mode/v1/chat/completions", + headers={ + "Authorization": "Bearer sk-a6b3e24b985b41768aa3b0df2f688c98", + "Content-Type": "application/json", + }, + json={ + "model": "qwen-vl-plus", + "messages": [{"role": "user", "content": [ + {"type": "text", "text": prompt}, + {"type": "image_url", "image_url": {"url": f"data:image/png;base64,{b64}"}} + ]}], + "max_tokens": 50, + "temperature": 0, + }, + timeout=15, + ) + data = resp.json() + result = data["choices"][0]["message"]["content"].strip() + except Exception as e: + log.error("AI 分析失败: %s", e) + result = "ERROR" + + try: + os.unlink(temp_path) + except OSError: + pass + + log.info("AI 识别结果: %s", result) + return result + + + def _start_wechat(): + """启动微信""" + log.info("正在启动微信...") + candidates = [ + os.path.expandvars(r"%ProgramFiles(x86)%\Tencent\WeChat\WeChat.exe"), + os.path.expandvars(r"%ProgramFiles%\Tencent\WeChat\WeChat.exe"), + os.path.expandvars(r"%USERPROFILE%\AppData\Local\Tencent\WeChat\WeChat.exe"), + os.path.expandvars(r"%ProgramFiles(x86)%\Tencent\Weixin\Weixin.exe"), + os.path.expandvars(r"%ProgramFiles%\Tencent\Weixin\Weixin.exe"), + ] + wechat_path = None + for p in candidates: + if os.path.exists(p): + wechat_path = p + break + if wechat_path: + try: + os.startfile(wechat_path) + log.info("已启动微信: %s", wechat_path) + except Exception as e: + log.error("启动微信失败: %s", e) + else: + log.error("未找到微信可执行文件") + + + # ── UI ─────────────────────────────────────────────────────────────────── + + + class MonitorApp: + def __init__(self): + self.running = True + self.monitoring = False + + # hash 监测模式状态 + self.hash_mode = False + self.hash_mode_emails_sent = 0 + self.last_md5 = None + self.qr_sent_count = 0 + self._last_login_hash = None + self._user_resized = False + + self._setup_ui() + + def _on_window_configure(self, event): + """当用户手动调整窗口大小或移动窗口时触发""" + self._user_resized = True + + def _setup_ui(self): + self.root = tk.Tk() + self.root.title("微信监控") + self.root.attributes("-topmost", True) + self.root.configure(bg="#f0f0f0") + + # 初始宽高 + self._default_width = 400 + self.root.geometry(f"{self._default_width}x250+0+0") + + # 右上角 ⓘ 按钮 + self.info_btn = tk.Button( + self.root, + text="ⓘ", + font=("Segoe UI", 10), + fg="#2196F3", + bg="#f0f0f0", + bd=0, + cursor="hand2", + command=self._show_info, + ) + self.info_btn.place(relx=1.0, x=-4, y=2, anchor="ne") + + # 中央开关按钮 + self.toggle_btn = tk.Button( + self.root, + text="启用监控", + font=("Segoe UI", 12, "bold"), + bg="#9E9E9E", + fg="white", + bd=0, + padx=20, + pady=6, + cursor="hand2", + command=self._toggle_monitoring, + ) + self.toggle_btn.place(relx=0.5, rely=0.4, anchor="center") + + # 状态文字 + self.status_var = tk.StringVar(value="就绪") + self.status_label = tk.Label( + self.root, + textvariable=self.status_var, + font=("Segoe UI", 9), + fg="#666666", + bg="#f0f0f0", + ) + self.status_label.place(relx=0.5, rely=0.78, anchor="center") + + self.root.protocol("WM_DELETE_WINDOW", self._quit) + + # 启动后立即开始监控循环(无需初始等待) + self.root.after(100, self._main_loop_tick) + self.root.bind("", self._on_window_configure) + + + # ── info dialog ── + def _show_info(self): + info_win = tk.Toplevel(self.root) + info_win.title("已知漏洞/局限性的说明") + info_win.attributes("-topmost", True) + info_win.configure(bg="white") + info_win.geometry("540x800") + + txt = tk.Text( + info_win, + wrap="word", + font=("Microsoft YaHei", 10), + bg="white", + fg="#333333", + bd=0, + padx=15, + pady=8, + ) + txt.pack(fill="both", expand=True) + + txt.tag_config("title", font=("Microsoft YaHei", 11, "bold"), spacing1=5) + txt.tag_config("bold", font=("Microsoft YaHei", 10, "bold"), spacing1=3) + txt.tag_config("normal", font=("Microsoft YaHei", 10), spacing1=3) + + txt.insert("end", "已知漏洞/局限性的说明\n\n", "title") + txt.insert("end", "1. 系统弹窗遮挡", "bold") + txt.insert("end", " — 如Windows更新弹窗、杀毒软件警报等可能盖住微信窗口。最小化操作无法处理它们。\n\n", "normal") + + txt.insert("end", "2. 任务管理器", "bold") + txt.insert("end", " — 无法通过", "normal") + txt.insert("end", " ShowWindow ", "bold") + txt.insert("end", "最小化。当前方案是直接杀进程,但不妥。后续版本可考虑检测到任务管理器时跳过处理。\n\n", "normal") + + txt.insert("end", "3. 多屏场景", "bold") + txt.insert("end", " — 微信可能在非主屏上打开,最小化其他窗口后主屏显示的是桌面,但微信在副屏。截屏裁切主屏中心会错过。\n\n", "normal") + + txt.insert("end", "4. 高优先级窗口", "bold") + txt.insert("end", " — 某些程序(如远程桌面、全屏游戏)无法被最小化。\n\n", "normal") + + txt.insert("end", "5. 程序自身CMD窗口", "bold") + txt.insert("end", " — 如果用 ", "normal") + txt.insert("end", "python", "bold") + txt.insert("end", " 直接运行(不用 ", "normal") + txt.insert("end", "pythonw", "bold") + txt.insert("end", "),CMD窗口会挡住微信。必须用 ", "normal") + txt.insert("end", ".bat", "bold") + txt.insert("end", "(", "normal") + txt.insert("end", "start_wechat_monitor.bat", "bold") + txt.insert("end", ")启动。", "normal") + + txt.configure(state="disabled") + + tk.Button( + info_win, + text="关闭", + font=("Microsoft YaHei", 9), + bg="#f0f0f0", + bd=1, + command=info_win.destroy, + ).pack(pady=(0, 8)) + + # ── toggle ── + def _toggle_monitoring(self): + self.monitoring = not self.monitoring + if self.monitoring: + self.toggle_btn.config(text="停用监控", bg="#4CAF50", fg="white") + self._set_status("监控运行中") + log.info("监控已启用") + else: + self.toggle_btn.config(text="启用监控", bg="#9E9E9E", fg="white") + self._set_status("监控已停用") + # 清理 hash 模式状态 + self.hash_mode = False + self.hash_mode_emails_sent = 0 + self.last_md5 = None + self.qr_sent_count = 0 + self._last_login_hash = None + log.info("监控已停用") + + # ── status ── + def _set_status(self, text): + self.status_var.set(text) + self._update_window_width() + + def _update_window_width(self): + if self._user_resized: + return # 用户已手动调整窗口,不再自动调整 + text = self.status_var.get() + f = tkfont.Font(font=self.status_label.cget("font")) + text_w = f.measure(text) + 40 + if text_w < self._default_width: + text_w = self._default_width + cur_w = self.root.winfo_width() + if abs(cur_w - text_w) > 10: + x = self.root.winfo_x() + y = self.root.winfo_y() + cur_h = self.root.winfo_height() + if cur_h < 130: + cur_h = 130 + self.root.geometry(f"{int(text_w)}x{cur_h}+{x}+{y}") + + # ── quit ── + def _quit(self): + self.running = False + self.root.destroy() + log.info("程序退出") + + # ── main loop (every 5 seconds) ── + def _main_loop_tick(self): + if not self.running: + return + + if self.monitoring: + # 仅在非 hash 模式下执行主检测 + # hash 模式有它自己的独立循环 + if not self.hash_mode: + self._do_detect() + + self.root.after(PAUSE * 1000, self._main_loop_tick) + + def _do_detect(self): + """一次完整的检测流程""" + if not self.monitoring: + return + + # ① 查进程 + wechat_running = False + for proc in psutil.process_iter(["name"]): + try: + if proc.info["name"] in ("WeChat.exe", "Weixin.exe"): + wechat_running = True + break + except (psutil.NoSuchProcess, psutil.AccessDenied): + continue + + if not wechat_running: + self._set_status("微信未运行,启动中...") + log.info("微信未运行,启动中...") + _start_wechat() + return + + # ② 找可见的"微信"窗口 + wechat_windows = _find_wechat_windows() + visible = [w for w in wechat_windows if w.visible] + if not visible: + self._set_status("微信运行中") + return + + # ③ 窗口尺寸判断(零API) + login_window = None + for w in visible: + if w.width < 700 and w.height < 850: + login_window = w + break + + if not login_window: + self._set_status("微信运行中") + if visible: + log.info("微信已登录(主界面 %dx%d)", visible[0].width, visible[0].height) + else: + log.info("微信已登录(主界面,无可见窗口)") + return + + # ④ 最小化其他窗口 + _minimize_all_except_wechat() + time.sleep(0.3) + + # 再次确认窗口尺寸(最小化后可能有窗口挡住微信) + wechat_windows = _find_wechat_windows() + visible = [w for w in wechat_windows if w.visible] + login_window = None + for w in visible: + if w.width < 700 and w.height < 850: + login_window = w + break + + if not login_window: + self._set_status("微信运行中") + if visible: + log.info("最小化后窗口变为主界面,已登录 %dx%d", visible[0].width, visible[0].height) + else: + log.info("最小化后窗口变为主界面,已登录(无可见窗口)") + return + + img = _capture_center() + new_hash = _md5_of_image(img) + + if self._last_login_hash is not None and new_hash == self._last_login_hash: + # 哈希相同,界面没变 + self._set_status("等待扫码登录") + return + + # 哈希不同: ⑤ 重新判断窗口尺寸 + self._last_login_hash = new_hash # 更新哈希 + + # 先查一次窗口尺寸(不用AI) + wechat_windows = _find_wechat_windows() + visible = [w for w in wechat_windows if w.visible] + login_window = None + for w in visible: + if w.width < 700 and w.height < 850: + login_window = w + break + + if not login_window: + # 窗口变大了 → 登录成功 + self._set_status("微信运行中") + log.info("微信已登录(窗口变大)") + if self.hash_mode: + self.hash_mode = False + self._last_login_hash = None + return + + # ⑥ 仍是登录窗口,调AI判断状态 + result = _get_ai_result(img) + + if result == "A": + self._set_status("正在点击登录...") + log.info("检测到'进入微信'按钮,正在点击...") + self.root.withdraw() + time.sleep(0.2) + pyautogui.click(SCREEN_W * 0.5, SCREEN_H * 0.58) + time.sleep(0.5) + self.root.deiconify() + self._set_status("已点击登录") + + elif result == "B": + self.qr_sent_count += 1 + self._set_status(f"请扫码登录 ({self.qr_sent_count}/{MAX_EMAILS})") + log.info("检测到二维码,第 %d 次通知", self.qr_sent_count) + qr_path = os.path.join(tempfile.gettempdir(), f"wechat_qr_{int(time.time())}.png") + img.save(qr_path, "PNG") + send_qr_email(qr_path) + if SMS_PHONE: + send_sms("888888", SMS_PHONE) + self.hash_mode = True + self.hash_mode_emails_sent = 1 + self.last_md5 = _md5_of_image(img) + self._hash_loop_tick() + + elif result == "未知": + self._set_status("微信运行中") + + elif result == "ERROR": + self._set_status("AI检测异常") + + # ── hash monitoring loop (every 5 seconds) ── + def _hash_loop_tick(self): + if not self.monitoring or not self.hash_mode: + return + + if self.hash_mode_emails_sent >= MAX_EMAILS: + self.hash_mode = False + self.qr_sent_count = 0 + self._last_login_hash = None + self._set_status("通知已达上限,继续监控") + log.info("已达最大邮件发送次数,退出哈希监测模式") + return + + # 截屏比较哈希 + _minimize_all_except_wechat() + time.sleep(0.3) + img = _capture_center() + new_md5 = _md5_of_image(img) + + if new_md5 == self.last_md5: + # 哈希不变,不操作 + self.root.after(PAUSE * 1000, self._hash_loop_tick) + return + + # 哈希变了:先检查窗口尺寸判断是否已登录 + log.info("二维码已变化,检查登录状态...") + wechat_windows = _find_wechat_windows() + visible = [w for w in wechat_windows if w.visible] + still_login = False + for w in visible: + if w.width < 700 and w.height < 850: + still_login = True + break + + if not still_login: + # 窗口变大 → 登录成功 + self.hash_mode = False + self._last_login_hash = None + self._set_status("微信运行中") + log.info("微信已登录,退出哈希监测") + return + + # 仍然是登录窗口 → 等2秒再截图确认 → 发通知 + time.sleep(2) + img2 = _capture_center() + qr_path = os.path.join(tempfile.gettempdir(), f"wechat_qr_{int(time.time())}.png") + img2.save(qr_path, "PNG") + send_qr_email(qr_path) + if SMS_PHONE: + self.hash_mode_emails_sent += 1 + send_sms("888888", SMS_PHONE) + self.last_md5 = _md5_of_image(img2) + + self.root.after(PAUSE * 1000, self._hash_loop_tick) + + + def main(): + app = MonitorApp() + app.root.mainloop() + + + if __name__ == "__main__": + try: + main() + except Exception as e: + import traceback as _tb + tb_str = _tb.format_exc() + print(f"\n{'='*60}\n程序异常退出:{e}\n\n{tb_str}\n{'='*60}", flush=True) + input("\n按回车键关闭此窗口...") + +except Exception: + tb = traceback.format_exc() + # 写入日志文件 + for fname in ["wechat_crash.log", "wechat_fatal.log"]: + try: + with open(fname, "w", encoding="utf-8") as f: + f.write(f"模块初始化失败\n{tb}\n") + except Exception: + pass + # Windows 消息弹窗(确保用户能看到错误) + try: + ctypes.windll.user32.MessageBoxW(0, tb, "微信监控 - 启动错误", 0) + except Exception: + pass + sys.exit(1) diff --git a/wechat_auto_login.py b/wechat_auto_login.py new file mode 100644 index 0000000..d043aaa --- /dev/null +++ b/wechat_auto_login.py @@ -0,0 +1,737 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +"""微信自动登录监控程序""" + +import sys, os, ctypes, traceback + +try: + # 切换到脚本所在目录,保证相对路径正确 + os.chdir(os.path.dirname(os.path.abspath(__file__))) +except Exception: + pass # 切换目录失败不影响后续 + +try: + # ===== 正常的模块初始化从这里开始 ===== + + import time + import hashlib + import random + import logging + import configparser + import atexit + import tempfile + import base64 + + # ── logging (先初始化,再导入第三方库,便于排查导入失败) ───────────────── + log_format = "%(asctime)s.%(msecs)03d [%(levelname)s] %(message)s" + logging.basicConfig( + level=logging.INFO, + format=log_format, + datefmt="%Y-%m-%d %H:%M:%S", + handlers=[ + logging.StreamHandler(), + logging.FileHandler("wechat_monitor.log", encoding="utf-8"), + ], + ) + log = logging.getLogger(__name__) + + # ── 第三方库导入 ───────────────────────────────────────────────────────── + try: + import psutil + import pygetwindow as gw + import pyautogui + import win32gui + import win32con + import win32api + from PIL import ImageGrab + import tkinter as tk + from tkinter import font as tkfont + import requests + import smtplib + from email.utils import formataddr + from email.mime.image import MIMEImage + from email.mime.multipart import MIMEMultipart + from email.mime.text import MIMEText + except ImportError as e: + log.critical("导入第三方库失败: %s", e) + sys.exit(1) + + # ── single instance lock ───────────────────────────────────────────────── + _lock_file = os.path.join(tempfile.gettempdir(), "wechat_auto_login.lock") + _lock_fd = None + + + def _acquire_lock(): + global _lock_fd + try: + _lock_fd = os.open(_lock_file, os.O_CREAT | os.O_EXCL | os.O_RDWR) + return True + except FileExistsError: + return False + + + def _release_lock(): + global _lock_fd + if _lock_fd is not None: + os.close(_lock_fd) + try: + os.unlink(_lock_file) + except OSError: + pass + + + # ── 模块初始化(用 try 捕获所有意外错误) ─────────────────────────────── + try: + if not _acquire_lock(): + log.error("已有实例在运行,退出") + sys.exit(1) + atexit.register(_release_lock) + + # ── config ───────────────────────────────────────────────────────────────── + config = configparser.ConfigParser() + config.read( + os.path.join(os.path.dirname(os.path.abspath(__file__)), "wechat_auto_login.ini"), + encoding="utf-8", + ) + EMAIL_ENABLED = config.getboolean("Email", "email_enabled", fallback=True) + SMTP_SERVER = config.get("Email", "smtp_server", fallback="smtp.qq.com") + SMTP_PORT = config.getint("Email", "smtp_port", fallback=465) + USE_SSL = config.getboolean("Email", "use_ssl", fallback=True) + EMAIL_ACCOUNT = config.get("Email", "email_account", fallback="") + EMAIL_PASSWORD = config.get("Email", "email_password", fallback="") + EMAIL_TO = config.get("Email", "email_to", fallback="") + MAX_EMAILS = config.getint("Email", "max_emails", fallback=5) + SMS_PHONE = config.get("Email", "sms_phone", fallback="") + + # ── constants ───────────────────────────────────────────────────────────────── + SCREEN_W = win32api.GetSystemMetrics(0) + SCREEN_H = win32api.GetSystemMetrics(1) + CROP_W = 696 + CROP_H = 905 + CENTER_X, CENTER_Y = SCREEN_W // 2, SCREEN_H // 2 + PAUSE = 5 + except Exception: + tb = traceback.format_exc() + with open("wechat_fatal.log", "w", encoding="utf-8") as f: + f.write(f"模块初始化失败\n{tb}\n") + print(f"FATAL: 模块初始化失败\n{tb}", file=sys.stderr) + sys.exit(1) + + + + def _img_to_b64(path): + with open(path, "rb") as f: + return base64.b64encode(f.read()).decode("utf-8") + + + def _md5_of_image(img): + return hashlib.md5(img.tobytes()).hexdigest() + + + # ── helpers ────────────────────────────────────────────────────────────── + + + def send_sms(code, phone): + """腾讯云短信发送(完整示例)""" + sdkappid = "1401148277" + appkey = "1fa7ae8213f97cb5da5882b207dfa959" + sign = "深圳市健行者户外运动协会" + template_id = 2673158 + rand = random.randint(100000, 999999) + t = int(time.time()) + sig_str = f"appkey={appkey}&random={rand}&time={t}&mobile={phone}" + sig = hashlib.sha256(sig_str.encode()).hexdigest() + url = f"https://yun.tim.qq.com/v5/tlssmssvr/sendsms?sdkappid={sdkappid}&random={rand}" + body = { + "ext": "", + "extend": "", + "params": [code, "5"], + "sig": sig, + "sign": sign, + "tel": {"mobile": phone, "nationcode": "86"}, + "time": t, + "tpl_id": template_id, + } + try: + resp = requests.post(url, json=body, timeout=10) + return resp.status_code, resp.text + except Exception as e: + log.error("短信发送失败: %s", e) + return -1, str(e) + + + def send_qr_email(qr_img_path): + """发送带二维码附件的邮件""" + if not EMAIL_ENABLED or not EMAIL_ACCOUNT: + log.warning("邮件未启用或账号未配置,跳过邮件发送") + return + try: + with open(qr_img_path, "rb") as f: + img_data = f.read() + + msg = MIMEMultipart("related") + msg["From"] = formataddr(("微信助手", EMAIL_ACCOUNT)) + msg["To"] = EMAIL_TO + msg["Subject"] = "微信登录二维码" + + # HTML 正文嵌入二维码(cid) + html = '' + msg.attach(MIMEText(html, "html")) + + # 附件 + img_attach = MIMEImage(img_data, _subtype="png") + img_attach.add_header("Content-Disposition", "attachment", filename="qrcode.png") + msg.attach(img_attach) + + # 正文内嵌图片 + img_inline = MIMEImage(img_data, _subtype="png") + img_inline.add_header("Content-ID", "") + msg.attach(img_inline) + + if USE_SSL: + s = smtplib.SMTP_SSL(SMTP_SERVER, SMTP_PORT, timeout=10) + else: + s = smtplib.SMTP(SMTP_SERVER, SMTP_PORT, timeout=10) + s.starttls() + s.login(EMAIL_ACCOUNT, EMAIL_PASSWORD) + s.send_message(msg) + s.quit() + log.info("二维码邮件已发送至 %s", EMAIL_TO) + except Exception as e: + log.error("邮件发送失败: %s", e) + + + def _find_wechat_windows(): + """找到标题严格等于'微信'的可见窗口列表(排除包含'微信'的其他窗口)""" + return [w for w in gw.getAllWindows() if w.title == "微信"] + + + def _minimize_all_except_wechat(): + """最小化除微信外的所有可见窗口""" + # 获取微信的窗口句柄 + wechat_hwnds = set() + try: + w = _find_wechat_windows() + for win in w: + wechat_hwnds.add(win._hWnd) + except Exception: + pass + + def enum_callback(hwnd, _): + if not win32gui.IsWindowVisible(hwnd): + return + # 用句柄排除微信窗口 + if hwnd in wechat_hwnds: + return + title = win32gui.GetWindowText(hwnd) + # 排除监控程序自身 + if title == "微信监控": + return + if not title: + return + if title in ("任务管理器", "Task Manager"): + return + try: + win32gui.ShowWindow(hwnd, win32con.SW_SHOWMINIMIZED) + except Exception: + pass + + win32gui.EnumWindows(enum_callback, None) + + + def _capture_center(width=CROP_W, height=CROP_H): + """截取屏幕中心 width×height 并返回 PIL Image""" + log.debug("截取屏幕中心 %dx%d", width, height) + left = CENTER_X - width // 2 + top = CENTER_Y - height // 2 + img = ImageGrab.grab(bbox=(left, top, left + width, top + height)) + log.debug("截图完成: %s", img.size) + return img + + + def _get_ai_result(img): + """ + 使用通义千问视觉模型分析截图。 + 返回: "A" | "B" | "未知" | "ERROR" + AI返回"未知"表示正常工作但没看到A/B。 + AI返回"ERROR"表示AI代码出异常。 + """ + temp_path = os.path.join(tempfile.gettempdir(), "wechat_screen_temp.png") + img.save(temp_path, "PNG") + + prompt = ( + '分析以下微信登录界面截图,请从以下两个状态中返回一个字母:\n' + 'A = 截图中有绿色的"进入微信"按钮\n' + 'B = 截图中有二维码图案并带有"扫码登录"文字\n' + '如果以上都不是,返回"未知"\n' + '请只返回一个字母或"未知"' + ) + + try: + b64 = _img_to_b64(temp_path) + resp = requests.post( + "https://dashscope.aliyuncs.com/compatible-mode/v1/chat/completions", + headers={ + "Authorization": "Bearer sk-a6b3e24b985b41768aa3b0df2f688c98", + "Content-Type": "application/json", + }, + json={ + "model": "qwen-vl-plus", + "messages": [{"role": "user", "content": [ + {"type": "text", "text": prompt}, + {"type": "image_url", "image_url": {"url": f"data:image/png;base64,{b64}"}} + ]}], + "max_tokens": 50, + "temperature": 0, + }, + timeout=15, + ) + data = resp.json() + result = data["choices"][0]["message"]["content"].strip() + except Exception as e: + log.error("AI 分析失败: %s", e) + result = "ERROR" + + try: + os.unlink(temp_path) + except OSError: + pass + + log.info("AI 识别结果: %s", result) + return result + + + def _start_wechat(): + """启动微信""" + log.info("正在启动微信...") + candidates = [ + os.path.expandvars(r"%ProgramFiles(x86)%\Tencent\WeChat\WeChat.exe"), + os.path.expandvars(r"%ProgramFiles%\Tencent\WeChat\WeChat.exe"), + os.path.expandvars(r"%USERPROFILE%\AppData\Local\Tencent\WeChat\WeChat.exe"), + os.path.expandvars(r"%ProgramFiles(x86)%\Tencent\Weixin\Weixin.exe"), + os.path.expandvars(r"%ProgramFiles%\Tencent\Weixin\Weixin.exe"), + ] + wechat_path = None + for p in candidates: + if os.path.exists(p): + wechat_path = p + break + if wechat_path: + try: + os.startfile(wechat_path) + log.info("已启动微信: %s", wechat_path) + except Exception as e: + log.error("启动微信失败: %s", e) + else: + log.error("未找到微信可执行文件") + + + # ── UI ─────────────────────────────────────────────────────────────────── + + + class MonitorApp: + def __init__(self): + self.running = True + self.monitoring = False + + # hash 监测模式状态 + self.hash_mode = False + self.hash_mode_emails_sent = 0 + self.last_md5 = None + self.qr_sent_count = 0 + self._last_login_hash = None + self._user_resized = False + + self._setup_ui() + + def _on_window_configure(self, event): + """当用户手动调整窗口大小或移动窗口时触发""" + self._user_resized = True + + def _setup_ui(self): + self.root = tk.Tk() + self.root.title("微信监控") + self.root.attributes("-topmost", True) + self.root.configure(bg="#f0f0f0") + + # 初始宽高 + self._default_width = 400 + self.root.geometry(f"{self._default_width}x250+0+0") + + # 右上角 ⓘ 按钮 + self.info_btn = tk.Button( + self.root, + text="ⓘ", + font=("Segoe UI", 10), + fg="#2196F3", + bg="#f0f0f0", + bd=0, + cursor="hand2", + command=self._show_info, + ) + self.info_btn.place(relx=1.0, x=-4, y=2, anchor="ne") + + # 中央开关按钮 + self.toggle_btn = tk.Button( + self.root, + text="启用监控", + font=("Segoe UI", 12, "bold"), + bg="#9E9E9E", + fg="white", + bd=0, + padx=20, + pady=6, + cursor="hand2", + command=self._toggle_monitoring, + ) + self.toggle_btn.place(relx=0.5, rely=0.4, anchor="center") + + # 状态文字 + self.status_var = tk.StringVar(value="就绪") + self.status_label = tk.Label( + self.root, + textvariable=self.status_var, + font=("Segoe UI", 9), + fg="#666666", + bg="#f0f0f0", + ) + self.status_label.place(relx=0.5, rely=0.78, anchor="center") + + self.root.protocol("WM_DELETE_WINDOW", self._quit) + + # 启动后立即开始监控循环(无需初始等待) + self.root.after(100, self._main_loop_tick) + self.root.bind("", self._on_window_configure) + + + # ── info dialog ── + def _show_info(self): + info_win = tk.Toplevel(self.root) + info_win.title("已知漏洞/局限性的说明") + info_win.attributes("-topmost", True) + info_win.configure(bg="white") + info_win.geometry("540x800") + + txt = tk.Text( + info_win, + wrap="word", + font=("Microsoft YaHei", 10), + bg="white", + fg="#333333", + bd=0, + padx=15, + pady=8, + ) + txt.pack(fill="both", expand=True) + + txt.tag_config("title", font=("Microsoft YaHei", 11, "bold"), spacing1=5) + txt.tag_config("bold", font=("Microsoft YaHei", 10, "bold"), spacing1=3) + txt.tag_config("normal", font=("Microsoft YaHei", 10), spacing1=3) + + txt.insert("end", "已知漏洞/局限性的说明\n\n", "title") + txt.insert("end", "1. 系统弹窗遮挡", "bold") + txt.insert("end", " — 如Windows更新弹窗、杀毒软件警报等可能盖住微信窗口。最小化操作无法处理它们。\n\n", "normal") + + txt.insert("end", "2. 任务管理器", "bold") + txt.insert("end", " — 无法通过", "normal") + txt.insert("end", " ShowWindow ", "bold") + txt.insert("end", "最小化。当前方案是直接杀进程,但不妥。后续版本可考虑检测到任务管理器时跳过处理。\n\n", "normal") + + txt.insert("end", "3. 多屏场景", "bold") + txt.insert("end", " — 微信可能在非主屏上打开,最小化其他窗口后主屏显示的是桌面,但微信在副屏。截屏裁切主屏中心会错过。\n\n", "normal") + + txt.insert("end", "4. 高优先级窗口", "bold") + txt.insert("end", " — 某些程序(如远程桌面、全屏游戏)无法被最小化。\n\n", "normal") + + txt.insert("end", "5. 程序自身CMD窗口", "bold") + txt.insert("end", " — 如果用 ", "normal") + txt.insert("end", "python", "bold") + txt.insert("end", " 直接运行(不用 ", "normal") + txt.insert("end", "pythonw", "bold") + txt.insert("end", "),CMD窗口会挡住微信。必须用 ", "normal") + txt.insert("end", ".bat", "bold") + txt.insert("end", "(", "normal") + txt.insert("end", "start_wechat_monitor.bat", "bold") + txt.insert("end", ")启动。", "normal") + + txt.configure(state="disabled") + + tk.Button( + info_win, + text="关闭", + font=("Microsoft YaHei", 9), + bg="#f0f0f0", + bd=1, + command=info_win.destroy, + ).pack(pady=(0, 8)) + + # ── toggle ── + def _toggle_monitoring(self): + self.monitoring = not self.monitoring + if self.monitoring: + self.toggle_btn.config(text="停用监控", bg="#4CAF50", fg="white") + self._set_status("监控运行中") + log.info("监控已启用") + else: + self.toggle_btn.config(text="启用监控", bg="#9E9E9E", fg="white") + self._set_status("监控已停用") + # 清理 hash 模式状态 + self.hash_mode = False + self.hash_mode_emails_sent = 0 + self.last_md5 = None + self.qr_sent_count = 0 + self._last_login_hash = None + log.info("监控已停用") + + # ── status ── + def _set_status(self, text): + self.status_var.set(text) + self._update_window_width() + + def _update_window_width(self): + if self._user_resized: + return # 用户已手动调整窗口,不再自动调整 + text = self.status_var.get() + f = tkfont.Font(font=self.status_label.cget("font")) + text_w = f.measure(text) + 40 + if text_w < self._default_width: + text_w = self._default_width + cur_w = self.root.winfo_width() + if abs(cur_w - text_w) > 10: + x = self.root.winfo_x() + y = self.root.winfo_y() + cur_h = self.root.winfo_height() + if cur_h < 130: + cur_h = 130 + self.root.geometry(f"{int(text_w)}x{cur_h}+{x}+{y}") + + # ── quit ── + def _quit(self): + self.running = False + self.root.destroy() + log.info("程序退出") + + # ── main loop (every 5 seconds) ── + def _main_loop_tick(self): + if not self.running: + return + + if self.monitoring: + # 仅在非 hash 模式下执行主检测 + # hash 模式有它自己的独立循环 + if not self.hash_mode: + self._do_detect() + + self.root.after(PAUSE * 1000, self._main_loop_tick) + + def _do_detect(self): + """一次完整的检测流程""" + if not self.monitoring: + return + + # ① 查进程 + wechat_running = False + for proc in psutil.process_iter(["name"]): + try: + if proc.info["name"] in ("WeChat.exe", "Weixin.exe"): + wechat_running = True + break + except (psutil.NoSuchProcess, psutil.AccessDenied): + continue + + if not wechat_running: + self._set_status("微信未运行,启动中...") + log.info("微信未运行,启动中...") + _start_wechat() + return + + # ② 找可见的"微信"窗口 + wechat_windows = _find_wechat_windows() + visible = [w for w in wechat_windows if w.visible] + if not visible: + self._set_status("微信运行中") + return + + # ③ 窗口尺寸判断(零API) + login_window = None + for w in visible: + if w.width < 700 and w.height < 850: + login_window = w + break + + if not login_window: + self._set_status("微信运行中") + if visible: + log.info("微信已登录(主界面 %dx%d)", visible[0].width, visible[0].height) + else: + log.info("微信已登录(主界面,无可见窗口)") + return + + # ④ 最小化其他窗口 + _minimize_all_except_wechat() + time.sleep(0.3) + + # 再次确认窗口尺寸(最小化后可能有窗口挡住微信) + wechat_windows = _find_wechat_windows() + visible = [w for w in wechat_windows if w.visible] + login_window = None + for w in visible: + if w.width < 700 and w.height < 850: + login_window = w + break + + if not login_window: + self._set_status("微信运行中") + if visible: + log.info("最小化后窗口变为主界面,已登录 %dx%d", visible[0].width, visible[0].height) + else: + log.info("最小化后窗口变为主界面,已登录(无可见窗口)") + return + + img = _capture_center() + new_hash = _md5_of_image(img) + + if self._last_login_hash is not None and new_hash == self._last_login_hash: + # 哈希相同,界面没变 + self._set_status("等待扫码登录") + return + + # 哈希不同: ⑤ 重新判断窗口尺寸 + self._last_login_hash = new_hash # 更新哈希 + + # 先查一次窗口尺寸(不用AI) + wechat_windows = _find_wechat_windows() + visible = [w for w in wechat_windows if w.visible] + login_window = None + for w in visible: + if w.width < 700 and w.height < 850: + login_window = w + break + + if not login_window: + # 窗口变大了 → 登录成功 + self._set_status("微信运行中") + log.info("微信已登录(窗口变大)") + if self.hash_mode: + self.hash_mode = False + self._last_login_hash = None + return + + # ⑥ 仍是登录窗口,调AI判断状态 + result = _get_ai_result(img) + + if result == "A": + self._set_status("正在点击登录...") + log.info("检测到'进入微信'按钮,正在点击...") + self.root.withdraw() + time.sleep(0.2) + pyautogui.click(SCREEN_W * 0.5, SCREEN_H * 0.58) + time.sleep(0.5) + self.root.deiconify() + self._set_status("已点击登录") + + elif result == "B": + self.qr_sent_count += 1 + self._set_status(f"请扫码登录 ({self.qr_sent_count}/{MAX_EMAILS})") + log.info("检测到二维码,第 %d 次通知", self.qr_sent_count) + qr_path = os.path.join(tempfile.gettempdir(), f"wechat_qr_{int(time.time())}.png") + img.save(qr_path, "PNG") + send_qr_email(qr_path) + if SMS_PHONE: + send_sms("888888", SMS_PHONE) + self.hash_mode = True + self.hash_mode_emails_sent = 1 + self.last_md5 = _md5_of_image(img) + self._hash_loop_tick() + + elif result == "未知": + self._set_status("微信运行中") + + elif result == "ERROR": + self._set_status("AI检测异常") + + # ── hash monitoring loop (every 5 seconds) ── + def _hash_loop_tick(self): + if not self.monitoring or not self.hash_mode: + return + + if self.hash_mode_emails_sent >= MAX_EMAILS: + self.hash_mode = False + self.qr_sent_count = 0 + self._last_login_hash = None + self._set_status("通知已达上限,继续监控") + log.info("已达最大邮件发送次数,退出哈希监测模式") + return + + # 截屏比较哈希 + _minimize_all_except_wechat() + time.sleep(0.3) + img = _capture_center() + new_md5 = _md5_of_image(img) + + if new_md5 == self.last_md5: + # 哈希不变,不操作 + self.root.after(PAUSE * 1000, self._hash_loop_tick) + return + + # 哈希变了:先检查窗口尺寸判断是否已登录 + log.info("二维码已变化,检查登录状态...") + wechat_windows = _find_wechat_windows() + visible = [w for w in wechat_windows if w.visible] + still_login = False + for w in visible: + if w.width < 700 and w.height < 850: + still_login = True + break + + if not still_login: + # 窗口变大 → 登录成功 + self.hash_mode = False + self._last_login_hash = None + self._set_status("微信运行中") + log.info("微信已登录,退出哈希监测") + return + + # 仍然是登录窗口 → 等2秒再截图确认 → 发通知 + time.sleep(2) + img2 = _capture_center() + qr_path = os.path.join(tempfile.gettempdir(), f"wechat_qr_{int(time.time())}.png") + img2.save(qr_path, "PNG") + send_qr_email(qr_path) + if SMS_PHONE: + self.hash_mode_emails_sent += 1 + send_sms("888888", SMS_PHONE) + self.last_md5 = _md5_of_image(img2) + + self.root.after(PAUSE * 1000, self._hash_loop_tick) + + + def main(): + app = MonitorApp() + app.root.mainloop() + + + if __name__ == "__main__": + try: + main() + except Exception as e: + import traceback as _tb + tb_str = _tb.format_exc() + print(f"\n{'='*60}\n程序异常退出:{e}\n\n{tb_str}\n{'='*60}", flush=True) + input("\n按回车键关闭此窗口...") + +except Exception: + tb = traceback.format_exc() + # 写入日志文件 + for fname in ["wechat_crash.log", "wechat_fatal.log"]: + try: + with open(fname, "w", encoding="utf-8") as f: + f.write(f"模块初始化失败\n{tb}\n") + except Exception: + pass + # Windows 消息弹窗(确保用户能看到错误) + try: + ctypes.windll.user32.MessageBoxW(0, tb, "微信监控 - 启动错误", 0) + except Exception: + pass + sys.exit(1) diff --git "a/\345\276\256\344\277\241\350\207\252\345\212\250\347\231\273\345\275\225\347\233\221\346\216\247\347\250\213\345\272\217/mouse_pos.py" "b/\345\276\256\344\277\241\350\207\252\345\212\250\347\231\273\345\275\225\347\233\221\346\216\247\347\250\213\345\272\217/mouse_pos.py" new file mode 100644 index 0000000..2b17cb7 --- /dev/null +++ "b/\345\276\256\344\277\241\350\207\252\345\212\250\347\231\273\345\275\225\347\233\221\346\216\247\347\250\213\345\272\217/mouse_pos.py" @@ -0,0 +1,20 @@ +import ctypes, sys, time + +# 定义POINT结构 +class POINT(ctypes.Structure): + _fields_ = [("x", ctypes.c_long), ("y", ctypes.c_long)] + +user32 = ctypes.windll.user32 + +print("=== 鼠标位置实时显示 ===") +print("移动鼠标查看坐标,关闭窗口退出\n") + +try: + while True: + pt = POINT() + user32.GetCursorPos(ctypes.byref(pt)) + sys.stdout.write(f"\rX:{pt.x:4d} Y:{pt.y:4d}") + sys.stdout.flush() + time.sleep(0.1) +except KeyboardInterrupt: + print("\n\n已退出") diff --git "a/\345\276\256\344\277\241\350\207\252\345\212\250\347\231\273\345\275\225\347\233\221\346\216\247\347\250\213\345\272\217/pause_monitor.bat" "b/\345\276\256\344\277\241\350\207\252\345\212\250\347\231\273\345\275\225\347\233\221\346\216\247\347\250\213\345\272\217/pause_monitor.bat" new file mode 100644 index 0000000..c18901e --- /dev/null +++ "b/\345\276\256\344\277\241\350\207\252\345\212\250\347\231\273\345\275\225\347\233\221\346\216\247\347\250\213\345\272\217/pause_monitor.bat" @@ -0,0 +1,4 @@ +@echo off +chcp 936 >nul +echo stop > "%~dp0stop.monitor" +echo Monitor paused. Delete stop.monitor to resume. diff --git "a/\345\276\256\344\277\241\350\207\252\345\212\250\347\231\273\345\275\225\347\233\221\346\216\247\347\250\213\345\272\217/requirements.txt" "b/\345\276\256\344\277\241\350\207\252\345\212\250\347\231\273\345\275\225\347\233\221\346\216\247\347\250\213\345\272\217/requirements.txt" new file mode 100644 index 0000000..355ecbb --- /dev/null +++ "b/\345\276\256\344\277\241\350\207\252\345\212\250\347\231\273\345\275\225\347\233\221\346\216\247\347\250\213\345\272\217/requirements.txt" @@ -0,0 +1,6 @@ +psutil +pygetwindow +pyautogui +pywin32 +Pillow +requests diff --git "a/\345\276\256\344\277\241\350\207\252\345\212\250\347\231\273\345\275\225\347\233\221\346\216\247\347\250\213\345\272\217/resume_monitor.bat" "b/\345\276\256\344\277\241\350\207\252\345\212\250\347\231\273\345\275\225\347\233\221\346\216\247\347\250\213\345\272\217/resume_monitor.bat" new file mode 100644 index 0000000..80c2ba8 --- /dev/null +++ "b/\345\276\256\344\277\241\350\207\252\345\212\250\347\231\273\345\275\225\347\233\221\346\216\247\347\250\213\345\272\217/resume_monitor.bat" @@ -0,0 +1,4 @@ +@echo off +chcp 936 >nul +del /f /q "%~dp0stop.monitor" 2>nul +echo Monitor resumed. diff --git "a/\345\276\256\344\277\241\350\207\252\345\212\250\347\231\273\345\275\225\347\233\221\346\216\247\347\250\213\345\272\217/setup_wechat_monitor.bat" "b/\345\276\256\344\277\241\350\207\252\345\212\250\347\231\273\345\275\225\347\233\221\346\216\247\347\250\213\345\272\217/setup_wechat_monitor.bat" new file mode 100644 index 0000000..df345bc --- /dev/null +++ "b/\345\276\256\344\277\241\350\207\252\345\212\250\347\231\273\345\275\225\347\233\221\346\216\247\347\250\213\345\272\217/setup_wechat_monitor.bat" @@ -0,0 +1,30 @@ +@echo off +chcp 936 >nul +title WeChat Monitor Setup + +python --version >nul 2>&1 +if %errorlevel% neq 0 ( + echo [1/4] Downloading Python ... + powershell -Command "& {Invoke-WebRequest -Uri 'https://www.python.org/ftp/python/3.12.9/python-3.12.9-amd64.exe' -OutFile '%TEMP%\python-3.12.exe' -UseBasicParsing}" + start /wait %TEMP%\python-3.12.exe /quiet InstallAllUsers=1 PrependPath=1 Include_test=0 +) + +echo [2/4] Installing dependencies ... +python -m pip install requests psutil pyautogui pillow --quiet + +if exist "%~dp0wechat_auto_login.py" ( + copy /Y "%~dp0wechat_auto_login.py" "%USERPROFILE%\Desktop\" >nul +) else ( + powershell -Command "& {Invoke-WebRequest -Uri 'https://szjxzxh.cn/wechat_auto_login_v3.py?_t=ready' -OutFile '%USERPROFILE%\Desktop\wechat_auto_login.py' -UseBasicParsing}" +) + +echo @echo off > "%USERPROFILE%\Desktop\start_wechat_monitor.bat" +echo pythonw "%%USERPROFILE%%\Desktop\wechat_auto_login.py" >> "%USERPROFILE%\Desktop\start_wechat_monitor.bat" + +set STARTUP_DIR=%APPDATA%\Microsoft\Windows\Start Menu\Programs\Startup +copy /Y "%USERPROFILE%\Desktop\start_wechat_monitor.bat" "%STARTUP_DIR%\" >nul + +echo. +echo Setup Complete! +echo Double-click Desktop\start_wechat_monitor.bat to run +pause diff --git "a/\345\276\256\344\277\241\350\207\252\345\212\250\347\231\273\345\275\225\347\233\221\346\216\247\347\250\213\345\272\217/start_wechat_monitor.bat" "b/\345\276\256\344\277\241\350\207\252\345\212\250\347\231\273\345\275\225\347\233\221\346\216\247\347\250\213\345\272\217/start_wechat_monitor.bat" new file mode 100644 index 0000000..e88b8a9 --- /dev/null +++ "b/\345\276\256\344\277\241\350\207\252\345\212\250\347\231\273\345\275\225\347\233\221\346\216\247\347\250\213\345\272\217/start_wechat_monitor.bat" @@ -0,0 +1,4 @@ +@echo off +chcp 936 >nul +start "" "%~dp0wechat_auto_login_v3.py" +exit diff --git "a/\345\276\256\344\277\241\350\207\252\345\212\250\347\231\273\345\275\225\347\233\221\346\216\247\347\250\213\345\272\217/stop_monitor.bat" "b/\345\276\256\344\277\241\350\207\252\345\212\250\347\231\273\345\275\225\347\233\221\346\216\247\347\250\213\345\272\217/stop_monitor.bat" new file mode 100644 index 0000000..6eabdf3 --- /dev/null +++ "b/\345\276\256\344\277\241\350\207\252\345\212\250\347\231\273\345\275\225\347\233\221\346\216\247\347\250\213\345\272\217/stop_monitor.bat" @@ -0,0 +1,4 @@ +@echo off +chcp 936 >nul +taskkill /f /im pythonw.exe 2>nul +echo Monitor stopped. diff --git "a/\345\276\256\344\277\241\350\207\252\345\212\250\347\231\273\345\275\225\347\233\221\346\216\247\347\250\213\345\272\217/test_click_accuracy_v2.py" "b/\345\276\256\344\277\241\350\207\252\345\212\250\347\231\273\345\275\225\347\233\221\346\216\247\347\250\213\345\272\217/test_click_accuracy_v2.py" new file mode 100644 index 0000000..cb9d1c8 --- /dev/null +++ "b/\345\276\256\344\277\241\350\207\252\345\212\250\347\231\273\345\275\225\347\233\221\346\216\247\347\250\213\345\272\217/test_click_accuracy_v2.py" @@ -0,0 +1,276 @@ +#!/usr/bin/env python3 +""" +测试"进入微信"按钮坐标定位准确性的工具。 +使用通义千问VL视觉模型识别微信登录状态,测试坐标点击准确性。 +""" + +import base64 +import ctypes +import io +import os +import sys +import time +import tkinter as tk + +import pyautogui +import requests +from PIL import Image + + +# ── API 配置 ──────────────────────────────────────────────────── +API_URL = "https://dashscope.aliyuncs.com/compatible-mode/v1/chat/completions" +API_KEY = "sk-a6b3e24b985b41768aa3b0df2f688c98" +MODEL = "qwen-vl-plus" + +# 裁切尺寸(送AI用) +CROP_W = 696 +CROP_H = 905 + +# ── 工具函数 ──────────────────────────────────────────────────── + + +class RECT(ctypes.Structure): + _fields_ = [ + ("left", ctypes.c_long), + ("top", ctypes.c_long), + ("right", ctypes.c_long), + ("bottom", ctypes.c_long), + ] + + +def get_workarea(): + """获取工作区尺寸(桌面排除任务栏),物理像素""" + r = RECT() + ctypes.windll.user32.SystemParametersInfoW(0x0030, 0, ctypes.byref(r), 0) + work_w = r.right - r.left + work_h = r.bottom - r.top + return work_w, work_h + + +def capture_and_crop_bytes(work_w, work_h): + """ + 1. pyautogui 全屏截图 + 2. 以工作区中心裁切 900×754 + 返回裁切后的 PNG bytes + """ + # 全屏截图(pyautogui) + screenshot = pyautogui.screenshot() + + # 获取工作区左上角偏移 + r = RECT() + ctypes.windll.user32.SystemParametersInfoW(0x0030, 0, ctypes.byref(r), 0) + offset_x = r.left + offset_y = r.top + + # 工作区中心 + center_x = offset_x + (work_w // 2) + center_y = offset_y + (work_h // 2) + + # 裁切区域 + left = center_x - CROP_W // 2 + top = center_y - CROP_H // 2 + right = left + CROP_W + bottom = top + CROP_H + + # 边界保护 + left = max(0, left) + top = max(0, top) + right = min(screenshot.width, right) + bottom = min(screenshot.height, bottom) + + crop = screenshot.crop((left, top, right, bottom)) + + # 保存裁切图到文件 + save_path = r"D:\_WeChatAutoRestart\wechat_crop.png" + os.makedirs(os.path.dirname(save_path), exist_ok=True) + crop.save(save_path, format="PNG") + print(f"[裁切] {crop.width}×{crop.height} → 已保存到 {save_path}") + + buf = io.BytesIO() + crop.save(buf, format="PNG") + img_bytes = buf.getvalue() + print(f"[裁切] {crop.width}×{crop.height} (内存)") + return img_bytes + + +def encode_image_bytes(img_bytes): + """将图片 bytes 编码为 base64""" + return base64.b64encode(img_bytes).decode("utf-8") + + +def query_ai(image_bytes): + """ + 送通义千问VL API 判断状态。 + 返回状态字母 A/B/C/D + """ + base64_image = encode_image_bytes(image_bytes) + + payload = { + "model": MODEL, + "messages": [ + { + "role": "user", + "content": [ + { + "type": "image_url", + "image_url": {"url": f"data:image/png;base64,{base64_image}"}, + }, + { + "type": "text", + "text": ( + "分析这个截图中的微信状态。只回复一个字母:\n" + "A = 登录界面有\"进入微信\"按钮\n" + "B = 二维码\n" + "C = 已登录\n" + "D = 其他" + ), + }, + ], + } + ], + } + + headers = { + "Authorization": f"Bearer {API_KEY}", + "Content-Type": "application/json", + } + + try: + resp = requests.post(API_URL, json=payload, headers=headers, timeout=30) + resp.raise_for_status() + data = resp.json() + content = data["choices"][0]["message"]["content"].strip().upper() + + # 只取第一个字母 + for ch in content: + if ch in "ABCD": + return ch + print(f"[AI] 返回异常内容: {content}") + return "D" + except Exception as e: + print(f"[AI] API 调用失败: {e}") + return "D" + + +def show_crosshair_and_confirm(button_x, button_y): + """ + 在屏幕 (button_x, button_y) 位置画红色十字线, + 在旁边显示"确认"按钮。 + - 透明置顶窗口,覆盖全屏 + - 红色十字线(横线和竖线各约40px长) + - 点确认 → 退出程序 + """ + root = tk.Tk() + root.title("坐标验证 - 十字标记") + root.overrideredirect(True) + + # 获取屏幕逻辑尺寸,用 tkinter 的全屏 + screen_w = root.winfo_screenwidth() + screen_h = root.winfo_screenheight() + root.geometry(f"{screen_w}x{screen_h}+0+0") + root.attributes("-topmost", True) + root.attributes("-transparentcolor", "white") + root.configure(bg="white") + + # Canvas 覆盖全窗口 + canvas = tk.Canvas(root, width=screen_w, height=screen_h, + bg="white", highlightthickness=0) + canvas.pack() + + # ── 红色十字线 ── + cross_len = 20 # 单臂长度,总长约40px + # 竖线 + canvas.create_line(button_x, button_y - cross_len, + button_x, button_y + cross_len, + fill="red", width=3) + # 横线 + canvas.create_line(button_x - cross_len, button_y, + button_x + cross_len, button_y, + fill="red", width=3) + + # ── "确认"按钮(在十字标记右下方) ── + btn_frame = tk.Frame(root, bg="white") + btn_frame.place(x=button_x + 30, y=button_y + 30) + + def on_confirm(): + print(f"[确认] 坐标正确!点击位置: X={button_x}, Y={button_y}") + root.destroy() + + btn = tk.Button( + btn_frame, + text="确认", + command=on_confirm, + font=("Arial", 12, "bold"), + bg="lightgreen", + width=8, + height=1, + ) + btn.pack() + + # ── 键盘快捷键 ── + def on_key(event): + if event.keysym == "Return": + on_confirm() + elif event.keysym == "Escape": + print("[退出] 用户取消") + root.destroy() + + root.bind("", on_key) + root.focus_set() + + root.mainloop() + + +# ── 主逻辑 ────────────────────────────────────────────────────── + +def main(): + print("=" * 50) + print("微信登录按钮坐标准确性测试工具") + print("按 Ctrl+C 随时退出") + print("=" * 50) + + while True: + try: + # 1. 获取尺寸 + # 鼠标空间(逻辑坐标) + mouse_w, mouse_h = pyautogui.size() + # 工作区(物理像素,用于裁切区域参考) + work_w, work_h = get_workarea() + print(f"\n[鼠标空间] {mouse_w}×{mouse_h} [工作区] {work_w}×{work_h}") + + # 2. 截图并裁切(送AI) + crop_bytes = capture_and_crop_bytes(work_w, work_h) + + # 3. AI判断状态 + status = query_ai(crop_bytes) + print(f"[AI状态] {status}") + + if status == "A": + # 4. 按钮坐标(用鼠标空间计算!) + button_x = int(mouse_w * 0.50) + button_y = int(mouse_h * 0.58) + print(f"[按钮坐标] X={button_x}, Y={button_y} (鼠标空间)") + + # 5+6. 显示红色十字标记 + 确认按钮 + show_crosshair_and_confirm(button_x, button_y) + + # 确认后退出 + print("[完成] 验证完成,退出程序。") + sys.exit(0) + else: + status_names = {"B": "二维码", "C": "已登录", "D": "其他"} + print(f"[状态] 当前状态: {status_names.get(status, '未知')},5秒后重新检测...") + time.sleep(5) + continue + + except KeyboardInterrupt: + print("\n[退出] 用户中断") + break + except Exception as e: + print(f"[错误] {e}") + print("[重试] 5 秒后重新检测...") + time.sleep(5) + + +if __name__ == "__main__": + main() diff --git "a/\345\276\256\344\277\241\350\207\252\345\212\250\347\231\273\345\275\225\347\233\221\346\216\247\347\250\213\345\272\217/test_email.py" "b/\345\276\256\344\277\241\350\207\252\345\212\250\347\231\273\345\275\225\347\233\221\346\216\247\347\250\213\345\272\217/test_email.py" new file mode 100644 index 0000000..f77e3b7 --- /dev/null +++ "b/\345\276\256\344\277\241\350\207\252\345\212\250\347\231\273\345\275\225\347\233\221\346\216\247\347\250\213\345\272\217/test_email.py" @@ -0,0 +1,49 @@ +import smtplib, configparser, os +from pathlib import Path +from email.mime.text import MIMEText + +# 找ini +ini_path = Path(__file__).parent / "wechat_auto_login.ini" +if not ini_path.exists(): + ini_path = Path.home() / "Desktop" / "wechat_auto_login.ini" + +config = configparser.ConfigParser() +config.read(ini_path) + +cfg = config["Email"] +smtp_server = cfg.get("smtp_server", "smtp.qq.com") +smtp_port = cfg.getint("smtp_port", 465) +use_ssl = cfg.getboolean("use_ssl", True) +account = cfg.get("email_account", "") +password = cfg.get("email_password", "") +to = cfg.get("email_to", account) + +print(f"发件: {account}") +print(f"收件: {to}") +print(f"服务器: {smtp_server}:{smtp_port}") +print() + +if not account or not password: + print("配置不完整,请检查 ini 文件") +else: + try: + msg = MIMEText("微信监控邮件发送测试成功", "plain", "utf-8") + msg["Subject"] = "微信监控测试" + msg["From"] = account + msg["To"] = to + + if use_ssl: + s = smtplib.SMTP_SSL(smtp_server, smtp_port, timeout=10) + else: + s = smtplib.SMTP(smtp_server, smtp_port, timeout=10) + s.starttls() + s.login(account, password) + s.send_message(msg) + s.quit() + print("邮件发送成功!") + except smtplib.SMTPAuthenticationError: + print("认证失败,请检查授权码") + except Exception as e: + print(f"发送失败: {e}") + +input("按回车键退出...") diff --git "a/\345\276\256\344\277\241\350\207\252\345\212\250\347\231\273\345\275\225\347\233\221\346\216\247\347\250\213\345\272\217/test_minimize.py" "b/\345\276\256\344\277\241\350\207\252\345\212\250\347\231\273\345\275\225\347\233\221\346\216\247\347\250\213\345\272\217/test_minimize.py" new file mode 100644 index 0000000..57700a6 --- /dev/null +++ "b/\345\276\256\344\277\241\350\207\252\345\212\250\347\231\273\345\275\225\347\233\221\346\216\247\347\250\213\345\272\217/test_minimize.py" @@ -0,0 +1,48 @@ +import ctypes + +user32 = ctypes.windll.user32 +current_hwnd = user32.GetForegroundWindow() + +@ctypes.WINFUNCTYPE(ctypes.c_bool, ctypes.c_int, ctypes.c_int) +def enum_callback(hwnd, _): + if user32.IsWindowVisible(hwnd): + title = ctypes.create_unicode_buffer(255) + length = user32.GetWindowTextW(hwnd, title, 255) + if length > 0: + windows.append((hwnd, title.value)) + return True + +print("=== 当前所有可见窗口 ===") +windows = [] +user32.EnumWindows(enum_callback, 0) +for hwnd, title in windows: + mark = " ← 本脚本" if hwnd == current_hwnd else "" + print(f" {title}{mark}") + +print("\n即将最小化除了微信和本窗口以外的所有窗口。") +print("请确认微信登录窗口是否可见。") +print("按回车键开始最小化...") +input() + +minimized = [] +for hwnd, title in windows: + if title == "微信" or title == "WeChat": + print(f" 保留: {title}") + elif hwnd == current_hwnd: + print(f" 保留本窗口") + else: + user32.ShowWindow(hwnd, 6) + minimized.append(title) + +print(f"\n✅ 已最小化 {len(minimized)} 个窗口。") +print("请确认:现在微信登录窗口是否在最前面?") +print("按回车键恢复所有窗口...") +input() + +for hwnd, title in windows: + if hwnd != current_hwnd: + if "任务管理" not in title and "Task Manager" not in title: + user32.ShowWindow(hwnd, 4) + user32.ShowWindow(hwnd, 9) + +print("✅ 已恢复") diff --git "a/\345\276\256\344\277\241\350\207\252\345\212\250\347\231\273\345\275\225\347\233\221\346\216\247\347\250\213\345\272\217/test_wechat_monitor.bat" "b/\345\276\256\344\277\241\350\207\252\345\212\250\347\231\273\345\275\225\347\233\221\346\216\247\347\250\213\345\272\217/test_wechat_monitor.bat" new file mode 100644 index 0000000..636a3c6 --- /dev/null +++ "b/\345\276\256\344\277\241\350\207\252\345\212\250\347\231\273\345\275\225\347\233\221\346\216\247\347\250\213\345\272\217/test_wechat_monitor.bat" @@ -0,0 +1,11 @@ +@echo off +chcp 936 >nul +echo =============================== +echo 微信自动登录监控 - 测试模式 +echo 按 Ctrl+C 退出 +echo =============================== +echo. +python "%~dp0wechat_auto_login_v3.py" +echo. +echo 程序已退出。 +pause diff --git "a/\345\276\256\344\277\241\350\207\252\345\212\250\347\231\273\345\275\225\347\233\221\346\216\247\347\250\213\345\272\217/wechat_auto_login.ini" "b/\345\276\256\344\277\241\350\207\252\345\212\250\347\231\273\345\275\225\347\233\221\346\216\247\347\250\213\345\272\217/wechat_auto_login.ini" new file mode 100644 index 0000000..b431888 --- /dev/null +++ "b/\345\276\256\344\277\241\350\207\252\345\212\250\347\231\273\345\275\225\347\233\221\346\216\247\347\250\213\345\272\217/wechat_auto_login.ini" @@ -0,0 +1,18 @@ +[General] +wechat_path = C:\Program Files (x86)\Tencent\WeChat\WeChat.exe +poll_interval = 5 +wait_for_window_timeout = 10 + +[QRCode] +capture_interval = 60 +max_retry = 10 +screenshot_dir = wechat_qrcodes + +[Email] +email_enabled = false +smtp_server = smtp.qq.com +smtp_port = 465 +use_ssl = true +email_account = +email_password = +email_to = diff --git "a/\345\276\256\344\277\241\350\207\252\345\212\250\347\231\273\345\275\225\347\233\221\346\216\247\347\250\213\345\272\217/wechat_auto_login.py" "b/\345\276\256\344\277\241\350\207\252\345\212\250\347\231\273\345\275\225\347\233\221\346\216\247\347\250\213\345\272\217/wechat_auto_login.py" new file mode 100644 index 0000000..a30c0af --- /dev/null +++ "b/\345\276\256\344\277\241\350\207\252\345\212\250\347\231\273\345\275\225\347\233\221\346\216\247\347\250\213\345\272\217/wechat_auto_login.py" @@ -0,0 +1,847 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +微信自动登录监控程序 +- 监控微信进程,未运行则启动 +- 截屏识别登录按钮或二维码 +- 自动点击"进入微信"按钮 +- 二维码刷新时发送邮件通知 +""" + +import ctypes +import ctypes.wintypes +import sys +import os +import time +import json +import base64 +import hashlib +import io +import logging +import threading +import configparser +import subprocess +import smtplib +import email +from email.mime.text import MIMEText +from email.mime.multipart import MIMEMultipart +from email.mime.base import MIMEBase +from email.mime.image import MIMEImage +from email import encoders +from email.utils import formataddr +from datetime import datetime +from pathlib import Path +from tkinter import Tk, Frame, Canvas, StringVar, Button, Event + +# 不使用DPI感知,所有坐标在PyAutoGUI自有坐标系中保持一致 + +import psutil +import pygetwindow as gw +import pyautogui +import requests +from PIL import Image + +# ============================================================ +# 路径常量 +# ============================================================ +SCRIPT_DIR = Path(__file__).parent.resolve() +CONFIG_FILE = SCRIPT_DIR / "wechat_auto_login.ini" +LOG_FILE = SCRIPT_DIR / "wechat_monitor.log" +LOCK_FILE = SCRIPT_DIR / "wechat_monitor.lock" +QR_DIR = SCRIPT_DIR / "wechat_qrcodes" + +# ============================================================ +# 日志 +# ============================================================ +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s [%(levelname)s] %(message)s", + handlers=[ + logging.FileHandler(str(LOG_FILE), encoding="utf-8"), + logging.StreamHandler(sys.stdout), + ], +) +log = logging.getLogger("wechat_monitor") + +# ============================================================ +# 单实例保护 +# ============================================================ +def acquire_lock(): + """尝试获取文件锁,防止多实例运行""" + try: + lock_fd = os.open(str(LOCK_FILE), os.O_CREAT | os.O_EXCL | os.O_RDWR) + return lock_fd + except FileExistsError: + # 检查锁文件进程是否还在运行 + try: + with open(str(LOCK_FILE), "r") as f: + pid = int(f.read().strip()) + try: + os.kill(pid, 0) # 检查进程是否存在 + log.error(f"已有实例在运行 (PID: {pid})") + sys.exit(1) + except OSError: + # 进程不存在,锁已过期 + os.remove(str(LOCK_FILE)) + lock_fd = os.open(str(LOCK_FILE), os.O_CREAT | os.O_EXCL | os.O_RDWR) + return lock_fd + except (ValueError, FileNotFoundError): + os.remove(str(LOCK_FILE)) + lock_fd = os.open(str(LOCK_FILE), os.O_CREAT | os.O_EXCL | os.O_RDWR) + return lock_fd + +def release_lock(lock_fd): + """释放文件锁""" + try: + os.close(lock_fd) + os.remove(str(LOCK_FILE)) + except Exception: + pass + +# ============================================================ +# 配置管理 +# ============================================================ +DEFAULT_CONFIG = { + "WeChat": { + "exe_path": "C:\\Program Files (x86)\\Tencent\\WeChat\\WeChat.exe", + }, + "Email": { + "email_enabled": "true", + "smtp_server": "smtp.qq.com", + "smtp_port": "465", + "use_ssl": "true", + "email_account": "", + "email_password": "", + "email_to": "", + "max_emails": "5", + "sms_phone": "", + }, +} + + +def load_config(): + """加载配置文件,不存在则自动生成""" + config = configparser.ConfigParser() + if CONFIG_FILE.exists(): + config.read(str(CONFIG_FILE)) + log.info("配置文件已加载") + else: + for section, options in DEFAULT_CONFIG.items(): + config[section] = options + with open(str(CONFIG_FILE), "w", encoding="utf-8") as f: + config.write(f) + log.info("已生成默认配置文件") + return config + + +# ============================================================ +# 通义千问VL API调用 +# ============================================================ +QIANWEN_API_KEY = "sk-a6b3e24b985b41768aa3b0df2f688c98" +QIANWEN_ENDPOINT = "https://dashscope.aliyuncs.com/compatible-mode/v1/chat/completions" +QIANWEN_MODEL = "qwen-vl-plus" + +# ── RECT结构体和获取工作区(同测试程序) ────────────────────────── + +class RECT(ctypes.Structure): + _fields_ = [ + ("left", ctypes.c_long), + ("top", ctypes.c_long), + ("right", ctypes.c_long), + ("bottom", ctypes.c_long), + ] + + +def get_workarea(): + """获取工作区尺寸(桌面排除任务栏),物理像素""" + r = RECT() + ctypes.windll.user32.SystemParametersInfoW(0x0030, 0, ctypes.byref(r), 0) + work_w = r.right - r.left + work_h = r.bottom - r.top + return work_w, work_h + + +QR_ANALYSIS_PROMPT = ( + "分析微信登录界面状态。" + "A=有绿色'进入微信'按钮。" + "B=有二维码图案且下方有'扫码登录'文字。" + "C=已登录主界面有聊天列表。" + "D=以上都不是。" + "只回复一个字母A/B/C/D。" +) + + +def encode_image(image_path): + """将图片编码为base64""" + with open(image_path, "rb") as f: + return base64.b64encode(f.read()).decode("utf-8") + + +def analyze_screenshot(image_path): + """调用通义千问VL API分析截图""" + base64_image = encode_image(image_path) + headers = { + "Authorization": f"Bearer {QIANWEN_API_KEY}", + "Content-Type": "application/json", + } + payload = { + "model": QIANWEN_MODEL, + "messages": [ + { + "role": "user", + "content": [ + {"type": "image_url", "image_url": {"url": f"data:image/png;base64,{base64_image}"}}, + {"type": "text", "text": QR_ANALYSIS_PROMPT}, + ], + } + ], + } + try: + resp = requests.post(QIANWEN_ENDPOINT, headers=headers, json=payload, timeout=30) + resp.raise_for_status() + result = resp.json() + content = result["choices"][0]["message"]["content"].strip() + log.info(f"AI分析结果: {content}") + return content + except Exception as e: + log.error(f"AI分析请求失败: {e}") + return "D" + + +# ============================================================ +# 邮件发送 +# ============================================================ +def send_qr_email(config, image_path): + """发送二维码截图邮件(HTML正文,CID嵌入二维码图片)""" + email_config = config["Email"] + if email_config.get("email_enabled", "true").lower() != "true": + log.info("邮件通知未启用,跳过") + return + + smtp_server = email_config.get("smtp_server", "smtp.qq.com") + smtp_port = int(email_config.get("smtp_port", "465")) + use_ssl = email_config.get("use_ssl", "true").lower() == "true" + account = email_config.get("email_account", "") + password = email_config.get("email_password", "") + to_addr = email_config.get("email_to", "") + + if not all([smtp_server, account, password, to_addr]): + log.warning("邮箱配置不完整,跳过邮件发送") + return + + timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S") + subject = f"微信二维码已刷新 - {timestamp}" + + msg = MIMEMultipart("related") + msg["From"] = formataddr(("微信助手", account)) + msg["To"] = to_addr + msg["Subject"] = subject + + # HTML正文,用cid嵌入二维码图片 + html_body = ( + f"" + f"

微信二维码已刷新

" + f"

时间:{timestamp}

" + f"

请用手机扫描下方二维码登录微信:

" + f"

" + f"" + ) + msg.attach(MIMEText(html_body, "html", "utf-8")) + + with open(image_path, "rb") as f: + img_data = f.read() + # CID嵌入图 + img_part = MIMEImage(img_data, _subtype="png") + img_part.add_header("Content-ID", "") + img_part.add_header("Content-Disposition", "inline", filename="qrcode.png") + msg.attach(img_part) + # 同时作为附件 + att = MIMEBase("image", "png") + att.set_payload(img_data) + encoders.encode_base64(att) + att.add_header( + "Content-Disposition", + "attachment", + filename=f"qrcode_{datetime.now().strftime('%Y%m%d_%H%M%S')}.png", + ) + msg.attach(att) + + try: + if use_ssl: + server = smtplib.SMTP_SSL(smtp_server, smtp_port, timeout=10) + else: + server = smtplib.SMTP(smtp_server, smtp_port, timeout=10) + server.starttls() + server.login(account, password) + server.send_message(msg) + server.quit() + log.info(f"邮件已发送到 {to_addr}") + except Exception as e: + log.error(f"邮件发送失败: {e}") + + +# ============================================================ +# 腾讯云短信 +# ============================================================ +def send_sms_notification(config): + """通过腾讯云短信API发送登录通知""" + sms_phone = config["Email"].get("sms_phone", "") + if not sms_phone: + log.info("未配置短信通知手机号,跳过") + return + try: + # 腾讯云短信API(POST JSON) + url = "https://yun.tim.qq.com/v5/tlssmssvr/sendsms" + import random + now = int(time.time()) + payload = { + "params": ["888888", "5"], + "tpl_id": 2673158, + "sign": "", + "tel": {"mobile": sms_phone, "nationcode": "86"}, + "time": now, + "sig": hashlib.sha256( + f"appkey=腾讯云APPKEY&random=123456&time={now}&mobile={sms_phone}".encode() + ).hexdigest(), + } + # 实际sdk_appid和appkey应在配置中 + resp = requests.post( + url, + json=payload, + timeout=10, + ) + log.info(f"短信发送结果: {resp.status_code} - {resp.text}") + except Exception as e: + log.error(f"短信发送失败: {e}") + + +# ============================================================ +# 微信监控器 +# ============================================================ +class WeChatMonitor: + """微信进程与界面监控逻辑""" + + def __init__(self, config): + self.config = config + self.running = False + self.thread = None + + # 二维码状态追踪 + self.qr_sent_count = 0 + self.max_qr_sends = int(self.config["Email"].get("max_emails", "5")) + self.last_qr_hash = None + self.qr_region_left = 0 + self.qr_region_top = 0 + + # 截屏区域参数 - 使用工作区中心裁切(同测试程序) + self.crop_w = 696 + self.crop_h = 905 + work_w, work_h = get_workarea() + self.work_w = work_w + self.work_h = work_h + r = RECT() + ctypes.windll.user32.SystemParametersInfoW(0x0030, 0, ctypes.byref(r), 0) + offset_x = r.left + offset_y = r.top + center_x = offset_x + (work_w // 2) + center_y = offset_y + (work_h // 2) + self.region_left = max(0, center_x - self.crop_w // 2) + self.region_top = max(0, center_y - self.crop_h // 2) + self.region_width = self.crop_w + self.region_height = self.crop_h + self.screen_width = pyautogui.size().width + self.screen_height = pyautogui.size().height + + # 确保二维码目录存在 + QR_DIR.mkdir(exist_ok=True) + + def start(self): + """启动监控线程""" + if self.running: + return + self.running = True + self.thread = threading.Thread(target=self._monitor_loop, daemon=True) + self.thread.start() + log.info("监控已启动") + + def stop(self): + """停止监控线程""" + self.running = False + if self.thread and self.thread.is_alive(): + self.thread.join(timeout=3) + log.info("监控已停止") + + def _monitor_loop(self): + """主监控循环(每5秒执行一次)""" + while self.running: + try: + self._check_step() + except Exception as e: + log.error(f"监控循环异常: {e}") + time.sleep(5) + + def _check_step(self): + """一次完整的监控检查""" + status_text = "" + + # ---- 第1步:检查微信进程 ---- + wechat_found = False + for proc in psutil.process_iter(["name"]): + try: + name = proc.info["name"] + if name and name.lower() in ("wechat.exe", "weixin.exe"): + wechat_found = True + break + except (psutil.NoSuchProcess, psutil.AccessDenied): + continue + + if not wechat_found: + wechat_exe = self.config["WeChat"].get("exe_path", "") + if wechat_exe and os.path.exists(wechat_exe): + try: + subprocess.Popen([wechat_exe], shell=True) + log.info(f"已启动微信: {wechat_exe}") + except Exception as e: + log.error(f"启动微信失败: {e}") + else: + log.warning(f"微信路径无效: {wechat_exe}") + status_text = "微信未运行,启动中..." + self._update_callback(status_text) + return + + # ---- 第2步:截屏并裁切(同测试程序算法) ---- + timestamp = datetime.now().strftime("%Y%m%d_%H%M%S_%f")[:-3] + screenshot_path = QR_DIR / f"screenshot_{timestamp}.png" + try: + screenshot = pyautogui.screenshot() + # 按工作区中心裁切 + r = RECT() + ctypes.windll.user32.SystemParametersInfoW(0x0030, 0, ctypes.byref(r), 0) + offset_x = r.left + offset_y = r.top + center_x = offset_x + (self.work_w // 2) + center_y = offset_y + (self.work_h // 2) + left = max(0, center_x - self.region_width // 2) + top = max(0, center_y - self.region_height // 2) + crop = screenshot.crop((left, top, left + self.region_width, top + self.region_height)) + crop.save(str(screenshot_path)) + log.info(f"截图已保存: {screenshot_path.name}") + except Exception as e: + log.error(f"截屏失败: {e}") + return + + # ---- 第3步:AI分析 ---- + result = analyze_screenshot(str(screenshot_path)) + + # ---- 第4步:根据结果执行 ---- + parts = result.strip().split() + result_type = parts[0] if parts else "D" + + if result_type == "A": + status_text = "正在点击登录..." + self._handle_login_button(result, screenshot_path) + + elif result_type == "B": + qr_display_count = min(self.qr_sent_count + 1, self.max_qr_sends) + status_text = f"请用手机扫码 ({qr_display_count}/{self.max_qr_sends})" + self._handle_qr_code(screenshot_path) + + elif result_type == "C": + status_text = "微信运行中" + self.qr_sent_count = 0 + self.last_qr_hash = None + + else: # D + # 无法识别:有微信窗口则重试最多3次,无则跳过 + wechat_windows = gw.getWindowsWithTitle("微信") + wechat_windows = [w for w in wechat_windows if w.title.strip()] + if wechat_windows: + retry_count = getattr(self, "_d_retry", 0) + 1 + self._d_retry = retry_count + if retry_count <= 3: + status_text = f"无法识别画面,重试 ({retry_count}/3)" + log.info(f"D状态:检测到微信窗口,第{retry_count}次重试") + time.sleep(5) + # 递归重试 + self._check_step() + return + else: + self._d_retry = 0 + status_text = "无法识别画面,已放弃重试" + else: + self._d_retry = 0 + status_text = "无法识别画面,无微信窗口,跳过" + + self._update_callback(status_text) + + def _handle_login_button(self, result, screenshot_path): + """处理A情况:点击'进入微信'按钮""" + self.qr_sent_count = 0 + self.last_qr_hash = None + + # 按钮坐标直接用鼠标空间固定比例(同测试程序) + mouse_w, mouse_h = pyautogui.size() + click_x = int(mouse_w * 0.5) + click_y = int(mouse_h * 0.58) + log.info(f"[按钮坐标] X={click_x}, Y={click_y} (鼠标空间)") + + # 最小化其他窗口 -> 隐藏自己 -> 点击 -> 恢复显示 + self._minimize_other_windows() + self._hide_and_click(click_x, click_y) + + def _handle_qr_code(self, screenshot_path): + """处理B情况:二维码检测与邮件通知(零API哈希监测模式)""" + try: + # 保存截图(已在第2步保存,这里使用同一文件) + qr_save_path = QR_DIR / f"qrcode_{datetime.now().strftime('%Y%m%d_%H%M%S_%f')[:-3]}.png" + # 复制截图作为二维码记录 + img = Image.open(str(screenshot_path)) + img.save(str(qr_save_path)) + log.info(f"二维码截图已保存: {qr_save_path.name}") + + # 取二维码区域(截图下半部分中央,约160x160区域作为哈希计算) + qr_crop = img.crop(( + self.region_width // 2 - 80, + self.region_height // 2 - 80, + self.region_width // 2 + 80, + self.region_height // 2 + 80, + )) + qr_resized = qr_crop.resize((16, 16), Image.LANCZOS) + qr_hash = hashlib.md5(qr_resized.tobytes()).hexdigest() + + # 如果哈希变了(二维码刷新)且未超过发送上限 + if qr_hash != self.last_qr_hash: + self.last_qr_hash = qr_hash + if self.qr_sent_count < self.max_qr_sends: + # 等待2秒后重新截图确认 + log.info("二维码哈希变化,等待2秒后重新截图确认...") + time.sleep(2) + confirm_path = QR_DIR / f"confirm_{datetime.now().strftime('%Y%m%d_%H%M%S_%f')[:-3]}.png" + try: + confirm_ss = pyautogui.screenshot() + r = RECT() + ctypes.windll.user32.SystemParametersInfoW(0x0030, 0, ctypes.byref(r), 0) + offset_x = r.left + offset_y = r.top + center_x = offset_x + (self.work_w // 2) + center_y = offset_y + (self.work_h // 2) + cl = max(0, center_x - self.region_width // 2) + ct = max(0, center_y - self.region_height // 2) + confirm_crop = confirm_ss.crop((cl, ct, cl + self.region_width, ct + self.region_height)) + confirm_crop.save(str(confirm_path)) + # 用确认截图发送通知 + send_qr_email(self.config, str(confirm_path)) + except Exception: + send_qr_email(self.config, str(screenshot_path)) + send_sms_notification(self.config) + self.qr_sent_count += 1 + log.info(f"二维码更新通知已发送 ({self.qr_sent_count}/{self.max_qr_sends})") + else: + log.info(f"已达二维码通知上限 ({self.max_qr_sends})") + except Exception as e: + log.error(f"处理二维码时出错: {e}") + + def _minimize_other_windows(self): + """最小化除微信外的所有窗口,让微信登录界面可见""" + try: + minimized = 0 + for w in gw.getAllWindows(): + title = w.title.strip() + if title and "微信" not in title: + try: + w.minimize() + minimized += 1 + except Exception: + pass + if minimized > 0: + log.info(f"已最小化 {minimized} 个其他窗口") + except Exception as e: + log.error(f"最小化其他窗口失败: {e}") + + def _show_click_marker(self, x, y): + """在点击位置附近显示确认窗口,等待用户点击确认后返回""" + confirm_event = threading.Event() + + def _marker_worker(): + try: + marker = Tk() + marker.title("点击位置确认") + marker.overrideredirect(True) # 无边框 + marker.attributes("-topmost", True) + marker.configure(bg="#2b2b2b") + + win_w, win_h = 250, 150 + # 窗口定位在点击坐标附近,确保不超出屏幕 + screen_w = marker.winfo_screenwidth() + screen_h = marker.winfo_screenheight() + win_x = min(x + 20, screen_w - win_w) + win_y = min(y + 20, screen_h - win_h) + + marker.geometry(f"{win_w}x{win_h}+{win_x}+{win_y}") + marker.resizable(False, False) + + # 画布: 绘制标记 + canvas = Canvas(marker, width=win_w, height=win_h - 45, bg="#2b2b2b", highlightthickness=0) + canvas.pack() + + cx, cy = win_w // 2, (win_h - 45) // 2 + r = 28 + # 红色外圆 + canvas.create_oval(cx - r, cy - r, cx + r, cy + r, outline="#FF4444", width=3) + # 十字线 + canvas.create_line(cx, cy - r - 5, cx, cy + r + 5, fill="#FF4444", width=2) + canvas.create_line(cx - r - 5, cy, cx + r + 5, cy, fill="#FF4444", width=2) + # 中心红点 + canvas.create_oval(cx - 4, cy - 4, cx + 4, cy + 4, fill="#FF4444", outline="") + # 坐标文字 + canvas.create_text( + cx, cy + r + 22, + text=f"点击位置:X:{x} Y:{y}", + fill="#FFFFFF", font=("Arial", 10), + ) + + # 确认按钮 + def on_confirm(): + confirm_event.set() + marker.destroy() + + btn_frame = Frame(marker, bg="#2b2b2b") + btn_frame.pack(fill="x", pady=(0, 6)) + + btn = Button(btn_frame, text="确认", command=on_confirm, + bg="#4CAF50", fg="white", font=("Arial", 10, "bold"), + width=10, height=1, relief="flat", cursor="hand2") + btn.pack() + + marker.focus_force() + marker.mainloop() + except Exception as e: + log.error(f"显示确认窗口失败: {e}") + confirm_event.set() # 出错时保证不阻塞 + + threading.Thread(target=_marker_worker, daemon=True).start() + confirm_event.wait() # 阻塞等待用户点击确认 + + def _hide_and_click(self, x, y): + """最小化其他窗口 -> 隐藏自己 -> 显示标记 -> 激活微信 -> 点击 -> 恢复显示""" + try: + # 最小化其他窗口 + self._minimize_other_windows() + time.sleep(0.2) + + # 显示确认窗口,等待用户确认后再继续 + self._show_click_marker(x, y) + + # 隐藏自己 + if hasattr(self, "_hide_callback") and self._hide_callback: + self._hide_callback() + + time.sleep(0.2) + + # 激活微信窗口 + wechat_windows = gw.getWindowsWithTitle("微信") + if wechat_windows: + win = wechat_windows[0] + hwnd = win._hWnd + # 方案1: ShowWindow激活 + ctypes.windll.user32.ShowWindow(hwnd, 1) # SW_SHOWNORMAL + time.sleep(0.1) + # 方案2: AttachThreadInput + SetForegroundWindow + try: + # 获取当前线程和窗口线程 + current_thread = ctypes.windll.kernel32.GetCurrentThreadId() + window_thread = ctypes.windll.user32.GetWindowThreadProcessId(hwnd, None) + # Attach输入线程 + ctypes.windll.user32.AttachThreadInput(current_thread, window_thread, 1) + ctypes.windll.user32.SetForegroundWindow(hwnd) + ctypes.windll.user32.AttachThreadInput(current_thread, window_thread, 0) + except Exception: + ctypes.windll.user32.SetForegroundWindow(hwnd) + # 方案3: BringWindowToTop + ctypes.windll.user32.BringWindowToTop(hwnd) + time.sleep(0.3) + + # 点击按钮 + pyautogui.click(int(x), int(y)) + log.info(f"已点击登录按钮: x={x:.0f}, y={y:.0f}") + time.sleep(1) + + # 恢复显示 + if hasattr(self, "_show_callback") and self._show_callback: + self._show_callback() + + except Exception as e: + log.error(f"隐藏/点击/恢复操作失败: {e}") + # 确保恢复显示 + if hasattr(self, "_show_callback") and self._show_callback: + self._show_callback() + + def set_ui_callbacks(self, hide_callback, show_callback, update_callback): + """设置UI回调函数""" + self._hide_callback = hide_callback + self._show_callback = show_callback + self._update_callback = update_callback + + +# ============================================================ +# UI界面 +# ============================================================ +class WeChatMonitorUI: + """tkinter监控界面,200x120,置顶,可拖拽""" + + def __init__(self, monitor): + self.monitor = monitor + + self.root = Tk() + self.root.title("微信自动登录") + self.root.geometry("200x120+0+0") # 默认左上角 + self.root.resizable(False, False) + self.root.attributes("-topmost", True) # 置顶 + self.root.protocol("WM_DELETE_WINDOW", self._on_close) + + # 自定义标题栏(去掉系统标题栏实现自定义拖拽) + self.root.overrideredirect(True) # 无边框 + + # 主框架 + self.main_frame = Frame(self.root, bg="#333333", highlightthickness=0) + self.main_frame.pack(fill="both", expand=True) + + # ---- 拖拽支持 ---- + self._drag_data = {"x": 0, "y": 0} + self.main_frame.bind("", self._drag_start) + self.main_frame.bind("", self._drag_move) + self.root.bind("", self._drag_start) + self.root.bind("", self._drag_move) + + # ---- 开关按钮 ---- + self.button_canvas = Canvas( + self.main_frame, + width=60, + height=60, + bg="#333333", + highlightthickness=0, + ) + self.button_canvas.pack(pady=(10, 2)) + self.button_canvas.bind("", self._toggle_monitor) + + # 状态: 默认ON + self.is_on = True + self._draw_button() + + # ---- 状态文字 ---- + self.status_var = StringVar() + self.status_var.set("监控中...") + self.status_label = Canvas( + self.main_frame, + width=180, + height=20, + bg="#333333", + highlightthickness=0, + ) + self.status_label.pack(pady=(0, 8)) + self._draw_status("监控中...") + + # ---- 设置监控回调 ---- + self.monitor.set_ui_callbacks( + hide_callback=self._hide_window, + show_callback=self._show_window, + update_callback=self._update_status, + ) + + # ---- 默认启动监控 ---- + self.monitor.start() + + def _draw_button(self): + """绘制开关按钮""" + self.button_canvas.delete("all") + x, y, r = 30, 30, 25 + color = "#4CAF50" if self.is_on else "#888888" + # 外圆 + self.button_canvas.create_oval(x-r, y-r, x+r, y+r, fill=color, outline="") + # 图标/文字 + self.button_canvas.create_text( + x, y, text="ON" if self.is_on else "OFF", + fill="white", font=("Arial", 10, "bold"), + ) + + def _draw_status(self, text): + """绘制状态文字(自动滚动文字)""" + self.status_label.delete("all") + self.status_label.create_text( + 90, 10, text=text, + fill="#CCCCCC", font=("Microsoft YaHei", 9), + anchor="w", + ) + + def _toggle_monitor(self, event=None): + """切换监控开关""" + self.is_on = not self.is_on + self._draw_button() + if self.is_on: + self.monitor.start() + self._draw_status("监控中...") + else: + self.monitor.stop() + self._draw_status("已暂停") + + def _update_status(self, text): + """更新状态文字(线程安全)""" + self.root.after(0, lambda: self._draw_status(text)) + + def _hide_window(self): + """隐藏窗口""" + self.root.after(0, lambda: self.root.withdraw()) + + def _show_window(self): + """显示窗口""" + self.root.after(0, lambda: self.root.deiconify()) + + def _drag_start(self, event): + """拖拽开始""" + self._drag_data["x"] = event.x + self._drag_data["y"] = event.y + + def _drag_move(self, event): + """拖拽移动""" + deltax = event.x - self._drag_data["x"] + deltay = event.y - self._drag_data["y"] + x = self.root.winfo_x() + deltax + y = self.root.winfo_y() + deltay + self.root.geometry(f"+{x}+{y}") + + def _on_close(self): + """关闭窗口 = 退出程序""" + log.info("用户关闭窗口,退出程序") + self.monitor.stop() + self.root.destroy() + + def run(self): + """启动主循环""" + self.root.mainloop() + + +# ============================================================ +# 主入口 +# ============================================================ +def main(): + lock_fd = acquire_lock() + # 写入当前PID到锁文件 + os.write(lock_fd, str(os.getpid()).encode()) + + try: + config = load_config() + monitor = WeChatMonitor(config) + ui = WeChatMonitorUI(monitor) + log.info("微信自动登录监控启动") + ui.run() + except KeyboardInterrupt: + log.info("收到中断信号") + except Exception as e: + log.exception(f"程序异常: {e}") + finally: + release_lock(lock_fd) + log.info("程序已退出") + + +if __name__ == "__main__": + main()