Source code for linkcheck.logger.failures

# 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
# 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.
A failures logger.

import os

from . import _Logger
from .. import log, LOG_CHECK
from ..configuration import get_user_data

[docs]class FailuresLogger(_Logger): """ Updates a list of failed links. If a link already on the list is found to be working, it is removed. After n days we only have links on the list which failed within those n days. """ LoggerName = "failures" LoggerArgs = { "filename": os.path.join(get_user_data(), "failures"), } def __init__(self, **kwargs): """Intialize with old failures data (if found).""" blacklist = os.path.join(get_user_data(), "blacklist") if os.path.isfile(blacklist): self.LoggerArgs["filename"] = blacklist log.warn( LOG_CHECK, _("%(blacklist)s file is deprecated please rename to failures") % {"blacklist": blacklist} ) args = self.get_args(kwargs) super().__init__(**args) self.init_fileoutput(args) self.failures = {} if self.filename is not None and os.path.exists(self.filename): self.read_failures()
[docs] def comment(self, s, **args): """ Write nothing. """ pass
[docs] def log_url(self, url_data): """ Add invalid url to failures, delete valid url from failures. """ key = (url_data.parent_url, url_data.cache_url) key = repr(key) if key in self.failures: if url_data.valid: del self.failures[key] else: self.failures[key] += 1 else: if not url_data.valid: self.failures[key] = 1
[docs] def end_output(self, **kwargs): """ Write failures file. """ self.write_failures()
[docs] def read_failures(self): """ Read a previously stored failures from file fd. """ with open(self.filename, 'r', encoding=self.output_encoding, errors=self.codec_errors) as fd: for line in fd: line = line.rstrip() if line.startswith('#') or not line: continue value, key = line.split(None, 1) key = key.strip('"') if not key.startswith('('): log.critical( LOG_CHECK, _("invalid line starting with '%(linestart)s' in %(failures)s") % {"linestart": line[:12], "failures": self.filename} ) raise SystemExit(2) self.failures[key] = int(value)
[docs] def write_failures(self): """ Write the failures file. """ oldmask = os.umask(0o077) for key, value in self.failures.items(): self.write("%d %s%s" % (value, repr(key), os.linesep)) self.close_fileoutput() # restore umask os.umask(oldmask)