mirror of
https://github.com/Hopiu/linkchecker.git
synced 2026-05-19 03:51:07 +00:00
commit
f0c03b129a
51 changed files with 118 additions and 133 deletions
|
|
@ -74,7 +74,7 @@ def get_link_pat(arg, strict=False):
|
|||
@rtype: dict
|
||||
@raises: re.error on invalid regular expressions
|
||||
"""
|
||||
log.debug(LOG_CHECK, "Link pattern %r strict=%s", arg, strict)
|
||||
log.debug(LOG_CHECK, _("Link pattern %r strict=%s"), arg, strict)
|
||||
if arg.startswith('!'):
|
||||
pattern = arg[1:]
|
||||
negate = True
|
||||
|
|
@ -84,7 +84,7 @@ def get_link_pat(arg, strict=False):
|
|||
try:
|
||||
regex = re.compile(pattern)
|
||||
except re.error as msg:
|
||||
log.warn(LOG_CHECK, "invalid regular expression %r: %s" % (pattern, msg))
|
||||
log.warn(LOG_CHECK, _("invalid regular expression %r: %s"), pattern, msg)
|
||||
raise
|
||||
return {
|
||||
"pattern": regex,
|
||||
|
|
|
|||
|
|
@ -232,7 +232,7 @@ def better_exchook(etype, value, tb, out=sys.stdout):
|
|||
if value is None or not valuestr:
|
||||
line = "%s" % etype
|
||||
else:
|
||||
line = "%s: %s" % (etype, valuestr)
|
||||
line = f"{etype}: {valuestr}"
|
||||
return line
|
||||
if (isinstance(etype, BaseException) or
|
||||
(hasattr(types, "InstanceType") and isinstance(etype, types.InstanceType)) or
|
||||
|
|
|
|||
|
|
@ -22,15 +22,13 @@ def parse_bookmark_data(data):
|
|||
Return iterator for bookmarks of the form (url, name).
|
||||
Bookmarks are not sorted.
|
||||
"""
|
||||
for url, name in parse_bookmark_json(json.loads(data)):
|
||||
yield url, name
|
||||
yield from parse_bookmark_json(json.loads(data))
|
||||
|
||||
|
||||
def parse_bookmark_json(data):
|
||||
"""Parse complete JSON data for Chromium Bookmarks."""
|
||||
for entry in data["roots"].values():
|
||||
for url, name in parse_bookmark_node(entry):
|
||||
yield url, name
|
||||
yield from parse_bookmark_node(entry)
|
||||
|
||||
|
||||
def parse_bookmark_node(node):
|
||||
|
|
@ -39,5 +37,4 @@ def parse_bookmark_node(node):
|
|||
yield node["url"], node["name"]
|
||||
elif node["type"] == "folder":
|
||||
for child in node["children"]:
|
||||
for entry in parse_bookmark_node(child):
|
||||
yield entry
|
||||
yield from parse_bookmark_node(child)
|
||||
|
|
|
|||
|
|
@ -48,8 +48,7 @@ def parse_plist(entry):
|
|||
yield (url, title)
|
||||
elif has_children(entry):
|
||||
for child in entry[KEY_CHILDREN]:
|
||||
for item in parse_plist(child):
|
||||
yield item
|
||||
yield from parse_plist(child)
|
||||
|
||||
|
||||
def is_leaf(entry):
|
||||
|
|
|
|||
|
|
@ -39,10 +39,10 @@ def guess_url(url):
|
|||
return url
|
||||
if url.lower().startswith("www."):
|
||||
# syntactic sugar
|
||||
return "http://%s" % url
|
||||
return f"http://{url}"
|
||||
elif url.lower().startswith("ftp."):
|
||||
# syntactic sugar
|
||||
return "ftp://%s" % url
|
||||
return f"ftp://{url}"
|
||||
return url
|
||||
|
||||
|
||||
|
|
@ -197,7 +197,7 @@ def get_index_html(urls):
|
|||
except KeyError:
|
||||
# Some unicode entries raise KeyError.
|
||||
url = name
|
||||
lines.append('<a href="%s">%s</a>' % (url, name))
|
||||
lines.append(f'<a href="{url}">{name}</a>')
|
||||
lines.extend(["</body>", "</html>"])
|
||||
return os.linesep.join(lines)
|
||||
|
||||
|
|
|
|||
|
|
@ -17,7 +17,6 @@
|
|||
Helper constants.
|
||||
"""
|
||||
import socket
|
||||
import select
|
||||
import nntplib
|
||||
import ftplib
|
||||
import requests
|
||||
|
|
@ -37,8 +36,6 @@ ExcCacheList = [
|
|||
OSError, # OSError is thrown on Windows when a file is not found
|
||||
LinkCheckerError,
|
||||
DNSException,
|
||||
socket.error,
|
||||
select.error,
|
||||
# nttp errors (including EOFError)
|
||||
nntplib.NNTPError,
|
||||
EOFError,
|
||||
|
|
|
|||
|
|
@ -357,10 +357,10 @@ class AnchorCheckFileUrl(FileUrl):
|
|||
if self.anchor:
|
||||
self.add_warning(
|
||||
_(
|
||||
f" URL `{self.url}' is a directory with an anchor."
|
||||
"URL `%s' is a directory with an anchor."
|
||||
" When checking local files AnchorCheck does not support"
|
||||
" anchors for directories."
|
||||
),
|
||||
) % self.url,
|
||||
tag=WARN_FILE_ANCHORCHECK_DIRECTORY,
|
||||
)
|
||||
|
||||
|
|
|
|||
|
|
@ -118,7 +118,7 @@ class FtpUrl(internpaturl.InternPatternUrl):
|
|||
# file found
|
||||
return
|
||||
# it could be a directory if the trailing slash was forgotten
|
||||
if "%s/" % self.filename in files:
|
||||
if f"{self.filename}/" in files:
|
||||
if not self.url.endswith('/'):
|
||||
self.add_warning(
|
||||
_("Missing trailing directory slash in ftp url."),
|
||||
|
|
@ -178,7 +178,7 @@ class FtpUrl(internpaturl.InternPatternUrl):
|
|||
data = get_index_html(self.files)
|
||||
else:
|
||||
# download file in BINARY mode
|
||||
ftpcmd = "RETR %s" % self.filename
|
||||
ftpcmd = f"RETR {self.filename}"
|
||||
buf = StringIO()
|
||||
|
||||
def stor_data(s):
|
||||
|
|
|
|||
|
|
@ -348,7 +348,7 @@ class HttpUrl(internpaturl.InternPatternUrl):
|
|||
"""Parse URLs in HTTP headers Link:."""
|
||||
for linktype, linkinfo in self.url_connection.links.items():
|
||||
url = linkinfo["url"]
|
||||
name = "Link: header %s" % linktype
|
||||
name = f"Link: header {linktype}"
|
||||
self.add_url(url, name=name)
|
||||
if 'Refresh' in self.headers:
|
||||
from ..htmlutil.linkparse import refresh_re
|
||||
|
|
|
|||
|
|
@ -247,7 +247,7 @@ class MailtoUrl(urlbase.UrlBase):
|
|||
)
|
||||
return
|
||||
for char in '@ \\",[]':
|
||||
if char in local.replace("\\%s" % char, ""):
|
||||
if char in local.replace(f"\\{char}", ""):
|
||||
self.set_result(
|
||||
_(
|
||||
"Local part of mail address `%(addr)s' contains"
|
||||
|
|
@ -370,7 +370,7 @@ class MailtoUrl(urlbase.UrlBase):
|
|||
The cache url is a comma separated list of emails.
|
||||
"""
|
||||
emails = ",".join(sorted(self.addresses))
|
||||
self.cache_url = "%s:%s" % (self.scheme, emails)
|
||||
self.cache_url = f"{self.scheme}:{emails}"
|
||||
|
||||
def can_get_content(self):
|
||||
"""
|
||||
|
|
|
|||
|
|
@ -423,7 +423,7 @@ ignored_schemes_other = r"""
|
|||
|whatsapp # WhatsApp
|
||||
"""
|
||||
|
||||
ignored_schemes = "^(%s%s%s%s)$" % (
|
||||
ignored_schemes = "^({}{}{}{})$".format(
|
||||
ignored_schemes_permanent,
|
||||
ignored_schemes_provisional,
|
||||
ignored_schemes_historical,
|
||||
|
|
|
|||
|
|
@ -23,7 +23,6 @@ from urllib.request import urlopen
|
|||
import time
|
||||
import errno
|
||||
import socket
|
||||
import select
|
||||
from io import BytesIO
|
||||
|
||||
from . import absolute_url, get_url_from
|
||||
|
|
@ -507,9 +506,9 @@ class UrlBase:
|
|||
if not self.port or self.port == urlutil.default_ports.get(self.scheme):
|
||||
host = self.host
|
||||
else:
|
||||
host = "%s:%d" % (self.host, self.port)
|
||||
host = f"{self.host}:{self.port}"
|
||||
if self.userinfo:
|
||||
urlparts[1] = "%s@%s" % (self.userinfo, host)
|
||||
urlparts[1] = f"{self.userinfo}@{host}"
|
||||
else:
|
||||
urlparts[1] = host
|
||||
# save anchor for later checking
|
||||
|
|
@ -538,7 +537,7 @@ class UrlBase:
|
|||
trace.trace_on()
|
||||
try:
|
||||
self.local_check()
|
||||
except (socket.error, select.error):
|
||||
except OSError:
|
||||
# on Unix, ctrl-c can raise
|
||||
# error: (4, 'Interrupted system call')
|
||||
etype, value = sys.exc_info()[:2]
|
||||
|
|
@ -624,7 +623,7 @@ class UrlBase:
|
|||
# format message "<exception name>: <error message>"
|
||||
errmsg = etype.__name__
|
||||
if evalue:
|
||||
errmsg += ": %s" % evalue
|
||||
errmsg += f": {evalue}"
|
||||
# limit length to 240
|
||||
return strformat.limit(errmsg, length=240)
|
||||
|
||||
|
|
@ -882,7 +881,7 @@ class UrlBase:
|
|||
@return: URL info
|
||||
@rtype: unicode
|
||||
"""
|
||||
return "<%s>" % self.serialized(sep=", ")
|
||||
return f'<{self.serialized(sep=", ")}>'
|
||||
|
||||
def to_wire_dict(self):
|
||||
"""Return a simplified transport object for logging and caching.
|
||||
|
|
|
|||
|
|
@ -119,7 +119,7 @@ def init():
|
|||
global _default_foreground, _default_background, _default_style
|
||||
try:
|
||||
attrs = GetConsoleScreenBufferInfo().wAttributes
|
||||
except (ArgumentError, WindowsError):
|
||||
except (ArgumentError, OSError):
|
||||
_default_foreground = GREY
|
||||
_default_background = BLACK
|
||||
_default_style = NORMAL
|
||||
|
|
|
|||
|
|
@ -180,7 +180,7 @@ file entry:
|
|||
)
|
||||
+ "\n".join(
|
||||
[
|
||||
" o %s - %s" % (tag, desc)
|
||||
f" o {tag} - {desc}"
|
||||
for tag, desc in sorted(checker.const.Warnings.items())
|
||||
]
|
||||
)
|
||||
|
|
|
|||
|
|
@ -46,7 +46,7 @@ HtmlCopyright = (
|
|||
HtmlAppInfo = App + ", " + HtmlCopyright
|
||||
Url = _release.__url__
|
||||
SupportUrl = _release.__support_url__
|
||||
UserAgent = "Mozilla/5.0 (compatible; %s/%s; +%s)" % (AppName, Version, Url)
|
||||
UserAgent = f"Mozilla/5.0 (compatible; {AppName}/{Version}; +{Url})"
|
||||
Freeware = (
|
||||
AppName
|
||||
+ """ comes with ABSOLUTELY NO WARRANTY!
|
||||
|
|
@ -87,7 +87,7 @@ def get_modules_info():
|
|||
if version_attr and hasattr(mod, version_attr):
|
||||
attr = getattr(mod, version_attr)
|
||||
version = attr() if callable(attr) else attr
|
||||
module_infos.append("%s %s" % (name, version))
|
||||
module_infos.append(f"{name} {version}")
|
||||
else:
|
||||
# ignore attribute errors in case library developers
|
||||
# change the version information attribute
|
||||
|
|
|
|||
|
|
@ -227,7 +227,7 @@ class LCConfigParser(RawConfigParser):
|
|||
self.config.add_auth(
|
||||
pattern=auth[0], user=auth[1], password=auth[2]
|
||||
)
|
||||
password_fields.append("entry/%s/%s" % (auth[0], auth[1]))
|
||||
password_fields.append(f"entry/{auth[0]}/{auth[1]}")
|
||||
elif len(auth) == 2:
|
||||
self.config.add_auth(pattern=auth[0], user=auth[1])
|
||||
else:
|
||||
|
|
|
|||
|
|
@ -53,7 +53,7 @@ class StatusLogger:
|
|||
|
||||
def writeln(self, msg):
|
||||
"""Write status message and line break to file descriptor."""
|
||||
self.fd.write("%s%s" % (msg, os.linesep))
|
||||
self.fd.write(f"{msg}{os.linesep}")
|
||||
|
||||
def flush(self):
|
||||
"""Flush file descriptor."""
|
||||
|
|
|
|||
|
|
@ -34,7 +34,7 @@ class Form:
|
|||
|
||||
def __repr__(self):
|
||||
"""Return string displaying URL and form data."""
|
||||
return "<url=%s data=%s>" % (self.url, self.data)
|
||||
return f"<url={self.url} data={self.data}>"
|
||||
|
||||
|
||||
def search_form(content, cgiuser, cgipassword):
|
||||
|
|
|
|||
|
|
@ -25,7 +25,7 @@ import sys
|
|||
import codecs
|
||||
|
||||
# more supported languages are added in init()
|
||||
supported_languages = set(['en'])
|
||||
supported_languages = {'en'}
|
||||
default_language = default_encoding = None
|
||||
default_directory = None
|
||||
default_domain = None
|
||||
|
|
@ -68,7 +68,7 @@ def init(domain, directory, loc=None):
|
|||
# get supported languages
|
||||
for lang in os.listdir(directory):
|
||||
path = os.path.join(directory, lang, 'LC_MESSAGES')
|
||||
mo_file = os.path.join(path, '%s.mo' % domain)
|
||||
mo_file = os.path.join(path, f"{domain}.mo")
|
||||
if os.path.exists(mo_file):
|
||||
supported_languages.add(lang)
|
||||
if loc is None:
|
||||
|
|
|
|||
|
|
@ -60,8 +60,7 @@ def application(environ, start_response):
|
|||
|
||||
status = '200 OK'
|
||||
start_response(status, get_response_headers())
|
||||
for output in checklink(form=form, env=environ):
|
||||
yield output
|
||||
yield from checklink(form=form, env=environ)
|
||||
|
||||
|
||||
_supported_langs = ('de', 'C')
|
||||
|
|
@ -113,7 +112,7 @@ class ThreadsafeIO:
|
|||
"""Write given unicode data to buffer."""
|
||||
assert isinstance(data, str)
|
||||
if self.closed:
|
||||
raise IOError("Write on closed I/O object")
|
||||
raise OSError("Write on closed I/O object")
|
||||
if data:
|
||||
self.buf.append(data)
|
||||
|
||||
|
|
@ -213,7 +212,7 @@ def checkform(form, env):
|
|||
locale.setlocale(locale.LC_ALL, localestr)
|
||||
init_i18n()
|
||||
except locale.Error as errmsg:
|
||||
log(env, "could not set locale %r: %s" % (localestr, errmsg))
|
||||
log(env, _("could not set locale %r: %s") % (localestr, errmsg))
|
||||
else:
|
||||
raise LCFormError(_("unsupported language %r") % lang)
|
||||
# check url syntax
|
||||
|
|
|
|||
|
|
@ -34,10 +34,10 @@ def get_package_modules(packagename, packagepath):
|
|||
for mod in pkgutil.iter_modules(packagepath):
|
||||
if not mod.ispkg:
|
||||
try:
|
||||
name = "..%s.%s" % (packagename, mod.name)
|
||||
name = f"..{packagename}.{mod.name}"
|
||||
yield importlib.import_module(name, __name__)
|
||||
except ImportError as msg:
|
||||
print("WARN: could not load module %s: %s" % (mod.name, msg))
|
||||
print(_("WARN: could not load module %s: %s") % (mod.name, msg))
|
||||
|
||||
|
||||
def get_folder_modules(folder, parentpackage):
|
||||
|
|
@ -54,7 +54,7 @@ def get_folder_modules(folder, parentpackage):
|
|||
spec.loader.exec_module(module)
|
||||
yield module
|
||||
except ImportError as msg:
|
||||
print("WARN: could not load file %s: %s" % (fullname, msg))
|
||||
print(_("WARN: could not load file %s: %s") % (fullname, msg))
|
||||
|
||||
|
||||
def get_importable_files(folder):
|
||||
|
|
@ -85,8 +85,7 @@ def get_plugins(modules, classes):
|
|||
@rtype: iterator of class objects
|
||||
"""
|
||||
for module in modules:
|
||||
for plugin in get_module_plugins(module, classes):
|
||||
yield plugin
|
||||
yield from get_module_plugins(module, classes)
|
||||
|
||||
|
||||
def get_module_plugins(module, classes):
|
||||
|
|
|
|||
|
|
@ -226,7 +226,7 @@ class _Logger(abc.ABC):
|
|||
os.makedirs(path)
|
||||
self.fd = self.create_fd()
|
||||
self.close_fd = True
|
||||
except IOError:
|
||||
except OSError:
|
||||
msg = sys.exc_info()[1]
|
||||
log.warn(
|
||||
LOG_CHECK,
|
||||
|
|
@ -255,13 +255,13 @@ class _Logger(abc.ABC):
|
|||
if self.fd is not None:
|
||||
try:
|
||||
self.flush()
|
||||
except IOError:
|
||||
except OSError:
|
||||
# ignore flush errors
|
||||
pass
|
||||
if self.close_fd:
|
||||
try:
|
||||
self.fd.close()
|
||||
except IOError:
|
||||
except OSError:
|
||||
# ignore close errors
|
||||
pass
|
||||
self.fd = None
|
||||
|
|
@ -308,7 +308,7 @@ class _Logger(abc.ABC):
|
|||
else:
|
||||
try:
|
||||
self.fd.write(s, **args)
|
||||
except IOError:
|
||||
except OSError:
|
||||
msg = sys.exc_info()[1]
|
||||
log.warn(
|
||||
LOG_CHECK,
|
||||
|
|
@ -325,7 +325,7 @@ class _Logger(abc.ABC):
|
|||
"""
|
||||
Write string to output descriptor plus a newline.
|
||||
"""
|
||||
self.write("%s%s" % (s, os.linesep), **args)
|
||||
self.write(f"{s}{os.linesep}", **args)
|
||||
|
||||
def has_part(self, name):
|
||||
"""
|
||||
|
|
@ -436,7 +436,7 @@ class _Logger(abc.ABC):
|
|||
if hasattr(self, "fd"):
|
||||
try:
|
||||
self.fd.flush()
|
||||
except (IOError, AttributeError):
|
||||
except (OSError, AttributeError):
|
||||
pass
|
||||
|
||||
def log_internal_error(self):
|
||||
|
|
@ -453,7 +453,7 @@ class _Logger(abc.ABC):
|
|||
@rtype: unicode
|
||||
"""
|
||||
if modified is not None:
|
||||
return modified.strftime("%Y-%m-%d{0}%H:%M:%S.%fZ".format(sep))
|
||||
return modified.strftime(f"%Y-%m-%d{sep}%H:%M:%S.%fZ")
|
||||
return ""
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -70,10 +70,10 @@ class DOTLogger(_GraphLogger):
|
|||
"""Write edge from parent to node."""
|
||||
source = dotquote(self.nodes[node["parent_url"]]["label"])
|
||||
target = dotquote(node["label"])
|
||||
self.writeln(' "%s" -> "%s" [' % (source, target))
|
||||
self.writeln(' label="%s",' % dotquote(node["edge"]))
|
||||
self.writeln(f' "{source}" -> "{target}" [')
|
||||
self.writeln(f' label="{dotquote(node["edge"])}",')
|
||||
if self.has_part("result"):
|
||||
self.writeln(" valid=%d," % node["valid"])
|
||||
self.writeln(f' valid={node["valid"]},')
|
||||
self.writeln(" ];")
|
||||
|
||||
def end_graph(self):
|
||||
|
|
|
|||
|
|
@ -85,7 +85,7 @@ class FailuresLogger(_Logger):
|
|||
"""
|
||||
Read a previously stored failures from file fd.
|
||||
"""
|
||||
with open(self.filename, 'r', encoding=self.output_encoding,
|
||||
with open(self.filename, encoding=self.output_encoding,
|
||||
errors=self.codec_errors) as fd:
|
||||
for line in fd:
|
||||
line = line.rstrip()
|
||||
|
|
|
|||
|
|
@ -112,8 +112,8 @@ class SQLLogger(_Logger):
|
|||
% {
|
||||
'table': self.dbname,
|
||||
'base_url': sqlify(url_data.base_url),
|
||||
'url_parent': sqlify((url_data.parent_url)),
|
||||
'base_ref': sqlify((url_data.base_ref)),
|
||||
'url_parent': sqlify(url_data.parent_url),
|
||||
'base_ref': sqlify(url_data.base_ref),
|
||||
'valid': intify(url_data.valid),
|
||||
'result': sqlify(url_data.result),
|
||||
'warning': sqlify(os.linesep.join(x[1] for x in url_data.warnings)),
|
||||
|
|
|
|||
|
|
@ -202,7 +202,7 @@ class TextLogger(_Logger):
|
|||
def write_warning(self, url_data):
|
||||
"""Write url_data.warning."""
|
||||
self.write(self.part("warning") + self.spaces("warning"))
|
||||
warning_msgs = ["[%s] %s" % x for x in url_data.warnings]
|
||||
warning_msgs = [f"[{tag}] {msg}" for tag, msg in url_data.warnings]
|
||||
self.writeln(self.wrap(warning_msgs, 65), color=self.colorwarning)
|
||||
|
||||
def write_result(self, url_data):
|
||||
|
|
|
|||
|
|
@ -113,4 +113,4 @@ class _XMLLogger(_Logger):
|
|||
for aname, avalue in attrs.items():
|
||||
args = (xmlquote(aname), xmlquoteattr(avalue))
|
||||
self.write(' %s="%s"' % args)
|
||||
self.writeln(">%s</%s>" % (xmlquote(content), xmlquote(name)))
|
||||
self.writeln(f">{xmlquote(content)}</{xmlquote(name)}>")
|
||||
|
|
|
|||
|
|
@ -47,7 +47,7 @@ def resolve_host(host):
|
|||
# canonical name, socket address)
|
||||
# add first ip of socket address
|
||||
ips.append(res[4][0])
|
||||
except socket.error:
|
||||
except OSError:
|
||||
log.info(LOG_CHECK, "Ignored invalid host %r", host)
|
||||
return ips
|
||||
|
||||
|
|
|
|||
|
|
@ -65,10 +65,8 @@ class _ParserPlugin(_PluginBase):
|
|||
def get_plugin_modules(folders):
|
||||
"""Get plugin modules for given folders."""
|
||||
for folder in folders:
|
||||
for module in loader.get_folder_modules(folder, 'linkcheck.dummy'):
|
||||
yield module
|
||||
for module in loader.get_package_modules('plugins', __path__):
|
||||
yield module
|
||||
yield from loader.get_folder_modules(folder, 'linkcheck.dummy')
|
||||
yield from loader.get_package_modules('plugins', __path__)
|
||||
|
||||
|
||||
def get_plugin_classes(modules):
|
||||
|
|
|
|||
|
|
@ -61,12 +61,12 @@ class UrlAnchorCheck:
|
|||
if any(x for x in self.anchors if x[0] == decoded_anchor):
|
||||
return
|
||||
if self.anchors:
|
||||
anchornames = sorted(set("`%s'" % x[0] for x in self.anchors))
|
||||
anchornames = sorted({f"`{x[0]}'" for x in self.anchors})
|
||||
anchors = ", ".join(anchornames)
|
||||
else:
|
||||
anchors = "-"
|
||||
args = {"name": url_data.anchor, "decoded": decoded_anchor, "anchors": anchors}
|
||||
msg = "%s %s" % (
|
||||
msg = "{} {}".format(
|
||||
_("Anchor `%(name)s' (decoded: `%(decoded)s') not found.") % args,
|
||||
_("Available anchors: %(anchors)s.") % args,
|
||||
)
|
||||
|
|
|
|||
|
|
@ -39,7 +39,7 @@ class HttpHeaderInfo(_ConnectionPlugin):
|
|||
headers.append(name.lower())
|
||||
if headers:
|
||||
items = [
|
||||
"%s=%s" % (name.capitalize(), url_data.headers[name])
|
||||
f"{name.capitalize()}={url_data.headers[name]}"
|
||||
for name in headers
|
||||
]
|
||||
info = "HTTP headers %s" % ", ".join(items)
|
||||
|
|
|
|||
|
|
@ -19,7 +19,6 @@ Store and retrieve country names for IPs.
|
|||
from . import _ConnectionPlugin
|
||||
import os
|
||||
import sys
|
||||
import socket
|
||||
from ..lock import get_lock
|
||||
from ..decorators import synchronized
|
||||
from .. import log, LOG_PLUGIN
|
||||
|
|
@ -105,7 +104,7 @@ def get_location(host):
|
|||
return None
|
||||
try:
|
||||
record = get_geoip_record(host)
|
||||
except (geoip_error, socket.error):
|
||||
except (geoip_error, OSError):
|
||||
log.debug(LOG_PLUGIN, "Geoip error for %r", host, exception=True)
|
||||
# ignore lookup errors
|
||||
return None
|
||||
|
|
|
|||
|
|
@ -85,7 +85,7 @@ class MarkdownCheck(_ContentPlugin):
|
|||
try:
|
||||
self.filename_re = re.compile(pattern)
|
||||
except re.error as msg:
|
||||
log.warn(LOG_PLUGIN, "Invalid regex pattern %r: %s" % (pattern, msg))
|
||||
log.warn(LOG_PLUGIN, _("Invalid regex pattern %r: %s"), pattern, msg)
|
||||
|
||||
@classmethod
|
||||
def read_config(cls, configparser):
|
||||
|
|
|
|||
|
|
@ -64,7 +64,7 @@ def has_word():
|
|||
key = winreg.OpenKey(winreg.HKEY_CLASSES_ROOT, "Word.Application")
|
||||
winreg.CloseKey(key)
|
||||
return True
|
||||
except (EnvironmentError, ImportError):
|
||||
except (OSError, ImportError):
|
||||
pass
|
||||
return False
|
||||
|
||||
|
|
|
|||
|
|
@ -42,7 +42,7 @@ class RegexCheck(_ContentPlugin):
|
|||
try:
|
||||
self.warningregex = re.compile(pattern)
|
||||
except re.error as msg:
|
||||
log.warn(LOG_PLUGIN, "Invalid regex pattern %r: %s" % (pattern, msg))
|
||||
log.warn(LOG_PLUGIN, _("Invalid regex pattern %r: %s"), pattern, msg)
|
||||
|
||||
def applies_to(self, url_data):
|
||||
"""Check for warningregex, extern flag and parseability."""
|
||||
|
|
|
|||
|
|
@ -94,7 +94,7 @@ class ClamdScanner:
|
|||
if i != -1:
|
||||
port = int(data[i + 5:])
|
||||
break
|
||||
except socket.error:
|
||||
except OSError:
|
||||
self.sock.close()
|
||||
raise
|
||||
if port is None:
|
||||
|
|
@ -103,7 +103,7 @@ class ClamdScanner:
|
|||
wsock = create_socket(socket.AF_INET, socket.SOCK_STREAM)
|
||||
try:
|
||||
wsock.connect(sockinfo[0][4])
|
||||
except socket.error:
|
||||
except OSError:
|
||||
wsock.close()
|
||||
raise
|
||||
return wsock
|
||||
|
|
@ -199,7 +199,7 @@ class ClamavConfig(dict):
|
|||
addr = self['LocalSocket']
|
||||
try:
|
||||
sock.connect(addr)
|
||||
except socket.error:
|
||||
except OSError:
|
||||
sock.close()
|
||||
raise
|
||||
return sock
|
||||
|
|
@ -211,7 +211,7 @@ class ClamavConfig(dict):
|
|||
sock = create_socket(socket.AF_INET, socket.SOCK_STREAM)
|
||||
try:
|
||||
sock.connect(sockinfo[0][4])
|
||||
except socket.error:
|
||||
except OSError:
|
||||
sock.close()
|
||||
raise
|
||||
return sock
|
||||
|
|
@ -225,7 +225,7 @@ def scan(data, clamconf):
|
|||
"""
|
||||
try:
|
||||
scanner = ClamdScanner(clamconf)
|
||||
except socket.error:
|
||||
except OSError:
|
||||
errmsg = _("Could not connect to ClamAV daemon.")
|
||||
return ([], [errmsg])
|
||||
try:
|
||||
|
|
|
|||
|
|
@ -25,7 +25,7 @@ if socket.has_ipv6:
|
|||
try:
|
||||
socket.socket(socket.AF_INET6, socket.SOCK_STREAM).close()
|
||||
has_ipv6 = True
|
||||
except socket.error as msg:
|
||||
except OSError as msg:
|
||||
# only catch these one:
|
||||
# socket.error: (97, 'Address family not supported by protocol')
|
||||
# socket.error: (10047, 'Address family not supported by protocol')
|
||||
|
|
|
|||
|
|
@ -73,10 +73,10 @@ def unquote(s, matching=False):
|
|||
return s
|
||||
|
||||
|
||||
_para_mac = r"(?:%(sep)s)(?:(?:%(sep)s)\s*)+" % {'sep': '\r'}
|
||||
_para_posix = r"(?:%(sep)s)(?:(?:%(sep)s)\s*)+" % {'sep': '\n'}
|
||||
_para_win = r"(?:%(sep)s)(?:(?:%(sep)s)\s*)+" % {'sep': '\r\n'}
|
||||
_para_ro = re.compile("%s|%s|%s" % (_para_mac, _para_posix, _para_win))
|
||||
_para_mac = r"(?:{sep})(?:(?:{sep})\s*)+".format(sep='\r')
|
||||
_para_posix = r"(?:{sep})(?:(?:{sep})\s*)+".format(sep='\n')
|
||||
_para_win = r"(?:{sep})(?:(?:{sep})\s*)+".format(sep='\r\n')
|
||||
_para_ro = re.compile(f"{_para_mac}|{_para_posix}|{_para_win}")
|
||||
|
||||
|
||||
def get_paragraphs(text):
|
||||
|
|
@ -104,7 +104,7 @@ def wrap(text, width, **kwargs):
|
|||
|
||||
def indent(text, indent_string=" "):
|
||||
"""Indent each line of text with the given indent string."""
|
||||
return os.linesep.join("%s%s" % (indent_string, x) for x in text.splitlines())
|
||||
return os.linesep.join(f"{indent_string}{x}" for x in text.splitlines())
|
||||
|
||||
|
||||
def paginate(text):
|
||||
|
|
@ -191,7 +191,7 @@ def strduration_long(duration, do_translate=True):
|
|||
time_str.reverse()
|
||||
if len(time_str) > 2:
|
||||
time_str.pop()
|
||||
return "%s%s" % (prefix, ", ".join(time_str))
|
||||
return "{}{}".format(prefix, ", ".join(time_str))
|
||||
|
||||
|
||||
def strtimezone():
|
||||
|
|
|
|||
|
|
@ -43,36 +43,34 @@ default_ports = {
|
|||
# adapted from David Wheelers "Secure Programming for Linux and Unix HOWTO"
|
||||
# http://www.dwheeler.com/secure-programs/Secure-Programs-HOWTO/\
|
||||
# filter-html.html#VALIDATING-URIS
|
||||
_basic = {
|
||||
"_path": r"\-\_\.\!\~\*\'\(\),",
|
||||
"_hex_safe": r"2-9a-f",
|
||||
"_hex_full": r"0-9a-f",
|
||||
"_part": r"([a-z0-9][-a-z0-9]{0,61}|[a-z])",
|
||||
}
|
||||
_path = r"\-\_\.\!\~\*\'\(\),"
|
||||
_hex_safe = r"2-9a-f"
|
||||
_hex_full = r"0-9a-f"
|
||||
_part = r"([a-z0-9][-a-z0-9]{0,61}|[a-z])"
|
||||
_safe_char = (
|
||||
r"([a-z0-9%(_path)s\+]|"
|
||||
r"(%%[%(_hex_safe)s][%(_hex_full)s]))" % _basic
|
||||
fr"([a-z0-9{_path}\+]|"
|
||||
fr"(%[{_hex_safe}][{_hex_full}]))"
|
||||
)
|
||||
_safe_scheme_pattern = r"(https?|ftp)"
|
||||
_safe_domain_pattern = r"(%(_part)s(\.%(_part)s)*\.?)" % _basic
|
||||
_safe_host_pattern = _safe_domain_pattern + r"(:(80|8080|8000|443))?" % _basic
|
||||
_safe_domain_pattern = fr"({_part}(\.{_part})*\.?)"
|
||||
_safe_host_pattern = fr"{_safe_domain_pattern}(:(80|8080|8000|443))?"
|
||||
_safe_path_pattern = (
|
||||
r"((/([a-z0-9%(_path)s]|"
|
||||
r"(%%[%(_hex_safe)s][%(_hex_full)s]))+)*/?)" % _basic
|
||||
fr"((/([a-z0-9{_path}]|"
|
||||
fr"(%[{_hex_safe}][{_hex_full}]))+)*/?)"
|
||||
)
|
||||
_safe_fragment_pattern = r"%s*" % _safe_char
|
||||
_safe_cgi = r"%s+(=(%s|/)+)?" % (_safe_char, _safe_char)
|
||||
_safe_query_pattern = r"(%s(&%s)*)?" % (_safe_cgi, _safe_cgi)
|
||||
_safe_param_pattern = r"(%s(;%s)*)?" % (_safe_cgi, _safe_cgi)
|
||||
safe_url_pattern = r"%s://%s%s(#%s)?" % (
|
||||
_safe_fragment_pattern = fr"{_safe_char}*"
|
||||
_safe_cgi = fr"{_safe_char}+(=({_safe_char}|/)+)?"
|
||||
_safe_query_pattern = fr"({_safe_cgi}(&{_safe_cgi})*)?"
|
||||
_safe_param_pattern = fr"({_safe_cgi}(;{_safe_cgi})*)?"
|
||||
safe_url_pattern = r"{}://{}{}(#{})?".format(
|
||||
_safe_scheme_pattern,
|
||||
_safe_host_pattern,
|
||||
_safe_path_pattern,
|
||||
_safe_fragment_pattern,
|
||||
)
|
||||
|
||||
is_safe_url = re.compile("(?i)^%s$" % safe_url_pattern).match
|
||||
is_safe_domain = re.compile("(?i)^%s$" % _safe_domain_pattern).match
|
||||
is_safe_url = re.compile(f"(?i)^{safe_url_pattern}$").match
|
||||
is_safe_domain = re.compile(f"(?i)^{_safe_domain_pattern}$").match
|
||||
|
||||
|
||||
# snatched form urlparse.py
|
||||
|
|
@ -195,7 +193,7 @@ def url_fix_host(urlparts, encoding):
|
|||
if not urlparts[2] or urlparts[2] == '/':
|
||||
urlparts[2] = comps
|
||||
else:
|
||||
urlparts[2] = "%s%s" % (
|
||||
urlparts[2] = "{}{}".format(
|
||||
comps,
|
||||
urllib.parse.unquote(urlparts[2], encoding=encoding),
|
||||
)
|
||||
|
|
@ -219,7 +217,7 @@ def url_fix_host(urlparts, encoding):
|
|||
if host.endswith("."):
|
||||
host = host[:-1]
|
||||
if port != dport:
|
||||
host = "%s:%d" % (host, port)
|
||||
host = f"{host}:{port}"
|
||||
netloc = host
|
||||
urlparts[1] = userpass + netloc
|
||||
return is_idn
|
||||
|
|
@ -255,12 +253,12 @@ def url_parse_query(query, encoding):
|
|||
k = urllib.parse.quote(k, safe='/-:,;')
|
||||
if v:
|
||||
v = urllib.parse.quote(v, safe='/-:,;')
|
||||
f.append("%s=%s%s" % (k, v, sep))
|
||||
f.append(f"{k}={v}{sep}")
|
||||
elif v is None:
|
||||
f.append("%s%s" % (k, sep))
|
||||
f.append(f"{k}{sep}")
|
||||
else:
|
||||
# some sites do not work when the equal sign is missing
|
||||
f.append("%s=%s" % (k, sep))
|
||||
f.append(f"{k}={sep}")
|
||||
return ''.join(f) + append
|
||||
|
||||
|
||||
|
|
@ -381,9 +379,9 @@ def url_quote(url, encoding):
|
|||
k = urllib.parse.quote(k, safe='/-:,;')
|
||||
if v:
|
||||
v = urllib.parse.quote(v, safe='/-:,;')
|
||||
f.append("%s=%s%s" % (k, v, sep))
|
||||
f.append(f"{k}={v}{sep}")
|
||||
else:
|
||||
f.append("%s%s" % (k, sep))
|
||||
f.append(f"{k}{sep}")
|
||||
urlparts[3] = ''.join(f)
|
||||
urlparts[4] = urllib.parse.quote(urlparts[4]) # anchor
|
||||
return urlunsplit(urlparts)
|
||||
|
|
@ -397,7 +395,7 @@ def document_quote(document):
|
|||
query = None
|
||||
doc = urllib.parse.quote(doc, safe='/=,')
|
||||
if query:
|
||||
return "%s?%s" % (doc, query)
|
||||
return f"{doc}?{query}"
|
||||
return doc
|
||||
|
||||
|
||||
|
|
@ -405,7 +403,7 @@ _nopathquote_chars = "-;/=,~*+()@!"
|
|||
if os.name == 'nt':
|
||||
_nopathquote_chars += "|"
|
||||
_safe_url_chars = re.escape(_nopathquote_chars + "_:.&#%?[]!") + "a-zA-Z0-9"
|
||||
_safe_url_chars_ro = re.compile(r"^[%s]*$" % _safe_url_chars)
|
||||
_safe_url_chars_ro = re.compile(fr"^[{_safe_url_chars}]*$")
|
||||
|
||||
|
||||
def url_needs_quoting(url):
|
||||
|
|
|
|||
|
|
@ -87,7 +87,7 @@ def main(args):
|
|||
|
||||
def get_regex(schemes):
|
||||
expr = [
|
||||
"|%s # %s" % (re.escape(scheme).ljust(10), description)
|
||||
f"|{re.escape(scheme).ljust(10)} # {description}"
|
||||
for scheme, description in sorted(schemes.items())
|
||||
]
|
||||
return "\n".join(expr)
|
||||
|
|
|
|||
|
|
@ -224,7 +224,7 @@ class LinkCheckTest(TestBase):
|
|||
if hasattr(self, "port"):
|
||||
d["port"] = self.port
|
||||
# all result files are encoded in utf-8
|
||||
with open(resultfile, "r", encoding="utf-8") as f:
|
||||
with open(resultfile, encoding="utf-8") as f:
|
||||
return [
|
||||
line.rstrip("\r\n") % d
|
||||
for line in f
|
||||
|
|
|
|||
|
|
@ -132,7 +132,7 @@ class NoQueryHttpRequestHandler(StoppableHttpRequestHandler):
|
|||
list = ["example1.txt", "example2.html", "example3"]
|
||||
for name in list:
|
||||
displayname = linkname = name
|
||||
list_item = '<li><a href="%s">%s</a>\n' % (
|
||||
list_item = '<li><a href="{}">{}</a>\n'.format(
|
||||
urllib.parse.quote(linkname),
|
||||
html.escape(displayname),
|
||||
)
|
||||
|
|
@ -247,7 +247,7 @@ def get_cookie(maxage=2000):
|
|||
("Version", "1"),
|
||||
("Foo", "Bar"),
|
||||
)
|
||||
return "; ".join('%s="%s"' % (key, value) for key, value in data)
|
||||
return "; ".join(f'{key}="{value}"' for key, value in data)
|
||||
|
||||
|
||||
class CookieRedirectHttpRequestHandler(NoQueryHttpRequestHandler):
|
||||
|
|
|
|||
|
|
@ -40,9 +40,9 @@ class TelnetServerTest(LinkCheckTest):
|
|||
def get_url(self, user=None, password=None):
|
||||
if user is not None:
|
||||
if password is not None:
|
||||
netloc = "%s:%s@%s" % (user, password, self.host)
|
||||
netloc = f"{user}:{password}@{self.host}"
|
||||
else:
|
||||
netloc = "%s@%s" % (user, self.host)
|
||||
netloc = f"{user}@{self.host}"
|
||||
else:
|
||||
netloc = self.host
|
||||
return "telnet://%s:%d" % (netloc, self.port)
|
||||
|
|
|
|||
|
|
@ -54,7 +54,7 @@ class TestHttpbin(LinkCheckTest):
|
|||
def test_basic_auth(self):
|
||||
user = "testuser"
|
||||
password = "testpassword"
|
||||
url = get_httpbin_url("/basic-auth/%s/%s" % (user, password))
|
||||
url = get_httpbin_url(f"/basic-auth/{user}/{password}")
|
||||
nurl = self.norm(url)
|
||||
entry = dict(user=user, password=password, pattern=re.compile(r".*"))
|
||||
confargs = dict(authentication=[entry])
|
||||
|
|
|
|||
|
|
@ -64,7 +64,7 @@ class TestConfig(TestBase):
|
|||
patterns = [x["pattern"].pattern for x in config["externlinks"]]
|
||||
for prefix in ("ignore_", "nofollow_"):
|
||||
for suffix in ("1", "2"):
|
||||
key = "%simadoofus%s" % (prefix, suffix)
|
||||
key = f"{prefix}imadoofus{suffix}"
|
||||
self.assertTrue(key in patterns)
|
||||
for key in ("url-unicode-domain",):
|
||||
self.assertTrue(key in config["ignorewarnings"])
|
||||
|
|
|
|||
|
|
@ -37,9 +37,9 @@ def pretty_print_html(fd, soup):
|
|||
if val is None:
|
||||
fd.write(" %s" % key)
|
||||
else:
|
||||
fd.write(' %s="%s"' % (key, quote_attrval(val)))
|
||||
fd.write(f' {key}="{quote_attrval(val)}"')
|
||||
if element_text:
|
||||
fd.write(">%s</%s>" % (element_text, tag))
|
||||
fd.write(f">{element_text}</{tag}>")
|
||||
else:
|
||||
fd.write("/>")
|
||||
|
||||
|
|
|
|||
|
|
@ -43,8 +43,8 @@ class TestLFUCache(unittest.TestCase):
|
|||
self.assertTrue(not self.d)
|
||||
self.d["a"] = 1
|
||||
self.d["b"] = 2
|
||||
self.assertEqual(set([1, 2]), set(self.d.values()))
|
||||
self.assertEqual(set([1, 2]), set(self.d.itervalues()))
|
||||
self.assertEqual({1, 2}, set(self.d.values()))
|
||||
self.assertEqual({1, 2}, set(self.d.itervalues()))
|
||||
|
||||
def test_popitem(self):
|
||||
self.assertTrue(not self.d)
|
||||
|
|
|
|||
|
|
@ -125,5 +125,5 @@ class TestFtpparse(unittest.TestCase):
|
|||
for line, expected in patterns:
|
||||
res = ftpparse(line)
|
||||
self.assertEqual(
|
||||
expected, res, "got %r\nexpected %r\n%r" % (res, expected, line)
|
||||
expected, res, f"got {res!r}\nexpected {expected!r}\n{line!r}"
|
||||
)
|
||||
|
|
|
|||
|
|
@ -165,7 +165,7 @@ class TestParser(unittest.TestCase):
|
|||
Check parse results.
|
||||
"""
|
||||
res = out.getvalue()
|
||||
msg = "Test error; in: %r, out: %r, expect: %r" % (_in, res, _out)
|
||||
msg = f"Test error; in: {_in!r}, out: {res!r}, expect: {_out!r}"
|
||||
self.assertEqual(res, _out, msg=msg)
|
||||
|
||||
def test_encoding_detection_utf_content(self):
|
||||
|
|
|
|||
|
|
@ -70,5 +70,5 @@ class TestGTranslator(unittest.TestCase):
|
|||
continue
|
||||
self.assertFalse(
|
||||
"·" in line,
|
||||
"Broken GTranslator copy/paste in %r:\n%s" % (f, line),
|
||||
f"Broken GTranslator copy/paste in {f!r}:\n{line}",
|
||||
)
|
||||
|
|
|
|||
|
|
@ -40,7 +40,7 @@ class TestRobotParser(unittest.TestCase):
|
|||
else:
|
||||
ac = "access allowed"
|
||||
if a != b:
|
||||
self.fail("%s != %s (%s)" % (a, b, ac))
|
||||
self.fail(f"{a} != {b} ({ac})")
|
||||
|
||||
@need_network
|
||||
def test_nonexisting_robots(self):
|
||||
|
|
|
|||
Loading…
Reference in a new issue