1
\$\begingroup\$

After encountering a suggestion that monitoring a log file for a webserver could allow for the implementation of automatically analyzing and handling of certain events (that might require altering firewall rules etc.), the following example application was written as a simple beginning point for processing the available data. It is not meant to be full featured or to have dependencies on anything outside of the standard library since those could be added by someone else who needed more than this example boilerplate, and the code should not favor any particular operating system in the assumptions it makes. The use of double underscores throughout the code, the long lines found in a few places throughout the program, and the fact that there are already programs that accomplish similar tasks already could be points of contention for some people but should be ignored for the purpose of this review.

While looking through this code, do you see any places where you think to yourself:

  1. Some functionality from a module in the standard library would be a perfect fit for this area to (a) increase readability, (b) increase performance, or (c) reduce the amount of code required?
  2. The standard library is not being used (a) properly, (b) efficiently, or (c) as intended in some part of the program and should either (a) be replaced with something else from the standard library or (b) have custom code written to improve that section of the program?
#! /usr/bin/env python3
# -*- coding: utf-8 -*-
"""Example Application for Monitoring a Server's Access Log.

This is free and unencumbered software released into the public domain.

Anyone is free to copy, modify, publish, use, compile, sell, or
distribute this software, either in source code form or as a compiled
binary, for any purpose, commercial or non-commercial, and by any
means.

In jurisdictions that recognize copyright laws, the author or authors
of this software dedicate any and all copyright interest in the
software to the public domain. We make this dedication for the benefit
of the public at large and to the detriment of our heirs and
successors. We intend this dedication to be an overt act of
relinquishment in perpetuity of all present and future rights to this
software under copyright law.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
IN NO EVENT SHALL THE AUTHORS BE LIABLE FOR ANY CLAIM, DAMAGES OR
OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE,
ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
OTHER DEALINGS IN THE SOFTWARE.

For more information, please refer to <http://unlicense.org/>"""

import collections
import datetime
import enum
import functools
import http
import ipaddress
import math
import pathlib
import re
import sys
import threading
import time
import urllib.parse

# Public Names
__all__ = (
    'SERVER_ROOT',
    'LOG_FILE',
    'COMBINED_LOG_FORMAT',
    'UPDATE_DELAY',
    'REPORT_DELAY',
    'LOG_PATH',
    'RE_CLEANUP_TABLE',
    'RE_COMBINED_LOG_FORMAT',
    'RE_CLF_PATTERN_SEARCH',
    'BILLION',
    'KB',
    'PREFIX',
    'main',
    'underline',
    'ExampleDatabase',
    'convert_datetime',
    'HTTPMethod',
    'convert_status',
    'LOG_ADAPTERS',
    'bytes_to_str',
    'UpdateEngine'
)

# Module Documentation
__version__ = 1, 0, 0
__date__ = datetime.date(2022, 6, 30)
__author__ = 'Stephen Paul Chappell'
__credits__ = 'vigy, for providing the inspiration for this program.'

# Configuration Constants
SERVER_ROOT = pathlib.Path('C:/Abyss Web Server')
LOG_FILE = pathlib.Path('log/access.log')
# Reference: https://www.oreilly.com/library/view/regular-expressions-cookbook/9781449327453/ch07s13.html
COMBINED_LOG_FORMAT = '''\
^(?P<client>\\S+)●\\S+●(?P<userid>\\S+)●\\[(?P<datetime>[^\\]]+)\\]↵
●"(?P<method>[A-Z]+)●(?P<request>[^●"]+)?●HTTP/[0-9.]+"↵
●(?P<status>[0-9]{3})●(?P<size>[0-9]+|-)●"(?P<referrer>[^"]*)"↵
●"(?P<useragent>[^"]*)"'''
UPDATE_DELAY = datetime.timedelta(seconds=1)
REPORT_DELAY = datetime.timedelta(seconds=5)

# Symbolic Constants
LOG_PATH = SERVER_ROOT / LOG_FILE
RE_CLEANUP_TABLE = str.maketrans('●', ' ', '↵\n')
RE_COMBINED_LOG_FORMAT = COMBINED_LOG_FORMAT.translate(RE_CLEANUP_TABLE)
RE_CLF_PATTERN_SEARCH = re.compile(RE_COMBINED_LOG_FORMAT).search
BILLION = 10 ** 9
KB = 1 << 10
PREFIX = '', 'K', 'M', 'G', 'T', 'P', 'E', 'Z', 'Y'


def main():
    """Periodically generate reports for overall analyzed requests."""
    db = ExampleDatabase(underline(f'\nREPORT FOR {LOG_PATH}'))
    engine = UpdateEngine(LOG_PATH, RE_CLF_PATTERN_SEARCH, UPDATE_DELAY, db)
    engine.start()
    db.report_forever(REPORT_DELAY, lambda: engine.is_alive())


def underline(display_text, underline_character='='):
    """Take a potentially multi-lined string and underline its width."""
    if not isinstance(underline_character, str):
        raise TypeError('underline_character must be of type str')
    if len(underline_character) != 1:
        raise ValueError('underline_character must be a single character')
    max_length = max(map(len, display_text.splitlines()))
    return f'{display_text}\n{underline_character * max_length}'


# Reference: https://goaccess.io/
class ExampleDatabase:
    """Class demonstrating an approach to processing _LogRecord instances."""

    def __init__(self, report_header):
        """Initialize the instance as an empty database."""
        self.__report_header = report_header
        self.__last_updated = datetime.datetime.now()
        self.__records = []
        self.__lock = threading.Lock()

    def __iter__(self):
        """Iterate over all records in the database."""
        yield from self.__records

    def add_record(self, record, parse_time, line_length):
        """Add a single record with metadata to the database."""
        if not isinstance(record, _LogRecord):
            raise TypeError('record must be of type _LogRecord')
        with self.__lock:
            self.__last_updated = datetime.datetime.now()
            # noinspection PyArgumentList
            self.__records.append(_EDRecord(*record, parse_time, line_length))

    @property
    def last_updated(self):
        """Timestamp of when the database was last updated."""
        return self.__last_updated

    @property
    def total_requests(self):
        """The total requests recorded so far."""
        return len(self.__records)

    @property
    def valid_requests(self):
        """The number of requested considered valid."""
        return sum(1 for record in self if record.valid)

    @property
    def failed_requests(self):
        """The number of requests considered failed."""
        return sum(1 for record in self if record.failed)

    @property
    def log_parsing_time(self):
        """The total time used for parsing the log file."""
        ns = sum(record.parse_time for record in self)
        return datetime.timedelta(seconds=ns / BILLION)

    @property
    def unique_visitors(self):
        """The count of unique client identifiers."""
        return len({record.client for record in self})

    @property
    def requested_files(self):
        """The total number of file requests made so far."""
        return sum(1 for record in self if record.file)

    @property
    def path_to_clients(self):
        """A simple database of who has requested each server path."""
        path_to_clients = {}
        for record in self:
            clients = path_to_clients.setdefault(record.path, set())
            clients.add(record.client)
        return path_to_clients

    @property
    def excl_ip_hits(self):
        """The number of paths that have been requested by only one client."""
        ptc_values = self.path_to_clients.values()
        return sum(1 for value in ptc_values if len(value) == 1)

    @property
    def referrers(self):
        """The count of unique referrers across all requests."""
        return len({record.referrer for record in self})

    @property
    def not_found(self):
        """The count of requests having a 404 error generated for them."""
        return sum(1 for record in self if record.not_found)

    @property
    def static_files(self):
        """The count of unique paths to files."""
        return len({record.path for record in self if record.file})

    @property
    def log_size(self):
        """The size of the log in characters processed so far."""
        return sum(record.line_length for record in self)

    @property
    def log_size_str(self):
        """A readable log size that assumes characters are bytes."""
        return bytes_to_str(self.log_size)

    @property
    def tx_amount(self):
        """The total bytes transferred from the server to its clients."""
        return sum(record.size for record in self)

    @property
    def tx_amount_str(self):
        """A readable transfer amount that assumes counts are in bytes."""
        return bytes_to_str(self.tx_amount)

    def create_report(self):
        """Generate an example report in a threadsafe way."""
        with self.__lock:
            return f'''\
{self.__report_header!s}
Last Updated     = {self.last_updated!s}
Total Requests   = {self.total_requests:,}
Valid Requests   = {self.valid_requests:,}
Failed Requests  = {self.failed_requests:,}
Log Parsing Time = {self.log_parsing_time!s}
Unique Visitors  = {self.unique_visitors:,}
Requested Files  = {self.requested_files:,}
Excl. IP Hits    = {self.excl_ip_hits:,}
Referrers        = {self.referrers:,}
Not Found        = {self.not_found:,}
Static Files     = {self.static_files:,}
Log Size         = {self.log_size_str}
Tx. Amount       = {self.tx_amount_str}'''

    def report_forever(self, interval, control=lambda: True, callback=print):
        """Create reports forever using the provided interval."""
        last_report = self.last_updated
        while control():
            time.sleep(interval.total_seconds())
            if last_report != self.last_updated:
                last_report = self.last_updated
                callback(self.create_report())


