Coverage for worker_scraper.py: 100%

1import time

2from fake_useragent import UserAgent

3import redis

4from selenium.webdriver.chrome.options import Options

5from selenium.webdriver.support.ui import WebDriverWait

6from selenium.webdriver.support import expected_conditions as EC

7from selenium.webdriver.common.by import By

8from seleniumwire import webdriver

9import os

10import signal

11import sys

12import json

13from constants import WAITING_TIME_FOR_ELEMENT as WT, PARSER_BIN_PATH as PP

14from logger import *

15from exceptions import *

16import subprocess

17from typing import Optional, Tuple, Dict

19from prometheus_client import disable_created_metrics, CollectorRegistry, Counter, Histogram, push_to_gateway

20import socket

22# metrics

23disable_created_metrics()

25REGISTRY = CollectorRegistry()

27JOBS_PROCESSED = Counter(

28 'scraper_worker_jobs_processed_total',

29 'Total jobs processed by this worker',

30 ['status'], # success | scrape_error | parse_error | internal_error

31 registry=REGISTRY

32)

33JOB_DURATION = Histogram(

34 'scraper_worker_job_duration_seconds',

35 'Total time to process a job',

36 buckets=[5, 15, 30, 60, 120, 300, 600],

37 registry=REGISTRY

38)

39SCRAPE_DURATION = Histogram(

40 'scraper_worker_scrape_duration_seconds',

41 'Time spent in the scraper',

42 buckets=[5, 15, 30, 60, 120],

43 registry=REGISTRY

44)

45SCRAPE_FAILED = Counter(

46 'scraper_worker_scrape_failed_total',

47 'Total scrapes failed by this worker',

48 registry=REGISTRY

49)

50PARSE_DURATION = Histogram(

51 'scraper_worker_parse_duration_seconds',

52 'Time spent in the parser subprocess',

53 buckets=[1, 2, 5, 10, 30],

54 registry=REGISTRY

55)

56PARSE_FAILED = Counter(

57 'scraper_worker_parse_failed_total',

58 'Total parses failed by this worker',

59 registry=REGISTRY

60)

62PUSHGATEWAY_URL = os.getenv('PUSHGATEWAY_URL', 'http://pushgateway:9091')

63WORKER_INSTANCE = socket.gethostname()

65# -----------------------

67def push_metrics():

68 try:

69 push_to_gateway(

70 PUSHGATEWAY_URL,

71 job='patent_worker',

72 grouping_key={'instance': WORKER_INSTANCE},

73 registry=REGISTRY

74 )

75 except Exception:

76 pass

78class DriverFactory:

79 def __init__(self, proxy: Optional[str] = None, user_agent_generator=None):

80 self.proxy = proxy

81 self.user_agent_generator = user_agent_generator or self._default_ua_generator

83 def _default_ua_generator(self):

84 ua = UserAgent()

85 return ua.random

87 def create_driver(self):

88 chrome_options = Options()

90 chrome_options.add_argument(f"user-agent={self.user_agent_generator()}")

92 chrome_options.add_argument('--headless=new')

93 chrome_options.add_argument('--no-sandbox')

94 chrome_options.add_argument('--disable-dev-shm-usage')

95 chrome_options.add_argument('--disable-gpu')

96 chrome_options.add_argument('--disable-dev-tools')

97 chrome_options.add_argument('--disable-software-rasterizer')

98 chrome_options.add_argument('--disable-extensions')

99 chrome_options.add_argument('--disable-background-networking')

100 chrome_options.add_argument('--disable-background-timer-throttling')

101 chrome_options.add_argument('--disable-backgrounding-occluded-windows')

102 chrome_options.add_argument('--disable-breakpad')

103 chrome_options.add_argument('--disable-client-side-phishing-detection')

104 chrome_options.add_argument('--disable-component-extensions-with-background-pages')

105 chrome_options.add_argument('--disable-default-apps')

106 chrome_options.add_argument('--disable-hang-monitor')

107 chrome_options.add_argument('--disable-ipc-flooding-protection')

108 chrome_options.add_argument('--disable-popup-blocking')

109 chrome_options.add_argument('--disable-prompt-on-repost')

110 chrome_options.add_argument('--disable-sync')

111 chrome_options.add_argument('--force-color-profile=srgb')

