Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/urlextract/urlextract_core.py: 66%

Shortcuts on this page

r m x   toggle line displays

j k   next/prev highlighted chunk

0   (zero) top of page

1   (one) first highlighted chunk

457 statements  

1#!/usr/bin/env python3 

2# -*- coding: utf-8 -*- 

3""" 

4urlextract_core.py - file with definition of URLExtract class and urlextract cli 

5 

6.. Created on 2016-07-29 

7.. Licence MIT 

8.. codeauthor:: Jan Lipovský <janlipovsky@gmail.com>, janlipovsky.cz 

9.. contributors: https://github.com/lipoja/URLExtract/graphs/contributors 

10""" 

11 

12from argparse import Namespace 

13import functools 

14import ipaddress 

15import logging 

16import re 

17import socket 

18from typing import Set, Iterable, Tuple, List, Union, NoReturn, Generator 

19import string 

20import sys 

21from collections import OrderedDict 

22from datetime import datetime, timedelta 

23 

24import uritools # type: ignore 

25 

26from urlextract.cachefile import CacheFile, CacheFileError 

27 

28# version of URLExtract (do not forget to change it in setup.py as well) 

29__version__ = "1.10.0" 

30 

31# default value for maximum count of processed URLs by find_url 

32DEFAULT_LIMIT = 10000 

33 

34 

35class URLExtract(CacheFile): 

36 """ 

37 Class for finding and extracting URLs from given string. 

38 

39 **Examples:** 

40 

41 .. code-block:: python 

42 

43 from urlextract import URLExtract 

44 

45 extractor = URLExtract() 

46 urls = extractor.find_urls("Let's have URL example.com example.") 

47 print(urls) # prints: ['example.com'] 

48 

49 # Another way is to get a generator over found URLs in text: 

50 for url in extractor.gen_urls(example_text): 

51 print(url) # prints: ['example.com'] 

52 

53 # Or if you want to just check if there is at least one URL in text: 

54 if extractor.has_urls(example_text): 

55 print("Given text contains some URL") 

56 """ 

57 

58 # compiled regexp for naive validation of host name 

59 _hostname_re = re.compile(r"^([a-zA-Z0-9]|[a-zA-Z0-9][a-zA-Z0-9\-]*[a-zA-Z0-9])$") 

60 

61 # list of enclosure of URL that should be removed 

62 _enclosure = { 

63 ("(", ")"), 

64 ("{", "}"), 

65 ("[", "]"), 

66 ('"', '"'), 

67 ("\\", "\\"), 

68 ("'", "'"), 

69 ("`", "`"), 

70 } 

71 

72 _ipv4_tld = [".{}".format(ip) for ip in reversed(range(256))] 

73 _ignore_list: Set[str] = set() 

74 _permit_list: Set[str] = set() 

75 

76 _limit = DEFAULT_LIMIT 

77 

78 def __init__( 

79 self, 

80 extract_email=False, 

81 cache_dns=True, 

82 extract_localhost=True, 

83 limit=DEFAULT_LIMIT, 

84 allow_mixed_case_hostname=True, 

85 **kwargs, # noqa E999 

86 ): 

87 """ 

88 Initialize function for URLExtract class. 

89 Tries to get cached TLDs, if cached file does not exist it will try 

90 to download new list from IANA and save it to cache file. 

91 

92 :param bool extract_email: True if we want to extract email from text. 

93 Disabled by default 

94 :param bool cache_dns: True replaces socket DNS lookup with caching 

95 equivalent provided by dnspython. 

96 Enabled by default 

97 :param bool extract_localhost: True if we want to extract 'localhost' 

98 as URL from text. 

99 Enabled by default 

100 :param int limit: maximum count of processed URLs by find_url function 

101 default value defined as global variable DEFAULT_LIMIT 

102 :param bool allow_mixed_case_hostname: True if hostname can contain mixed case letters 

103 (upper-case and lower-case). 

104 Disabled by default 

105 """ 

106 super(URLExtract, self).__init__(**kwargs) 

107 

108 self._tlds_re = None 

109 self._extract_localhost = extract_localhost 

110 self._extract_email = extract_email 

111 self._cache_dns = cache_dns 

112 self._limit = limit 

113 self._allow_mixed_case_hostname = allow_mixed_case_hostname 

114 self._reload_tlds_from_file() 

115 

116 # general stop characters 

117 general_stop_chars = {'"', "<", ">", ";", "|"} 

118 # defining default stop chars left 

119 self._stop_chars_left = set(string.whitespace) 

120 self._stop_chars_left |= general_stop_chars | {"=", "]", ")", "}"} 

121 

122 # default stop characters on left side from schema 

123 self._stop_chars_left_from_schema = self._stop_chars_left.copy() | {":"} 

124 

125 # defining default stop chars left 

126 self._stop_chars_right = set(string.whitespace) 

127 self._stop_chars_right |= general_stop_chars 

128 

129 # characters that are allowed to be right after TLD 

130 self._after_tld_chars = self._get_after_tld_chars() 

131 

132 def _get_after_tld_chars(self) -> Set[str]: 

133 """Initialize after tld characters""" 

134 after_tld_chars = set(string.whitespace) 

