rpctest/main.py

import requests
import time
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.chrome.service import Service
from selenium.webdriver.support.ui import Select
from selenium.webdriver.support import expected_conditions as EC
from selenium.webdriver.support.ui import WebDriverWait
from selenium.common.exceptions import TimeoutException
from multiprocessing import Process  # 替换 threading 为 multiprocessing
# PROFILE_ID为环境ID

def start_Ads(PROFILE_ID):
    BASE_URL = "http://127.0.0.1:54345"
    API_KEY = "dc954750b89edd7e421639832bff3151"  # 替换为你的 API Key
    # 启动 Adspower
    start_profile_url = f"{BASE_URL}/api/v1/browser/start?user_id={PROFILE_ID}&api_key={API_KEY}"
    response = requests.get(start_profile_url).json()

    # 检查是否启动成功
    if response["code"] == 0:
        print(f"Adspower 启动成功，返回信息")
        chrome_driver_path = response["data"]["webdriver"]
        debugger_address = response["data"]["ws"]["selenium"]
        print(chrome_driver_path)
    else:
        print(f"启动失败: {response['msg']}")
        exit(1)
    # 打开 Selenium -> Adspower
    options = webdriver.ChromeOptions()
    options.add_experimental_option("debuggerAddress", debugger_address)
    web = webdriver.Chrome(service=Service("/Users/zangtuo/Library/Application Support/adspower_global/cwd_global/chrome_131/chromedriver.app/Contents/MacOS/chromedriver"),options=options)
    web.implicitly_wait(30)
    # 访问页面
    web.get("https://claimform.savingsclubsettlement.com/consumerb-claimants")
    print("当前页面标题:", web.title)
    time.sleep(10)
    print("等待验证码识别")
    # while True:
    #     try:
    #         WebDriverWait(web, 5).until(
    #             EC.frame_to_be_available_and_switch_to_it(
    #                 (By.XPATH, "//iframe[contains(@title, 'recaptcha challenge')]"))
    #         )
    #         # 切换回主界面
    #         # web.switch_to.default_content()
    #         print("等待 10 秒后再次尝试")
    #         time.sleep(10)
    #     except TimeoutException:
    #         print("验证码识别成功")
    #         break  # 超时后退出循环，继续执行后续代码
    try:
        iframe = WebDriverWait(web, 10).until(
            EC.frame_to_be_available_and_switch_to_it((By.CSS_SELECTOR, "iframe[src*='google.com/recaptcha']"))
        )
    except TimeoutException:
        print("未找到reCAPTCHA iframe")
        exit()
    while True:
        try:
            # 等待直到响应值非空
            WebDriverWait(web, 10).until(
                lambda d: d.find_element(By.ID, 'g-recaptcha-response').get_attribute('value') != ''
            )
            print("reCAPTCHA验证成功")
            break
        except TimeoutException:
            print("验证未完成或失败")
        finally:
            web.switch_to.default_content()  # 切回主页面
    web.switch_to.default_content()
    # 姓名
    web.find_element(By.XPATH, "//input[@type='text']").send_keys("测试数据")  # Claimants Name: *
    # 删除
    web.find_element(By.XPATH, "//div[3]/div/div/input").send_keys("测试数据")  # Claimants Name: *
    # street
    web.find_element(By.ID,"street1").send_keys("测试数据")
    # city
    web.find_element(By.ID, "city").send_keys("测试数据")
    # 选择州
    Select(web.find_element(By.XPATH, "(.//*[normalize-space(text()) and normalize-space(.)='*'])[5]/preceding::select[1]")).select_by_index(37)
    # 选择code
    web.find_element(By.ID, "zip").send_keys("测试数据")
    #
    web.find_element(By.XPATH, "//div[6]/div/div/input").send_keys("测试数据")
    # 邮箱
    web.find_element(By.ID, "email").send_keys("8888888888@gmail.com")
    # 确认
    web.find_element(By.XPATH, "//input[@type='checkbox']").click()
    # 金额
    web.find_element(By.XPATH, "//input[@value='$501-$1,000']").click()
    # 签名
    web.find_element(By.ID, "signature").send_keys("测试数据")
    # 切换frame，点击验证按钮
    # WebDriverWait(web, 10).until(EC.frame_to_be_available_and_switch_to_it((By.XPATH, "//iframe[contains(@title, 'reCAPTCHA')]")))
    # recaptcha_anchor = WebDriverWait(web, 10).until(
    #     EC.element_to_be_clickable((By.XPATH, "//span[@id='recaptcha-anchor']/div"))
    # )
    # recaptcha_anchor.click()
    # # 验证码处理逻辑，切换页面
    # EC.frame_to_be_available_and_switch_to_it((By.XPATH, "//iframe[contains(@title, 'reCAPTCHA')]"))
    # # 测试
    # EC.frame_to_be_available_and_switch_to_it((By.XPATH, "/html/body/div/div[4]/iframe"))
    # # 验证码指令
    # dataw = web.find_element(By.XPATH,"//*[@id='rc-imageselect']/div[2]/div[1]/div[1]/div/text()[1]")
    # # 加载图片
    # img = web.find_element(By.XPATH, "//div[@id='rc-imageselect-target']/table/tbody/tr/td[2]/div/div/img")
    # 点击提交按钮
    submit_button = WebDriverWait(web, 10).until(EC.element_to_be_clickable((By.XPATH, "//button[@type='submit']")))
    submit_button.click()
    time.sleep(3)
    web.quit()
    print("完成一次")

