当前位置：首页 > article >正文

本地ocr实现监听屏幕弹窗并进行邮箱通知

article 2025/1/21 1:19:54

现在需要解决一个问题，就是电脑屏幕跳出抖音的无人值守验证滑动页面，需要使用代码来完成邮件通知我

~~注意监控画面需要实现实时监控~~

python

初步代码

要解决这个问题，可以使用Python中的pyautogui库来模拟鼠标和键盘操作，opencv（OpenCV）来进行图像识别，以及smptlib和email库来发送邮件通知。

实时屏幕监控：你可以通过定期截取屏幕并进行图像识别，来检测抖音的无人值守验证滑动页面。
邮件通知：一旦检测到这种情况，通过SMTP协议发送邮件通知。

安装必要的库

pip install pyautogui opencv-python pillow smtplib email

完整代码

import requests
import base64
import json
import cv2
import pyautogui
import time
import logging
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
import smtplib
import threading
import os
from datetime import datetime
from email.mime.multipart import MIMEMultipart
import numpy as np

# 配置日志
logging.basicConfig(filename='ocr_monitor.log', level=logging.INFO,
                    format='%(asctime)s - %(levelname)s - %(message)s')

# 本地OCR接口配置
LOCAL_OCR_URL = "http://127.0.0.1:5000/readImage"


# 屏幕截图
def capture_screen(region=None):
    screenshot = pyautogui.screenshot(region=region)  # region参数可以限定截图的区域
    screenshot = np.array(screenshot)
    screenshot = cv2.cvtColor(screenshot, cv2.COLOR_RGB2BGR)  # 转换为BGR格式
    return screenshot


# 保存截图到本地
def save_screenshot(screenshot, save_path):
    # 确保路径正确，包含扩展名
    if not save_path.lower().endswith(('.png', '.jpg', '.jpeg')):
        save_path += ".png"  # 默认保存为.png格式

    if not os.path.exists(os.path.dirname(save_path)):
        os.makedirs(os.path.dirname(save_path))  # 如果目录不存在，创建目录

    if screenshot is not None and screenshot.size > 0:
        result = cv2.imwrite(save_path, screenshot)  # 保存为文件
        if result:
            print(f"截图已保存到: {save_path}")  # 控制台输出
            logging.info(f"截图已保存到: {save_path}")
        else:
            print(f"无法保存截图到: {save_path}")  # 控制台输出
            logging.error(f"无法保存截图到: {save_path}")
    else:
        print("截图无效，无法保存。")  # 控制台输出
        logging.error("截图无效，无法保存。")


# 请求本地OCR接口进行识别
def ocr_with_local_api(image, username="user"):
    files = {
        'file': ('screenshot.png', cv2.imencode('.png', image)[1].tobytes(), 'image/png'),
    }
    data = {'username': username}

    try:
        response = requests.post(LOCAL_OCR_URL, files=files, data=data)
        if response.status_code == 200:
            result = response.json()
            if 'answer' in result:
                return result['answer'], result['img_stream']
            else:
                print(f"OCR接口未返回有效结果: {response.text}")  # 控制台输出
                logging.error(f"OCR接口未返回有效结果: {response.text}")
                return None, None
        else:
            print(f"请求本地OCR接口失败: {response.status_code} - {response.text}")  # 控制台输出
            logging.error(f"请求本地OCR接口失败: {response.status_code} - {response.text}")
            return None, None
    except Exception as e:
        print(f"请求OCR接口时发生错误: {str(e)}")  # 控制台输出
        logging.error(f"请求OCR接口时发生错误: {str(e)}")
        return None, None


# 定期监控并识别
def monitor_screen(interval=5, target_texts=None, save_dir="saved_screenshots"):
    if target_texts is None:
        target_texts = ["10s完成", "验证码", "请输入", "验证", "挂机", "完成"]

    email_sent = False  # 记录邮件是否已发送
    while True:
        screenshot = capture_screen()  # 获取屏幕截图
        timestamp = time.strftime("%Y%m%d_%H%M%S")
        save_path = os.path.join(save_dir, f"monitor_{target_texts[0]}_{timestamp}.png")

        # 保存截图
        save_screenshot(screenshot, save_path)

        # 调用本地OCR接口进行识别
        recognized_text, _ = ocr_with_local_api(screenshot)

        if recognized_text:
            print(f"识别文本: {recognized_text}")  # 控制台输出
            logging.info(f"识别文本: {recognized_text}")
            for target_text in target_texts:
                if target_text in recognized_text:
                    print(f"检测到目标文本: {target_text}")  # 控制台输出
                    logging.info(f"检测到目标文本: {target_text}")
                    # 发送邮件通知，只发送一次
                    if not email_sent:
                        send_email("触发验证，请及时验证", f"检测到的文本为: {recognized_text}\n截图已保存到: {save_path}")
                        email_sent = True
                        print("邮件已发送，停止监控.")  # 控制台输出
                        logging.info("邮件已发送，停止监控.")
                        return  # 停止监控

        # 删除未匹配到目标的截图
        if not any(target in recognized_text for target in target_texts):
            os.remove(save_path)
            print(f"未检测到目标文本，已删除截图: {save_path}")  # 控制台输出
            logging.info(f"未检测到目标文本，已删除截图: {save_path}")

        time.sleep(interval)  # 每隔interval秒检测一次，默认5秒


# 邮件通知
def send_email(subject, body):
    sender_email = "@qq.com"  # 发送邮箱地址
    receiver_email = "@qq.com"  # 收件邮箱地址
    password = ""  # 发送邮箱授权码

    msg = MIMEMultipart()
    msg['From'] = sender_email
    msg['To'] = receiver_email
    msg['Subject'] = subject
    msg.attach(MIMEText(body, 'plain'))

    try:
        server = smtplib.SMTP_SSL('smtp.qq.com', 465)
        server.login(sender_email, password)
        server.sendmail(sender_email, receiver_email, msg.as_string())
        server.close()
        print("邮件发送成功！")  # 控制台输出
        logging.info("邮件发送成功！")
    except Exception as e:
        print(f"邮件发送失败: {str(e)}")  # 控制台输出
        logging.error(f"邮件发送失败: {str(e)}")


# 启动监控线程
def start_monitoring_thread(interval=5, target_texts=None):
    monitor_thread = threading.Thread(target=monitor_screen, args=(interval, target_texts))
    monitor_thread.daemon = True
    monitor_thread.start()


if __name__ == "__main__":
    # 配置多个目标文本
    target_texts = ["10s完成", "验证码", "请输入", "验证", "挂机"]  # 可以在此添加或修改监控的文本

    print("任务正在运行监控中...")  # 控制台输出
    logging.info("任务正在运行监控中...")
    start_monitoring_thread(interval=5, target_texts=target_texts)

    # 主线程保持运行，每10秒输出任务状态
    while True:
        print("任务运行中...")  # 控制台输出
        time.sleep(10)
        logging.info("任务运行中...")