Coverage for human_requests/browsers/families/base.py: 86%

1from __future__ import annotations 

2 

3from abc import ABC, abstractmethod 

4from pathlib import Path 

5from typing import Any, Dict, Literal, Optional 

6 

7from playwright.async_api import Browser, BrowserContext, StorageState 

8 

9Family = Literal["playwright", "patchright", "camoufox"] 

10PlaywrightEngine = Literal["chromium", "firefox", "webkit"] 

11 

12 

13class DesiredConfig: 

14 """Unified "desired" configuration for all families.""" 

15 

16 __slots__ = ("family", "engine", "headless", "stealth", "launch_opts") 

17 

18 def __init__( 

19 self, 

20 *, 

21 family: Family, 

22 engine: PlaywrightEngine | None, 

23 stealth: bool, 

24 launch_opts: Dict[str, Any], 

25 ) -> None: 

26 self.family = family 

27 self.engine = engine 

28 self.stealth = stealth 

29 self.launch_opts = dict(launch_opts) # копия 

30 

31 

32class BrowserFamily(ABC): 

33 """Family interface. Implements idempotent start and soft restarts internally.""" 

34 

35 @property 

36 @abstractmethod 

37 def name(self) -> Family: # noqa: D401 

38 """Family name.""" 

39 raise NotImplementedError 

40 

41 @abstractmethod 

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

43 """Idempotent launch/restart according to cfg.""" 

44 raise NotImplementedError 

45 

46 @abstractmethod 

47 async def close(self) -> None: 

48 """Close all family resources (browser + runtime).""" 

49 raise NotImplementedError 

50 

51 @property 

52 @abstractmethod 

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

54 """Current Browser or None if not started.""" 

55 raise NotImplementedError 

56 

57 async def new_context( 

58 self, 

59 *, 

60 storage_state: StorageState | str | Path | None = None, 

61 ) -> BrowserContext: 

62 await self._ensure() 

63 assert self.browser is not None 

64 return await self.browser.new_context(storage_state=storage_state) 

65 

66 async def _ensure(self) -> None: 

67 if self.browser is None: 

68 # Семейство само знает «последнюю cfg». Упрощённо: бросаем, если не стартовало. 

69 raise RuntimeError(f"{self.name}: not started yet")