1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 414 415 416 417 418 419 420 421 422 423 424 425 426 427 428 429 430 431 432 433 434 435 436 437 438 439 440 441 442 443 444 445 446 447 448 449 450 451 452 453 454 455 456 457 458 459 460 461 462 463 464 465 466 467 468 469 470 471 472 473 474 475 476 477 478 479 480 481 482 483 484 485 486 487 488 489 490 491 492 493 494 495 496 497 498 499 500 501 502 503 504 505 506 507 508 509 510 511 512 513 514 515 516 517 518 519 520 521 522 523 524 525 526 527 528 529 530 531 532 533 534 535 536 537 538 539 540 541 542 543 544 545 546 547 548 549 550 551 552 553 554 555 556 557 558 559 560 561 562 563 564 565 566 567 568 569 570 571 572 573 574 575 576 577 578 579 580 581 582 583 584 585 586 587 588 589 590 591 592 593 594 595 596 597 598 599 600 601 602 603 604 605 606 607 608 609 610 611 612 613 614 615 616 617 618 619 620 621 622 623 624 625 626 627 628 629 630 631 632 633 634 635 636 637 638 639 640 641 642 643 644 645 646 647 648 649 650 651 652 653 654 655 656 657 658 659 660 661 662 663 664 665 666 667 668 669 670 671 672 673 674 675 676 677 678 679 680 681 682 683 684 685 686 687 688 689 690 691 692 693 694 695 696 697 698 699 700 701 702 703 704 705 706 707 708 709 710 711 712 713 714 715 716 717 718 719 720 721 722 723 724 725 726 727 728 729 730 731 732 733 734 735 736 737 738 739 740 741 742 743 744 745 746 747 748 749 750 751 752 753 754 755 756 757 758 759 760 761 762 763 764 765 766 767 768 769 770 771 772 773 774 775 776 777 778 779 780 781 782 783 784 785 786 787 788 789 790 791 792 793 794 795 796 797 798 799 800 801 802 803 804 805 806 807 808 809 810 811 812 813 814 815 816 817 818 819 820 821 822 823 824 825 826 827 828 829 830 831 832 833 834 835 836 837 838 839 840 841 842 843 844 845 846 847 848 849 850 851 852 853 854 855 856 857 858 859 860 861 862 863 864 865 866 867 868 869 870 871 872 873 874 875 876 877 878 879 880 881 882 883 884 885 886 887 888 889 890 891 892 893 894 895 896 897 898 899 900 901 902 903 904 905 906 907 908 909 910 911 912 913 914 915 916 917 918 919 920 921 922 923 924 925 926 927 928 929 930 931 932 933 934 935 936 937 938 939 940 941 942 943 944 945 946 947 948 949 950 951 952 953 954 955 956 957 958 959 960 961 962 963 964 965 966 967 968 969 970 971 972 973 974 975 976 977 978 979 980 981 982 983 984 985 986 987 988 989 990 991 992 993 994 995 996 997 998 999 1000 1001 1002 1003 1004 1005 1006 1007 1008 1009 1010 1011 1012 1013 1014 1015 1016 1017 1018 1019 1020 1021
|
# Copyright (C) 2000-2014 Bastian Kleineidam
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with this program; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
"""
Base URL handler.
"""
# pylint: disable=assignment-from-none, catching-non-exception, no-member
import sys
import os
import urllib.parse
from urllib.request import urlopen
import time
import errno
import socket
from io import BytesIO
from . import absolute_url, get_url_from
from .. import (
log,
LOG_CHECK,
strformat,
LinkCheckerError,
url as urlutil,
trace,
get_link_pat,
)
from ..htmlutil import htmlsoup
from ..network import iputil
from .const import (
WARN_URL_EFFECTIVE_URL,
WARN_URL_ERROR_GETTING_CONTENT,
WARN_URL_OBFUSCATED_IP,
WARN_URL_CONTENT_SIZE_ZERO,
WARN_URL_CONTENT_SIZE_TOO_LARGE,
WARN_URL_CONTENT_TYPE_UNPARSEABLE,
WARN_URL_WHITESPACE,
URL_MAX_LENGTH,
WARN_URL_TOO_LONG,
ExcList,
ExcSyntaxList,
ExcNoCacheList,
)
from ..url import url_fix_wayback_query
# schemes that are invalid with an empty hostname
scheme_requires_host = ("ftp", "http")
def urljoin(parent, url):
"""
If url is relative, join parent and url. Else leave url as-is.
@return: joined url
"""
if urlutil.url_is_absolute(url):
return url
return urllib.parse.urljoin(parent, url)
def url_norm(url, encoding):
"""Wrapper for url.url_norm() to convert UnicodeError in
LinkCheckerError."""
try:
return urlutil.url_norm(url, encoding=encoding)
except UnicodeError:
msg = _("URL has unparsable domain name: %(name)s") % {
"name": sys.exc_info()[1]
}
raise LinkCheckerError(msg)
class UrlBase:
"""An URL with additional information like validity etc."""
# file types that can be parsed recursively
ContentMimetypes = {
"text/html": "html",
"application/xhtml+xml": "html",
# Include PHP file which helps when checking local .php files.
# It does not harm other URL schemes like HTTP since HTTP servers
# should not send this content type. They send text/html instead.
"application/x-httpd-php": "html",
"text/css": "css",
"application/vnd.adobe.flash.movie": "swf",
"application/x-shockwave-flash": "swf",
"application/msword": "word",
"text/plain+linkchecker": "text",
"text/plain+opera": "opera",
"text/plain+chromium": "chromium",
"application/x-plist+safari": "safari",
"text/vnd.wap.wml": "wml",
"application/xml+sitemap": "sitemap",
"application/xml+sitemapindex": "sitemapindex",
"application/pdf": "pdf",
"application/x-pdf": "pdf",
}
# Read in 16kb chunks
ReadChunkBytes = 1024 * 16
def __init__(
self,
base_url,
recursion_level,
aggregate,
parent_url=None,
base_ref=None,
line=-1,
column=-1,
page=-1,
name="",
url_encoding=None,
extern=None,
):
"""
Initialize check data, and store given variables.
@param base_url: unquoted and possibly unnormed url
@param recursion_level: on what check level lies the base url
@param aggregate: aggregate instance
@param parent_url: quoted and normed url of parent or None
@param base_ref: quoted and normed url of <base href=""> or None
@param line: line number of url in parent content
@param column: column number of url in parent content
@param page: page number of url in parent content
@param name: name of url or empty
@param url_encoding: encoding of URL or None
@param extern: None or (is_extern, is_strict)
"""
self.reset()
self.init(
base_ref,
base_url,
parent_url,
recursion_level,
aggregate,
line,
column,
page,
name,
url_encoding,
extern,
)
self.check_syntax()
if recursion_level == 0:
self.add_intern_pattern()
self.set_extern(self.url)
if self.extern[0] and self.extern[1]:
self.add_info(
_("The URL is outside of the domain filter, checked only syntax.")
)
if not self.has_result:
self.set_result(_("filtered"))
def init(
self,
base_ref,
base_url,
parent_url,
recursion_level,
aggregate,
line,
column,
page,
name,
url_encoding,
extern,
):
"""
Initialize internal data.
"""
self.base_ref = base_ref
if self.base_ref is not None:
assert isinstance(self.base_ref, str), repr(self.base_ref)
self.base_url = base_url.strip() if base_url else base_url
if self.base_url is not None:
assert isinstance(self.base_url, str), repr(self.base_url)
self.parent_url = parent_url
if self.parent_url is not None:
assert isinstance(self.parent_url, str), repr(self.parent_url)
self.recursion_level = recursion_level
self.aggregate = aggregate
self.line = line
self.column = column
self.page = page
self.name = name
assert isinstance(self.name, str), repr(self.name)
self.encoding = url_encoding
self.extern = extern
if self.base_ref:
assert not urlutil.url_needs_quoting(self.base_ref), (
"unquoted base reference URL %r" % self.base_ref
)
if self.parent_url:
assert not urlutil.url_needs_quoting(self.parent_url), (
"unquoted parent URL %r" % self.parent_url
)
url = absolute_url(self.base_url, base_ref, parent_url)
# assume file link if no scheme is found
self.scheme = url.split(":", 1)[0].lower() or "file"
if self.base_url != base_url:
self.add_warning(
_("Leading or trailing whitespace in URL `%(url)s'.")
% {"url": base_url},
tag=WARN_URL_WHITESPACE,
)
self.ignore_errors = self.aggregate.config['ignoreerrors']
def reset(self):
"""
Reset all variables to default values.
"""
# self.url is constructed by self.build_url() out of base_url
# and (base_ref or parent) as absolute and normed url.
# This the real url we use when checking so it also referred to
# as 'real url'
self.url = None
# a split version of url for convenience
self.urlparts = None
# the scheme, host, port and anchor part of url
self.scheme = self.host = self.port = self.anchor = None
# the result message string and flag
self.result = ""
self.has_result = False
# valid or not
self.valid = True
# list of warnings (without duplicates)
self.warnings = []
# list of infos
self.info = []
# content size
self.size = -1
# last modification time of content in HTTP-date format
# as specified in RFC2616 chapter 3.3.1
self.modified = None
# download time
self.dltime = -1
# check time
self.checktime = 0
# connection object
self.url_connection = None
# data of url content, (data == None) means no data is available
self.data = None
# url content data encoding
self.content_encoding = None
# url content as a Unicode string
self.text = None
# url content as a Beautiful Soup object
self.soup = None
# cache url is set by build_url() calling set_cache_url()
self.cache_url = None
# extern flags (is_extern, is_strict)
self.extern = None
# flag if the result should be cached
self.caching = True
# title is either the URL or parsed from content
self.title = None
# flag if content should be checked or not
self.do_check_content = True
# MIME content type
self.content_type = ""
# URLs seen through redirections
self.aliases = []
# error messages (regular expressions) to ignore
self.ignore_errors = []
def set_result(self, msg, valid=True, overwrite=False):
"""
Set result string and validity.
"""
if self.has_result and not overwrite:
log.warn(
LOG_CHECK,
"Double result %r (previous %r) for %s",
msg,
self.result,
self,
)
else:
self.has_result = True
if not msg:
log.warn(LOG_CHECK, "Empty result for %s", self)
self.result = msg
self.valid = valid
if not self.valid:
for url_regex, msg_regex in self.ignore_errors:
if not url_regex.search(self.url):
continue
if not msg_regex.search(self.result):
continue
self.valid = True
self.result = f"Ignored: {self.result}"
# free content data
self.data = None
def get_title(self):
"""Return title of page the URL refers to.
This is per default the filename or the URL."""
if self.title is None:
url = ""
if self.base_url:
url = self.base_url
elif self.url:
url = self.url
self.title = url
if "/" in url:
title = url.rsplit("/", 1)[1]
if title:
self.title = title
return self.title
def is_content_type_parseable(self):
"""
Return True iff the content type of this url is parseable.
"""
if self.content_type in self.ContentMimetypes:
return True
log.debug(
LOG_CHECK,
"URL with content type %r is not parseable",
self.content_type,
)
if self.recursion_level == 0:
self.add_warning(
_("The URL with content type %r is not parseable.") % self.content_type,
tag=WARN_URL_CONTENT_TYPE_UNPARSEABLE,
)
return False
def is_parseable(self):
"""
Return True iff content of this url is parseable.
"""
return False
def is_html(self):
"""Return True iff content of this url is HTML formatted."""
return self._is_ctype("html")
def is_css(self):
"""Return True iff content of this url is CSS stylesheet."""
return self._is_ctype("css")
def _is_ctype(self, ctype):
"""Return True iff content is valid and of the given type."""
if not self.valid:
return False
mime = self.content_type
return self.ContentMimetypes.get(mime) == ctype
def is_http(self):
"""Return True for *http://* or *https://* URLs."""
return self.scheme in ("http", "https")
def is_file(self):
"""Return True for *file://* URLs."""
return self.scheme == "file"
def is_directory(self):
"""Return True if current URL represents a directory."""
return False
def is_local(self):
"""Return True for local (ie. *file://*) URLs."""
return self.is_file()
def should_ignore_warning(self, tag):
"""Return true if a warning should be ignored"""
log.debug(LOG_CHECK,
"ignorewarnings(%s): %s, %s", self.url, tag,
repr(self.aggregate.config["ignorewarnings"]))
if tag in self.aggregate.config["ignorewarnings"]:
return True
ignore_warnings = self.aggregate.config["ignorewarningsforurls"]
for url_regex, name_regex in ignore_warnings:
log.debug(LOG_CHECK,
"ignorewarningsforurl: considering '%s, %s'",
url_regex, name_regex)
if not url_regex.search(self.url):
continue
log.debug(LOG_CHECK, "ignorewarningsforurl: URL matches '%s'", self.url)
if not name_regex.search(tag):
log.debug(LOG_CHECK,
"ignorewarningsforurl: tag doesn't match '%s'",
tag)
continue
log.debug(LOG_CHECK, "ignorewarningsforurl: tag matches '%s'", tag)
return True
return False
def add_warning(self, s, tag=None):
"""
Add a warning string.
"""
item = (tag, s)
if item not in self.warnings:
if self.should_ignore_warning(tag):
self.add_info(s)
else:
self.warnings.append(item)
def add_info(self, s):
"""
Add an info string.
"""
if s not in self.info:
self.info.append(s)
def set_cache_url(self):
"""Set the URL to be used for caching."""
if "AnchorCheck" in self.aggregate.config["enabledplugins"]:
self.cache_url = self.url
else:
# remove anchor from cached target url since we assume
# URLs with different anchors to have the same content
self.cache_url = urlutil.urlunsplit(self.urlparts[:4] + [''])
log.debug(LOG_CHECK, "cache_url '%s'", self.cache_url)
def check_syntax(self):
"""
Called before self.check(), this function inspects the
url syntax. Success enables further checking, failure
immediately logs this url. Syntax checks must not
use any network resources.
"""
log.debug(LOG_CHECK, "checking syntax")
if self.base_url is None:
self.base_url = ""
if not (self.base_url or self.parent_url):
self.set_result(_("URL is empty"), valid=False)
return
try:
self.build_url()
self.check_url_warnings()
except tuple(ExcSyntaxList) as msg:
self.set_result(str(msg), valid=False)
else:
self.set_cache_url()
def check_url_warnings(self):
"""Check URL name and length."""
effectiveurl = urlutil.urlunsplit(self.urlparts)
if self.url != effectiveurl:
self.add_warning(
_("Effective URL %(url)r.") % {"url": effectiveurl},
tag=WARN_URL_EFFECTIVE_URL,
)
self.url = effectiveurl
if len(self.url) > URL_MAX_LENGTH and self.scheme != "data":
args = dict(len=len(self.url), max=URL_MAX_LENGTH)
self.add_warning(
_("URL length %(len)d is longer than %(max)d.") % args,
tag=WARN_URL_TOO_LONG,
)
def build_url(self):
"""
Construct self.url and self.urlparts out of the given base
url information self.base_url, self.parent_url and self.base_ref.
"""
# norm base url - can raise UnicodeError from url.idna_encode()
base_url, is_idn = url_norm(self.base_url, self.encoding)
# make url absolute
if self.base_ref:
# use base reference as parent url
if ":" not in self.base_ref:
# some websites have a relative base reference
self.base_ref = urljoin(self.parent_url, self.base_ref)
self.url = urljoin(self.base_ref, base_url)
elif self.parent_url:
# strip the parent url anchor
urlparts = list(urllib.parse.urlsplit(self.parent_url))
urlparts[4] = ""
parent_url = urlutil.urlunsplit(urlparts)
self.url = urljoin(parent_url, base_url)
else:
self.url = base_url
# urljoin can unnorm the url path, so norm it again
urlparts = list(urllib.parse.urlsplit(self.url))
if urlparts[2]:
urlparts[2] = urlutil.collapse_segments(urlparts[2])
if not urlparts[0].startswith("feed"):
# restore second / in http[s]:// in wayback path
urlparts[2] = url_fix_wayback_query(urlparts[2])
self.url = urlutil.urlunsplit(urlparts)
self.urlparts = self.build_url_parts(self.url)
# and unsplit again
self.url = urlutil.urlunsplit(self.urlparts)
def build_url_parts(self, url):
"""Set userinfo, host, port and anchor from url and return urlparts.
Also checks for obfuscated IP addresses.
"""
split = urllib.parse.urlsplit(url)
urlparts = list(split)
# check userinfo@host:port syntax
self.userinfo, host = urlutil.split_netloc(split.netloc)
try:
port = split.port
except ValueError:
raise LinkCheckerError(
_("URL host %(host)r has invalid port") % {"host": host}
)
if port is None:
port = urlutil.default_ports.get(self.scheme, 0)
if port is None:
raise LinkCheckerError(
_("URL host %(host)r has invalid port") % {"host": host}
)
self.port = port
# urllib.parse.SplitResult.hostname is lowercase
self.host = split.hostname
if self.scheme in scheme_requires_host:
if not self.host:
raise LinkCheckerError(_("URL has empty hostname"))
self.check_obfuscated_ip()
if not self.port or self.port == urlutil.default_ports.get(self.scheme):
host = self.host
else:
host = f"{self.host}:{self.port}"
if self.userinfo:
urlparts[1] = f"{self.userinfo}@{host}"
else:
urlparts[1] = host
# save anchor for later checking
self.anchor = split.fragment
if self.anchor is not None:
assert isinstance(self.anchor, str), repr(self.anchor)
return urlparts
def check_obfuscated_ip(self):
"""Warn if host of this URL is obfuscated IP address."""
# check if self.host can be an IP address
# check for obfuscated IP address
if iputil.is_obfuscated_ip(self.host):
ips = iputil.resolve_host(self.host)
if ips:
self.host = ips[0]
self.add_warning(
_("URL %(url)s has obfuscated IP address %(ip)s")
% {"url": self.base_url, "ip": ips[0]},
tag=WARN_URL_OBFUSCATED_IP,
)
def check(self):
"""Main check function for checking this URL."""
if self.aggregate.config["trace"]:
trace.trace_on()
try:
self.local_check()
except OSError:
# on Unix, ctrl-c can raise
# error: (4, 'Interrupted system call')
etype, value = sys.exc_info()[:2]
if etype == errno.EINTR:
raise KeyboardInterrupt(value)
else:
raise
def local_check(self):
"""Local check function can be overridden in subclasses."""
log.debug(LOG_CHECK, "Checking %s", self)
# strict extern URLs should not be checked
assert not self.extern[1], 'checking strict extern URL'
# check connection
log.debug(LOG_CHECK, "checking connection")
try:
self.check_connection()
self.set_content_type()
self.add_size_info()
self.aggregate.plugin_manager.run_connection_plugins(self)
except tuple(ExcList) as exc:
value = self.handle_exception()
# make nicer error msg for unknown hosts
if isinstance(exc, socket.error) and exc.args[0] == -2:
value = _('Hostname not found')
elif isinstance(exc, UnicodeError):
# idna.encode(host) failed
value = _('Bad hostname %(host)r: %(msg)s') % {
'host': self.host,
'msg': value,
}
self.set_result(value, valid=False)
def check_content(self):
"""Check content of URL.
@return: True if content can be parsed, else False
"""
if self.do_check_content and self.valid:
# check content and recursion
try:
if self.can_get_content():
self.aggregate.plugin_manager.run_content_plugins(self)
if self.allows_recursion():
return True
except tuple(ExcList):
value = self.handle_exception()
self.add_warning(
_("could not get content: %(msg)s") % {"msg": value},
tag=WARN_URL_ERROR_GETTING_CONTENT,
)
return False
def close_connection(self):
"""
Close an opened url connection.
"""
if self.url_connection is None:
# no connection is open
return
try:
self.url_connection.close()
except Exception:
# ignore close errors
pass
self.url_connection = None
def handle_exception(self):
"""
An exception occurred. Log it and set the cache flag.
"""
etype, evalue = sys.exc_info()[:2]
log.debug(
LOG_CHECK, "Error in %s: %s %s", self.url, etype, evalue, exception=True
)
# note: etype must be the exact class, not a subclass
if (
(etype in ExcNoCacheList)
or (etype == socket.error and evalue.args[0] == errno.EBADF)
or not evalue
):
# EBADF occurs when operating on an already socket
self.caching = False
# format message "<exception name>: <error message>"
errmsg = etype.__name__
if evalue:
errmsg += f": {evalue}"
# limit length to 240
return strformat.limit(errmsg, length=240)
def check_connection(self):
"""
The basic connection check uses urlopen to initialize
a connection object.
"""
self.url_connection = urlopen(self.url)
def add_size_info(self):
"""Set size of URL content (if any)..
Should be overridden in subclasses."""
maxbytes = self.aggregate.config["maxfilesizedownload"]
if self.size > maxbytes:
self.add_warning(
_("Content size %(size)s is larger than %(maxbytes)s.")
% dict(
size=strformat.strsize(self.size),
maxbytes=strformat.strsize(maxbytes),
),
tag=WARN_URL_CONTENT_SIZE_TOO_LARGE,
)
def allows_simple_recursion(self):
"""Check recursion level and extern status."""
rec_level = self.aggregate.config["recursionlevel"]
if rec_level >= 0 and self.recursion_level >= rec_level:
log.debug(LOG_CHECK, "... no, maximum recursion level reached.")
return False
if self.extern[0]:
log.debug(LOG_CHECK, "... no, extern.")
return False
return True
def allows_recursion(self):
"""
Return True iff we can recurse into the url's content.
"""
log.debug(LOG_CHECK, "checking recursion of %r ...", self.url)
if not self.valid:
log.debug(LOG_CHECK, "... no, invalid.")
return False
if not self.can_get_content():
log.debug(LOG_CHECK, "... no, cannot get content.")
return False
if not self.allows_simple_recursion():
return False
if self.size > self.aggregate.config["maxfilesizeparse"]:
log.debug(LOG_CHECK, "... no, maximum parse size.")
return False
if not self.is_parseable():
log.debug(LOG_CHECK, "... no, not parseable.")
return False
if not self.content_allows_robots():
log.debug(LOG_CHECK, "... no, robots.")
return False
log.debug(LOG_CHECK, "... yes, recursion.")
return True
def content_allows_robots(self):
"""Returns True: only check robots.txt on HTTP links."""
return True
def set_extern(self, url):
"""
Match URL against extern and intern link patterns. If no pattern
matches the URL is extern. Sets self.extern to a tuple (bool,
bool) with content (is_extern, is_strict).
@return: None
"""
if self.extern:
return
if not url:
self.extern = (1, 1)
return
for entry in self.aggregate.config["externlinks"]:
match = entry['pattern'].search(url)
if (entry['negate'] and not match) or (match and not entry['negate']):
log.debug(LOG_CHECK, "Extern URL %r", url)
self.extern = (1, entry['strict'])
return
for entry in self.aggregate.config["internlinks"]:
match = entry['pattern'].search(url)
if (entry['negate'] and not match) or (match and not entry['negate']):
log.debug(LOG_CHECK, "Intern URL %r", url)
self.extern = (0, 0)
return
if self.aggregate.config['checkextern']:
self.extern = (1, 0)
else:
self.extern = (1, 1)
def set_content_type(self):
"""Set content MIME type.
Should be overridden in subclasses."""
pass
def can_get_content(self):
"""Indicate whether url get_content() can be called."""
return self.size <= self.aggregate.config["maxfilesizedownload"]
def download_content(self):
log.debug(LOG_CHECK, "Get content of %r", self.url)
t = time.time()
content = self.read_content()
self.size = len(content)
self.dltime = time.time() - t
if self.size == 0:
self.add_warning(_("Content size is zero."), tag=WARN_URL_CONTENT_SIZE_ZERO)
else:
self.aggregate.add_downloaded_bytes(self.size)
return content
def get_soup(self):
if self.soup is None:
self.get_content()
return self.soup
def get_raw_content(self):
if self.data is None:
self.data = self.download_content()
return self.data
def get_content(self, encoding=None):
if self.text is None:
self.get_raw_content()
self.soup = htmlsoup.make_soup(self.data, encoding)
# Sometimes soup.original_encoding is None! Better mangled text
# than an internal crash, eh? ISO-8859-1 is a safe fallback in the
# sense that any binary blob can be decoded, it'll never cause a
# UnicodeDecodeError.
log.debug(
LOG_CHECK, "Beautiful Soup detected %s", self.soup.original_encoding
)
self.content_encoding = self.soup.original_encoding or 'ISO-8859-1'
log.debug(LOG_CHECK, "Content encoding %s", self.content_encoding)
self.text = self.data.decode(self.content_encoding)
return self.text
def read_content(self):
"""Return data for this URL. Can be overridden in subclasses."""
buf = BytesIO()
data = self.read_content_chunk()
while data:
if buf.tell() + len(data) > self.aggregate.config["maxfilesizedownload"]:
raise LinkCheckerError(_("File size too large"))
buf.write(data)
data = self.read_content_chunk()
return buf.getvalue()
def read_content_chunk(self):
"""Read one chunk of content from this URL.
Precondition: url_connection is an opened URL.
"""
return self.url_connection.read(self.ReadChunkBytes)
def get_user_password(self):
"""Get tuple (user, password) from configured authentication.
Both user and password can be None.
"""
if self.userinfo:
# URL itself has authentication info
split = urllib.parse.urlsplit(self.url)
return (split.username, split.password)
return self.aggregate.config.get_user_password(self.url)
def add_url(self, url, line=0, column=0, page=0, name="", base=None, parent=None):
"""Add new URL to queue."""
if base:
base_ref = urlutil.url_norm(base, encoding=self.content_encoding)[0]
else:
base_ref = None
url_data = get_url_from(
url,
self.recursion_level + 1,
self.aggregate,
parent_url=self.url if parent is None else parent,
base_ref=base_ref,
line=line,
column=column,
page=page,
name=name,
parent_content_type=self.content_type,
url_encoding=self.content_encoding,
)
self.aggregate.urlqueue.put(url_data)
def serialized(self, sep=os.linesep):
"""
Return serialized url check data as unicode string.
"""
return sep.join(
[
"%s link" % self.scheme,
"base_url=%r" % self.base_url,
"parent_url=%r" % self.parent_url,
"base_ref=%r" % self.base_ref,
"recursion_level=%d" % self.recursion_level,
"url_connection=%s" % self.url_connection,
"line=%s" % self.line,
"column=%s" % self.column,
"page=%d" % self.page,
"name=%r" % self.name,
"anchor=%r" % self.anchor,
"cache_url=%s" % self.cache_url,
]
)
def get_intern_pattern(self, url=None):
"""Get pattern for intern URL matching.
@param url: the URL to set intern pattern for, else self.url
@type url: unicode or None
@return: non-empty regex pattern or None
@rtype: String or None
"""
return None
def add_intern_pattern(self, url=None):
"""Add intern URL regex to config."""
try:
pat = self.get_intern_pattern(url=url)
if pat:
log.debug(LOG_CHECK, "Add intern pattern %r", pat)
self.aggregate.config['internlinks'].append(get_link_pat(pat))
except UnicodeError as msg:
res = _("URL has unparsable domain name: %(domain)s") % {"domain": msg}
self.set_result(res, valid=False)
def __str__(self):
"""
Get URL info.
@return: URL info
@rtype: unicode
"""
return self.serialized()
def __bytes__(self):
"""
Get URL info.
@return: URL info, encoded with the output logger encoding
@rtype: string
"""
s = str(self)
return self.aggregate.config['logger'].encode(s)
def __repr__(self):
"""
Get URL info.
@return: URL info
@rtype: unicode
"""
return f'<{self.serialized(sep=", ")}>'
def to_wire_dict(self):
"""Return a simplified transport object for logging and caching.
The transport object must contain these attributes:
- url_data.valid: bool
Indicates if URL is valid
- url_data.result: unicode
Result string
- url_data.warnings: list of tuples (tag, warning message)
List of tagged warnings for this URL.
- url_data.name: unicode string or None
name of URL (eg. filename or link name)
- url_data.parent_url: unicode or None
Parent URL
- url_data.base_ref: unicode
HTML base reference URL of parent
- url_data.url: unicode
Fully qualified URL.
- url_data.domain: unicode
URL domain part.
- url_data.checktime: int
Number of seconds needed to check this link, default: zero.
- url_data.dltime: int
Number of seconds needed to download URL content, default: -1
- url_data.size: int
Size of downloaded URL content, default: -1
- url_data.info: list of unicode
Additional information about this URL.
- url_data.line: int
Line number of this URL at parent document, or None
- url_data.column: int
Column number of this URL at parent document, or None
- url_data.page: int
Page number of this URL at parent document, or -1
- url_data.cache_url: unicode
Cache url for this URL.
- url_data.content_type: unicode
MIME content type for URL content.
- url_data.level: int
Recursion level until reaching this URL from start URL
- url_data.last_modified: datetime
Last modification date of retrieved page (or None).
"""
return dict(
valid=self.valid,
extern=self.extern[0],
result=self.result,
warnings=self.warnings[:],
name=self.name or "",
title=self.get_title(),
parent_url=self.parent_url or "",
base_ref=self.base_ref or "",
base_url=self.base_url or "",
url=self.url or "",
domain=(self.urlparts[1] if self.urlparts else ""),
checktime=self.checktime,
dltime=self.dltime,
size=self.size,
info=self.info,
line=self.line,
column=self.column,
page=self.page,
cache_url=self.cache_url,
content_type=self.content_type,
level=self.recursion_level,
modified=self.modified,
)
def to_wire(self):
"""Return compact UrlData object with information from to_wire_dict().
"""
return CompactUrlData(self.to_wire_dict())
urlDataAttr = [
'valid',
'extern',
'result',
'warnings',
'name',
'title',
'parent_url',
'base_ref',
'base_url',
'url',
'domain',
'checktime',
'dltime',
'size',
'info',
'modified',
'line',
'column',
'page',
'cache_url',
'content_type',
'level',
]
class CompactUrlData:
"""Store selected UrlData attributes in slots to minimize memory usage."""
__slots__ = urlDataAttr
def __init__(self, wired_url_data):
'''Set all attributes according to the dictionary wired_url_data'''
for attr in urlDataAttr:
setattr(self, attr, wired_url_data[attr])
|