135 after_tld_chars |= {"/", '"', "'", "<", ">", "?", ":", ".", ",", "|"} 

136 # get left enclosure characters 

137 _, right_enclosure = zip(*self._enclosure) 

138 # add right enclosure characters to be valid after TLD 

139 # for correct parsing of URL e.g. (example.com) 

140 after_tld_chars |= set(right_enclosure) 

141 

142 return after_tld_chars 

143 

144 def _reload_tlds_from_file(self): 

145 """ 

146 Reloads TLDs from file and compile regexp. 

147 :raises: CacheFileError when cached file is not readable for user 

148 """ 

149 

150 tlds = sorted(self._load_cached_tlds(), key=len, reverse=True) 

151 tlds += self._ipv4_tld 

152 if self._extract_localhost: 

153 tlds.append("localhost") 

154 re_escaped = [re.escape(str(tld)) for tld in tlds] 

155 self._tlds_re = re.compile("|".join(re_escaped), flags=re.IGNORECASE) 

156 

157 @property 

158 def extract_email(self) -> bool: 

159 """ 

160 If set to True email will be extracted from text 

161 

162 :rtype: bool 

163 """ 

164 return self._extract_email 

165 

166 @extract_email.setter 

167 def extract_email(self, extract: bool): 

168 """ 

169 Set if emails will be extracted from text 

170 

171 :param bool extract: True if emails should be extracted False otherwise 

172 """ 

173 self._extract_email = extract 

174 

175 @property 

176 def allow_mixed_case_hostname(self) -> bool: 

177 """ 

178 If set to True host should contain mixed case letters (upper-case and lower-case) 

179 

180 :rtype: bool 

181 """ 

182 return self._allow_mixed_case_hostname 

183 

184 @allow_mixed_case_hostname.setter 

185 def allow_mixed_case_hostname(self, allow_mixed_case: bool): 

186 """ 

187 Set if mixed case hostnames are allowed 

188 

189 :param bool allow_mixed_case: True if we should allow mixed case hostnames False otherwise 

190 """ 

191 self._allow_mixed_case_hostname = allow_mixed_case 

192 

193 @property 

194 def extract_localhost(self) -> bool: 

195 """ 

196 If set to True 'localhost' will be extracted as URL from text 

197 

198 :rtype: bool 

199 """ 

200 return self._extract_localhost 

201 

202 @extract_localhost.setter 

203 def extract_localhost(self, enable: bool): 

204 """ 

205 Set if 'localhost' will be extracted as URL from text 

206 

207 :param bool enable: True if 'localhost' should be extracted 

208 False otherwise 

209 """ 

210 self._extract_localhost = enable 

211 

212 @property 

213 def ignore_list(self) -> Set[str]: 

214 """ 

215 Set of URLs to be ignored (not returned) while extracting from text 

216 

217 :return: Returns set of ignored URLs 

218 :rtype: set(str) 

219 """ 

220 return self._ignore_list 

221 

222 @ignore_list.setter 

223 def ignore_list(self, ignore_list: Set[str]): 

224 """ 

225 Set of URLs to be ignored (not returned) while extracting from text 

226 

227 :param set(str) ignore_list: set of URLs 

228 """ 

229 self._ignore_list = ignore_list 

230 

231 def load_ignore_list(self, file_name): 

232 """ 

233 Load URLs from file into ignore list 

234 

235 :param str file_name: path to file containing URLs 

236 """ 

237 with open(file_name) as f: 

238 for line in f: 

239 url = line.strip() 

240 if not url: 

241 continue 

242 self._ignore_list.add(url) 

243 

244 @property 

245 def permit_list(self): 

246 """ 

247 Set of URLs that can be processed 

248 

249 :return: Returns set of URLs that can be processed 

250 :rtype: set(str) 

251 """ 

252 return self._permit_list 

253 

254 @permit_list.setter 

255 def permit_list(self, permit_list): 

256 """ 

257 Set of URLs that can be processed 

258 

259 :param set(str) permit_list: set of URLs 

260 """ 

261 self._permit_list = permit_list 

262 

263 def load_permit_list(self, file_name): 

264 """ 

265 Load URLs from file into permit list 

266 

267 :param str file_name: path to file containing URLs 

268 """ 

269 with open(file_name) as f: 

270 for line in f: 

271 url = line.strip() 

272 if not url: 

273 continue 

274 self._permit_list.add(url) 

275 

276 def update(self): 

277 """ 

278 Update TLD list cache file. 

279 

280 :return: True if update was successful False otherwise 

281 :rtype: bool 

282 """ 

283 if not self._download_tlds_list(): 

284 return False 

285 

286 self._reload_tlds_from_file() 

287 

288 return True 

289 

290 def update_when_older(self, days: int) -> bool: 

291 """ 

292 Update TLD list cache file if the list is older than 

293 number of days given in parameter `days` or if it does not exist. 

294 

295 :param int days: number of days from last change 

296 :return: True if update was successful, False otherwise 

297 :rtype: bool 

298 """ 

299 

300 last_cache = self._get_last_cachefile_modification() 

301 if last_cache is None: 

302 return self.update() 

303 