def convert_datetime(value):
    """Convert an access log timestamp into a datetime object."""
    return datetime.datetime.strptime(value, '%d/%b/%Y:%H:%M:%S %z')


# Reference: https://www.w3schools.com/tags/ref_httpmethods.asp
@enum.unique
class HTTPMethod(enum.Enum):
    """Enumeration of all legal HTTP methods recognized by servers."""

    GET = 'GET is used to request data from a specified resource.'
    POST = 'POST is used to send data to a server to create/update a resource.'
    PUT = 'PUT is used to send data to a server to create/update a resource.'
    HEAD = 'HEAD is almost identical to GET, but without the response body.'
    DELETE = 'The DELETE method deletes the specified resource.'
    PATCH = 'The PATCH method is used to apply partial modifications to a resource.'
    OPTIONS = 'The OPTIONS method describes the communication options for the target resource.'
    CONNECT = 'The CONNECT method is used to start a two-way communications (a tunnel) with the requested resource.'
    TRACE = 'The TRACE method method is used to perform a message loop-back test that tests the path for the target resource (useful for debugging purposes).'


def convert_status(value):
    """Convert an HTTP status code into an enumeration instance."""
    return http.HTTPStatus(int(value))


LOG_ADAPTERS = collections.OrderedDict((
    ('client', ipaddress.ip_address),
    ('userid', sys.intern),
    ('datetime', convert_datetime),
    ('method', HTTPMethod.__getitem__),
    ('request', urllib.parse.urlparse),
    ('status', convert_status),
    ('size', int),
    ('referrer', sys.intern),
    ('useragent', sys.intern)
))

_ED_RECORD_FIELDS = tuple(LOG_ADAPTERS) + ('parse_time', 'line_length')


class _EDRecord(collections.namedtuple('_EDRecordBase', _ED_RECORD_FIELDS)):
    VALID_STATUS_PREFIX = {1, 2, 3}
    FAILED_STATUS_PREFIX = {4, 5}

    @functools.cached_property
    def status_prefix(self):
        """The first digit of a three-digit status code."""
        # noinspection PyUnresolvedReferences
        return self.status.value // 100

    @functools.cached_property
    def valid(self):
        """The indicator if the request was valid."""
        return self.status_prefix in self.VALID_STATUS_PREFIX

    @functools.cached_property
    def failed(self):
        """The indicator if the request was a failure."""
        return self.status_prefix in self.FAILED_STATUS_PREFIX

    @functools.cached_property
    def path(self):
        """The path of what was requested sans other request data."""
        # noinspection PyUnresolvedReferences
        return pathlib.PurePath(self.request.path)

    @functools.cached_property
    def file(self):
        """A guess whether the path represents a file or not."""
        return bool(self.path.suffix)

    @functools.cached_property
    def not_found(self):
        """The indicator if a request caused a 404 error or not."""
        # noinspection PyUnresolvedReferences
        return self.status is http.HTTPStatus.NOT_FOUND


def bytes_to_str(size):
    """Convert an item's size in bytes to an easily readable representation."""
    size = int(size)
    if size:
        selector = round(math.log(size, KB) - 0.4)
        if selector:
            return f'{size / KB ** selector:.2f} {PREFIX[selector]}B'
        return f'{size:d} {PREFIX[selector]}B'
    return '-'


class UpdateEngine(threading.Thread):
    """Core of this program that handles changes in the access log file."""

    def __init__(self, log_path, search, sleep_interval, database):
        """Initialize the instance with data for detecting log changes."""
        super().__init__(daemon=True)
        self.__log_path = log_path
        self.__search = search
        self.__sleep_interval = sleep_interval
        self.__database = database
        self.__stream_position = 0
        self.__line_number = 1

    @property
    def stream_position(self):
        """The place in the file where the next update will resume."""
        return self.__stream_position

    @property
    def line_number(self):
        """The line in the file where the next update will resume."""
        return self.__line_number

    def run(self):
        """Detect when the log file is updated and trigger updates."""
        last_modified = None
        while True:
            modified_time = self.__log_path.stat().st_mtime
            next_modified = datetime.datetime.fromtimestamp(modified_time)
            if last_modified != next_modified:
                last_modified = next_modified
                self.__handle_update()
            time.sleep(self.__sleep_interval.total_seconds())

    def __handle_update(self):
        """Create and add records to the database when updates occur."""
        with self.__log_path.open() as file:
            file.seek(self.stream_position)
            for line in file:
                start_time = time.perf_counter_ns()
                if match := self.__search(line):
                    record = self.__create_record(match)
                    stop_time = time.perf_counter_ns()
                    parse_time = stop_time - start_time
                    self.__database.add_record(record, parse_time, len(line))
                else:
                    self.handle_search_error()
                self.__line_number += 1
            self.__stream_position = file.tell()

    def __create_record(self, match):
        """Make a log record and convert fields to their proper datatypes."""
        groups = match.groupdict()
        record = _LogRecord(*(
            value(groups.pop(key)) for key, value in LOG_ADAPTERS.items()
        ))
        if groups:
            self.handle_match_error(groups)
        return record

    def handle_search_error(self):
        """Provide a default behavior when a search fails."""
        raise ValueError(f'starting at position {self.stream_position}, '
                         f'line {self.line_number} could not be handled')

    def handle_match_error(self, groups):
        """Provide a default behavior when groups are not processed."""
        raise ValueError(f'groups not processed: {", ".join(groups)}')


_LogRecord = collections.namedtuple('LogRecord', LOG_ADAPTERS)

if __name__ == '__main__':
    main()
