使用python的剧作家-从CloudFlare下载文件

Question

我正在使用Python中的CloudFlare：https://www.historique-meteo.net/site/export.php?ville_id=1067中的PlayWright下载这个文件

但不可能成功..。你能帮上忙吗？

注意:没有点击执行下载，只需等待5秒进行JS自动检查。

这是我的代码：

from playwright.sync_api import sync_playwright

def run(playwright):
    browser = playwright.chromium.launch(headless=False)
    context = browser.new_context(accept_downloads=True)

    # Open new page
    page = context.new_page()

    # Go to site
    page.goto("https://www.historique-meteo.net/site/export.php?ville_id=1067")

    # Download
    page.on("download", lambda download: download.save_as(download.suggested_filename))

    context.close()
    browser.close()

with sync_playwright() as playwright:
    run(playwright)

非常感谢！)

Answer 1

不能保证成功。如果函数async_cf_retry容易失败，则更改它的参数。我是通过修改我的重模块来编写这段代码的，因此它可能无法很好地工作。

import re
import asyncio
from playwright.async_api import async_playwright, Error, Page
from cf_clearance import stealth_async
import httpx

# local_client = httpx.AsyncClient(verify=False)


async def async_cf_retry(page: Page, tries=10) -> bool:
    # use tries=-1 for infinite retries
    # excerpted from `from cf_clearance import async_retry`
    success = False
    while tries != 0:
        try:
            title = await page.title()
        except Error:
            tries -= 1
            await asyncio.sleep(1)
        else:
            # print(title)
            if title == 'Please Wait... | Cloudflare':
                await page.close()
                raise NotImplementedError('Encountered recaptcha. Check whether your proxy is an elite proxy.')
            elif title == 'Just a moment...':
                tries -= 1
                await asyncio.sleep(5)
            elif "www." in title:
                await page.reload()
                tries -= 1
                await asyncio.sleep(5)
            else:
                success = True
                break
    return success


async def get_client_with_clearance(proxy: str = None):
    async def get_one_clearance(proxy=proxy, logs=False):
        # proxy = {"server": "socks5://localhost:7890"}
        if type(proxy) is str:
            proxy = {'server': proxy}
        async with async_playwright() as p:
            browser = await p.chromium.launch(headless=False, proxy=proxy,
                                              args=["--window-position=1000,800", "--disable-web-security",
                                                    "--disable-webgl"])
            page = await browser.new_page()#viewport={"width": 0, "height": 0})
            await stealth_async(page)
            
            # Download
            async def on_download(download):
                print('download', download)
                await download.save_as(download.suggested_filename)
            page.on("download", on_download)

            if logs:
                def log_response(intercepted_response):
                    print("a response was received:", intercepted_response.url)
                
                page.on("response", log_response)
            
            await page.goto("https://www.historique-meteo.net/site/export.php?ville_id=1067")
            res = await async_cf_retry(page)
            if res:
                cookies = await page.context.cookies()
                cookies_for_httpx = {cookie['name']: cookie['value'] for cookie in cookies}
                ua = await page.evaluate('() => {return navigator.userAgent}')
                # print(ua)
            else:
                await page.close()
                raise InterruptedError("cf challenge fail")
            await asyncio.sleep(10000)
            return ua, cookies_for_httpx
    
    ua, cookies_for_httpx = await get_one_clearance(logs=True)
    print(cookies_for_httpx)


print(asyncio.get_event_loop().run_until_complete(get_client_with_clearance(
    # proxy='http://localhost:8888'
    # use proxifier on windows as an elite proxy
)))

如果控制台中显示了download <Download url='https://www.historique-meteo.net/site/export.php?ville_id=1067' suggested_filename='export-aix-les-bains.csv'>，浏览器必须下载您的文件。

即使文件已经下载，此代码也可能一直试图获得cloudflare权限。如果发生这种情况，请在async_cf_retry中编辑您的策略。

使用剧作家提供的persisent context保存和重用您的cloudflare许可。考虑使用httpx下载。