Coverage for src/extratools_html/__init__.py: 50%

1from __future__ import annotations 

2 

3import asyncio 

4import ssl 

5from collections.abc import Iterable 

6from contextlib import suppress 

7from datetime import timedelta 

8from enum import StrEnum 

9from http import HTTPStatus 

10from typing import Any, cast 

11from urllib.parse import urlparse 

12 

13import backoff 

14import httpx 

15import minify_html 

16import truststore 

17from blob_dict.blob import StrBlob 

18from blob_dict.dict.path import LocalPath, PathBlobDict 

19from extratools_core.path import cleanup_dir_by_ttl 

20from extratools_core.typing import PathLike 

21from html2text import HTML2Text 

22 

23with suppress(ImportError): 

24 from playwright.async_api import Browser, async_playwright, expect 

25 from playwright.async_api import TimeoutError as PlaywrightTimeoutError 

26 

27from .cleanup import cleanup_page 

28 

29MAX_TRIES: int = 3 

30MAX_TIMEOUT: int = 60 

31REQUEST_TIMEOUT: int = 10 

32# In milliseconds 

33PRE_ACTION_TIMEOUT: int = 10 * 1_000 

34 

35ctx = truststore.SSLContext(ssl.PROTOCOL_TLS_CLIENT) 

36 

37# TODO: Make cache path/TTL configurable via configuration file (in TOML). 

38# It will also allow support for other non-local path (like `CloudPath`). 

39CACHE_PATH: PathLike = LocalPath("~/.http-cache").expanduser() 

40CACHE_TTL: timedelta = timedelta(days=1) 

41 

42cache = PathBlobDict(CACHE_PATH, blob_class=StrBlob, ttl=CACHE_TTL) 

43cache.create() 

44 

45# Trigger cleanup here as we do not have CRON job or daemon process 

46# If available, native solution is preferred (like S3's object lifecycle management). 

47# Use longer TTL here than cache TTL in case we still somehow need raw data 

48list(cleanup_dir_by_ttl(CACHE_PATH, timedelta(days=30))) 

49 

50 

51KEEP_CLICK_INTERVAL: timedelta = timedelta(seconds=5) 

52KEEP_CLICK_MAX_TRIES: int = 100 

53 

54KEEP_SCROLL_TO_BE_VISIBLE_INTERVAL: timedelta = timedelta(seconds=5) 

55KEEP_SCROLL_TO_BE_VISIBLE_MAX_TRIES: int = 100 

56 

57 

58class PageElementAction(StrEnum): 

59 CLICK = "click" 

60 KEEP_CLICK = "keep_click" 

61 TO_BE_VISIBLE = "to_be_visible" 

62 KEEP_SCROLL_TO_BE_VISIBLE = "keep_scroll_to_be_visible" 

63 

64 

65async def __download_via_request( 

66 page_url: str, 

67 *, 

68 user_agent: str | None = None, 

69) -> str | None: 

70 # https://www.python-httpx.org/advanced/ssl/ 

71 async with httpx.AsyncClient(verify=ctx) as client: 

72 response: httpx.Response = await client.get( 

73 page_url, 

74 follow_redirects=True, 

75 timeout=REQUEST_TIMEOUT, 

76 headers=( 

77 { 

78 "User-Agent": user_agent, 

79 } if user_agent 

80 else {} 

81 ), 

82 ) 

83 

84 if response.status_code == HTTPStatus.TOO_MANY_REQUESTS: 

85 # It also triggers backoff if necessary 

86 return None 

87 

88 response.raise_for_status() 

89 

90 return response.text 

91 

92 

93async def __download_via_browser( 

94 page_url: str, 

95 *, 

96 user_agent: str | None = None, 

97 pre_actions: Iterable[tuple[str, PageElementAction]] | None = None, 

98) -> str | None: 

99 async with async_playwright() as playwright: # pyright: ignore [reportPossiblyUnboundVariable] 

100 browser: Browser = await playwright.chromium.launch() 

101 await browser.new_context( 

102 user_agent=user_agent, 

103 ) 

104 

105 page = await browser.new_page() 

106 await page.route( 

107 "**/*", 

108 lambda route: ( 

109 route.abort() 

110 # https://playwright.dev/python/docs/api/class-request#request-resource-type 

111 if route.request.resource_type in { 

112 "font", 

113 "image", 

114 "media", 

115 } 

116 else route.continue_() 

117 ), 

118 ) 

119 response = await page.goto(page_url) 