\$\endgroup\$
2
  • \$\begingroup\$ [Some things] could be points of contention for some people but should be ignored for the purpose of this review is not how the site works. Any insightful observations are on-topic. \$\endgroup\$
    – Reinderien
    Jul 1 at 14:12
  • \$\begingroup\$ Your previous question on the same topic - Abyss web server - has answers that have received no feedback one way or the other. You should address this. \$\endgroup\$
    – Reinderien
    Jul 1 at 14:22

2 Answers 2

0
\$\begingroup\$

This does not appear to be an I/O bound application, so using threading will not contribute to its performance. To accomplish the task of having multiple actions executed concurrently, you could use the sched module instead. If you want to make the program more modular, you might consider creating an abstract base class using the abc module.

Depending on the interface that is created using the abstract class, its children may need modifications. The constructor/initializer may take extra arguments, and some methods may need to be reorganized. In the solution provided down below, a new Repeating class is introduced; and it is used in both the ExampleDatabase and UpdateEngine classes.

With the use of the scheduler class, the threading module can be dropped from the program. Neither its Lock class nor Thread class are needed anymore to enable the program's concurrent tasks to operate together. However, refactoring is required since run methods return now after having completed one round of their associated actions.

In child classes of Repeating, the three methods that can be overridden are:

  • __init__ (required) - marking the initializer as abstract helps to enforce the abstract nature of the class
  • gate (optional) - the default implementation always allows the run method to be executed as its action
  • run (required) - this represents the main activity of the class and is simular to run in the Thread class

While gate was not strictly required, both child classes of Repeating had obvious use for such a concept; and it allows for better separation of concerns. UpdateEngine benefited the most from gate and run with one of its methods only needing to be renamed. However, new instance attributes were required to assist with the logic in gate.

#! /usr/bin/env python3
# -*- coding: utf-8 -*-
"""Example Application for Monitoring a Server's Access Log.

This is free and unencumbered software released into the public domain.

Anyone is free to copy, modify, publish, use, compile, sell, or
distribute this software, either in source code form or as a compiled
binary, for any purpose, commercial or non-commercial, and by any
means.

In jurisdictions that recognize copyright laws, the author or authors
of this software dedicate any and all copyright interest in the
software to the public domain. We make this dedication for the benefit
of the public at large and to the detriment of our heirs and
successors. We intend this dedication to be an overt act of
relinquishment in perpetuity of all present and future rights to this
software under copyright law.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
IN NO EVENT SHALL THE AUTHORS BE LIABLE FOR ANY CLAIM, DAMAGES OR
OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE,
ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
OTHER DEALINGS IN THE SOFTWARE.

For more information, please refer to <http://unlicense.org/>"""

import abc
import collections
import datetime
import enum
import functools
import http
import ipaddress
import math
import pathlib
import re
import sched
import sys
import time
import urllib.parse

# Public Names
__all__ = (
    'SERVER_ROOT',
    'LOG_FILE',
    'COMBINED_LOG_FORMAT',
    'UPDATE_DELAY',
    'REPORT_DELAY',
    'LOG_PATH',
    'RE_CLEANUP_TABLE',
    'RE_COMBINED_LOG_FORMAT',
    'RE_CLF_PATTERN_SEARCH',
    'BILLION',
    'KB',
    'PREFIX',
    'main',
    'underline',
    'Repeating',
    'ExampleDatabase',
    'convert_datetime',
    'HTTPMethod',
    'convert_status',
    'LOG_ADAPTERS',
    'bytes_to_str',
    'UpdateEngine'
)

# Module Documentation
__version__ = 1, 1, 0
__date__ = datetime.date(2022, 7, 20)
__author__ = 'Stephen Paul Chappell'
__credits__ = 'vigy, for providing the inspiration for this program.'

# Configuration Constants
SERVER_ROOT = pathlib.Path('C:/Abyss Web Server')
LOG_FILE = pathlib.Path('log/access.log')
# Reference: https://www.oreilly.com/library/view/regular-expressions-cookbook/9781449327453/ch07s13.html
COMBINED_LOG_FORMAT = '''\
^(?P<client>\\S+)●\\S+●(?P<userid>\\S+)●\\[(?P<datetime>[^\\]]+)\\]↵
●"(?P<method>[A-Z]+)●(?P<request>[^●"]+)?●HTTP/[0-9.]+"↵
●(?P<status>[0-9]{3})●(?P<size>[0-9]+|-)●"(?P<referrer>[^"]*)"↵
●"(?P<useragent>[^"]*)"'''
UPDATE_DELAY = datetime.timedelta(seconds=1)
REPORT_DELAY = datetime.timedelta(seconds=5)

# Symbolic Constants
LOG_PATH = SERVER_ROOT / LOG_FILE
RE_CLEANUP_TABLE = str.maketrans('●', ' ', '↵\n')
RE_COMBINED_LOG_FORMAT = COMBINED_LOG_FORMAT.translate(RE_CLEANUP_TABLE)
RE_CLF_PATTERN_SEARCH = re.compile(RE_COMBINED_LOG_FORMAT).search
BILLION = 10 ** 9
KB = 1 << 10
PREFIX = '', 'K', 'M', 'G', 'T', 'P', 'E', 'Z', 'Y'


def main():
    """Periodically generate reports for overall analyzed requests."""
    db = ExampleDatabase(underline(f'\nREPORT FOR {LOG_PATH}'), REPORT_DELAY)
    UpdateEngine(LOG_PATH, RE_CLF_PATTERN_SEARCH, UPDATE_DELAY, db)
    Repeating.start()


def underline(display_text, underline_character='='):
    """Take a potentially multi-lined string and underline its width."""
    if not isinstance(underline_character, str):
        raise TypeError('underline_character must be of type str')
    if len(underline_character) != 1:
        raise ValueError('underline_character must be a single character')
    max_length = max(map(len, display_text.splitlines()))
    return f'{display_text}\n{underline_character * max_length}'


class Repeating(abc.ABC):
    """Abstract class for tasks that need to be repeated on a schedule."""

    PRIORITY = math.inf  # Default priority for class instances.
    __scheduler = sched.scheduler()  # Central queue for all task schedules.

    @classmethod
    def start(cls):
        """Begin running schedules specified by the child instances."""
        cls.__scheduler.run()

    @abc.abstractmethod
    def __init__(self, delay):
        """Initialize the instance and schedule the first run."""
        self.delay = delay
        self.__enter()

    @property
    def delay(self):
        """The interval between repeating an instance's primary action."""
        return self.__delay

    @delay.setter
    def delay(self, value):
        if not isinstance(value, datetime.timedelta):
            raise TypeError('delay must be of type timedelta')
        if value.total_seconds() < 0:
            raise ValueError('total seconds must be zero or greater')
        self.__delay = value

    def __enter(self):
        """Schedule the next time the primary action should be run."""
        delay = self.__delay.total_seconds()
        self.__scheduler.enter(delay, self.PRIORITY, self.__action)

    def __action(self):
        """Get permission to run, execute as needed, and reschedule."""
        try:
            if self.gate():
                self.run()
        finally:
            self.__enter()

    def gate(self):
        """Check if the run method should be executed."""
        return True

    @abc.abstractmethod
    def run(self):
        """Execute the primary action an instance is meant to repeat."""
        pass


