根据指定子流程名称进行并发,然后阻塞主流程进行后续操作

感谢【耐家军】DC 全程监听随机验证码滑块出现处理-适用于win和web通用不卡死方案 和蕾恩恩 影刀RPA过随机验证码 并发子流程暂停另一个子流程 并发暂停与恢复

因为复现耐家军的指令,发现触发了子流程检测元素是否存在和处理关闭元素,但主流程还是继续运行,没有阻塞.所以结合原版发给AI又改了一版本,且可以根据指定的子流程的名称来进行操作,而不是process7的命名.

NO3:

import sys
import time
from xbot import print, sleep
import xbot
import threading
from datetime import datetime
from .package import variables as glv
import os
import json
import re
import xbot_visual

def check_state(检测流程名称, 处理流程名称):
    package_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "package.json")
    
    if "xbot_extensions" not in package_path:
        pattern = r'(xbot_extensions\[^\]+\\package\.json)'
        package_path = re.sub(pattern, r'xbot_robot\\package.json', package_path)
    
    with open(package_path, 'r', encoding='utf-8') as f:
        flows = json.loads(f.read())['flows']
    
    流程映射 = {}
    for item in flows:
        流程映射[item['name']] = item['filename']
    
    # 处理滑块的事件对象
    slider_being_processed = threading.Event()
    
    def process_slider():
        if 处理流程名称 in 流程映射:
            print(f"[{datetime.now()}] 开始处理滑块...")
            xbot_visual.process.run(
                process=f"xbot_robot.{流程映射[处理流程名称]}",
                package=__name__,
                inputs={},
                outputs=[],
                timeout=300
            )
            print(f"[{datetime.now()}] 滑块处理完成...")
        slider_being_processed.clear()
    
    def poll_slider_state():
        while True:
            if not slider_being_processed.is_set():
                if 检测流程名称 in 流程映射:
                    xbot_visual.process.run(
                        process=f"xbot_robot.{流程映射[检测流程名称]}",
                        package=__name__,
                        inputs={},
                        outputs=[],
                        timeout=300
                    )
                if glv['判断滑块出现'] and not slider_being_processed.is_set():
                    slider_being_processed.set()
                    # 启动处理线程
                    threading.Thread(target=process_slider, daemon=True).start()
            time.sleep(0.1)
    
    # 启动轮询线程
    polling_thread = threading.Thread(target=poll_slider_state, daemon=True)
    polling_thread.start()
    
    # 使用sys.settrace阻塞主线程
    def trace_calls(frame, event, arg):
        if glv['判断滑块出现']:
            while glv['判断滑块出现']:
                time.sleep(0.1)  # 阻塞主线程
        return trace_calls
    
    sys.settrace(trace_calls)

def main(args):
    检测流程名称 = args.get("检测是否出现", "")
    处理流程名称 = args.get("处理出现", "")
    check_state(检测流程名称, 处理流程名称)

↓是主流程

检测流程：1 判断是否出现

处理流程名称：2 处理出现

窗口处理完要设置布尔值变量为False,

原理相同所以不挑是否是windows端

我认为把他封装为指令很重要,此外代码设置轮询为0.1秒,如果不想那么高可以手动设置,或者修改为变自己选择