ca_auto_table/spider/main.py

from math import log
import random
from re import S
import time
from tkinter import N
from DrissionPage import Chromium
from loguru import logger
from work import get_random_canada_info
from mail_ import mail_
from bit_browser import bit_browser
from api import api


class Auto:
    def __init__(self, http: str = None):
        self.browser = Chromium(http)
        self.tab = self.browser.latest_tab
        pass

    # cf打码
    def solve_cloudflare(self, is_ok: bool = False):
        tab = self.browser.latest_tab
        for _ in range(5):
            tab.wait(1)
            res = tab.ele(
                't:h1@text()=Sorry, you have been blocked', timeout=1)
            if res:
                logger.error("Cloudflare验证失败")
                return False

            try:
                shadow1 = tab.ele(
                    'x://*[@name="cf-turnstile-response"]').parent().shadow_root
                iframe = shadow1.get_frame(1)
                if iframe:
                    logger.debug("找到Cloudflare iframe")
                    shadow2 = iframe.ele('x:/html/body').shadow_root
                    if shadow2:
                        logger.debug("找到Cloudflare iframe body shadow root")
                        status = shadow2.ele(
                            'x://span[text()="Verifying..."]', timeout=1.5)
                        if status:
                            tab.wait(3)
                        status = shadow2.ele(
                            'x://span[text()="Success!"]', timeout=1.5)
                        if status:
                            logger.debug("Cloudflare验证成功")
                            return True
                        checkbox = shadow2.ele(
                            'x://input[@type="checkbox"]', timeout=1.5)
                        if checkbox:
                            checkbox.click()
                            logger.debug("点击Cloudflare复选框")
                            tab.wait(3)
                        logger.debug("重新获取状态")
                        # return False
            except Exception as e:
                # logger.error(f"处理Cloudflare异常: {e}")
                if is_ok:
                    logger.debug(f"cloudflare处理通过: {e}")
                    return True
                return self.solve_cloudflare(is_ok=True)
            tab.wait(1)
        return False

    # 打开URL
    def open_url(self, url: str):
        self.tab.get(url)

    # 等待进入首页
    def wait_home(self):
        logger.debug("等待进入首页")
        jc = 0
        while True:
            if jc > 3:
                logger.error("等待进入首页超过5次，未成功")
                return False
            self.tab.wait(1)
            # 判断cf是否通过
            bol = self.solve_cloudflare()
            if not bol:
                logger.debug("Cloudflare验证失败.")
                # 刷新网页
                self.tab.refresh()
                self.tab.wait(1.5)
                jc += 1
                continue
            else:
                logger.debug("Cloudflare验证成功.")
            self.tab.wait(1.5)
            bol = self.tab.ele(
                't:h1@text()=Sorry, you have been blocked', timeout=1)
            if bol:
                logger.debug("ip被ban秒")
                return False
            bol = self.tab.ele(
                't:div@text():ERR_SSL_PROTOCOL_ERROR', timeout=1)
            if bol:
                logger.debug("刷新网页")
                self.tab.refresh()
                self.tab.wait(1.5)
            html = self.tab.url
            logger.debug(f"当前URL: {html}")
            if 'https://veritaconnect.ca/canadianbreadsettlement/en-us' == html:
                logger.debug("成功进入首页")
                return True
            jc += 1

    # 点击continue按钮
    def click_continue(self, bl: bool = False):
        logger.debug("点击Continue按钮")
        jc = 0
        while True:
            if jc > 3:
                logger.error("点击Continue按钮超过5次，未成功")
                return False
            try:
                continue_button = self.tab.ele(
                    't:button@text():Continue', timeout=1)
                if continue_button:
                    # 滚动到最底部
                    self.tab.scroll.to_bottom()
                    self.tab.wait(1)
                    # 判断cf是否通过
                    bol = self.solve_cloudflare()
                    if not bol:
                        logger.debug("Cloudflare验证失败..")
                        self.tab.refresh()
                        self.tab.wait(1.5)
                        jc += 1
                        continue
                    else:
                        logger.debug("Cloudflare验证成功..")
                    self.tab.wait(3)
                    continue_button.click()
                    logger.debug("点击Continue按钮成功")
                    self.tab.wait(1.5)
                # bol = self.tab.ele('@text():Loading', timeout=1)
                # if bol:
                    # logger.debug("Loading...")
                    # if bl:
                    #     logger.debug("多次异常界面, 结束继续点击")
                    #     return False
                    # logger.debug("异常界面")
                    # self.tab.wait(1)
                    # return self.click_continue(bl=True)

                bol = self.tab.ele(
                    't:h1@text()=Sorry, you have been blocked', timeout=1)
                if bol:
                    logger.debug("ip被ban秒")
                    # 刷新网页
                    self.tab.refresh()
                    self.tab.wait(1.5)

                bol = self.tab.ele(
                    't:h2@text()=You are being rate limited', timeout=1)
                if bol:
                    logger.debug("被限流, 退出")
                    return False
                bol = self.tab.ele(
                    't:li@text():There was a problem, please try again.', timeout=1)
                if bol:
                    if bl:
                        logger.debug("多次异常界面, 结束继续点击")
                        return False
                    logger.debug("异常界面")
                    self.tab.wait(1)
                    return self.click_continue(bl=True)
                html = self.tab.url
                logger.debug(f"当前URL: {html}")
                if 'https://veritaconnect.ca/canadianbreadsettlement/en-us/Claimant/UnknownClaimForm' in html:
                    logger.debug("成功进入问卷界面")
                    return True
            except Exception as e:
                logger.error(f"点击Continue按钮异常: {e}")
            self.tab.wait(1)
        return False

    # 随机取城市
    def get_random_city(self, province: str | None = None):
        cities = {
            "Alberta": ["Calgary", "Edmonton"],
            "British Columbia": ["Vancouver"],
            # "Manitoba": ["Winnipeg", "Rochester"],
            # "New Brunswick": ["Fredericton", "Moncton"],
            # "Newfoundland and Labrador": ["St. John's", "Halifax"],
            "Nova Scotia": ["Halifax"],
            "Ontario": ["Toronto"],
            # "Prince Edward Island": ["Charlottetown", "St. John's"],
            # "Quebec": ["Quebec City", "Montreal"],
            # "Saskatchewan": ["Saskatoon", "Regina"],
        }
        if province is None:
            province = random.choice(list(cities.keys()))
        return province, random.choice(cities.get(province, []))

    def get_province_by_city(self) -> str | None:
        """
        根据城市名称解析对应省份

        参数:
            city (str): 城市名称，例如 `Calgary`、`Edmonton` 等

        返回值:
            str | None: 对应的省份名称；未匹配返回 None
        """
        mapping = {
            "Calgary": "Alberta",
            "Edmonton": "Alberta",
            "Vancouver": "British Columbia",
            "Halifax": "Nova Scotia",
            "Toronto": "Ontario",
            "Ottawa": "Ontario",
            "Mississauga": "Ontario",
            "Brampton": "Ontario",
            "Hamilton": "Ontario",
            "Kitchener": "Ontario",
            "London": "Ontario",
            "Markham": "Ontario",
            "Vaughan": "Ontario",
            "Windsor": "Ontario",
            "Oshawa": "Ontario",
            "Brantford": "Ontario",
            "Barrie": "Ontario",
            "Sudbury": "Ontario",
            "Kingston": "Ontario",
            "Guelph": "Ontario",
            "Cambridge": "Ontario",
            "Sarnia": "Ontario",
            "Peterborough": "Ontario",
            "Waterloo": "Ontario",
            "Belleville": "Ontario",
            "Brockville": "Ontario",
            "Burlington": "Ontario",
            "Cornwall": "Ontario",
            "Kawartha Lakes": "Ontario",
            "North Bay": "Ontario",
            "Orillia": "Ontario",
            "Pickering": "Ontario",
            "Sault Ste. Marie": "Ontario",
            "Stratford": "Ontario",
            "Durham": "Ontario",
            "Norfolk County": "Ontario",
            "Prince Edward County": "Ontario",
            "Quinte West": "Ontario",
            "St. Catharines": "Ontario",
            "Welland": "Ontario",
            "Thorold": "Ontario",
            "Niagara Falls": "Ontario",
            "Pelham": "Ontario",
            "Port Colborne": "Ontario",
        }
        # 随机返回一条 key 和 value
        return random.choice(list(mapping.items()))

    # 随机实物

    def get_random_food(self, city: str, shop: str) -> list[str]:
        """
        随机选择 1~2 种食物类别，并为每个类别至少选择 1 个具体产品

        参数:
            shop (str): 商店名称（当前未使用，占位参数）

        返回值:
            list[str]: 随机选取的产品名称列表
        """
        categories = [
            [
                'Wonder Bread White',
                'Villaggio White Bread',
                'No Name Sliced White Bread',
                "President's Choice White Sliced Bread",
            ],
            [
                "Ben's Original Whole Wheat Bread",
                "POM Whole Wheat Bread",
                "Silver Hills Bakery Whole Wheat Sliced Bread",
                "Country Harvest Whole Wheat Bread",
            ],
            [
                "Wonder Bread Hot Dog Buns",
                "Villaggio Hamburger Buns",
                "Dempster's Dinner Rolls",
                "No Frills Hot Dog Buns",
            ],
            [
                "Stonemill Bakehouse Bagels",
                "Wonder Bagels",
                "Montreal Bagels (pre-packaged, e.g., St. Lawrence brand)",
                "President's Choice Bagels",
            ],
            [
                "Silver Hills Multi-Grain Sliced Bread",
                "POM Multi-Grain Bread",
                "Country Harvest Multi-Grain Loaf",
            ],
            [
                "President's Choice French Stick",
                "Dempster's Italian Style Bread",
                "Wonder Italian Bread",
                "Villaggio Country Style Loaf",
            ],
        ]

        # 随机选择 1~2 个类别（不重复）
        category_count = random.randint(1, 2)
        chosen_categories = random.sample(categories, k=category_count)

        # 每个类别至少选择 1 个产品，最多选择 3 个以避免过多
        selected_products: list[str] = []
        for cat in chosen_categories:
            max_pick = min(3, len(cat))
            pick_count = random.randint(1, max_pick)
            selected_products.extend(random.sample(cat, k=pick_count))
        logger.debug(f"随机选择的产品: {selected_products}")
        text = f'{shop}, {city} buy: '
        for p in selected_products:
            text += f'{p} * {random.randint(1, 3)}, '
        text = text[:-2]
        text = text + '.'
        logger.debug(f'随机选择的产品文本: {text}')
        return text

    # 填写问卷
    def fill_questionnaire(self):
        """
        完成问卷填写

        参数:
            city (str): 线程启动时传入的城市名称，用于匹配省份并填写数据
        """
        try:
            city, province = self.get_province_by_city()
            if province is None:
                logger.error(f"未找到城市对应省份")
                return
            j = 0
            while True:
                if j > 3:
                    return False
                info = get_random_canada_info(province, city)
                if len(info.get('postcode')) > 5:
                    break
                j += 1
            first_name = info["firstname"]
            last_name = info["lastname"]
            # 将生日格式从 '8/28/1995' 转为 'yyyy-mm-dd'，日月不足两位补0
            birthday = info["birthday"]
            current_address = info["address_str"]
            # 保持使用线程传入的城市与解析出的省份
            postal_code = info["postcode"]
            email = mail_.email_create_random()
            phone = info["phone"]
            shop = api.get_random_shop()
            if shop is None:
                return None
            street = shop.get('street')
            if street is None:
                return None
            text = self.get_random_food(shop.get('city'), street)
            # 人数
            person_count = str(random.randint(3, 5))
            logger.debug("填写问卷")
            self.tab.wait(0.1)
            logger.debug(f"填写first_name: {first_name}")
            self.tab.ele('t:input@id=FirstName').set.value(first_name)
            self.tab.wait(0.1)
            logger.debug(f"填写last_name: {last_name}")
            self.tab.ele('t:input@id=LastName').set.value(last_name)
            self.tab.wait(0.1)
            logger.debug(f"填写birthday: {birthday}")
            self.tab.ele('t:input@id=DateOfBirth').set.value(birthday)
            self.tab.wait(0.1)
            logger.debug(f"填写current_address: {current_address}")
            self.tab.ele('t:input@id=AddressLine1').set.value(current_address)
            self.tab.wait(0.1)
            logger.debug(f"填写city: {city}")
            self.tab.ele('t:input@id=City').set.value(city)
            self.tab.wait(0.1)
            logger.debug(f"填写province: {province}")
            self.tab.ele(
                't:select@id=CanProv').ele(f't:option@text()={province}').click()
            self.tab.wait(0.1)
            logger.debug(f"填写postal_code: {postal_code}")
            self.tab.ele('t:input@id=CanPostal').set.value(postal_code)
            self.tab.wait(0.1)
            logger.debug(f"填写NumberOfAdults: {person_count}")
            self.tab.ele(
                't:select@id=NumberOfAdults').ele(f't:option@text()={person_count}').click()
            self.tab.wait(0.1)
            logger.debug(f"选择地址没变")
            self.tab.eles('t:input@id=IsDifferentAddress')[1].click()
            self.tab.wait(0.1)
            logger.debug(f"填写email: {email}")
            self.tab.ele('t:input@id=EmailAddress').set.value(email)
            self.tab.wait(0.1)
            logger.debug(f"填写ConfirmEmailAddress: {email}")
            self.tab.ele('t:input@id=ConfirmEmailAddress').set.value(email)
            self.tab.wait(0.1)
            logger.debug(f"填写phone: {phone}")
            self.tab.ele('t:input@id=PhoneNumber').set.value(phone)
            self.tab.wait(0.1)
            logger.debug(f"选择同意条款")
            self.tab.ele('t:input@id=IVerify').click()
            self.tab.wait(0.1)
            logger.debug(f"选择没有申请过")
            self.tab.eles('t:input@id=IsCompensated')[1].click()
            self.tab.wait(0.1)
            logger.debug(f"填写text: {text}")
            self.tab.ele('t:textarea@id=MetaAnswerA').set.value(text)
            self.tab.wait(0.1)
            logger.debug(f"勾选同意我的名字")
            self.tab.ele('t:input@id=IDeclare').click()
            self.tab.wait(0.1)
            logger.debug(f"填写PrintName: {last_name+' '+first_name}")
            self.tab.ele(
                't:input@id=PrintName').set.value(last_name+' '+first_name)
            self.tab.wait(0.1)
            return self.submit_file(first_name, last_name, birthday, current_address, city, phone, postal_code, province, email, text)
            
        except Exception as e:
            logger.error(f"填写问卷失败: {e}")

    # 提交问卷
    def submit_file(self, first_name: str, last_name: str, birthday: str, current_address: str, city: str, phone: str, postal_code: str, province: str, email: str, text: str):
        jc = 0
        while True:
            if jc >= 3:
                logger.error("提交问卷失败")
                return False
            res = self.tab.ele(
                't:h2@text()=CLAIM SUBMISSION CONFIRMATION', timeout=3)
            if res:
                logger.info("提交问卷成功")
                api.create_info(
                    first_name=first_name,
                    last_name=last_name,
                    birthday=birthday,
                    current_address=current_address,
                    city=city,
                    phone=phone,
                    postal_code=postal_code,
                    province=province,
                    email=email,
                    text=text
                )
                return True
            
            bol = self.solve_cloudflare()
            if not bol:
                logger.debug("Cloudflare验证失败.")
                self.tab.wait(1)
            else:
                logger.debug("Cloudflare验证成功.")
                logger.debug(f"点击Submit按钮")
                self.tab.ele('t:button@text():Submit').click()
                self.tab.wait(3)
            jc += 1