304 time_to_update = last_cache + timedelta(days=days) 

305 

306 if datetime.now() >= time_to_update: 

307 return self.update() 

308 

309 return True 

310 

311 @staticmethod 

312 def get_version() -> str: 

313 """ 

314 Returns version number. 

315 

316 :return: version number 

317 :rtype: str 

318 """ 

319 

320 return __version__ 

321 

322 def get_after_tld_chars(self) -> List[str]: 

323 """ 

324 Returns list of chars that are allowed after TLD 

325 

326 :return: list of chars that are allowed after TLD 

327 :rtype: list 

328 """ 

329 

330 return list(self._after_tld_chars) 

331 

332 def set_after_tld_chars(self, after_tld_chars: Iterable[str]): 

333 """ 

334 Set chars that are allowed after TLD. 

335 

336 :param list after_tld_chars: list of characters 

337 """ 

338 

339 self._after_tld_chars = set(after_tld_chars) 

340 

341 def get_stop_chars_left(self) -> Set[str]: 

342 """ 

343 Returns set of stop chars for text on left from TLD. 

344 

345 :return: set of stop chars 

346 :rtype: set 

347 """ 

348 return self._stop_chars_left 

349 

350 def set_stop_chars_left(self, stop_chars: Set[str]): 

351 """ 

352 Set stop characters for text on left from TLD. 

353 Stop characters are used when determining end of URL. 

354 

355 :param set stop_chars: set of characters 

356 :raises: TypeError 

357 """ 

358 if not isinstance(stop_chars, set): 

359 raise TypeError( 

360 "stop_chars should be type set " 

361 "but {} was given".format(type(stop_chars)) 

362 ) 

363 

364 self._stop_chars_left = stop_chars 

365 

366 def get_stop_chars_left_from_scheme(self) -> Set[str]: 

367 """ 

368 Returns set of stop chars for text on left from scheme. 

369 

370 :return: set of stop chars 

371 :rtype: set 

372 """ 

373 return self._stop_chars_left_from_schema 

374 

375 def set_stop_chars_left_from_scheme(self, stop_chars: Set[str]): 

376 """ 

377 Set stop characters for text on left from scheme. 

378 Stop characters are used when determining end of URL. 

379 

380 :param set stop_chars: set of characters 

381 :raises: TypeError 

382 """ 

383 if not isinstance(stop_chars, set): 

384 raise TypeError( 

385 "stop_chars should be type set " 

386 "but {} was given".format(type(stop_chars)) 

387 ) 

388 

389 self._stop_chars_left_from_schema = stop_chars 

390 

391 def get_stop_chars_right(self) -> Set[str]: 

392 """ 

393 Returns set of stop chars for text on right from TLD. 

394 

395 :return: set of stop chars 

396 :rtype: set 

397 """ 

398 return self._stop_chars_right 

399 

400 def set_stop_chars_right(self, stop_chars: Set[str]): 

401 """ 

402 Set stop characters for text on right from TLD. 

403 Stop characters are used when determining end of URL. 

404 

405 :param set stop_chars: set of characters 

406 :raises: TypeError 

407 """ 

408 if not isinstance(stop_chars, set): 

409 raise TypeError( 

410 "stop_chars should be type set " 

411 "but {} was given".format(type(stop_chars)) 

412 ) 

413 

414 self._stop_chars_right = stop_chars 

415 

416 def get_enclosures(self) -> Set[Tuple[str, str]]: 

417 """ 

418 Returns set of enclosure pairs that might be used to enclosure URL. 

419 For example brackets (example.com), [example.com], {example.com} 

420 

421 :return: set of tuple of enclosure characters 

422 :rtype: set(tuple(str,str)) 

423 """ 

424 return self._enclosure 

425 

426 def add_enclosure(self, left_char: str, right_char: str): 

427 """ 

428 Add new enclosure pair of characters. That and should be removed 

429 when their presence is detected at beginning and end of found URL 

430 

431 :param str left_char: left character of enclosure pair - e.g. "(" 

432 :param str right_char: right character of enclosure pair - e.g. ")" 

433 """ 

434 assert len(left_char) == 1, "Parameter left_char must be character not string" 

435 assert len(right_char) == 1, "Parameter right_char must be character not string" 

436 self._enclosure.add((left_char, right_char)) 

437 

438 self._after_tld_chars = self._get_after_tld_chars() 

439 

440 def remove_enclosure(self, left_char: str, right_char: str): 

441 """ 

442 Remove enclosure pair from set of enclosures. 

443 

444 :param str left_char: left character of enclosure pair - e.g. "(" 

445 :param str right_char: right character of enclosure pair - e.g. ")" 

446 """ 

447 assert len(left_char) == 1, "Parameter left_char must be character not string" 

448 assert len(right_char) == 1, "Parameter right_char must be character not string" 

449 rm_enclosure = (left_char, right_char) 

450 if rm_enclosure in self._enclosure: 

451 self._enclosure.remove(rm_enclosure) 

452 

453 self._after_tld_chars = self._get_after_tld_chars() 

454 

455 def _complete_url( 

456 self, text: str, tld_pos: int, tld: str, check_dns=False, with_schema_only=False 

457 ) -> str: 

