自动充值脚本 - 多线程处理订单并监控卡券状态

import queue
import concurrent.futures
import time
import datetime
import os
import winsound
from retry import retry
from selenium.webdriver.common.by import By
import requests
from selenium import webdriver

'v1.0 bate --重构项目 --增加多api支持'

def beep():
    frequency_beep = 2500
    duration = 1000
    winsound.Beep(frequency_beep, duration)

def start_driver(url):
    options = webdriver.ChromeOptions()
    options.add_argument('start-maximized')
    options.add_argument('--headless')
    options.add_experimental_option('excludeSwitches', ['enable-automation'])
    options.add_experimental_option('useAutomationExtension', False)
    driver_tmp = webdriver.Chrome(options=options)
    with open('./stealth.min.js') as f:
        js = f.read()
    driver_tmp.execute_cdp_cmd('Page.addScriptToEvaluateOnNewDocument', {'source': js})
    driver_tmp.implicitly_wait(5)
    driver_tmp.set_window_size(700, 900)
    driver_tmp.set_window_position(0, 0)
    # driver_tmp.set_page_load_timeout(2)
    driver_tmp.get(url)
    return driver_tmp

@retry(tries=5, delay=1)
def keep_alive(driver):
    driver.find_element(By.XPATH, '//*[@id="root"]/div/div[4]/div[1]/div[1]').click()
    time.sleep(0.1)
    driver.find_element(By.XPATH, '//*[@id="root"]/div/div[4]/div[1]/div[2]').click()
    # print('[调试]: keep_alive done')
    return True

def info_output(printf):
    while True:
        line = printf.get()
        print(line)

def log(status, order_number, phone_number, message):
    today = datetime.date.today()
    log_dir = './logs'
    if not os.path.exists(log_dir):
        os.makedirs(log_dir)
    log_file = os.path.join(log_dir, today.strftime('%Y-%m-%d') + '.txt')
    # 将日志写入文件
    with open(log_file, 'a', encoding='utf-8') as f:
        date_now = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        log_line = f'{status}: Date: {date_now} - Order: {order_number}, Phone: {phone_number}, Message: {message}
'
        f.write(log_line)
    # print('[调试]: log done')

@retry(tries=5, delay=3)
def get_orders():
    url = f'http://{api_domain}/clientsupplier/orderlist/selectorderlist.html'
    params = {
        'token': api_token,
        'status': 98
    }
    response = requests.get(url=url, params=params).json()
    orders = [[order['attach'], order['orderno']] for order in response['data']]
    # print('[调试]: get_orders done')
    return orders

def complete_order(order_number):
    url = f'http://{api_domain}/clientsupplier/orderlist/delivergoods'
    params = {
        'token': api_token,
        'orderno': order_number,
        'status': '5',
        'receipt': receipt,
    }
    response = requests.get(url=url, params=params).json()
    if response['code'] == 1:
        return True
    # print('[调试]: complete done')
    return response['msg']

