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()