458 """ 

459 Expand string in both sides to match whole URL. 

460 

461 :param str text: text where we want to find URL 

462 :param int tld_pos: position of TLD 

463 :param str tld: matched TLD which should be in text 

464 :param bool check_dns: filter results to valid domains 

465 :param bool with_schema_only: get domains with schema only 

466 (e.g. https://janlipovsky.cz but not example.com) 

467 :return: returns URL 

468 :rtype: str 

469 """ 

470 

471 left_ok = True 

472 right_ok = True 

473 

474 # hack to fix Markdown link match 

475 possible_markdown = False 

476 right_enclosure_pos = None 

477 

478 max_len = len(text) - 1 

479 end_pos = tld_pos 

480 start_pos = tld_pos 

481 in_scheme = False 

482 while left_ok or right_ok: 

483 if left_ok: 

484 if start_pos <= 0: 

485 left_ok = False 

486 else: 

487 # For Markdown link is typical to have "](" these 

488 # brackets next to each other without white space 

489 if text[start_pos] == "(" and text[start_pos - 1] == "]": 

490 possible_markdown = True 

491 if ( 

492 in_scheme 

493 and text[start_pos - 1] in self._stop_chars_left_from_schema 

494 ): 

495 left_ok = False 

496 if ( 

497 left_ok 

498 and text[start_pos - 1] not in self._stop_chars_left 

499 # Allow only ASCII characters in authority and schema 

500 and ord(text[start_pos - 1]) <= 127 

501 ): 

502 start_pos -= 1 

503 else: 

504 left_ok = False 

505 if right_ok: 

506 if end_pos >= max_len: 

507 right_ok = False 

508 else: 

509 if text[end_pos + 1] not in self._stop_chars_right: 

510 # correcting Markdown matches 

511 if right_enclosure_pos is None and text[end_pos + 1] == ")": 

512 right_enclosure_pos = end_pos + 1 

513 end_pos += 1 

514 else: 

515 right_ok = False 

516 

517 if text[start_pos : start_pos + 3] == "://": 

518 in_scheme = True 

519 

520 # correcting Markdown matches 

521 if possible_markdown and right_enclosure_pos is not None: 

522 end_pos = right_enclosure_pos 

523 

524 complete_url = text[start_pos : end_pos + 1].lstrip("/") 

525 # remove last character from url 

526 # when it is allowed character right after TLD (e.g. dot, comma) 

527 temp_tlds = {tld + c for c in self._after_tld_chars} 

528 # get only dot+tld+one_char and compare 

529 extended_tld = complete_url[len(complete_url) - len(tld) - 1 :] 

530 if extended_tld in temp_tlds: 

531 # We do not want to change found URL 

532 if not extended_tld.endswith("/"): 

533 complete_url = complete_url[:-1] 

534 

535 complete_url = self._split_markdown(complete_url, tld_pos - start_pos) 

536 complete_url = self._remove_enclosure_from_url( 

537 complete_url, tld_pos - start_pos, tld 

538 ) 

539 

540 # search for enclosures before URL ignoring space character " " 

541 # when URL contains right enclosure character (issue #77) 

542 enclosure_map = { 

543 left_char: right_char for left_char, right_char in self._enclosure 

544 } 

545 if any( 

546 enclosure in complete_url[tld_pos - start_pos :] 

547 for enclosure in enclosure_map.values() 

548 ): 

549 enclosure_space_char = True 

550 enclosure_found = False 

551 tmp_start_pos = start_pos 

552 while enclosure_space_char: 

553 if tmp_start_pos <= 0: 

554 break 

555 if text[tmp_start_pos - 1] == " ": 

556 tmp_start_pos -= 1 

557 elif text[tmp_start_pos - 1] in enclosure_map.keys(): 

558 tmp_start_pos -= 1 

559 enclosure_found = True 

560 else: 

561 enclosure_space_char = False 

562 

563 if enclosure_found: 

564 pre_url = text[tmp_start_pos:start_pos] 

565 extended_complete_url = pre_url + complete_url 

566 complete_url = self._remove_enclosure_from_url( 

567 extended_complete_url, tld_pos - tmp_start_pos, tld 

568 ) 

569 # URL should not start/end with whitespace 

570 complete_url = complete_url.strip() 

571 # URL should not start with two backslashes 

572 if complete_url.startswith("//"): 

573 complete_url = complete_url[2:] 

574 # URL should not start with unreserved characters 

575 if complete_url.startswith(("-", ".", "~", "_")): 

576 complete_url = complete_url[1:] 

577 if not self._is_domain_valid( 

578 complete_url, tld, check_dns=check_dns, with_schema_only=with_schema_only 

579 ): 

580 return "" 

581 

582 return complete_url 

583 

584 def _validate_tld_match(self, text: str, matched_tld: str, tld_pos: int) -> bool: 

585 """ 

586 Validate TLD match - tells if at found position is really TLD. 

587 

588 :param str text: text where we want to find URLs 

589 :param str matched_tld: matched TLD 

590 :param int tld_pos: position of matched TLD 

591 :return: True if match is valid, False otherwise 

592 :rtype: bool 

593 """ 

594 if tld_pos > len(text): 