def start_Ads_data(PROFILE_ID, data:list):
    BASE_URL = "http://127.0.0.1:54345"
    API_KEY = "dc954750b89edd7e421639832bff3151"  # 替换为你的 API Key
    # 启动 Adspower
    start_profile_url = f"{BASE_URL}/api/v1/browser/start?user_id={PROFILE_ID}&api_key={API_KEY}"
    response = requests.get(start_profile_url).json()
    # 检查是否启动成功
    if response["code"] == 0:
        print(f"Adspower 启动成功，返回信息")
        chrome_driver_path = response["data"]["webdriver"]
        debugger_address = response["data"]["ws"]["selenium"]
        print(chrome_driver_path)
    else:
        print(f"启动失败: {response['msg']}")
        exit(1)
    # 打开 Selenium -> Adspower
    options = webdriver.ChromeOptions()
    options.add_experimental_option("debuggerAddress", debugger_address)
    web = webdriver.Chrome(service=Service("/Users/zangtuo/Library/Application Support/adspower_global/cwd_global/chrome_131/chromedriver.app/Contents/MacOS/chromedriver"),options=options)
    web.implicitly_wait(30)
    # 访问页面
    web.get("https://claimform.savingsclubsettlement.com/consumerb-claimants")
    print("当前页面标题:", web.title)
    time.sleep(10)
    print("等待验证码识别")
    try:
        iframe = WebDriverWait(web, 10).until(
            EC.frame_to_be_available_and_switch_to_it((By.CSS_SELECTOR, "iframe[src*='google.com/recaptcha']"))
        )
    except TimeoutException:
        print("未找到reCAPTCHA iframe")
        exit()
    while True:
        try:
            # 等待直到响应值非空
            WebDriverWait(web, 10).until(
                lambda d: d.find_element(By.ID, 'g-recaptcha-response').get_attribute('value') != ''
            )
            print("reCAPTCHA验证成功")
            break
        except TimeoutException:
            print("验证未完成或失败")
        finally:
            web.switch_to.default_content()  # 切回主页面
    web.switch_to.default_content()
    # 姓名
    web.find_element(By.XPATH, "//input[@type='text']").send_keys(data[1])
    # 删除
    # web.find_element(By.XPATH, "//div[3]/div/div/input").send_keys("测试数据")
    # street
    web.find_element(By.ID,"street1").send_keys(data[2])
    # city
    web.find_element(By.ID, "city").send_keys(data[3])
    # 选择州
    # TODO: 缺个逻辑根据州名判断点击哪个按钮
    Select(web.find_element(By.XPATH, "(.//*[normalize-space(text()) and normalize-space(.)='*'])[5]/preceding::select[1]")).select_by_index(37)
    # 选择code
    web.find_element(By.ID, "zip").send_keys(data[5])
    # web.find_element(By.XPATH, "//div[6]/div/div/input").send_keys("测试数据")
    # 邮箱
    web.find_element(By.ID, "email").send_keys(data[0])
    # 确认
    web.find_element(By.XPATH, "//input[@type='checkbox']").click()
    # 金额
    web.find_element(By.XPATH, "//input[@value='$501-$1,000']").click()
    # 签名
    web.find_element(By.ID, "signature").send_keys(data[1])
    submit_button = WebDriverWait(web, 10).until(EC.element_to_be_clickable((By.XPATH, "//button[@type='submit']")))
    submit_button.click()
    time.sleep(3)
    web.quit()
    time.sleep(3)
    print("完成一次")