112 chrome_options.add_argument('--metrics-recording-only')

113 chrome_options.add_argument('--no-first-run')

114 chrome_options.add_argument('--password-store=basic')

115 chrome_options.add_argument('--use-mock-keychain')

116 chrome_options.add_argument('--disable-features=TranslateUI,BlinkGenPropertyTrees,IsolateOrigins,site-per-process')

117 chrome_options.add_argument('--disable-blink-features=AutomationControlled')

118 chrome_options.add_argument('--js-flags=--max-old-space-size=256')

119

120 # disable as much stuff as possible

121 prefs = {

122 'profile.default_content_setting_values': {

123 'images': 2,

124 'stylesheet': 2,

125 'plugins': 2,

126 'popups': 2,

127 'geolocation': 2,

128 'notifications': 2,

129 'media_stream': 2,

130 },

131 'profile.managed_default_content_settings': {

132 'images': 2,

133 'stylesheets': 2

134 }

135 }

136

137 chrome_options.add_experimental_option('prefs', prefs)

138 chrome_options.page_load_strategy = 'none' # prevents endless loading

139 chrome_options.add_argument('--window-size=1280,720')

140

141 try:

142 if self.proxy:

143 seleniumwire_options = {

144 "proxy": {

145 "http": self.proxy,

146 },

147 }

148 driver = webdriver.Chrome(

149 options=chrome_options,

150 seleniumwire_options=seleniumwire_options

151 )

152 else:

153 driver = webdriver.Chrome(options=chrome_options)

154

155 driver.set_page_load_timeout(60)

156 driver.implicitly_wait(5)

157

158 return driver

159 except Exception as e:

160 return None

161

162

163class PatentScraper:

164 def __init__(self, driver_factory: DriverFactory, logger, wait_time=WT):

165 self.driver_factory = driver_factory

166 self.logger = logger

167 self.wait_time = wait_time

168 self.driver = None

169

170 def scrape(self, url: str) -> Tuple[Dict, str]:

171 if not self.driver:

172 self.driver = self.driver_factory.create_driver()

173

174 if not self.driver:

175 raise Exception("Could not initialize WebDriver")

176

177 try:

178 self.driver.get(url)

179

180 id_main = 'detailMainForm:MyTabViewId'

181 WebDriverWait(self.driver, self.wait_time).until(

182 EC.presence_of_element_located((By.ID, id_main))

183 )

184

185 np_data = self._scrape_national_phase()

186

187 xml_content = self._scrape_xml_document()

188

189 np_json_dict = {"np": np_data}

190

191 return (np_json_dict, xml_content)

192

193 except Exception as e:

194 raise Exception(f"WebDriver error. {type(e).__name__}.")

195 finally:

196 if self.driver:

197 self.driver.quit()

198 self.driver = None

199

200 def _scrape_national_phase(self) -> Dict:

201 np_data = {}

202

203 try:

204 href_value = '#detailMainForm:MyTabViewId:NATIONALPHASE'

205 anchor_tag = self.driver.find_element(By.CSS_SELECTOR, f'a[href="{href_value}"]')

206 anchor_tag.click()

207

208 WebDriverWait(self.driver, 15).until(

209 EC.presence_of_element_located((By.CLASS_NAME, 'b-table'))

210 )

211

212 np = self.driver.find_element(By.ID, "detailMainForm:MyTabViewId:NATIONALPHASE")

213 table = np.find_element(By.CLASS_NAME, 'b-table')

214

215 headers = [header.text for header in table.find_elements(By.TAG_NAME, "th")]

216 rows = table.find_elements(By.CSS_SELECTOR, "tbody tr")

217

218 for row in rows:

219 cells = [cell.text for cell in row.find_elements(By.TAG_NAME, "td")]

220

221 if cells:

222 country = cells[0]

223 row_data = dict(zip(headers[1:], cells[1:]))

224

225 if country not in np_data:

226 np_data[country] = {}

227

228 np_data[country] = row_data

229 except:

230 pass

231

232 return np_data

233

234 def _scrape_xml_document(self) -> str:

235 href_value = '#detailMainForm:MyTabViewId:PCTDOCUMENTS'

236 anchor_tag = self.driver.find_element(By.CSS_SELECTOR, f'a[href="{href_value}"]')