# Reference: https://goaccess.io/
class ExampleDatabase(Repeating):
    """Class demonstrating an approach to processing _LogRecord instances."""

    PRIORITY = 2  # Lower numbers have higher priority.

    def __init__(self, report_header, delay):
        """Initialize the instance as an empty database."""
        super().__init__(delay)
        self.__report_header = report_header
        self.__last_updated = datetime.datetime.now()
        self.__last_report = None
        self.__records = []

    def __iter__(self):
        """Iterate over all records in the database."""
        yield from self.__records

    def add_record(self, record, parse_time, line_length):
        """Add a single record with metadata to the database."""
        if not isinstance(record, _LogRecord):
            raise TypeError('record must be of type _LogRecord')
        self.__last_updated = datetime.datetime.now()
        # noinspection PyArgumentList
        self.__records.append(_EDRecord(*record, parse_time, line_length))

    @property
    def last_updated(self):
        """Timestamp of when the database was last updated."""
        return self.__last_updated

    @property
    def total_requests(self):
        """The total requests recorded so far."""
        return len(self.__records)

    @property
    def valid_requests(self):
        """The number of requested considered valid."""
        return sum(1 for record in self if record.valid)

    @property
    def failed_requests(self):
        """The number of requests considered failed."""
        return sum(1 for record in self if record.failed)

    @property
    def log_parsing_time(self):
        """The total time used for parsing the log file."""
        ns = sum(record.parse_time for record in self)
        return datetime.timedelta(seconds=ns / BILLION)

    @property
    def unique_visitors(self):
        """The count of unique client identifiers."""
        return len({record.client for record in self})

    @property
    def requested_files(self):
        """The total number of file requests made so far."""
        return sum(1 for record in self if record.file)

    @property
    def path_to_clients(self):
        """A simple database of who has requested each server path."""
        path_to_clients = {}
        for record in self:
            clients = path_to_clients.setdefault(record.path, set())
            clients.add(record.client)
        return path_to_clients

    @property
    def excl_ip_hits(self):
        """The number of paths that have been requested by only one client."""
        ptc_values = self.path_to_clients.values()
        return sum(1 for value in ptc_values if len(value) == 1)

    @property
    def referrers(self):
        """The count of unique referrers across all requests."""
        return len({record.referrer for record in self})

    @property
    def not_found(self):
        """The count of requests having a 404 error generated for them."""
        return sum(1 for record in self if record.not_found)

    @property
    def static_files(self):
        """The count of unique paths to files."""
        return len({record.path for record in self if record.file})

    @property
    def log_size(self):
        """The size of the log in characters processed so far."""
        return sum(record.line_length for record in self)

    @property
    def log_size_str(self):
        """A readable log size that assumes characters are bytes."""
        return bytes_to_str(self.log_size)

    @property
    def tx_amount(self):
        """The total bytes transferred from the server to its clients."""
        return sum(record.size for record in self)

    @property
    def tx_amount_str(self):
        """A readable transfer amount that assumes counts are in bytes."""
        return bytes_to_str(self.tx_amount)

    def create_report(self):
        """Generate an example report in a threadsafe way."""
        return f'''\
{self.__report_header!s}
Last Updated     = {self.last_updated!s}
Total Requests   = {self.total_requests:,}
Valid Requests   = {self.valid_requests:,}
Failed Requests  = {self.failed_requests:,}
Log Parsing Time = {self.log_parsing_time!s}
Unique Visitors  = {self.unique_visitors:,}
Requested Files  = {self.requested_files:,}
Excl. IP Hits    = {self.excl_ip_hits:,}
Referrers        = {self.referrers:,}
Not Found        = {self.not_found:,}
Static Files     = {self.static_files:,}
Log Size         = {self.log_size_str}
Tx. Amount       = {self.tx_amount_str}'''

    def gate(self):
        """Generate a new report only after an update is detected."""
        if self.__last_report != self.__last_updated:
            self.__last_report = self.__last_updated
            return True
        return False

    def run(self):
        """Create a report and display it to the screen."""
        print(self.create_report())


def convert_datetime(value):
    """Convert an access log timestamp into a datetime object."""
    return datetime.datetime.strptime(value, '%d/%b/%Y:%H:%M:%S %z')


# Reference: https://www.w3schools.com/tags/ref_httpmethods.asp
@enum.unique
class HTTPMethod(enum.Enum):
    """Enumeration of all legal HTTP methods recognized by servers."""

    GET = 'GET is used to request data from a specified resource.'
    POST = 'POST is used to send data to a server to create/update a resource.'
    PUT = 'PUT is used to send data to a server to create/update a resource.'
    HEAD = 'HEAD is almost identical to GET, but without the response body.'
    DELETE = 'The DELETE method deletes the specified resource.'
    PATCH = 'The PATCH method is used to apply partial modifications to a resource.'
    OPTIONS = 'The OPTIONS method describes the communication options for the target resource.'
    CONNECT = 'The CONNECT method is used to start a two-way communications (a tunnel) with the requested resource.'
    TRACE = 'The TRACE method method is used to perform a message loop-back test that tests the path for the target resource (useful for debugging purposes).'


def convert_status(value):
    """Convert an HTTP status code into an enumeration instance."""
    return http.HTTPStatus(int(value))


LOG_ADAPTERS = collections.OrderedDict((
    ('client', ipaddress.ip_address),
    ('userid', sys.intern),
    ('datetime', convert_datetime),
    ('method', HTTPMethod.__getitem__),
    ('request', urllib.parse.urlparse),
    ('status', convert_status),
    ('size', int),
    ('referrer', sys.intern),
    ('useragent', sys.intern)
))

_ED_RECORD_FIELDS = tuple(LOG_ADAPTERS) + ('parse_time', 'line_length')


class _EDRecord(collections.namedtuple('_EDRecordBase', _ED_RECORD_FIELDS)):
    VALID_STATUS_PREFIX = {1, 2, 3}
    FAILED_STATUS_PREFIX = {4, 5}

    @functools.cached_property
    def status_prefix(self):
        """The first digit of a three-digit status code."""
        # noinspection PyUnresolvedReferences
        return self.status.value // 100

    @functools.cached_property
    def valid(self):
        """The indicator if the request was valid."""
        return self.status_prefix in self.VALID_STATUS_PREFIX

    @functools.cached_property
    def failed(self):
        """The indicator if the request was a failure."""
        return self.status_prefix in self.FAILED_STATUS_PREFIX

    @functools.cached_property
    def path(self):
        """The path of what was requested sans other request data."""
        # noinspection PyUnresolvedReferences
        return pathlib.PurePath(self.request.path)

    @functools.cached_property
    def file(self):
        """A guess whether the path represents a file or not."""
        return bool(self.path.suffix)

    @functools.cached_property
    def not_found(self):
        """The indicator if a request caused a 404 error or not."""
        # noinspection PyUnresolvedReferences
        return self.status is http.HTTPStatus.NOT_FOUND


def bytes_to_str(size):
    """Convert an item's size in bytes to an easily readable representation."""
    size = int(size)
    if size:
        selector = round(math.log(size, KB) - 0.4)
        if selector:
            return f'{size / KB ** selector:.2f} {PREFIX[selector]}B'
        return f'{size:d} {PREFIX[selector]}B'
    return '-'


