Clean up, remove compat with py < 3.6

[pyi2ncommon] / src / log_read.py

# The software in this package is distributed under the GNU General
# Public License version 2 (with a special exception described below).
#
# A copy of GNU General Public License (GPL) is included in this distribution,
# in the file COPYING.GPL.
#
# As a special exception, if other files instantiate templates or use macros
# or inline functions from this file, or you compile this file and link it
# with other works to produce a work based on this file, this file
# does not by itself cause the resulting work to be covered
# by the GNU General Public License.
#
# However the source code for this file must still be made available
# in accordance with section (3) of the GNU General Public License.
#
# This exception does not invalidate any other reasons why a work based
# on this file might be covered by the GNU General Public License.
#
# Copyright (c) 2016-2018 Intra2net AG <info@intra2net.com>

"""

SUMMARY
------------------------------------------------------
Iterative reading of log files, similar to shell command `tail -f`.

Copyright: Intra2net AG


CONTENTS
------------------------------------------------------

Basic Functionality (class :py:class:`IterativeReader`):
Runs stat in a loop to find out whether file size has changed. Then reads the
new data and forwards that

.. todo:: Want to also use lsof to find out whether file/pipe/socket was
          closed, so can return from read loop

:py:class:`LineReader` takes output of :py:class:`IterativeReader` and returns
it line-wise as is normal for log files

:py:class:`LogParser` takes those lines and tries to parse them into fields
like date, time, module name, urgency and message.

.. todo:: auto-detect log line layout


INTERFACE
------------------------------------------------------

"""

import os
import os.path
import re
from warnings import warn
import logging
from .iter_helpers import zip_longest
from .type_helpers import is_str_or_byte, is_file_obj


class LogReadWarning(UserWarning):
    """Warnings issued by classes in this module."""
    pass


def true_func(_):
    """Replacement for :py:func:`check_is_used`. Returns `True` always."""
    return True


def check_is_used(file_handle):
    """
    Check whether file is being written to.

    To be implemented, e.g. using lsof.

    If beneficial could also easily supply python file object as arg.

    :param int file_handle: OS-level file descriptor
    """
    raise NotImplementedError(file_handle)


#: counter for unknown sources in :py:func:`create_description`
_create_description_unknown_counter = 0


def create_description(file_obj, file_handle):
    """
    Create some description for given file-like object / file descriptor.

    :param file_obj: file-like object
    :param int file_handle: os-level file descriptor
    :returns: Short description for file-like object
    :rtype: string
    """
    global _create_description_unknown_counter

    try:
        desc = file_obj.name
        if desc:
            return desc
    except AttributeError:
        pass

    if file_handle is not None:
        return 'file{0}'.format(file_handle)
    else:
        _create_description_unknown_counter += 1
        return 'unknown{0}'.format(_create_description_unknown_counter)


#: error message for IterativeReader constructor
_STR_ERR = 'not accepting file name "{0}" since cannot guarantee closing ' \
           'files --> use with open(file_name)!'


