Source code for linkcheck.logger.sql

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

import os
from . import _Logger
from .. import url as urlutil


[docs]def sqlify(s): """ Escape special SQL chars and strings. """ if not s: return "NULL" return "'%s'" % s.replace("'", "''").replace(os.linesep, r"\n")
[docs]def intify(s): """ Coerce a truth value to 0/1. @param s: an object (usually a string) @type s: object @return: 1 if object truth value is True, else 0 @rtype: number """ if s: return 1 return 0
[docs]class SQLLogger(_Logger): """ SQL output, should work with any SQL database (not tested). """ LoggerName = 'sql' LoggerArgs = { "filename": "linkchecker-out.sql", 'separator': ';', 'dbname': 'linksdb', } def __init__(self, **kwargs): """Initialize database access data.""" args = self.get_args(kwargs) super().__init__(**args) self.init_fileoutput(args) self.dbname = args['dbname'] self.separator = args['separator']
[docs] def comment(self, s, **args): """ Write SQL comment. """ self.write("-- ") self.writeln(s=s, **args)
[docs] def start_output(self): """ Write start of checking info as sql comment. """ super().start_output() if self.has_part("intro"): self.write_intro() self.writeln() self.flush()
[docs] def log_url(self, url_data): """ Store url check info into the database. """ self.writeln( "insert into %(table)s(urlname," "parentname,baseref,valid,result,warning,info,url,line,col," "name,checktime,dltime,size,cached,level,modified) values (" "%(base_url)s," "%(url_parent)s," "%(base_ref)s," "%(valid)d," "%(result)s," "%(warning)s," "%(info)s," "%(url)s," "%(line)s," "%(column)s," "%(name)s," "%(checktime)d," "%(dltime)d," "%(size)d," "%(cached)d," "%(level)d," "%(modified)s" ")%(separator)s" % { 'table': self.dbname, 'base_url': sqlify(url_data.base_url), '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)), 'info': sqlify(os.linesep.join(url_data.info)), 'url': sqlify(urlutil.url_quote(url_data.url, encoding="utf-8")), 'line': 'NULL' if url_data.line is None else url_data.line, 'column': 'NULL' if url_data.column is None else url_data.column, 'name': sqlify(url_data.name), 'checktime': url_data.checktime, 'dltime': url_data.dltime, 'size': url_data.size, 'cached': 0, 'separator': self.separator, "level": url_data.level, "modified": sqlify(self.format_modified(url_data.modified)), } ) self.flush()
[docs] def end_output(self, **kwargs): """ Write end of checking info as sql comment. """ if self.has_part("outro"): self.write_outro() self.close_fileoutput()