class UpdateEngine(Repeating):
    """Core of this program that handles changes in the access log file."""

    PRIORITY = 1  # Lower numbers have higher priority.

    def __init__(self, log_path, search, sleep_interval, database):
        """Initialize the instance with data for detecting log changes."""
        super().__init__(sleep_interval)
        self.__log_path = log_path
        self.__search = search
        self.__database = database
        self.__stream_position = 0
        self.__line_number = 1
        self.__last_modified = None

    @property
    def stream_position(self):
        """The place in the file where the next update will resume."""
        return self.__stream_position

    @property
    def line_number(self):
        """The line in the file where the next update will resume."""
        return self.__line_number

    def gate(self):
        """Detect when the log file is updated and trigger updates."""
        modified_time = self.__log_path.stat().st_mtime
        next_modified = datetime.datetime.fromtimestamp(modified_time)
        if self.__last_modified != next_modified:
            self.__last_modified = next_modified
            return True
        return False

    def run(self):
        """Create and add records to the database when updates occur."""
        with self.__log_path.open() as file:
            file.seek(self.stream_position)
            for line in file:
                start_time = time.perf_counter_ns()
                if match := self.__search(line):
                    record = self.__create_record(match)
                    stop_time = time.perf_counter_ns()
                    parse_time = stop_time - start_time
                    self.__database.add_record(record, parse_time, len(line))
                else:
                    self.handle_search_error()
                self.__line_number += 1
            self.__stream_position = file.tell()

    def __create_record(self, match):
        """Make a log record and convert fields to their proper datatypes."""
        groups = match.groupdict()
        record = _LogRecord(*(
            value(groups.pop(key)) for key, value in LOG_ADAPTERS.items()
        ))
        if groups:
            self.handle_match_error(groups)
        return record

    def handle_search_error(self):
        """Provide a default behavior when a search fails."""
        raise ValueError(f'starting at position {self.stream_position}, '
                         f'line {self.line_number} could not be handled')

    def handle_match_error(self, groups):
        """Provide a default behavior when groups are not processed."""
        raise ValueError(f'groups not processed: {", ".join(groups)}')


_LogRecord = collections.namedtuple('LogRecord', LOG_ADAPTERS)

if __name__ == '__main__':
    main()
\$\endgroup\$
0
\$\begingroup\$

Replacing the threading module with the sched module and introducing an abstract class using the abc module provides a better showcase of the standard library and demonstrates an alternative to using threads in Python. However, the program could use a proper database that stores the data retrieved from the access log in a pre-processed format that is easily queryable. A few more modules from the standard library (such as sqlite3) could be incorporated into the monitor so that the work of parsing the access log is not entirely lost once the application is closed.

The constants near the top of the program have been reorganized a little based on what a developer might be expected to have the most interest in changing, and several have been made private if they only help with the creation of other constants. The largest new constant is named SQL and is meant to be a collection point for any queries needed by the program. Parameters are provided as ? where appropriate, and {} found in the queries are meant to be placeholders for either the table's name (3 times) or the name of the column(s) that will be selected in the middle of a query (1 time).

The main function needed to be refactored to deal with two new classes, and it also has the responsibility of initiating adjustments to the sqlite3 engine to account for new custom datatypes. The ExampleDatabase class remains mostly the same as last time but switches to using a different style for signifying private attributes and methods (see Reserved classes of identifiers from the official documentation for more information on the subject). The implementations of several methods were also simplified to take advantage of the fact that Boolean values can be interpreted as 0 (False) and 1 (True).

A JournalMode enumeration is added to support the next class, PersistentDatabase, which uses the sqlite3 module to add support for a proper database that exists after the application is closed. Many new methods in the database class are meant to provide a simple interface to Connection and Cursor objects coming from SQLite, and a few constant attributes near the top allow customization of the SQLite table name and the number of records that are retrieved at one time from the engine. A few methods and properties were inherited from ExampleDatabase, but most properties required a new implementation.