def using_code(driver, phone, order_num):
    # 刷新兑换码
    try:
        # 进入主页
        driver.find_element(By.XPATH, '//*[@id="root"]/div/div[4]/div[1]/div[1]').click()
        time.sleep(0.1)
        # 进入我的权益
        driver.find_element(By.XPATH, '//*[@id="root"]/div/div[4]/div[1]/div[2]').click()
        time.sleep(0.1)
    except Exception as e:
        return False, 1, '错误：切换标签卡出错。', e
    # 检测兑换码是否存在
    try:
        # 进入第一个权益
        driver.find_element(By.XPATH, '//*[@id="root"]/div/div[2]/div[2]/div/div[1]/div[1]/div[1]/div[1]/div[3]').click()
        time.sleep(0.1)
    except Exception as e:
        driver.quit()
        return False, 2, '错误：兑换卡券用完，请及时补充。', e
    # 使用兑换码
    try:
        # 立即使用
        driver.find_element(By.XPATH, '//*[@id="root"]/div/div/div/div[3]/div[1]/div/div').click()
        time.sleep(0.1)
        # 定位输入框
        driver.find_element(By.XPATH, '//*[@id="root"]/div/div/div/div[2]/div[1]/div[2]/input').send_keys(phone)
        time.sleep(0.1)
        # 同意使用协议
        driver.find_element(By.XPATH, '//*[@id="root"]/div/div/div/div[2]/div[2]/div[1]/img').click()
        time.sleep(0.5)
        # 确认无误
        driver.find_element(By.XPATH, '//*[@id="root"]/div/div/div/div[2]/div[2]/div[2]/div[2]').click()
        time.sleep(0.3)
        # 确认使用
        driver.find_element(By.XPATH, '/html/body/div[3]/div[1]/div[3]/span[2]').click()
        time.sleep(0.1)
    except Exception as e:
        return False, 3, '错误：订单使用过程出错。', e
    # 获取订单信息
    driver.find_element(By.XPATH, '//*[@id="wrapper-root"]/div[1]/div[8]/div[2]')
    file_name = os.path.join(path_screen_shoot, f'{order_num}.png')
    time.sleep(0.1)
    driver.save_screenshot(file_name)
    # 返回我的权益
    driver.back()
    # print('[调试]: using_code done')
    return True, 0, '成功：订单已完成。', ''

def orders_detection(stack, printf):
    orders_is_in_stack = []
    while True:
        try:
            orders = get_orders()
            printf.put(f'【信息】：当前订单总数量 {len(orders)} 当前核内数量 {stack.qsize()}')
            if stack.qsize() >= 50:
                beep()
                # print(f'[调试]: 当前栈数量 {stack.qsize()}')
        except Exception as e:
            printf.put(f'【错误】：订单API出错')
            time.sleep(0.1)
            continue
        if not orders:
            print('【信息】：无订单等待中')
        for order in orders:
            if order in orders_is_in_stack:
                continue
            orders_is_in_stack.append(order)
            stack.put(order)
            time.sleep(0.1)

def orders_processed(index, driver, stack, printf):
    start_time = time.time() # 获取开始时间
    while True:
        try:
            order = stack.get(block=False)
        except queue.Empty:
            keep_alive(driver=driver)
            time.sleep(0.1)
            # 检查是否已超过5分钟，如果是，重启订单处理
            if time.time() - start_time > 300:
                printf.put('【信息】：订单处理时间超过5分钟，重启订单处理')
                return True, '订单处理时间超过5分钟，重启订单处理。'
            continue
        checked, status, message, exception = using_code(driver=driver, phone=order[0], order_num=order[1])
        if checked is True:
            printf.put(f'【充值成功】: 订单编号[{order[1]}] 号码[{order[0]}]')
            complete_order(order_number=order[1])
            log(status='Finish', order_number=order[1], phone_number=order[0], message='')
            continue
        if status == 0:
            printf.put(f'【充值成功】: 订单编号[{order[1]}] 号码[{order[0]}]')
            complete_order(order_number=order[1])
            log(status='Finish', order_number=order[1], phone_number=order[0], message='')
            continue
        if status == 1:
            driver.get(using_code_urls[index])
            stack.put(order)
            continue
        if status == 2:
            printf.put(f'【订单提交错误】: 编号[{index}]链接，卡券已耗尽')
            remove_empty_link(index=index)
            stack.put(order)
            return False, '错误：卡券已耗尽。'
        if status == 3:
            printf.put(f'【订单提交错误】: 订单编号[{order[1]} 号码[{order[0]}]]')
            stack.put(order)
            continue

def remove_empty_link(index):
    global using_code_urls
    del using_code_urls[index]

