Python 3 script to iterate all SRCDS logs, and store connected players' IPv4 addresses to a single file, one per line.

srcds_ip_grepper.py

#!/usr/bin/env python3

"""Python 3 script to iterate all SRCDS logs,
   and store connected players' IPv4 addresses to a single file, one per line.

   Note that this script currently only handles IPv4 addresses."""

# Copyright (c) 2020 Rainyan https://github.com/Rainyan
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# 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 OR COPYRIGHT HOLDERS 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.

import os
import re

# Whether to print non-errors and informational status messages.
SRCDS_GREPPER_META_VERBOSE = True
# Meta info to print on invokation (if set to verbose).
SRCDS_GREPPER_META_NAME = 'SRCDS IP Grepper'
SRCDS_GREPPER_META_VERSION = '0.2'
SRCDS_GREPPER_META_URL = 'https://gist.github.com/Rainyan/323a468fc348e24a8' \
                         'ed9d639887db214'
SRCDS_GREPPER_META_DESC = "Iterate all SRCDS logs, and store connected " \
                          "players' IPv4 addresses to a single file, line " \
                          "per line."
SRCDS_GREPPER_META_LINEBREAK = '= = = = = = = = = = = = = = = = = = = ='

# Don't write to file.
DEBUG_DRYRUN = False

# Folder where to look for logs.
LOG_FILE_DIRECTORY = 'logs'
# Log file extension.
LOG_FILE_EXTENSION = '.log'
# Log file encoding.
LOG_FILE_ENCODING = 'utf-8'
# File to write output into.
OUT_IP_LOG = 'ips.txt'


def get_ip(line):
    """Takes a grep match string as input, and outputs list of
       (success boolean, ip string).
       IP is guaranteed valid only if success was True and line input was a
       valid SRCDS log line format of:
       [0 or more chars]"[valid ipv4 address]:[ipv4 port number]"
    """
    # pylint: disable=invalid-name
    found_first_quote = False
    found_port = False
    ip = ''
    success = False
    for c in reversed(line):
        if c == '\"':
            if not found_first_quote:
                found_first_quote = True
                continue
            success = found_first_quote and found_port and len(ip) > 0
            break
        if not found_port:
            if c == ':':
                found_port = True
            continue
        if not c.isdigit() and not c == '.':
            return False, ip
        ip += c
    # Flip the IP string because we iterated backwards.
    return success, ip[::-1]


def run_grep(log_file_path):
    """Opens a file, greps it for IPs, and returns a list of IP strings.
    """
    # pylint: disable=invalid-name
    match = []
    with open(log_file_path, 'r', encoding=LOG_FILE_ENCODING) as file:
        for line in file:
            line = line.rstrip()  # remove whitespace
            pattern = re.compile(r'\b(?:[0-9]{1,3}\.){3}[0-9]{1,3}\b')
            if pattern.search(line) is not None:
                match.append(line)
    ips = []
    for item in match:
        result = get_ip(item)
        was_valid_ip = result[0]
        if was_valid_ip:
            ip = result[1]
            ips.append(ip)
    return ips


def write_ips(write_path, ips):
    """Writes a list of IP strings to path.
    """
    # pylint: disable=invalid-name
    num_ips = len(ips)
    if num_ips < 1:
        if SRCDS_GREPPER_META_VERBOSE:
            print('No IPs to process, cancelling operation.')
        return

    if DEBUG_DRYRUN:
        print('# DRYRUN START -- Would have written to file: \"' +
              write_path + '\"')
        for ip in ips:
            print('## DRYRUN -- Would have written: \"' + ip + '\"')
        print('# DRYRUN FINISH -- Would have written ' + str(num_ips) +
              ' IPs to file.')
    else:
        if os.path.isfile(write_path):
            os.remove(write_path)
        with open(write_path, 'a', encoding=LOG_FILE_ENCODING) as file:
            first_ip = True
            for ip in ips:
                file.write(ip if first_ip else '\n' + ip)
                first_ip = False
        if SRCDS_GREPPER_META_VERBOSE:
            print('Wrote ' + str(num_ips) + ' IPs to file: ' + write_path)


def process_logs():
    """Returns a list of IP strings discovered from the logs.
    """
    # pylint: disable=invalid-name
    all_ips = []
    for file in os.listdir(LOG_FILE_DIRECTORY):
        filename = os.fsdecode(file)
        if filename.endswith(LOG_FILE_EXTENSION):
            this_log_ips = run_grep(os.path.join(LOG_FILE_DIRECTORY, filename))
            for ip in this_log_ips:
                all_ips.append(ip)
    return all_ips


def main():
    """Entry point.
    """
    if SRCDS_GREPPER_META_VERBOSE:
        print(SRCDS_GREPPER_META_LINEBREAK)
        print(SRCDS_GREPPER_META_NAME + ' v.' + SRCDS_GREPPER_META_VERSION)
        print(SRCDS_GREPPER_META_URL)
        print(SRCDS_GREPPER_META_DESC)
        print(SRCDS_GREPPER_META_LINEBREAK)
        print('Processing log files...')

    all_ips = process_logs()
    write_ips(OUT_IP_LOG, all_ips)

    if SRCDS_GREPPER_META_VERBOSE:
        print('All done.')
        print(SRCDS_GREPPER_META_LINEBREAK)


main()