Coverage for human_requests/browsers/families/playwright_family.py: 93%

1from __future__ import annotations 

2 

3from typing import Any, Dict, Optional 

4 

5from playwright.async_api import Browser, async_playwright 

6 

7from .base import BrowserFamily, DesiredConfig, Family, PlaywrightEngine 

8 

9 

10class PlaywrightFamily(BrowserFamily): 

11 """ 

12 Standard Playwright (with stealth wrapper on demand). 

13 Restarts only what has changed: PW engine when stealth changes, 

14 browser when engine/headless/launch_opts change. 

15 """ 

16 

17 def __init__(self) -> None: 

18 self._pw: Any | None = None 

19 self._stealth_cm: Any | None = None 

20 self._browser: Browser | None = None 

21 

22 # кэш использованных опций 

23 self._engine_used: PlaywrightEngine | None = None 

24 self._stealth_used: bool | None = None 

25 self._launch_opts_used: Dict[str, Any] | None = None 

26 

27 @property 

28 def name(self) -> Family: 

29 return "playwright" 

30 

31 @property 

32 def browser(self) -> Optional[Browser]: 

33 return self._browser 

34 

35 async def start(self, cfg: DesiredConfig) -> None: 

36 assert cfg.family == "playwright", "wrong family for PlaywrightFamily" 

37 assert cfg.engine in ("chromium", "firefox", "webkit") 

38 

39 # Нужен ли перезапуск PW (stealth изменился / ещё не поднят) 

40 need_pw_restart = self._pw is None or ( 

41 self._stealth_used is not None and self._stealth_used != cfg.stealth 

42 ) 

43 if need_pw_restart: 

44 await self._stop_pw() # мягко закрыть PW-уровень 

45 if cfg.stealth: 

46 try: 

47 from playwright_stealth import Stealth # type: ignore[import-untyped] 

48 except Exception: 

49 raise RuntimeError( 

50 "stealth=True, but the 'playwright-stealth' package is not installed. " 

51 "Install it with: pip install playwright-stealth" 

52 ) 

53 self._stealth_cm = Stealth().use_async(async_playwright()) 

54 self._pw = await self._stealth_cm.__aenter__() 

55 else: 

56 self._pw = await async_playwright().__aenter__() 

57 

58 # Нужен ли перелонч браузера 

59 need_browser_relaunch = ( 

60 need_pw_restart 

61 or self._browser is None 

62 or self._engine_used != cfg.engine 

63 or self._launch_opts_used != cfg.launch_opts 

64 ) 

65 if need_browser_relaunch: 

66 if self._browser is not None: 

67 await self._browser.close() 

68 self._browser = None 

69 

70 assert self._pw is not None 

71 launcher = getattr(self._pw, cfg.engine) 

72 

73 kwargs = dict(cfg.launch_opts) 

74 self._browser = await launcher.launch(**kwargs) 

75 

76 # обновить кэш 

77 self._engine_used = cfg.engine 

78 self._stealth_used = cfg.stealth 

79 self._launch_opts_used = dict(cfg.launch_opts) 

80 

81 async def close(self) -> None: 

82 # Закрыть браузер 

83 if self._browser is not None: 

84 await self._browser.close() 

85 self._browser = None 

86 

87 await self._stop_pw() 

88 

89 # сброс кэша 

90 self._engine_used = None 

91 self._stealth_used = None 

92 self._launch_opts_used = None 

93 

94 async def _stop_pw(self) -> None: 

95 # Сначала закрыть stealth CM (если был) — он закрывает и свой PW 

96 if self._stealth_cm is not None: 

97 await self._stealth_cm.__aexit__(None, None, None) 

98 self._stealth_cm = None 

99 self._pw = None 

100 elif self._pw is not None: 

101 await self._pw.stop() 

102 self._pw = None