class IterativeReader(object):
    """
    Read continuously from a given file.

    Use `os.stat(file_obj.fileno()).st_size` as measure whether file has
    changed or not; Always reads as much data as possible.

    Does not care about closing files, so does not accept file names.

    This is the base for class :py:class:`LineReader` that just has to
    implement a different :py:meth:`prepare_result` method.
    """

    def __init__(self, sources, descs=None, return_when_done=False):
        """
        Create a reader; do some basic checks on args.

        :param sources: iterable over sources. Sources can be opened file
                        objects or read-opened os-level file descriptors.
                        Calling code has to ensure they are closed properly, so
                        best use this within a "with open(file_name) as
                        file_handle:"-context. If sources is a single file
                        obj/descriptor, both source and desc will be converted
                        to lists of length 1
        :param descs: can be anything of same length as sources. If sources is
                      a single source, then descs is also converted to a list
                      of length 1. If not given (i.e. None), will use
                      :py:func:`create_description` to guess descriptions
        :param bool return_when_done: ignore file_handle if no-one is writing
                                      to it any more. Return from iterator when
                                      all watched files are done (not
                                      implemented yet)
        :raises: OSError when testing fstat on source
        """
        if not sources:
            raise ValueError('need at least some source!')
        elif is_str_or_byte(sources):
            raise ValueError(_STR_ERR.format(sources))
        elif is_file_obj(sources) or isinstance(sources, int):
            source_input = [sources, ]
            desc_input = [descs, ]
        else:
            source_input = sources  # assume some iterable
            desc_input = descs

        # now divide sources into os-level file descriptors for os.fstat,
        # and file objects for read()
        self.file_objs = []
        self.file_handles = []          # file descriptOR, not descriptION
        for source in source_input:
            if is_file_obj(source):
                self.file_objs.append(source)
                self.file_handles.append(source.fileno())
            elif isinstance(source, int):
                self.file_objs.append(os.fdopen(source))
                self.file_handles.append(source)
            elif is_str_or_byte(source):
                raise ValueError(_STR_ERR.format(source))
            else:
                raise ValueError('source {0} is neither file obj nor file '
                                 'descriptor!')

            # try to fstat the new file descriptor just for testing
            os.fstat(self.file_handles[-1])

        # guess descriptions if not given
        if not desc_input:
            self.descriptions = [create_description(obj, file_handle)
                                 for obj, file_handle
                                 in zip(self.file_objs, self.file_handles)]
        else:
            try:
                if len(desc_input) != len(self.file_objs):
                    raise ValueError('need same number of sources and '
                                     'descriptions!')
            except TypeError:
                pass  # desc_input is generator or so

            self.descriptions = []
            for obj, file_handle, description in \
                    zip_longest(self.file_objs, self.file_handles, desc_input):
                if obj is None:
                    raise ValueError('more descriptions than sources!')
                elif description is None:
                    self.descriptions.append(create_description(obj,
                                                                file_handle))
                else:
                    self.descriptions.append(description)

        self.last_sizes = [0 for _ in self.file_objs]
        self.ignore = [False for _ in self.file_objs]

        if return_when_done:
            self.is_used_func = check_is_used
        else:
            self.is_used_func = true_func

        for obj, file_handle, description in \
                zip(self.file_objs, self.file_handles, self.descriptions):
            logging.debug('log_read initialized with file descriptor {0}, '
                          'file obj {1}, description "{2}"'
                          .format(file_handle, obj, description))

    def n_sources(self):
        """Return number of sources given to constructor."""
        return len(self.file_objs)

    def n_active_sources(self):
        """Return number of sources we are actually watching."""
        return len(self.ignore) - sum(self.ignore)

    def __iter__(self):
        """
        Continue reading from sources, yield results.

        yields result of :py:meth:`prepare_result`, which depends on what sub
        class you called this function from.
        """
        while True:
            for idx, (obj, file_handle, description, last_size, do_ignore) in \
                    enumerate(zip(self.file_objs, self.file_handles,
                                  self.descriptions, self.last_sizes,
                                  self.ignore)):
                if do_ignore:
                    continue

                # get new file size
                new_size = os.fstat(file_handle).st_size

                # compare to old size
                if new_size == last_size:
                    if not self.is_used_func(file_handle):
                        warn('no one is writing to {0} / {1} -- '
                             'stop watching it!'
                             .format(file_handle, description),
                             category=LogReadWarning)
                        self.ignore[idx] = True
                else:
                    if new_size < last_size:  # happened at start of some tests
                        warn('{0} / {1} has become smaller ({2} --> {3})! '
                             .format(obj, description, last_size, new_size)
                             + 'Maybe you are reading from a half-initialized '
                             + 'file?',
                             category=LogReadWarning)
                    try:
                        new_data = obj.read()
                    except OSError as ose:    # includes IOErrors
                        warn('io error reading from {0} / {1}: {2})'
                             .format(obj, description, ose),
                             category=LogReadWarning)
                        new_data = str(ose)
                    except UnicodeDecodeError as ude:
                        warn('unicode error reading from {0} / {1}: {2}'
                             .format(obj, description, ude),
                             category=LogReadWarning)
                        new_data = str(ude)

                    # post-processing
                    to_yield = self.prepare_result(description, new_data, idx)
                    for result in to_yield:
                        yield result

                    # prepare next iteration
                    self.last_sizes[idx] = new_size

    def prepare_result(self, description, data, idx):
        """
        From raw new data create some yield-able results.

        Intended for overwriting in subclasses.

        This function is called from __iter__ for each new data that becomes
        available. It has to return some iterable whose entries are yielded
        from iteration over objects of this class.

        This base implementation just returns its input in a list, so new data
        is yielded from __iter__ as-is. Subclass implementations can also yield
        tuples.

        :param str description: Description of source of lines, one of
                                :py:data:`self.descriptions`
        :param str data: Text data read from source
        :param idx: Index of data source
        :returns: [(description, data, idx], same as input
        :rtype [(str, str, int)]
        """
        return [(description, data, idx), ]