# 取对应城市的代理
def get_proxy(city: str):
    if city == "Calgary":
        return "us.novproxy.io:1000:ozua8623-region-CA-st-Alberta-city-Calgary:6wdcv4gq".split(':')
    elif city == 'Edmonton':
        return 'us.novproxy.io:1000:ozua8623-region-CA-st-Alberta-city-Edmonton:6wdcv4gq'.split(':')
    elif city == 'Vancouver':
        return 'us.novproxy.io:1000:ozua8623-region-CA-st-British Columbia-city-Vancouver:6wdcv4gq'.split(':')
    elif city == 'Halifax':
        return 'us.novproxy.io:1000:ozua8623-region-CA-st-Nova Scotia-city-Halifax:6wdcv4gq'.split(':')
    elif city == 'Toronto':
        return 'us.novproxy.io:1000:ozua8623-region-CA-st-Ontario-city-Toronto:6wdcv4gq'.split(':')
    else:
        return None


def get_random_proxy() -> list[str] | None:
    """
    随机选择一个代理配置（按指纹浏览器数量随机取 IP）

    返回值:
        list[str] | None: 代理参数列表 `[host, port, user, pwd]`；无可用代理返回 None
    """
    proxy_list = [
        "us.novproxy.io:1000:zhiyu111-region-CA:zhiyu111",
        "us.novproxy.io:1000:zhiyu111-region-US:zhiyu111",
        "us.novproxy.io:1000:zhiyu222-region-CA:zhiyu222",
        "us.novproxy.io:1000:zhiyu222-region-US:zhiyu222",
    ]
    try:
        return random.choice(proxy_list).split(':')
    except Exception:
        return None