237 anchor_tag.click()

238

239 id_doc = 'detailMainForm:MyTabViewId:relatedDocumentsTab'

240 WebDriverWait(self.driver, self.wait_time).until(

241 EC.presence_of_element_located((By.ID, id_doc))

242 )

243

244 anchors = self.driver.find_elements(By.CSS_SELECTOR, 'a.ps-downloadables')

245

246 for anchor in anchors:

247 href = anchor.get_attribute('href')

248 if 'iasr' in href and 'xml' in href and 'filename=' not in href:

249 anchor.click()

250

251 root_element = self.driver.find_element(By.TAG_NAME, "wo-international-application-status")

252 xml_content = root_element.get_attribute('outerHTML')

253

254 return xml_content

255

256 raise Exception("XML document not found")

257

258

259class JobProcessor:

260 def __init__(

261 self,

262 redis_conn,

263 logger,

264 scraper: PatentScraper,

265 parser_bin_path: str,

266 subprocess_runner=None

267 ):

268 self.redis_conn = redis_conn

269 self.logger = logger

270 self.scraper = scraper

271 self.parser_bin_path = parser_bin_path

272 self.subprocess_runner = subprocess_runner or subprocess.run

273

274 def _safe_hset(self, job_id: str, mapping: dict, expected_url: str):

275 existing_url = self.redis_conn.hget(f"job:{job_id}", "url")

276

277 if existing_url is None:

278 raise Exception(f"Job {job_id} URL not found during status update. Aborting job.")

279

280 if existing_url != expected_url:

281 raise Exception(f"Job {job_id} URL mismatch! Expected: {expected_url}, Found: {existing_url}. Aborting job.")

282

283 self.redis_conn.hset(f"job:{job_id}", mapping=mapping)

284 self.redis_conn.publish(f"job_events:{job_id}", mapping["status"])

285

286 def process_job(self, job_id: str):

287 start = time.time()

288

289 self.logger.info(f"Processing job {job_id}")

290

291 job_data = self.redis_conn.hgetall(f"job:{job_id}")

292 url = job_data.get("url")

293

294 if not url:

295 self.redis_conn.hset(f"job:{job_id}", mapping={

296 "status": "FAILED",

297 "err_type": "CRITICAL",

298 "err_msg": "No URL provided"

299 })

300

301 self.logger.exception(f"Job {job_id} failed: No URL provided")

302

303 return

304

305 self.scraper.driver_factory.proxy = job_data.get("proxy")

306

307 try:

308 self._safe_hset(job_id, mapping={"status": "FETCHING"}, expected_url=url)

309

310 result = None

311 try:

312 scrape_start = time.time()

313 result = self.scraper.scrape(url)

314 SCRAPE_DURATION.observe(time.time() - scrape_start)

315 except Exception as e:

316 SCRAPE_FAILED.inc()

317 raise ScrapeError(f"Failed fetch for {url}") from e

318

319 np_json_input, xml_content = result

320 combined_input = f"{xml_content}\n---JSON---\n{json.dumps(np_json_input)}"

321

322 try:

323 parse_start = time.time()

324 completed = self.subprocess_runner(

325 [self.parser_bin_path],

326 input=combined_input,

327 text=True,

328 capture_output=True,

329 check=True

330 )

331 PARSE_DURATION.observe(time.time() - parse_start)

332 except subprocess.CalledProcessError as e:

333 PARSE_FAILED.inc()

334 raise ParseError(f"Failed parse for {url}: (from exec): {e.stderr}") from None

335 except Exception as e:

336 PARSE_FAILED.inc()

337 raise ParseError(f"Failed parse for {url}: {str(e)}")

338

339 if completed.returncode == 0:

340 result = completed.stdout

341 else:

342 PARSE_FAILED.inc()

343 raise ParseError(f"Failed parse for {url}: {completed.stderr}")

344

345 self._safe_hset(job_id, mapping={

346 "status": "COMPLETED",

347 "result": result

348 }, expected_url=url)

349

350 JOBS_PROCESSED.labels(status='success').inc()

351 self.logger.info(f"Job {job_id}: Completed successfully")

352

353 except ScrapeError as e:

354 JOBS_PROCESSED.labels(status='scrape_error').inc()

355 self.logger.exception(f"Job {job_id}: Failed with error: {str(e)}")