#: characters to `rstrip()` from end of complete lines
LINE_SPLITTERS = '\n\r'


class LineReader(IterativeReader):
    """
    An :py:class:`IterativeReader` that returns new data line-wise.

    This means buffering partial line data.
    """

    def __init__(self, *args, **kwargs):
        """Create an :py:class:`IterativeReader and buffers for sources."""
        super(LineReader, self).__init__(*args, **kwargs)
        self.line_buffers = ['' for _ in range(self.n_sources())]

    def prepare_result(self, description, new_data, idx):
        """
        Take raw new data and split it into lines.

        If line is not complete, then buffer it.

        Args: see super class method :py:meth:`IterativeReader.prepare_result`
        :returns: list of 3-tuples `(description, line, idx)` where
                  `description` and `idx` are same as args, and `line` is
                  without trailing newline characters
        :rtype: [(str, str, int)]
        """
        all_data = self.line_buffers[idx] + new_data
        self.line_buffers[idx] = ''
        result = []
        should_be_no_new_lines = False
        for line in all_data.splitlines(True):
            if line[-1] in LINE_SPLITTERS:
                result.append((description, line.rstrip(LINE_SPLITTERS), idx))
            elif should_be_no_new_lines:
                # self-check
                raise ValueError('Programming error: something went wrong with '
                                 'line splitting/buffering.')
            else:
                self.line_buffers[idx] = line
                should_be_no_new_lines = True  # (this should be the last)

        return result


class LogParser(LineReader):
    """
    Takes lines from :py:class:`LineReader` and parses their contents.

    Requires a pattern for log lines, auto-detection is not implemented yet.

    Iteration returns re.match result or -- if matching failed -- the original
    raw line.
    """

    def __init__(self, log_file, pattern=None):
        """
        Create a LogParser.

        :param str log_file: name of log file to parse (required!)
        :param pattern: regexp to split log lines; None (default) to return
                        line as they are
        :type pattern: str or None (default)
        """
        super(LogParser, self).__init__(log_file)

        self.pattern = pattern

    def prepare_result(self, *args):
        """
        Try parsing lines.

        Args: see super class method :py:meth:`IterativeReader.prepare_result`
        :returns: 3-tuples `(description, line, idx)` where `description` and
                  `idx` are same as input args and `line` is either a
                  :py:class:`re.Match` if line matched :py:data:`self.pattern`
                  or just str if line did not match.
        :rtype: [(str, :py:class:`re.Match` OR str, int)]
        """
        # let super class split data into lines
        for description, raw_line, idx in \
                super(LogParser, self).prepare_result(*args):
            result = re.match(self.pattern, raw_line)
            if result:
                return description, result, idx
            else:
                return description, raw_line, idx