595 return False 

596 

597 right_tld_pos = tld_pos + len(matched_tld) 

598 if len(text) > right_tld_pos: 

599 if text[right_tld_pos] in self._after_tld_chars: 

600 if tld_pos > 0 and text[tld_pos - 1] not in self._stop_chars_left: 

601 return True 

602 else: 

603 if tld_pos > 0 and text[tld_pos - 1] not in self._stop_chars_left: 

604 return True 

605 

606 return False 

607 

608 def _is_domain_valid( 

609 self, url: str, tld: str, check_dns=False, with_schema_only=False 

610 ): 

611 """ 

612 Checks if given URL has valid domain name (ignores subdomains) 

613 

614 :param str url: complete URL that we want to check 

615 :param str tld: TLD that should be found at the end of URL (hostname) 

616 :param bool check_dns: filter results to valid domains 

617 :param bool with_schema_only: URL must contain schema (protocol) 

618 to be considered valid 

619 :return: True if URL is valid, False otherwise 

620 :rtype: bool 

621 

622 >>> extractor = URLExtract() 

623 >>> extractor._is_domain_valid("janlipovsky.cz", ".cz") 

624 True 

625 

626 >>> extractor._is_domain_valid("https://janlipovsky.cz", ".cz") 

627 True 

628 

629 >>> extractor._is_domain_valid("invalid.cz.", ".cz") 

630 False 

631 

632 >>> extractor._is_domain_valid("invalid.cz,", ".cz") 

633 False 

634 

635 >>> extractor._is_domain_valid("in.v_alid.cz", ".cz") 

636 False 

637 

638 >>> extractor._is_domain_valid("-is.valid.cz", ".cz") 

639 True 

640 

641 >>> extractor._is_domain_valid("not.valid-.cz", ".cz") 

642 False 

643 

644 >>> extractor._is_domain_valid("http://blog/media/path.io.jpg", ".cz") 

645 False 

646 """ 

647 

648 if not url: 

649 return False 

650 

651 scheme_pos = url.find("://") 

652 if scheme_pos == -1: 

653 if with_schema_only: 

654 return False 

655 url = "http://" + url 

656 added_schema = True 

657 else: 

658 added_schema = False 

659 

660 url_parts = uritools.urisplit(url) 

661 # <scheme>://<authority>/<path>?<query>#<fragment> 

662 

663 # authority can't start with @ 

664 if url_parts.authority and url_parts.authority.startswith("@"): 

665 return False 

666 

667 # if URI contains user info and schema was automatically added 

668 # the url is probably an email 

669 if url_parts.getuserinfo() and added_schema: 

670 # do not collect emails 

671 if not self._extract_email: 

672 return False 

673 else: 

674 # if we want to extract email we have to be sure that it 

675 # really is email -> given URL does not have other parts 

676 if ( 

677 url_parts.getport() 

678 or url_parts.getpath() 

679 or url_parts.getquery() 

680 or url_parts.getfragment() 

681 ): 

682 return False 

683 

684 try: 

685 host = url_parts.gethost() 

686 except ValueError: 

687 self._logger.info( 

688 "Invalid host '%s'. " "If the host is valid report a bug.", url 

689 ) 

690 return False 

691 

692 if not host: 

693 return False 

694 

695 if not self.allow_mixed_case_hostname: 

696 # we have to take url_parts.host instead of host variable because url_parts.host is not normalized 

697 if not ( 

698 all(s.islower() for s in url_parts.host if s.isalpha()) 

699 or all(s.isupper() for s in url_parts.host if s.isalpha()) 

700 ): 

701 return False 

702 

703 if self._permit_list and host not in self._permit_list: 

704 return False 

705 

706 if host in self._ignore_list: 

707 return False 

708 

709 # IP address are valid hosts 

710 is_ipv4 = isinstance(host, ipaddress.IPv4Address) 

711 if is_ipv4: 

712 return True 

713 

714 # when TLD is a number the host must be IP 

715 if tld in self._ipv4_tld and not is_ipv4: 

716 return False 

717 

718 host_parts = host.split(".") 

719 

720 if self._extract_localhost and host_parts == ["localhost"]: 

721 return True 

722 

723 if len(host_parts) <= 1: 

724 return False 

725 

726 host_tld = "." + host_parts[-1] 

727 if host_tld.lower() != tld.lower(): 

728 return False 

729 

730 top = host_parts[-2] 

731 

732 if self._hostname_re.match(top) is None: 

733 return False 

734 

735 if check_dns: 

736 if self._cache_dns is True: 

737 dns_cache_install() 

738 self._cache_dns = False 

739 

740 try: 

741 socket.gethostbyname(host) 

742 except socket.herror as err: 

743 if err.errno == 0: 

744 self._logger.info( 

745 "Unable to resolve address {}: {}".format(host, err) 

746 ) 

747 else: 

748 self._logger.info(err) 

749 return False 

750 except Exception as err: 

751 self._logger.info( 

752 "Unknown exception during gethostbyname({}) {!r}".format(host, err) 

753 ) 

754 return False 

755 

756 return True 

757 

758 def _remove_enclosure_from_url(self, text_url: str, tld_pos: int, tld: str) -> str: 

