How to use test_video_should_exist method in Playwright Python

The Windows implementation of asyncio can use two event loop implementations: SelectorEventLoop, default before Python 3.8, required when using Twisted. ProactorEventLoop, default since Python 3.8, cannot work with Twisted.

So on Python 3.8+ the event loop class needs to be changed.

Changed in version 2.6.0: The event loop class is changed automatically when you change the TWISTED_REACTOR setting or call install_reactor().

To change the event loop class manually, call the following code before installing the reactor:

import asyncio
asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())

You can put this in the same function that installs the reactor, if you do that yourself, or in some code that runs before the reactor is installed, e.g. settings.py.

Documentation: The Windows implementation of asyncio

It looks like there is some kind of lazy load, at first, first 8 elements are loaded, if you scroll down, more content will load

Not guaranteed for success. Change the parameters in function async_cf_retry if it tends to fail. I wrote this code by modifying my heavy modules, so it may not work elegantly.

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
)))

If download <Download url='https://www.historique-meteo.net/site/export.php?ville_id=1067' suggested_filename='export-aix-les-bains.csv'> is shown in the console, the browser must be downloading your file.
This code may keep trying to get a cloudflare clearance even if the file has been downloaded. Edit your strategies in async_cf_retry if this happens.
Use persisent context offered by playwright to save and reuse your cloudflare clearance. Consider downloading with httpx.

The best way to do it is using playwright docker image as base image, then you won't need to install its dependencies. Take a look to the documentation here: https://playwright.dev/docs/docker

Problem is, that these are not text or input elements, but divs, which inner HTML is being changed when typing. You can check that, by opening WhatsApp in the browser, opening the dev tools, and typing something into the field.

You can identify those divs via the title attribute, as those seem to be unique: //div[@title="Search text box"]

Playwright cannot change the inner HTML of a div, but you could do via Javascript, which you let evaluate by Playwright.

The Javascript would be (you can evaluate that in the dev tools' console):

document.querySelector('div[title^=\'Search text box\']').innerHTML = 'contact name';

To evaluate this from your python code:

page.evaluate('() => document.querySelector('div[title^=\'Search text box\']').innerHTML = 'contact name'')

See documentation for more details: https://playwright.dev/python/docs/evaluating

afaik, playwright uses a virtual keyboard and does not exactly send the same signal as a real keyboard to the browser application.

consider going native and using something closer to kernel like pyautogui, after you've focused you window:

import pyautogui
pyautogui.press('f5')

There is no outer_html out of the box. But it's not hard to implement it:

locator.evaluate("el => el.outerHTML") 

Here's how to initialize playwright browser globally without context managers.

In your particular code, this will work like below:

from playwright.sync_api import Playwright, sync_playwright

locationList = [
    'London Luton Airport',
    'London Hilton',
    'London City',
    'London Wembley',
    'London Battersea',
    'London Shepherds Bush'
]

playwright = sync_playwright().start()  # <-- Use this to initialize playwright globally

chromium = playwright.chromium
browser = chromium.launch(channel="chrome", headless=False)
page = browser.new_page()
page.set_viewport_size({"width": 1920, "height": 1080})
page.goto("https://www.sixt.co.uk/")
page.locator("[data-testid=\"uc-accept-all-button\"]").click()
page.locator("[placeholder=\"Find a location\"]").click()
page.locator("[placeholder=\"Find a location\"]").fill("luton")
page.locator("text=London Luton Airport (GB)").click()
page.locator("button:has-text(\"Show offers\")").click()

def run(playwright: Playwright):

    page.locator("[placeholder=\"Find a location\"]").click()
    page.locator("[placeholder=\"Find a location\"]").fill(loc)
    page.locator('//div[text()="' + loc + '"]').click()
    carnames = page.locator("//h2[@class='vehicle-item__title']")
    #Get all carnames in a list
    carnamelist = [cars.text for cars in carnames]

for loc in locationList:
    run(playwright)

playwright.stop()  # --> Cleanup resources properly when done

However, you can also initialize page and browser from inside the context manager you create in the original question, and it will work as expected.

Not guaranteed for success. Change the parameters in function async_cf_retry if it tends to fail. I wrote this code by modifying my heavy modules, so it may not work elegantly.

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
)))

If download <Download url='https://www.historique-meteo.net/site/export.php?ville_id=1067' suggested_filename='export-aix-les-bains.csv'> is shown in the console, the browser must be downloading your file.
This code may keep trying to get a cloudflare clearance even if the file has been downloaded. Edit your strategies in async_cf_retry if this happens.
Use persisent context offered by playwright to save and reuse your cloudflare clearance. Consider downloading with httpx.