120 if not response: 

121 return None 

122 if response.status == HTTPStatus.TOO_MANY_REQUESTS: 

123 # It also triggers backoff if necessary 

124 return None 

125 

126 for selector, action in pre_actions or []: 

127 with suppress(AssertionError, PlaywrightTimeoutError): # pyright: ignore [reportPossiblyUnboundVariable] 

128 match action: 

129 case PageElementAction.CLICK: 

130 await page.locator(selector).click( 

131 timeout=PRE_ACTION_TIMEOUT, 

132 # Allow click even current element is covered by other elements. 

133 # Otherwise, other pre-actions are needed before this pre-action 

134 # to dismiss those covering elements. 

135 # However, it is possible that dismissing those covering elements 

136 # is necessary logic for page to function properly. 

137 force=True, 

138 ) 

139 case PageElementAction.KEEP_CLICK: 

140 for _ in range(KEEP_CLICK_MAX_TRIES): 

141 if not await page.locator(selector).is_visible(): 

142 break 

143 

144 await asyncio.sleep(KEEP_CLICK_INTERVAL.total_seconds()) 

145 

146 await page.locator(selector).click( 

147 timeout=PRE_ACTION_TIMEOUT, 

148 force=True, 

149 ) 

150 

151 case PageElementAction.TO_BE_VISIBLE: 

152 await expect(page.locator(selector)).to_be_visible( # pyright: ignore [reportPossiblyUnboundVariable] 

153 timeout=PRE_ACTION_TIMEOUT, 

154 ) 

155 case PageElementAction.KEEP_SCROLL_TO_BE_VISIBLE: 

156 for _ in range(KEEP_SCROLL_TO_BE_VISIBLE_MAX_TRIES): 

157 await asyncio.sleep(KEEP_SCROLL_TO_BE_VISIBLE_INTERVAL.total_seconds()) 

158 

159 while not await page.locator(selector).is_visible(): 

160 await page.mouse.wheel(0, 1000) 

161 

162 html: str = await page.content() 

163 

164 await browser.close() 

165 

166 return html 

167 

168 

169def get_cache_key(page_url: str) -> str: 

170 parse_result = urlparse(page_url) 

171 

172 # Need to handle reserved characters for filename 

173 # https://en.wikipedia.org/wiki/Filename#Reserved_characters_and_words 

174 root_dir: str = parse_result.netloc.replace(":", "_") 

175 

176 path: str = parse_result.path or "/" 

177 # Add default filename 

178 if path.endswith("/"): 

179 path += "?" 

180 

181 if parse_result.query: 

182 if not path.endswith("/?"): 

183 path += "?" 

184 path += parse_result.query 

185 

186 return root_dir + path 

187 

188 

189@backoff.on_predicate( 

190 backoff.expo, 

191 max_tries=MAX_TRIES, 

192 max_time=MAX_TIMEOUT, 

193) 

194async def download_page_async( 

195 page_url: str, 

196 *, 

197 cleanup: bool = False, 

198 text_only: bool = False, 

199 minify: bool = True, 

200 user_agent: str | None = None, 

201 use_browser: bool = False, 

202 pre_actions: Iterable[tuple[str, PageElementAction]] | None = None, 

203 use_cache: bool = True, 

204) -> str | None: 

205 page_html: str | None 

206 cache_key: str = get_cache_key(page_url) 

207 

208 if use_cache and (cache_blob := cache.get(cache_key)): 

209 page_html = cast("StrBlob", cache_blob).as_str() 

210 elif use_browser: 

211 page_html = await __download_via_browser( 

212 page_url, 

213 user_agent=user_agent, 

214 pre_actions=pre_actions, 

215 ) 

216 else: 

217 page_html = await __download_via_request( 

218 page_url, 

219 user_agent=user_agent, 

220 ) 

221 if page_html is None: 

222 return None 

223 

224 cache[cache_key] = StrBlob(page_html) 

225 

226 if minify: 

227 page_html = minify_html.minify(page_html) 

228 

229 if cleanup: 

230 page_html = await cleanup_page(page_html) 

231 

232 if text_only: 

233 h = HTML2Text() 

234 h.ignore_images = True 

235 h.ignore_links = True 

236 return h.handle(page_html) 

237 

238 return page_html 

239 

240 

241def download_page( 

242 page_url: str, 

243 **kwargs: Any, 

244) -> str | None: 

245 return asyncio.run(download_page_async( 

246 page_url, 

247 **kwargs, 

248 ))