ayaka.playwright

浏览器截图

View Source

 1'''浏览器截图'''
 2from .driver import get_driver
 3import platform
 4from loguru import logger
 5from .config import ayaka_root_config
 6
 7running_on_windows = platform.system() == "Windows"
 8fastapi_reload = getattr(get_driver().config, "fastapi_reload", True)
 9
10if ayaka_root_config.use_playwright:
11    import datetime
12    from playwright.async_api import async_playwright, Browser, Page, Playwright
13    from contextlib import asynccontextmanager
14    from typing import AsyncIterator
15    from .driver import get_driver
16
17    driver = get_driver()
18    _browser: Browser = None
19    _playwright: Playwright = None
20
21    @asynccontextmanager
22    async def get_new_page(width=None, high_quality=False, **kwargs) -> AsyncIterator[Page]:
23        ''' 获取playwright Page对象，width设置屏幕宽度
24
25            使用示例：
26            ```
27            async with get_new_page(width=200) as p:
28                await p.goto(...)
29                await p.screenshot(...)
30            ```
31        '''
32        if width:
33            kwargs["viewport"] = {"width": width, "height": width}
34        if high_quality:
35            kwargs["device_scale_factor"] = 2
36        page = await _browser.new_page(**kwargs)
37        yield page
38        await page.close()
39
40    @driver.on_startup
41    async def startup():
42        with open("test.log", "a+", encoding="utf8") as f:
43            d = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
44            f.write(f"[{d}] 记录时间  playwright init\n")
45        print('''
46
47    playwright init
48
49        ''')
50        if running_on_windows and fastapi_reload:
51            logger.warning("playwright未加载，win平台请关闭fastapi reload功能")
52            return
53
54        global _browser, _playwright
55        _playwright = await async_playwright().start()
56        _browser = await _playwright.chromium.launch()
57
58    @driver.on_shutdown
59    async def shutdown():
60        # 查bug 怀疑卡死原因是没有正确退出
61        with open("test.log", "a+", encoding="utf8") as f:
62            d = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
63            f.write(f"[{d}] 记录时间  playwright close\n")
64        print('''
65
66    playwright close
67
68        ''')
69        if _browser:
70            await _browser.close()
71        if _playwright:
72            await _playwright.stop()
73else:
74    async def get_new_page(width=None, high_quality=False, **kwargs):
75        logger.warning("ayaka未开启playwright功能")

@asynccontextmanager

def get_new_page( width=None, high_quality=False, **kwargs) -> AsyncIterator[playwright.async_api._generated.Page]: View Source

22    @asynccontextmanager
23    async def get_new_page(width=None, high_quality=False, **kwargs) -> AsyncIterator[Page]:
24        ''' 获取playwright Page对象，width设置屏幕宽度
25
26            使用示例：
27            ```
28            async with get_new_page(width=200) as p:
29                await p.goto(...)
30                await p.screenshot(...)
31            ```
32        '''
33        if width:
34            kwargs["viewport"] = {"width": width, "height": width}
35        if high_quality:
36            kwargs["device_scale_factor"] = 2
37        page = await _browser.new_page(**kwargs)
38        yield page
39        await page.close()

获取playwright Page对象，width设置屏幕宽度

使用示例：

async with get_new_page(width=200) as p:
    await p.goto(...)
    await p.screenshot(...)

@driver.on_startup

async def startup(): View Source

41    @driver.on_startup
42    async def startup():
43        with open("test.log", "a+", encoding="utf8") as f:
44            d = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
45            f.write(f"[{d}] 记录时间  playwright init\n")
46        print('''
47
48    playwright init
49
50        ''')
51        if running_on_windows and fastapi_reload:
52            logger.warning("playwright未加载，win平台请关闭fastapi reload功能")
53            return
54
55        global _browser, _playwright
56        _playwright = await async_playwright().start()
57        _browser = await _playwright.chromium.launch()

@driver.on_shutdown

async def shutdown(): View Source

59    @driver.on_shutdown
60    async def shutdown():
61        # 查bug 怀疑卡死原因是没有正确退出
62        with open("test.log", "a+", encoding="utf8") as f:
63            d = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
64            f.write(f"[{d}] 记录时间  playwright close\n")
65        print('''
66
67    playwright close
68
69        ''')
70        if _browser:
71            await _browser.close()
72        if _playwright:
73            await _playwright.stop()