def get_all_proxies() -> list[list[str]]:
    """
    返回固定代理列表（与提供的代理一一对应）

    返回值:
        list[list[str]]: 每个元素为 `[host, port, user, pwd]`
    """
    proxy_list = [
        "us.novproxy.io:1000:zhiyu111-region-CA:zhiyu111",
        "us.novproxy.io:1000:zhiyu111-region-US:zhiyu111",
        "us.novproxy.io:1000:zhiyu222-region-CA:zhiyu222",
        "us.novproxy.io:1000:zhiyu222-region-US:zhiyu222",
    ]
    return [p.split(":") for p in proxy_list]


"""指纹浏览器操作"""
# 创建指纹浏览器


def create_fingerprint_browser(city: str | None = None):
    """
    创建指纹浏览器并执行一次流程（支持随机 IP 与指定城市）

    参数:
        city (str | None): 指定城市使用其对应代理；None 则使用随机代理并随机选择城市
    """
    browser_id = None
    try:
        if city is not None:
            proxy = get_proxy(city)
            if proxy is None:
                logger.error(f"{city} 未配置对应代理，结束该线程")
                return
            remark = city
        else:
            proxy = get_random_proxy()
            if proxy is None:
                logger.error("未获取到随机代理，结束该线程")
                return
            remark = "random-ip"
        logger.info("准备创建指纹浏览器")
        browser_id = bit_browser.bit_browser_create(
            remark=remark,
            host=proxy[0],
            port=proxy[1],
            proxy_user=proxy[2],
            proxy_pwd=proxy[3],
            proxy_type='socks5'
        )
        logger.debug(browser_id)
        # 打开指纹浏览器
        http = bit_browser.bit_browser_open(browser_id)
        logger.debug(http)
        auto = Auto(http)
        auto.open_url(
            "https://veritaconnect.ca/canadianbreadsettlement/en-us/Claimant/UnknownClaimForm")
        bol = auto.wait_home()
        if not bol:
            logger.error("进入首页失败，结束该线程")
            return

        bol = auto.click_continue()
        if not bol:
            logger.error("点击 Continue 失败，结束该线程")
            return
        auto.fill_questionnaire()
        # fill_city = city if city is not None else random.choice(['Calgary', 'Edmonton', 'Vancouver', 'Halifax', 'Toronto'])
        # auto.fill_questionnaire(fill_city)
        # time.sleep(5)
    finally:
        if browser_id:
            # 关闭指纹浏览器
            try:
                bit_browser.bit_browser_close(browser_id)
            except Exception as e:
                logger.error(f"关闭浏览器异常: {e}")
            # 删除指纹浏览器
            try:
                bit_browser.bit_browser_delete(browser_id)
            except Exception as e:
                logger.error(f"删除浏览器异常: {e}")


