Source code for linkcheck.logger.csvlog

# 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 CSV logger.
import csv
from io import StringIO
import os

from . import _Logger

Columns = (

[docs] class CSVLogger(_Logger): """ CSV output, consisting of one line per entry. Entries are separated by a separator (a semicolon per default). """ LoggerName = "csv" LoggerArgs = { "filename": "linkchecker-out.csv", 'separator': ';', "quotechar": '"', "dialect": "excel", } def __init__(self, **kwargs): """Store default separator and (os dependent) line terminator.""" args = self.get_args(kwargs) super().__init__(**args) self.init_fileoutput(args) self.separator = args['separator'] self.quotechar = args['quotechar'] self.dialect = args['dialect'] self.linesep = os.linesep
[docs] def comment(self, s, **args): """Write CSV comment.""" self.writeln(s="# %s" % s, **args)
[docs] def start_output(self): """Write checking start info as csv comment.""" super().start_output() row = [] if self.has_part("intro"): self.write_intro() self.flush() else: # write empty string to initialize file output self.write("") self.queue = StringIO() self.writer = csv.writer( self.queue, dialect=self.dialect, delimiter=self.separator, lineterminator=self.linesep, quotechar=self.quotechar, ) for s in Columns: if self.has_part(s): row.append(s) if row: self.writerow(row)
[docs] def log_url(self, url_data): """Write csv formatted url check info.""" row = [] if self.has_part("urlname"): row.append(url_data.base_url) if self.has_part("parentname"): row.append(url_data.parent_url) if self.has_part("base"): row.append(url_data.base_ref) if self.has_part("result"): row.append(url_data.result) if self.has_part("warningstring"): row.append(self.linesep.join(x[1] for x in url_data.warnings)) if self.has_part("infostring"): row.append(self.linesep.join( if self.has_part("valid"): row.append(url_data.valid) if self.has_part("url"): row.append(url_data.url) if self.has_part("line") and url_data.line is not None: row.append(url_data.line) if self.has_part("column") and url_data.column is not None: row.append(url_data.column) if self.has_part("name"): row.append( if self.has_part("dltime"): row.append(url_data.dltime) if self.has_part("dlsize"): row.append(url_data.size) if self.has_part("checktime"): row.append(url_data.checktime) if self.has_part("cached"): row.append(0) if self.has_part("level"): row.append(url_data.level) if self.has_part("modified"): row.append(self.format_modified(url_data.modified)) self.writerow(row) self.flush()
[docs] def writerow(self, row): """Write one row in CSV format.""" self.writer.writerow(row) # Fetch UTF-8 output from the queue ... data = self.queue.getvalue() try: data = data.decode("utf-8") except AttributeError: pass # ... and write to the target stream self.write(data) # empty queue self.queue.truncate(0)
[docs] def end_output(self, **kwargs): """Write end of checking info as csv comment.""" if self.has_part("outro"): self.write_outro() self.close_fileoutput()