759 """ 

760 Removes enclosure characters from URL given in text_url. 

761 For example: (example.com) -> example.com 

762 

763 :param str text_url: text with URL that we want to extract from 

764 enclosure of two characters 

765 :param int tld_pos: position of TLD in text_url 

766 :param str tld: matched TLD which should be in text 

767 :return: URL that has removed enclosure 

768 :rtype: str 

769 """ 

770 

771 enclosure_map = { 

772 left_char: right_char for left_char, right_char in self._enclosure 

773 } 

774 # get position of most right left_char of enclosure pairs 

775 left_pos = max( 

776 [ 

777 text_url.rfind(left_char, 0, tld_pos) 

778 for left_char in enclosure_map.keys() 

779 ] 

780 ) 

781 left_char = text_url[left_pos] if left_pos >= 0 else "" 

782 right_char = enclosure_map.get(left_char, "") 

783 # get count of left and right enclosure characters and 

784 left_char_count = text_url[: left_pos + 1].count(left_char) 

785 right_char_count = text_url[left_pos:].count(right_char) 

786 # we want to find only pairs and ignore rest (more occurrences) 

787 min_count = min(left_char_count, right_char_count) 

788 

789 right_pos = len(text_url) + 1 

790 # find position of Nth occurrence of right enclosure character 

791 for i in range(max(min_count, 1)): 

792 right_pos = text_url[:right_pos].rfind(right_char) 

793 

794 if right_pos < 0 or right_pos < tld_pos: 

795 right_pos = len(text_url) 

796 

797 new_url = text_url[left_pos + 1 : right_pos] 

798 tld_pos -= left_pos + 1 

799 

800 # Get valid domain when we have input as: example.com)/path 

801 # we assume that if there is enclosure character after TLD it is 

802 # the end URL itself therefore we remove the rest 

803 after_tld_pos = tld_pos + len(tld) 

804 if after_tld_pos < len(new_url): 

805 if new_url[after_tld_pos] in enclosure_map.values(): 

806 new_url_tmp = new_url[:after_tld_pos] 

807 return self._remove_enclosure_from_url(new_url_tmp, tld_pos, tld) 

808 

809 return new_url 

810 

811 @staticmethod 

812 def _split_markdown(text_url: str, tld_pos: int) -> str: 

813 """ 

814 Split markdown URL. There is an issue wen Markdown URL is found. 

815 Parsing of the URL does not stop on right place so wrongly found URL 

816 has to be split. 

817 

818 :param str text_url: URL that we want to extract from enclosure 

819 :param int tld_pos: position of TLD 

820 :return: URL that has removed enclosure 

821 :rtype: str 

822 """ 

823 # Markdown url can looks like: 

824 # [http://example.com/](http://example.com/status/210) 

825 

826 left_bracket_pos = text_url.find("[") 

827 # subtract 3 because URL is never shorter than 3 characters 

828 if left_bracket_pos > tld_pos - 3: 

829 return text_url 

830 

831 right_bracket_pos = text_url.find(")") 

832 if right_bracket_pos < tld_pos: 

833 return text_url 

834 

835 middle_pos = text_url.rfind("](") 

836 if middle_pos > tld_pos: 

837 return text_url[left_bracket_pos + 1 : middle_pos] 

838 return text_url 

839 

840 @staticmethod 

841 # TODO: fix DOC to accomodate to return value 

842 def _get_tld_pos(url: str, tld: str) -> int: 

843 """ 

844 Return position of TLD in hostname. 

845 

846 :param str url: URL in which TLD should be located 

847 :param str tld: TLD we want ot find 

848 :return: 

849 """ 

850 tpm_url = "http://" + url if url.find("://") == -1 else url 

851 

852 url_parts = uritools.urisplit(tpm_url) 

853 host = str(url_parts.gethost()) 

854 # `host` is always returned in lowercase, 

855 # so make sure `url` & `tld` must also be lowercase, 

856 # otherwise the `find()` may fail. 

857 offset = url.lower().find(host) 

858 return host.rfind(tld.lower()) + offset 

859 

860 # TODO: move type assertion to be Generator based 

861 # found https://stackoverflow.com/a/38423388/14669675 

862 def gen_urls( 

863 self, text: str, check_dns=False, get_indices=False, with_schema_only=False 

864 ) -> Generator[Union[str, Tuple[str, Tuple[int, int]]], None, None]: 

865 """ 

866 Creates generator over found URLs in given text. 

867 

868 :param str text: text where we want to find URLs 

869 :param bool check_dns: filter results to valid domains 

870 :param bool get_indices: whether to return beginning and 

871 ending indices as (<url>, (idx_begin, idx_end)) 

872 :param bool with_schema_only: get domains with schema only 

873 :yields: URL or URL with indices found in text or empty string if nothing was found 

874 :rtype: str|tuple(str, tuple(int, int)) 

875 """ 

876 tld_pos = 0 

877 matched_tlds = self._tlds_re.findall(text) 

878 

879 while matched_tlds: 

880 tld = matched_tlds.pop(0) 

881 tmp_text = text[tld_pos:] 

882 offset = tld_pos 

883 tld_pos = tmp_text.find(tld) 

