from pathlib import Path import os from loguru import logger import get_all_model from playwright.async_api import BrowserContext as ASyncContext, async_playwright import hcaptcha_challenger as solver from hcaptcha_challenger.agents import AgentT, Malenia get_all_model.download_all() # Init local-side of the ModelHub solver.install(upgrade=True) # Save dataset to current working directory tmp_dir = Path(__file__).parent.joinpath("tmp_dir") @logger.catch async def hit_challenge(context: ASyncContext, host, sitekey, user_data_dir, times: int = 8): await context.route('**/*', lambda route, request: route_continuation(route, request, host, sitekey)) page = context.pages[0] agent = AgentT.from_page(page=page, tmp_dir=tmp_dir) await page.goto(f"https://{host}") await agent.handle_checkbox() for pth in range(1, times): result = await agent() print(f">> {pth} - Challenge Result: {result}") match result: case agent.status.CHALLENGE_BACKCALL: await page.wait_for_timeout(500) fl = page.frame_locator(agent.HOOK_CHALLENGE) await fl.locator("//div[@class='refresh button']").click() case agent.status.CHALLENGE_SUCCESS: rqdata = agent.cr.__dict__ print(os.system(f"rm -rf {tmp_dir}")) print(os.system(f"rm -rf {user_data_dir}")) return rqdata["generated_pass_UUID"] async def route_continuation(route, request, host, sitekey): # 检查请求的URL,只拦截特定网站的请求 if request.url == f"https://{host}/": print("start to solve") # 修改DNS解析结果 await route.fulfill(status=200, body=""" hCAPTCHA 演示

""".replace("%%%%%%%%%%%", sitekey)) else: # 对于其他网站的请求,不做修改 await route.continue_() async def bytedance(host, sitekey, user_data_dirs): # playwright install firefox --with-deps async with async_playwright() as p: context = await p.firefox.launch_persistent_context( user_data_dir=Path(__file__).parent.joinpath(user_data_dirs), headless=True, locale="en-US" ) await Malenia.apply_stealth(context) token = await hit_challenge(context, host, sitekey, Path(__file__).parent.joinpath(user_data_dirs)) return token