def run_city_forever(city: str):
    """
    持续循环运行指定城市流程：完成一次即关闭并删除浏览器，然后重新创建继续运行

    参数:
        city (str): 城市名称
    """
    while True:
        try:
            create_fingerprint_browser(city)
        except Exception as e:
            logger.error(f"{city} 流程异常: {e}")
        time.sleep(2)


def run_all_cities_concurrently(num: int):
    """
    多线程并发运行城市流程（支持随机选择）

    参数:
        num (int | None): 随机选择并启动的城市数量；None 表示全部
    """
    import threading
    threads = []
    for i in range(num):
        t = threading.Thread(target=run_random_ips_forever,
                             name=f"random-ip-thread-{i}")
        t.start()
        threads.append(t)
        logger.info(f"随机 IP 线程 {i} 已启动")
    for t in threads:
        t.join()
    logger.info("所有随机 IP 流程执行完成")


def run_random_ips_forever():
    """
    持续使用随机 IP 执行流程：每次完成后关闭并删除浏览器再重建
    """
    while True:
        try:
            create_fingerprint_browser(None)
        except Exception as e:
            logger.error(f"随机 IP 流程异常: {e}")
        time.sleep(2)


def run_random_ips_concurrently(num: int):
    """
    根据指纹浏览器数量并发运行流程（随机取 IP）

    参数:
        num (int): 并发指纹浏览器数量（每个使用随机代理）
    """
    import threading
    if num <= 0:
        logger.warning("num 不合法（<=0），不启动任何线程")
        return
    threads = []
    for i in range(num):
        t = threading.Thread(target=run_random_ips_forever,
                             name=f"random-ip-thread-{i}")
        t.start()
        threads.append(t)
        logger.info(f"随机 IP 线程 {i} 已启动")
    for t in threads:
        t.join()
    logger.info("随机 IP 并发流程执行完成")