The old UpdateEngine class had some name changes applied so that private variables and methods could be more easily accessed from child classes when needed. Specifically, the new StreamEngine class inherits from UpdateEngine and requires access to specific attributes and the ability to use the _create_record method when handling data coming from the access log file. A few methods are implemented that allow for streaming records to the database via an iterator, but both classes could probably use some refactoring to simplify the implementations and help make the code a bit more DRY (don't repeat yourself).

The following is an extended implementation of the previous program that adds some new standard library modules and changes to make use of the functionality they provide. The new classes, PersistentDatabase and StreamEngine, could probably be improved by modifying both the implementation of their parents along with their implementations (after making ExampleDatabase and UpdateEngine easier to extend). Further work could also be done to account for events where internal variables may need to be updated (such as when the program is shut down and reloaded, the access log file is rotated, or the database file is corrupted).

#! /usr/bin/env python3
# -*- coding: utf-8 -*-
"""Example Application for Monitoring a Server's Access Log.

This is free and unencumbered software released into the public domain.

Anyone is free to copy, modify, publish, use, compile, sell, or
distribute this software, either in source code form or as a compiled
binary, for any purpose, commercial or non-commercial, and by any
means.

In jurisdictions that recognize copyright laws, the author or authors
of this software dedicate any and all copyright interest in the
software to the public domain. We make this dedication for the benefit
of the public at large and to the detriment of our heirs and
successors. We intend this dedication to be an overt act of
relinquishment in perpetuity of all present and future rights to this
software under copyright law.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
IN NO EVENT SHALL THE AUTHORS BE LIABLE FOR ANY CLAIM, DAMAGES OR
OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE,
ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
OTHER DEALINGS IN THE SOFTWARE.

For more information, please refer to <http://unlicense.org/>"""

import abc
import collections
import contextlib
import datetime
import enum
import functools
import http
import ipaddress
import math
import pathlib
import pickle
import re
import sched
import sqlite3
import sys
import time
import urllib.parse

# Public Names
__all__ = (
    'SERVER_ROOT',
    'LOG_FILE',
    'DATABASE_FILE',
    'UPDATE_DELAY',
    'REPORT_DELAY',
    'LOG_PATH',
    'DATABASE_PATH',
    'RE_CLF_PATTERN_SEARCH',
    'BILLION',
    'KB',
    'PREFIX',
    'SQL',
    'main',
    'register_adapters_and_converters',
    'sql_connect',
    'underline',
    'Repeating',
    'ExampleDatabase',
    'JournalMode',
    'PersistentDatabase',
    'convert_datetime',
    'HTTPMethod',
    'convert_status',
    'LOG_ADAPTERS',
    'bytes_to_str',
    'UpdateEngine',
    'StreamEngine'
)

# Module Documentation
__version__ = 1, 2, 0
__date__ = datetime.date(2022, 8, 10)
__author__ = 'Stephen Paul Chappell'
__credits__ = 'vigy, for providing the inspiration for this program.'

# Configuration Constants
SERVER_ROOT = pathlib.Path('C:/Abyss Web Server')
LOG_FILE = pathlib.Path('log/access.log')
DATABASE_FILE = pathlib.Path('access_log.db')
UPDATE_DELAY = datetime.timedelta(seconds=1)
REPORT_DELAY = datetime.timedelta(seconds=5)

# Symbolic Constants
LOG_PATH = SERVER_ROOT / LOG_FILE
_ALM_ROOT = pathlib.Path(sys.argv[0]).parent
DATABASE_PATH = _ALM_ROOT / DATABASE_FILE
# Reference: https://www.oreilly.com/library/view/regular-expressions-cookbook/9781449327453/ch07s13.html
_COMBINED_LOG_FORMAT = '''\
^(?P<client>\\S+)●\\S+●(?P<userid>\\S+)●\\[(?P<datetime>[^\\]]+)\\]↵
●"(?P<method>[A-Z]+)●(?P<request>[^●"]+)?●HTTP/[0-9.]+"↵
●(?P<status>[0-9]{3})●(?P<size>[0-9]+|-)●"(?P<referrer>[^"]*)"↵
●"(?P<useragent>[^"]*)"'''
_RE_CLEANUP_TABLE = str.maketrans('●', ' ', '↵\n')
_RE_COMBINED_LOG_FORMAT = _COMBINED_LOG_FORMAT.translate(_RE_CLEANUP_TABLE)
RE_CLF_PATTERN_SEARCH = re.compile(_RE_COMBINED_LOG_FORMAT).search
BILLION = 10 ** 9
KB = 1 << 10
PREFIX = '', 'K', 'M', 'G', 'T', 'P', 'E', 'Z', 'Y'
SQL = dict(
    count_table='''\
SELECT count(*) AS total
  FROM sqlite_master
 WHERE type = 'table'
   AND name = ?''',
    create_table='''\
CREATE TABLE {} (
    record_id   INTEGER PRIMARY KEY,
    client      IPADDRESS,
    userid      TEXT,
    datetime    DATETIMEZONE,
    method      HTTPMETHOD,
    request     URL,
    status      HTTPSTATUS,
    size        INTEGER,
    referrer    TEXT,
    useragent   TEXT,
    parse_time  INTEGER,
    line_length INTEGER
)''',
    insert_record='''\
INSERT INTO {} (
    client,
    userid,
    datetime,
    method,
    request,
    status,
    size,
    referrer,
    useragent,
    parse_time,
    line_length
) VALUES (
    ?,
    ?,
    ?,
    ?,
    ?,
    ?,
    ?,
    ?,
    ?,
    ?,
    ?
)''',
    select_record='''\
SELECT {}
  FROM {}'''
)


def main():
    """Periodically generate reports for overall analyzed requests."""
    register_adapters_and_converters()
    report_header = underline(f'\nREPORT FOR {LOG_PATH}')
    with PersistentDatabase(report_header, REPORT_DELAY, DATABASE_PATH) as db:
        StreamEngine(LOG_PATH, RE_CLF_PATTERN_SEARCH, UPDATE_DELAY, db)
        Repeating.start()


def register_adapters_and_converters():
    """Add support for several new SQLite types for use in its tables."""
    sqlite_types = dict(
        IPADDRESS=(ipaddress.IPv4Address, ipaddress.IPv6Address),
        DATETIMEZONE=datetime.datetime,
        HTTPMETHOD=HTTPMethod,
        URL=urllib.parse.ParseResult,
        HTTPSTATUS=http.HTTPStatus
    )

    def make_adapter(type_):
        """Create a function to convert objects of type into bytes."""

        def adapter(object_):
            if not isinstance(object_, type_):
                raise TypeError(f'found {type(object_)}; requires {type_}')
            return pickle.dumps(object_, pickle.HIGHEST_PROTOCOL)

        return adapter

    def make_converter(type_):
        """Create a function to coverts bytes into objects of type."""

        def converter(bytes_):
            object_ = pickle.loads(bytes_)
            if not isinstance(object_, type_):
                raise TypeError(f'found {type(object_)}; requires {type_}')
            return object_

        return converter

    for key, value in sqlite_types.items():
        if isinstance(value, tuple):
            shared_adapter = make_adapter(value)
            for item in value:
                sqlite3.register_adapter(item, shared_adapter)
        else:
            sqlite3.register_adapter(value, make_adapter(value))
        sqlite3.register_converter(key, make_converter(value))


def sql_connect(database):
    """Opens a connection to the SQLite database file *database*."""
    db = sqlite3.connect(database, detect_types=sqlite3.PARSE_DECLTYPES)
    db.row_factory = sqlite3.Row
    return db


def underline(display_text, underline_character='='):
    """Take a potentially multi-lined string and underline its width."""
    if not isinstance(underline_character, str):
        raise TypeError('underline_character must be of type str')
    if len(underline_character) != 1:
        raise ValueError('underline_character must be a single character')
    max_length = max(map(len, display_text.splitlines()))
    return f'{display_text}\n{underline_character * max_length}'


class Repeating(abc.ABC):
    """Abstract class for tasks that need to be repeated on a schedule."""

    PRIORITY = math.inf  # Default priority for class instances.
    __scheduler = sched.scheduler()  # Central queue for all task schedules.

    @classmethod
    def start(cls):
        """Begin running schedules specified by the child instances."""
        cls.__scheduler.run()

    @abc.abstractmethod
    def __init__(self, delay):
        """Initialize the instance and schedule the first run."""
        self.delay = delay
        self.__enter()

    @property
    def delay(self):
        """The interval between repeating an instance's primary action."""
        return self.__delay

    @delay.setter
    def delay(self, value):
        if not isinstance(value, datetime.timedelta):
            raise TypeError('delay must be of type timedelta')
        if value.total_seconds() < 0:
            raise ValueError('total seconds must be zero or greater')
        self.__delay = value

    def __enter(self):
        """Schedule the next time the primary action should be run."""
        delay = self.__delay.total_seconds()
        self.__scheduler.enter(delay, self.PRIORITY, self.__action)

    def __action(self):
        """Get permission to run, execute as needed, and reschedule."""
        try:
            if self.gate():
                self.run()
        finally:
            self.__enter()

    def gate(self):
        """Check if the run method should be executed."""
        return True

    @abc.abstractmethod
    def run(self):
        """Execute the primary action an instance is meant to repeat."""
        pass


# Reference: https://goaccess.io/
class ExampleDatabase(Repeating):
    """Class demonstrating an approach to processing _LogRecord instances."""

    PRIORITY = 2  # Lower numbers have higher priority.

    def __init__(self, report_header, delay):
        """Initialize the instance as an empty database."""
        super().__init__(delay)
        self._report_header = report_header
        self._last_updated = datetime.datetime.now()
        self._last_report = None
        self._records = []

    def __iter__(self):
        """Iterate over all records in the database."""
        yield from self._records

    def add_record(self, record, parse_time, line_length):
        """Add a single record with metadata to the database."""
        if not isinstance(record, _LogRecord):
            raise TypeError('record must be of type _LogRecord')
        self._last_updated = datetime.datetime.now()
        # noinspection PyArgumentList
        self._records.append(_EDRecord(*record, parse_time, line_length))

    @property
    def last_updated(self):
        """Timestamp of when the database was last updated."""
        return self._last_updated

    @property
    def total_requests(self):
        """The total requests recorded so far."""
        return len(self._records)

    @property
    def valid_requests(self):
        """The number of requested considered valid."""
        return sum(record.valid for record in self)

    @property
    def failed_requests(self):
        """The number of requests considered failed."""
        return sum(record.failed for record in self)

    @property
    def log_parsing_time(self):
        """The total time used for parsing the log file."""
        ns = sum(record.parse_time for record in self)
        return datetime.timedelta(seconds=ns / BILLION)

    @property
    def unique_visitors(self):
        """The count of unique client identifiers."""
        return len({record.client for record in self})

    @property
    def requested_files(self):
        """The total number of file requests made so far."""
        return sum(record.file for record in self)

    @property
    def path_to_clients(self):
        """A simple database of who has requested each server path."""
        path_to_clients = {}
        for record in self:
            clients = path_to_clients.setdefault(record.path, set())
            clients.add(record.client)
        return path_to_clients

    @property
    def excl_ip_hits(self):
        """The number of paths that have been requested by only one client."""
        ptc_values = self.path_to_clients.values()
        return sum(len(value) == 1 for value in ptc_values)

    @property
    def referrers(self):
        """The count of unique referrers across all requests."""
        return len({record.referrer for record in self})

    @property
    def not_found(self):
        """The count of requests having a 404 error generated for them."""
        return sum(record.not_found for record in self)

    @property
    def static_files(self):
        """The count of unique paths to files."""
        return len({record.path for record in self if record.file})

    @property
    def log_size(self):
        """The size of the log in characters processed so far."""
        return sum(record.line_length for record in self)

    @property
    def log_size_str(self):
        """A readable log size that assumes characters are bytes."""
        return bytes_to_str(self.log_size)

    @property
    def tx_amount(self):
        """The total bytes transferred from the server to its clients."""
        return sum(record.size for record in self)

    @property
    def tx_amount_str(self):
        """A readable transfer amount that assumes counts are in bytes."""
        return bytes_to_str(self.tx_amount)

    def create_report(self):
        """Generate an example report via the instance's many properties."""
        return f'''\
{self._report_header!s}
Last Updated     = {self.last_updated!s}
Total Requests   = {self.total_requests:,}
Valid Requests   = {self.valid_requests:,}
Failed Requests  = {self.failed_requests:,}
Log Parsing Time = {self.log_parsing_time!s}
Unique Visitors  = {self.unique_visitors:,}
Requested Files  = {self.requested_files:,}
Excl. IP Hits    = {self.excl_ip_hits:,}
Referrers        = {self.referrers:,}
Not Found        = {self.not_found:,}
Static Files     = {self.static_files:,}
Log Size         = {self.log_size_str}
Tx. Amount       = {self.tx_amount_str}'''

    def gate(self):
        """Generate a new report only after an update is detected."""
        if self._last_report != self._last_updated:
            self._last_report = self._last_updated
            return True
        return False

    def run(self):
        """Create a report and display it to the screen."""
        print(self.create_report())


JournalMode = enum.Enum(
    'JournalMode',
    'DELETE | TRUNCATE | PERSIST | MEMORY | WAL | OFF'.split(' | ')
)


class PersistentDatabase(ExampleDatabase):
    """Class showing how to create and use a SQLite database for records."""

    TABLE_NAME = 'combined_log_record'
    CURSOR_ARRAY_SIZE = 100

    def __init__(self, report_header, delay, database_path):
        """Initialize the instance as an empty database."""
        super().__init__(report_header, delay)
        del self._records
        self._database_path = database_path
        self._init_database()
        self._connection = None

    def _init_database(self):
        """Create and initialize the database file if it does not exist."""
        if not self._database_path.exists():
            with contextlib.closing(sql_connect(self._database_path)) as db:
                # noinspection PyUnresolvedReferences
                self._set_journal_mode(db, JournalMode.WAL)
                if not self._table_exists(db):
                    with db:
                        db.execute(SQL['create_table'].format(self.TABLE_NAME))
                    if not self._table_exists(db):
                        raise RuntimeError('table was not properly created')

    @staticmethod
    def _set_journal_mode(db, mode):
        """Set the journal mode for the provided database connection."""
        if not isinstance(mode, JournalMode):
            raise TypeError('the mode argument must be of type JournalMode')
        with db:
            cursor = db.execute(f'PRAGMA journal_mode = {mode.name}')
            if cursor.fetchone()['journal_mode'] != mode.name.casefold():
                raise RuntimeError('journal mode was not properly set')

    def _table_exists(self, db):
        """Find out if a table exists in a SQLite database."""
        with db:
            cursor = db.execute(SQL['count_table'], [self.TABLE_NAME])
            total = cursor.fetchone()['total']
        return {0: False, 1: True}[total]

    def __enter__(self):
        """Open the database connection when used as a context manager."""
        self._connection = sql_connect(self._database_path)
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        """Close the database connection after the context manager block."""
        self._connection.close()
        self._connection = None
        return False

    def __iter__(self):
        """Iterate over all records in the database."""
        return self._get_columns('*')

    def _get_columns(self, *columns):
        """Iterate over all the columns the caller has asked for."""
        columns = ',\n       '.join(columns)
        sql = SQL['select_record'].format(columns, self.TABLE_NAME)
        with self._db as db:
            cursor = db.execute(sql)
            # noinspection SpellCheckingInspection
            cursor.arraysize = self.CURSOR_ARRAY_SIZE
            yield from self._fetch_iter(cursor)

    @property
    def _db(self):
        """The database connection after being validated."""
        db = self._connection
        if db is None:
            raise RuntimeError('database must be used as a context manager')
        if not isinstance(db, sqlite3.Connection):
            raise TypeError('database connection was not properly initialized')
        return db

    @staticmethod
    def _fetch_iter(cursor):
        """Fetch all remaining rows of a query result using an iterator."""
        while True:
            array = cursor.fetchmany()
            if not array:
                break
            yield from array

    def add_record(self, record, parse_time, line_length):
        """Add a single record with metadata to the database."""
        if not isinstance(record, _LogRecord):
            raise TypeError('record must be of type _LogRecord')
        record += parse_time, line_length
        with self._db as db:
            db.execute(SQL['insert_record'].format(self.TABLE_NAME), record)
        self._last_updated = datetime.datetime.now()

    def add_records(self, records):
        """Add a group of records with metadata to the database."""
        sql = SQL['insert_record'].format(self.TABLE_NAME)
        with self._db as db:
            db.executemany(sql, records)
        self._last_updated = datetime.datetime.now()

    @property
    def total_requests(self):
        """The total requests recorded so far."""
        sql = SQL['select_record'].format('count(*) AS total', self.TABLE_NAME)
        with self._db as db:
            cursor = db.execute(sql)
            total = cursor.fetchone()['total']
        return total

    @property
    def valid_requests(self):
        """The number of requested considered valid."""
        return sum(status_prefix in _EDRecord.VALID_STATUS_PREFIX
                   for status_prefix in self._all_status_prefix)

    @property
    def _all_status_prefix(self):
        """The first digit of a three-digit status code."""
        return (column.value // 100 for column in self._get_column('status'))

    def _get_column(self, column):
        """Iterate over a single column the caller has asked for."""
        return (row[column] for row in self._get_columns(column))

    @property
    def failed_requests(self):
        """The number of requests considered failed."""
        return sum(status_prefix in _EDRecord.FAILED_STATUS_PREFIX
                   for status_prefix in self._all_status_prefix)

    @property
    def log_parsing_time(self):
        """The total time used for parsing the log file."""
        ns = sum(self._get_column('parse_time'))
        return datetime.timedelta(seconds=ns / BILLION)

    @property
    def unique_visitors(self):
        """The count of unique client identifiers."""
        return len(set(self._get_column('client')))

    @property
    def requested_files(self):
        """The total number of file requests made so far."""
        return sum(bool(path.suffix) for path in self._all_path)

    @property
    def _all_path(self):
        """The path of what was requested sans other request data."""
        return (pathlib.PurePath(column.path)
                for column in self._get_column('request'))

    @property
    def path_to_clients(self):
        """A simple database of who has requested each server path."""
        path_to_clients = {}
        for row in self._get_columns('request', 'client'):
            path = pathlib.PurePath(row['request'].path)
            clients = path_to_clients.setdefault(path, set())
            clients.add(row['client'])
        return path_to_clients

    @property
    def referrers(self):
        """The count of unique referrers across all requests."""
        return len(set(self._get_column('referrer')))

    @property
    def not_found(self):
        """The count of requests having a 404 error generated for them."""
        return sum(column is http.HTTPStatus.NOT_FOUND
                   for column in self._get_column('status'))

    @property
    def static_files(self):
        """The count of unique paths to files."""
        return len({path for path in self._all_path if path.suffix})

    @property
    def log_size(self):
        """The size of the log in characters processed so far."""
        return sum(self._get_column('line_length'))

    @property
    def tx_amount(self):
        """The total bytes transferred from the server to its clients."""
        return sum(self._get_column('size'))


def convert_datetime(value):
    """Convert an access log timestamp into a datetime object."""
    return datetime.datetime.strptime(value, '%d/%b/%Y:%H:%M:%S %z')


# Reference: https://www.w3schools.com/tags/ref_httpmethods.asp
@enum.unique
class HTTPMethod(enum.Enum):
    """Enumeration of all legal HTTP methods recognized by servers."""

    GET = 'GET is used to request data from a specified resource.'
    POST = 'POST is used to send data to a server to create/update a resource.'
    PUT = 'PUT is used to send data to a server to create/update a resource.'
    HEAD = 'HEAD is almost identical to GET, but without the response body.'
    DELETE = 'The DELETE method deletes the specified resource.'
    PATCH = 'The PATCH method is used to apply partial modifications to a resource.'
    OPTIONS = 'The OPTIONS method describes the communication options for the target resource.'
    CONNECT = 'The CONNECT method is used to start a two-way communications (a tunnel) with the requested resource.'
    TRACE = 'The TRACE method method is used to perform a message loop-back test that tests the path for the target resource (useful for debugging purposes).'


def convert_status(value):
    """Convert an HTTP status code into an enumeration instance."""
    return http.HTTPStatus(int(value))


LOG_ADAPTERS = collections.OrderedDict((
    ('client', ipaddress.ip_address),
    ('userid', sys.intern),
    ('datetime', convert_datetime),
    ('method', HTTPMethod.__getitem__),
    ('request', urllib.parse.urlparse),
    ('status', convert_status),
    ('size', int),
    ('referrer', sys.intern),
    ('useragent', sys.intern)
))

_ED_RECORD_FIELDS = tuple(LOG_ADAPTERS) + ('parse_time', 'line_length')


class _EDRecord(collections.namedtuple('_EDRecordBase', _ED_RECORD_FIELDS)):
    """Record type that implements multiple virtual, cached columns."""

    VALID_STATUS_PREFIX = {1, 2, 3}
    FAILED_STATUS_PREFIX = {4, 5}

    @functools.cached_property
    def status_prefix(self):
        """The first digit of a three-digit status code."""
        # noinspection PyUnresolvedReferences
        return self.status.value // 100

    @functools.cached_property
    def valid(self):
        """The indicator if the request was valid."""
        return self.status_prefix in self.VALID_STATUS_PREFIX

    @functools.cached_property
    def failed(self):
        """The indicator if the request was a failure."""
        return self.status_prefix in self.FAILED_STATUS_PREFIX

    @functools.cached_property
    def path(self):
        """The path of what was requested sans other request data."""
        # noinspection PyUnresolvedReferences
        return pathlib.PurePath(self.request.path)

    @functools.cached_property
    def file(self):
        """A guess whether the path represents a file or not."""
        return bool(self.path.suffix)

    @functools.cached_property
    def not_found(self):
        """The indicator if a request caused a 404 error or not."""
        # noinspection PyUnresolvedReferences
        return self.status is http.HTTPStatus.NOT_FOUND


def bytes_to_str(size):
    """Convert an item's size in bytes to an easily readable representation."""
    size = int(size)
    if size:
        selector = round(math.log(size, KB) - 0.4)
        if selector:
            return f'{size / KB ** selector:.2f} {PREFIX[selector]}B'
        return f'{size:d} {PREFIX[selector]}B'
    return '-'


class UpdateEngine(Repeating):
    """Core of this program that handles changes in the access log file."""

    PRIORITY = 1  # Lower numbers have higher priority.

    def __init__(self, log_path, search, sleep_interval, database):
        """Initialize the instance with data for detecting log changes."""
        super().__init__(sleep_interval)
        self._log_path = log_path
        self._search = search
        self._database = database
        self._stream_position = 0
        self._line_number = 1
        self._last_modified = None

    @property
    def stream_position(self):
        """The place in the file where the next update will resume."""
        return self._stream_position

    @property
    def line_number(self):
        """The line in the file where the next update will resume."""
        return self._line_number

    def gate(self):
        """Detect when the log file is updated and trigger updates."""
        modified_time = self._log_path.stat().st_mtime
        next_modified = datetime.datetime.fromtimestamp(modified_time)
        if self._last_modified != next_modified:
            self._last_modified = next_modified
            return True
        return False

    def run(self):
        """Create and add records to the database when updates occur."""
        with self._log_path.open() as file:
            file.seek(self.stream_position)
            for line in file:
                start_time = time.perf_counter_ns()
                if match := self._search(line):
                    record = self._create_record(match)
                    stop_time = time.perf_counter_ns()
                    parse_time = stop_time - start_time
                    self._database.add_record(record, parse_time, len(line))
                else:
                    self.handle_search_error()
                self._line_number += 1
            self._stream_position = file.tell()

    def _create_record(self, match):
        """Make a log record and convert fields to their proper datatypes."""
        groups = match.groupdict()
        record = _LogRecord(*(
            value(groups.pop(key)) for key, value in LOG_ADAPTERS.items()
        ))
        if groups:
            self.handle_match_error(groups)
        return record

    def handle_search_error(self):
        """Provide a default behavior when a search fails."""
        raise ValueError(f'starting at position {self.stream_position}, '
                         f'line {self.line_number} could not be handled')

    def handle_match_error(self, groups):
        """Provide a default behavior when groups are not processed."""
        raise ValueError(f'groups not processed: {", ".join(groups)}')


class StreamEngine(UpdateEngine):
    """Refactored class that supports iterative record creation."""

    def run(self):
        """Create and add records to the database when updates occur."""
        self._database.add_records(self._create_records())

    def _create_records(self):
        """Iterate over all records that need to be added to the database."""
        with self._log_path.open() as file:
            file.seek(self.stream_position)
            for line in file:
                start_time = time.perf_counter_ns()
                if match := self._search(line):
                    record = self._create_record(match)
                    stop_time = time.perf_counter_ns()
                    parse_time = stop_time - start_time
                    yield record + (parse_time, len(line))
                else:
                    self.handle_search_error()
                self._line_number += 1
            self._stream_position = file.tell()


_LogRecord = collections.namedtuple('LogRecord', LOG_ADAPTERS)

if __name__ == '__main__':
    main()
\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.