884 validated = self._validate_tld_match(text, tld, offset + tld_pos) 

885 if tld_pos != -1 and validated: 

886 tmp_url = self._complete_url( 

887 text, 

888 offset + tld_pos, 

889 tld, 

890 check_dns=check_dns, 

891 with_schema_only=with_schema_only, 

892 ) 

893 

894 if tmp_url: 

895 # do not search for TLD in already extracted URL 

896 tld_pos_url = self._get_tld_pos(tmp_url, tld) 

897 # move cursor right after found TLD 

898 tld_pos += len(tld) + offset 

899 # move cursor after end of found URL 

900 rest_url = tmp_url[tld_pos_url + len(tld) :] 

901 tld_pos += len(rest_url) 

902 

903 # remove all matched TLDs that were found in currently 

904 # extracted URL (tmp_url resp. rest_url) 

905 while matched_tlds: 

906 new_tld = matched_tlds[0] 

907 tmp_tld_pos_url = rest_url.find(new_tld) 

908 if tmp_tld_pos_url < 0: 

909 break 

910 rest_url = rest_url[tmp_tld_pos_url + len(new_tld) :] 

911 matched_tlds.pop(0) 

912 

913 if get_indices: 

914 yield tmp_url, (tld_pos - len(tmp_url), tld_pos) 

915 else: 

916 yield tmp_url 

917 

918 continue 

919 

920 # move cursor right after found TLD 

921 tld_pos += len(tld) + offset 

922 

923 def find_urls( 

924 self, 

925 text: str, 

926 only_unique=False, 

927 check_dns=False, 

928 get_indices=False, 

929 with_schema_only=False, 

930 ) -> List[Union[str, Tuple[str, Tuple[int, int]]]]: 

931 """ 

932 Find all URLs in given text. 

933 

934 :param str text: text where we want to find URLs 

935 :param bool only_unique: return only unique URLs 

936 :param bool check_dns: filter results to valid domains 

937 :return: list of URLs found in text 

938 :param bool get_indices: whether to return beginning and 

939 ending indices as (<url>, (idx_begin, idx_end)) 

940 :param bool with_schema_only: get domains with schema only 

941 (e.g. https://janlipovsky.cz but not example.com) 

942 :rtype: list 

943 

944 :raises URLExtractError: Raised when count of found URLs reaches 

945 given limit. Processed URLs are returned in `data` argument. 

946 """ 

947 urls = self.gen_urls( 

948 text, 

949 check_dns=check_dns, 

950 get_indices=get_indices, 

951 with_schema_only=with_schema_only, 

952 ) 

953 if self._limit is None: 

954 if only_unique: 

955 return list(OrderedDict.fromkeys(urls)) 

956 return list(urls) 

957 

958 result_urls: List[Union[str, Tuple[str, Tuple[int, int]]]] = [] 

959 url = next(urls, "") 

960 url_count = 1 

961 while url: 

962 if url_count > self._limit: 

963 err = "Limit for extracting URLs was reached. [{} URLs]".format( 

964 self._limit 

965 ) 

966 self._logger.error(err) 

967 

968 raise URLExtractError(err, data=result_urls) 

969 

970 result_urls.append(url) 

971 url = next(urls, "") 

972 url_count += 1 

973 

974 if only_unique: 

975 return list(OrderedDict.fromkeys(result_urls)) 

976 return result_urls 

977 

978 def has_urls(self, text: str, check_dns=False, with_schema_only=False) -> bool: 

979 """ 

980 Checks if text contains any valid URL. 

981 Returns True if text contains at least one URL. 

982 

983 :param text: text where we want to find URLs 

984 :param bool check_dns: filter results to valid domains 

985 :param bool with_schema_only: consider domains with schema only 

986 :return: True if et least one URL was found, False otherwise 

987 :rtype: bool 

988 """ 

989 

990 return any( 

991 self.gen_urls(text, check_dns=check_dns, with_schema_only=with_schema_only) 

992 ) 

993 

994 

995class URLExtractError(Exception): 

996 """ 

997 Raised when some error occurred during processing URLs. 

998 

999 Attributes: 

1000 message -- explanation of the error 

1001 data -- input expression in which the error occurred 

1002 """ 

1003 

1004 def __init__(self, message, data): 

1005 self.data = data 

1006 self.message = message 

1007 

1008 

1009def report_issue(func): 

1010 """Friendly message with link to GitHub for easier reporting""" 

1011 

1012 @functools.wraps(func) 

1013 def wrapper_urlextract_cli(*args, **kwargs): 

1014 try: 

1015 return func(*args, **kwargs) 

1016 except Exception: 

1017 print( 

1018 "Error: An unexpected error occurred. " 

1019 "If you can't resolve this issue please report it to: " 

1020 "https://github.com/lipoja/URLExtract/issues " 

1021 "and help us improve urlextract!", 

1022 file=sys.stderr, 

1023 ) 

1024 raise 

1025 

1026 return wrapper_urlextract_cli 

1027 

1028 

1029@report_issue 

1030def _urlextract_cli(): 