356 self._safe_hset(job_id, mapping={

357 "status": "FAILED",

358 "err_type": "RECOVERABLE",

359 "err_msg": "Fetching failed"

360 }, expected_url=url)

361

362 except ParseError as e:

363 JOBS_PROCESSED.labels(status='parse_error').inc()

364 self.logger.exception(f"Job {job_id}: Failed with error: {str(e)}")

365 self._safe_hset(job_id, mapping={

366 "status": "FAILED",

367 "err_type": "CRITICAL",

368 "err_msg": "Parsing failed"

369 }, expected_url=url)

370

371 except Exception as e:

372 JOBS_PROCESSED.labels(status='internal_error').inc()

373 self.logger.exception(f"Job {job_id}: Failed with error: {str(e)}")

374 self._safe_hset(job_id, mapping={

375 "status": "FAILED",

376 "err_type": "INTERNAL",

377 "err_msg": f"{str(e)}"

378 }, expected_url=url)

379 finally:

380 JOB_DURATION.observe(time.time() - start)

381 push_metrics()

382

383

384class Worker:

385 def __init__(

386 self,

387 redis_conn,

388 logger,

389 job_processor: JobProcessor,

390 jobs_before_dying: Optional[int] = None,

391 sleep_fn=None

392 ):

393 self.redis_conn = redis_conn

394 self.logger = logger

395 self.job_processor = job_processor

396 self.jobs_before_dying = jobs_before_dying

397 self.jobs_processed = 0

398 self.running = True

399 self.sleep_fn = sleep_fn or time.sleep

400

401 def shutdown(self):

402 self.logger.info("Shutdown signal received, stopping worker...")

403 self.running = False

404

405 def run(self):

406 self.logger.info("Worker started, waiting for jobs...")

407

408 try:

409 while self.running:

410 try:

411 result = self.redis_conn.brpop("job_queue", timeout=1)

412

413 if result:

414 _, job_id = result

415 self.job_processor.process_job(job_id)

416 self.jobs_processed += 1

417 if self.jobs_before_dying and self.jobs_processed >= self.jobs_before_dying:

418 self.logger.info(

419 f"Reached max jobs ({self.jobs_before_dying}), exiting for restart"

420 )

421 self.running = False

422

423 except redis.ConnectionError:

424 self.logger.info("Lost connection to Redis, retrying in 5 seconds...")

425 self.sleep_fn(5)

426 except Exception as e:

427 if self.running:

428 self.logger.exception(str(e))

429 self.sleep_fn(1)

430 finally:

431 if self.redis_conn:

432 self.redis_conn.close()

433 self.logger.info("Worker stopped")

434

435

436def create_redis_connection(host: str, port: int) -> redis.Redis:

437 return redis.Redis(

438 host=host,

439 port=port,

440 decode_responses=True,

441 socket_connect_timeout=5,

442 socket_timeout=5

443 )

444

445

446def main(): # pragma: no cover

447 redis_host = os.getenv('REDIS_HOST', 'localhost')

448 redis_port = int(os.getenv('REDIS_PORT', 6379))

449

450 logger = setup_logging()

451

452 redis_conn = create_redis_connection(redis_host, redis_port)

453

454 try:

455 redis_conn.ping()

456 logger.info(f"Connected to Redis at {redis_host}:{redis_port}")

457 except redis.ConnectionError:

458 logger.exception(f"Failed to connect to Redis at {redis_host}:{redis_port}")

459 sys.exit(1)

460

461 driver_factory = DriverFactory()

462 scraper = PatentScraper(driver_factory, logger)

463 job_processor = JobProcessor(redis_conn, logger, scraper, PP)

464 max_jobs_env = os.getenv('JOBS_BEFORE_CLEANUP')

465 max_jobs_before_dying = int(max_jobs_env) if max_jobs_env else None

466 worker = Worker(redis_conn, logger, job_processor, jobs_before_dying=max_jobs_before_dying)

467

468 def shutdown_handler(signum, frame):

469 worker.shutdown()

470

471 signal.signal(signal.SIGINT, shutdown_handler)

472 signal.signal(signal.SIGTERM, shutdown_handler)

473

474 worker.run()

475

476if __name__ == "__main__": # pragma: no cover

477 main()