def read_data(path):
    records = []
    with open(path, 'r', encoding='utf-8') as f:
        lines = [line.strip() for line in f if line.strip()]

    for i in range(0, len(lines), 2):
        if i + 1 >= len(lines):
            break
        email_line = lines[i]
        data_line = lines[i + 1]

        # 处理电子邮件
        email = email_line.replace('\t', '').strip()

        # 分割数据行字段
        fields = data_line.split('\t')
        fields = [f.strip() for f in fields if f.strip()]

        # 确保字段足够
        if len(fields) < 5:
            print(f"跳过字段不足的行: {data_line}")
            continue
        # 提取姓名等字段
        name_parts = fields[:len(fields) - 5]
        name = ' '.join(name_parts)
        try:
            address = fields[-5]
            city = fields[-4]
            state = fields[-3]
            zip_code = fields[-2]
            phone = fields[-1]
        except IndexError as e:
            print(f"字段提取错误: {e}, 行内容: {data_line}")
            continue

        records.append({
            'email': email,
            'name': name,
            'address': address,
            'city': city,
            'state': state,
            'zip': zip_code,
            'phone': phone
        })
    return records

def read_data(path):
    records = []
    with open(path, 'r', encoding='utf-8') as f:
        lines = [line.strip() for line in f if line.strip()]

    for i in range(0, len(lines), 2):
        if i + 1 >= len(lines):
            break
        email_line = lines[i]
        data_line = lines[i + 1]

        # 处理电子邮件
        email = email_line.replace('\t', '').strip()

        # 分割数据行字段
        fields = data_line.split('\t')
        fields = [f.strip() for f in fields if f.strip()]

        # 确保字段足够
        if len(fields) < 5:
            print(f"跳过字段不足的行: {data_line}")
            continue
        # 提取姓名等字段
        name_parts = fields[:len(fields) - 5]
        name = ' '.join(name_parts)
        try:
            address = fields[-5]
            city = fields[-4]
            state = fields[-3]
            zip_code = fields[-2]
            phone = fields[-1]
        except IndexError as e:
            print(f"字段提取错误: {e}, 行内容: {data_line}")
            continue

        records.append({
            'email': email,
            'name': name,
            'address': address,
            'city': city,
            'state': state,
            'zip': zip_code,
            'phone': phone
        })
    return records
def read_txt_file(file_path):
    """
    读取并解析制表符分隔的文本文件
    :param file_path: 文件路径（例如：data.txt）
    :return: 包含所有记录的列表，每个记录是字段列表
    """
    data = []

    try:
        with open(file_path, 'r', encoding='utf-8') as file:
            for line in file:
                # 移除首尾空白字符并按制表符分割
                cleaned_line = line.strip()
                if not cleaned_line:  # 跳过空行
                    continue

                fields = cleaned_line.split('\t')

                # 验证字段数量（示例数据每行6个字段）
                if len(fields) != 6:
                    print(f"警告：第 {len(data) + 1} 行字段数量异常: {len(fields)}")
                    continue

                data.append(fields)

    except FileNotFoundError:
        print(f"错误：文件 {file_path} 未找到")
        return None
    except Exception as e:
        print(f"读取文件时发生错误: {str(e)}")
        return None

    return data

# start_Ads('kvavcrf')
# path = '模拟数据.txt'
#
# records = read_data(path)
# for record in records:
#     start_Ads_data('kvavcrf',record)
# print("全部完成")
def cont():
    datas = read_txt_file("generated_data.txt")
    for data in datas:
        start_Ads_data('kvavcrf', data)
        print("进入下次循环")
    return 0

if __name__ == '__main__':
    cont()