1031 """ 

1032 urlextract - command line program that will print all URLs to stdout 

1033 Usage: urlextract [input_file] [-u] [-v] 

1034 

1035 input_file - text file with URLs to extract 

1036 """ 

1037 import argparse 

1038 

1039 # TODO: add type checking here 

1040 def get_args() -> Namespace: 

1041 """Parse programs arguments""" 

1042 parser = argparse.ArgumentParser( 

1043 description="urlextract - prints out all URLs that were " 

1044 "found in input file or stdin based on locating " 

1045 "their TLDs" 

1046 ) 

1047 

1048 ver = URLExtract.get_version() 

1049 parser.add_argument( 

1050 "-v", 

1051 "--version", 

1052 action="version", 

1053 version="%(prog)s - version {}".format(ver), 

1054 ) 

1055 

1056 parser.add_argument( 

1057 "-u", 

1058 "--unique", 

1059 dest="unique", 

1060 action="store_true", 

1061 help="print out only unique URLs found in file", 

1062 ) 

1063 

1064 parser.add_argument( 

1065 "-dl", 

1066 "--disable-localhost", 

1067 dest="disable_localhost", 

1068 action="store_true", 

1069 help='disable extracting "localhost" as URL', 

1070 ) 

1071 

1072 parser.add_argument( 

1073 "-c", 

1074 "--check-dns", 

1075 dest="check_dns", 

1076 action="store_true", 

1077 help="print out only URLs for existing domain names", 

1078 ) 

1079 

1080 parser.add_argument( 

1081 "-i", 

1082 "--ignore-file", 

1083 metavar="<ignore_file>", 

1084 type=str, 

1085 default=None, 

1086 help="input text file with URLs to exclude from extraction", 

1087 ) 

1088 

1089 parser.add_argument( 

1090 "-p", 

1091 "--permit-file", 

1092 metavar="<permit_file>", 

1093 type=str, 

1094 default=None, 

1095 help="input text file with URLs that can be processed", 

1096 ) 

1097 

1098 parser.add_argument( 

1099 "-l", 

1100 "--limit", 

1101 dest="limit", 

1102 type=int, 

1103 default=DEFAULT_LIMIT, 

1104 help="Maximum count of URLs that can be processed. " 

1105 "Set 0 to disable the limit. " 

1106 "Default: {}".format(DEFAULT_LIMIT), 

1107 ) 

1108 

1109 parser.add_argument( 

1110 "input_file", 

1111 nargs="?", 

1112 metavar="<input_file>", 

1113 type=argparse.FileType(), 

1114 default=sys.stdin, 

1115 help="input text file with URLs to extract", 

1116 ) 

1117 

1118 parsed_args = parser.parse_args() 

1119 return parsed_args 

1120 

1121 args = get_args() 

1122 logging.basicConfig( 

1123 level=logging.WARNING, 

1124 stream=sys.stderr, 

1125 format="%(asctime)s - %(levelname)s (%(name)s): %(message)s", 

1126 ) 

1127 logger = logging.getLogger("urlextract") 

1128 

1129 try: 

1130 limit = None if args.limit <= 0 else args.limit 

1131 urlextract = URLExtract(limit=limit) 

1132 if args.disable_localhost: 

1133 urlextract.extract_localhost = False 

1134 if args.ignore_file: 

1135 urlextract.load_ignore_list(args.ignore_file) 

1136 if args.permit_file: 

1137 urlextract.load_permit_list(args.permit_file) 

1138 urlextract.update_when_older(30) 

1139 content = args.input_file.read() 

1140 try: 

1141 for url in urlextract.find_urls(content, args.unique, args.check_dns): 

1142 print(url) 

1143 except URLExtractError as e: 

1144 logger.error( 

1145 "You can set limit using --limit parameter. " 

1146 "See --help for more details." 

1147 ) 

1148 for url in e.data: 

1149 print(url) 

1150 

1151 except CacheFileError as e: 

1152 logger.error(str(e)) 

1153 sys.exit(-1) 

1154 finally: 

1155 args.input_file.close() 

1156 

1157 

1158def dns_cache_install() -> None: 

1159 try: 

1160 from dns import resolver as dnspython_resolver_module # type: ignore 

1161 from dns_cache.resolver import ExceptionCachingResolver # type: ignore 

1162 

1163 if not dnspython_resolver_module.default_resolver: 

1164 dnspython_resolver_module.default_resolver = ExceptionCachingResolver() 

1165 del dnspython_resolver_module 

1166 except ImportError: 

1167 pass 

1168 

1169 try: 

1170 from dns.resolver import ( # type: ignore 

1171 LRUCache, 

1172 Resolver, 

1173 _resolver, 

1174 default_resolver, 

1175 override_system_resolver, 

1176 ) 

1177 except ImportError: 

1178 return 

1179 

1180 if default_resolver: 

1181 if not default_resolver.cache: 

1182 default_resolver.cache = LRUCache() 

1183 resolver = default_resolver 

1184 elif _resolver and _resolver.cache: 

1185 resolver = _resolver 

1186 else: 

1187 resolver = Resolver() 

1188 resolver.cache = LRUCache() 

1189 override_system_resolver(resolver) 

1190 

1191 

1192if __name__ == "__main__": 

1193 _urlextract_cli()