def main():
    # with multiprocessing.Pool(processes=process_count) as pool:
    #     drivers = pool.starmap(start_driver, using_code_urls)
    drivers = [start_driver(url=url) for url in using_code_urls]
    with concurrent.futures.ThreadPoolExecutor() as executor:
        future_orders_detection = executor.submit(orders_detection, queue_order, queue_output)
        futures_orders_processed = []
        for index, driver in enumerate(drivers):
            future_orders_processed = executor.submit(orders_processed, index, driver, queue_order, queue_output)
            futures_orders_processed.append(future_orders_processed)
        future_info_output = executor.submit(info_output, queue_output)

        # 等待所有 future 完成
        concurrent.futures.wait([future_orders_detection, *futures_orders_processed, future_info_output], return_when=concurrent.futures.FIRST_EXCEPTION)
        
        # 检查是否有任何 future 出现异常
        for future in [future_orders_detection, *futures_orders_processed, future_info_output]:
            if future.exception():
                printf.put(f'【错误】：{future.exception()}')

if __name__ == '__main__':
    # 创建队列
    queue_order = queue.Queue()
    queue_output = queue.Queue()
    # 域名
    api_domain = 'longteng.xinyavip.com'
    # 令牌
    api_token = 'f96017ab9f62b7d9ffa4c73a8de431d8'
    # 填写兑换链接
    using_code_urls = [
        'https://m.yqb.com/interestplatform/#/home?mzoneToken=8154fb7ae7566cd17185c9a36c82bc1eee74fa19478dd53483b16c2369877692&zoneMno=600002224016&channel=YQB',
        'https://m.yqb.com/interestplatform/#/home?mzoneToken=6ffc2c12c972a5432475a08c54bbf9a3ddeb5691bdec1ca34662f0a5db0ed069&zoneMno=600002224016&channel=YQB',
        'https://m.yqb.com/interestplatform/#/home?mzoneToken=7a148d1f0cf7ed872b8fb870ef1587078e84421af211551de23b289d0244b8ad&zoneMno=600002224016&channel=YQB',
        'https://m.yqb.com/interestplatform/#/home?mzoneToken=d21e29382f0398abc839463cc806c254c721f65d7a3225aa06cd6f5ef9926e61&zoneMno=600002224016&channel=YQB',
        'https://m.yqb.com/interestplatform/#/home?mzoneToken=ca89d6d2529e30aceb38ce15f3e0c9d9d67a50451ff836ba1bd1d5b9edaf07c6&zoneMno=600002224016&channel=YQB',
        # 'https://m.yqb.com/interestplatform/#/home?mzoneToken=645989c608ece0d889f18af4e543371a7850c8bfb7ed28e92d9edbfc888e71ae&zoneMno=600002224016&channel=YQB',
        # 'https://m.yqb.com/interestplatform/#/home?mzoneToken=64b7cecd0586130a27e03523710a57a123993db07d376a5cb24810d6c934c1d9&zoneMno=600002224016&channel=YQB',
        # 'https://m.yqb.com/interestplatform/#/home?mzoneToken=003e650bbe64d13e75b7cd79eb0b6a2f3d9310d41c737ddd7b7d9dc5e56297c8&zoneMno=600002224016&channel=YQB',
        # 'https://m.yqb.com/interestplatform/#/home?mzoneToken=b38e74152f7f92b52213818d8024b45d53ad43131d3c9ed94e5231b3b109f76f&zoneMno=600002224016&channel=YQB',
        # 'https://m.yqb.com/interestplatform/#/home?mzoneToken=3b0b51ebb3578344413aab23c65fc7d41aea2eb7872ed85bbb1f58ec333549f3&zoneMno=600002224016&channel=YQB',
        # 'https://m.yqb.com/interestplatform/#/home?mzoneToken=8907feb0bb9a0d2f762dec55801dc0406cbe5b23e43b678f0193ace55afd9f38&zoneMno=600002224016&channel=YQB',
    ]
    # 客户回执
    receipt = '充值成功'
    # 线程进程数量计算
    process_count = len(using_code_urls)
    # 截图保存文件夹
    path_screen_shoot = './logs'
    # 程序启动
    main()