hanzhiyu/ca_auto_table
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

0.0.4

Browse Source
This commit is contained in:
bvwl
2025-11-20 21:42:20 +08:00
parent e2d2b0b75b
commit 150d45e3eb
10 changed files with 491 additions and 126 deletions
Download Patch File Download Diff File Expand all files Collapse all files

329
spider/main.py
View File

@@ -8,19 +8,19 @@ 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):
def __init__(self,http:str=None):
self.browser = Chromium(http)
self.tab = self.browser.latest_tab
pass
# cf打码
def solve_cloudflare(self):
tab = self.browser.latest_tab
for _ in range(8):
self.tab.wait(1)
for _ in range(5):
self.tab.wait(0.5)
try:
shadow1 = tab.ele(
'x://*[@name="cf-turnstile-response"]').parent().shadow_root
@@ -31,16 +31,16 @@ class Auto:
if shadow2:
logger.debug("找到Cloudflare iframe body shadow root")
status = shadow2.ele(
'x://span[text()="Success!"]', timeout=1)
'x://span[text()="Success!"]', timeout=0.5)
if status:
logger.debug("Cloudflare验证成功")
return True
checkbox = shadow2.ele(
'x://input[@type="checkbox"]', timeout=1)
'x://input[@type="checkbox"]', timeout=0.5)
if checkbox:
checkbox.click()
logger.debug("点击Cloudflare复选框")
tab.wait(2)
tab.wait(3)
logger.debug("重新获取状态")
# return False
except Exception as e:
@@ -59,7 +59,7 @@ class Auto:
logger.debug("等待进入首页")
jc = 0
while True:
if jc > 5:
if jc > 3:
logger.error("等待进入首页超过5次，未成功")
return False
self.tab.wait(1)
@@ -83,7 +83,7 @@ class Auto:
logger.debug("点击Continue按钮")
jc = 0
while True:
if jc > 5:
if jc > 3:
logger.error("点击Continue按钮超过5次，未成功")
return False
try:
@@ -100,7 +100,19 @@ class Auto:
continue_button.click()
logger.debug("点击Continue按钮成功")
self.tab.wait(1.5)
bol = self.tab.ele('t:div@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: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:
@@ -110,10 +122,6 @@ class Auto:
logger.debug("异常界面")
self.tab.wait(1)
return self.click_continue(bl=True)
# bol = self.tab.ele('t:h2@text()=Claim Form', timeout=1)
# if bol:
# logger.debug("成功进入问卷界面")
# return True
html = self.tab.url
logger.debug(f"当前URL: {html}")
if 'https://veritaconnect.ca/canadianbreadsettlement/en-us/Claimant/UnknownClaimForm' in html:
@@ -142,81 +150,215 @@ class Auto:
province = random.choice(list(cities.keys()))
return province,random.choice(cities.get(province, []))
def get_province_by_city(self, city: str) -> str | None:
"""
根据城市名称解析对应省份
参数:
city (str): 城市名称，例如 `Calgary`、`Edmonton` 等
返回值:
str | None: 对应的省份名称；未匹配返回 None
"""
mapping = {
"Calgary": "Alberta",
"Edmonton": "Alberta",
"Vancouver": "British Columbia",
"Halifax": "Nova Scotia",
"Toronto": "Ontario",
}
return mapping.get(city)
# 随机实物
def get_random_food(self, 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} 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):
province, city = self.get_random_city()
info = get_random_canada_info(province, city)
first_name = info["firstname"]
last_name = info["lastname"]
# 将生日格式从 '8/28/1995' 转为 'yyyy-mm-dd'，日月不足两位补0
birthday = info["birthday"]
current_address = info["address_str"]
city = info["city_name"]
province = info["province"]
postal_code = info["postcode"]
email = 'sfsf@qq.com'
phone = info["phone"]
text = '3333'
# 人数
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)
# logger.debug(f"点击Submit按钮")
# self.tab.ele('t:button@text():Submit').click()
def fill_questionnaire(self, city: str):
"""
根据传入的城市解析省份并完成问卷填写
参数:
city (str): 线程启动时传入的城市名称，用于匹配省份并填写数据
"""
try:
province = self.get_province_by_city(city)
if province is None:
logger.error(f"未找到城市对应省份: {city}")
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(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)
for i in range(3):
bol = self.solve_cloudflare()
if not bol:
logger.debug("Cloudflare验证失败.")
self.tab.wait(0.1)
else:
logger.debug("Cloudflare验证成功.")
logger.debug(f"点击Submit按钮")
self.tab.ele('t:button@text():Submit').click()
break
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
)
self.tab.wait(2)
except Exception as e:
logger.error(f"填写问卷失败: {e}")
# 取对应城市的代理
@@ -247,6 +389,9 @@ def create_fingerprint_browser(city: str):
browser_id = None
try:
proxy = get_proxy(city)
if proxy is None:
logger.error(f"{city} 未配置对应代理，结束该线程")
return
logger.info(f"{city} 准备创建指纹浏览器")
browser_id = bit_browser.bit_browser_create(
remark=city,
@@ -272,7 +417,7 @@ def create_fingerprint_browser(city: str):
if not bol:
logger.error(f"{city} 点击 Continue 失败，结束该线程")
return
auto.fill_questionnaire()
auto.fill_questionnaire(city)
time.sleep(5)
finally:
if browser_id:
@@ -307,17 +452,19 @@ def run_all_cities_concurrently():
"""
import threading
cities = ['Calgary', 'Edmonton', 'Vancouver', 'Halifax', 'Toronto']
# cities = ['Calgary']
threads = []
for city in cities:
t = threading.Thread(target=run_city_forever, args=(city,), name=f"{city}-thread")
t.start()
threads.append(t)
logger.info(f"{city} 线程已启动")
time.sleep(2)
# time.sleep(2)
for t in threads:
t.join()
logger.info("所有城市流程执行完成")
if __name__ == "__main__":
# auto = Auto()
# auto.get_random_food('a')
run_all_cities_concurrently()