def create_fingerprint_browser_with_proxy(proxy: list[str]):
    """
    使用指定代理创建指纹浏览器并执行一次流程（一一对应）

    参数:
        proxy (list[str]): `[host, port, user, pwd]`
    """
    browser_id = None
    try:
        if not proxy or len(proxy) < 4:
            logger.error("代理参数不完整，结束该线程")
            return
        logger.info(f"使用代理 {proxy[2]} 创建浏览器")
        browser_id = bit_browser.bit_browser_create(
            remark=f"{proxy[2]}",
            host=proxy[0],
            port=proxy[1],
            proxy_user=proxy[2],
            proxy_pwd=proxy[3],
            proxy_type='socks5'
        )
        logger.debug(f"创建浏览器 {browser_id}")
        http = bit_browser.bit_browser_open(browser_id)
        logger.debug(f"打开浏览器 {browser_id}")
        auto = Auto(http)
        auto.open_url(
            "https://veritaconnect.ca/canadianbreadsettlement/en-us/Claimant/UnknownClaimForm")
        bol = auto.wait_home()
        if not bol:
            logger.error("进入首页失败，结束该线程")
            return
        bol = auto.click_continue()
        if not bol:
            logger.error("点击 Continue 失败，结束该线程")
            return
        auto.fill_questionnaire()
    finally:
        if browser_id:
            try:
                bit_browser.bit_browser_close(browser_id)
            except Exception as e:
                logger.error(f"关闭浏览器异常: {e}")
            try:
                bit_browser.bit_browser_delete(browser_id)
            except Exception as e:
                logger.error(f"删除浏览器异常: {e}")


def run_proxies_forever(proxy: list[str]):
    """
    持续使用指定代理执行流程：完成后关闭并删除浏览器再重建

    参数:
        proxy (list[str]): `[host, port, user, pwd]`
    """
    while True:
        try:
            create_fingerprint_browser_with_proxy(proxy)
        except Exception as e:
            logger.error(f"固定代理流程异常: {e}")
        time.sleep(2)


def run_all_proxies_concurrently():
    """
    按固定代理列表一一创建并发浏览器
    """
    import threading
    proxies = get_all_proxies()
    if not proxies:
        logger.warning("未找到可用代理，结束执行")
        return
    threads = []
    for i, proxy in enumerate(proxies):
        t = threading.Thread(target=run_proxies_forever, args=(proxy,), name=f"proxy-thread-{i}")
        t.start()
        threads.append(t)
        logger.info(f"固定代理线程 {i} 已启动: {proxy[0]}:{proxy[1]} @ {proxy[2]}")
    for t in threads:
        t.join()
    logger.info("固定代理并发流程执行完成")


if __name__ == "__main__":
    # auto  = Auto()
    # auto.get_random_food('a')
    run_all_proxies_concurrently()