Related to my previous post on launching 16 scripts in tmux, I also needed to provide a script to monitor the output files from some scripts and alert if they are not updated. For the same reasons, so the researcher can maintain his own scripts, this was written in Python rather than Bash.

This worked by checking the files and outputting a message if the files had not been updated - the intention is that the script is run via cron and the feature that cron emails the output to the user whose crontab ran the script to mean the user gets alerted.

The script itself is quite simple and uses Python’s logging for the output:

#!/usr/bin/env python3

import argparse
from datetime import datetime
import logging
from pathlib import Path

logger = logging.getLogger(__name__)

def check_output_updates(path, timeout, pattern='*'):
    """
    Looks into each sub-directory in the directory given and
    checks if a file matching the pattern has been updated in
    the last timeout seconds.

    Args:
        path: string or path-like object which is the directory
            to search.
        timeout: number of seconds within which the file must
            have been updated.
        pattern: glob pattern to match (defaults to '*').

    Returns: nothing
    """
    path = Path(path)  # Make it path-like

    # Loop over each sub-directory and check for files updated
    threshold = datetime.now().timestamp() - timeout
    for d in [x for x in path.iterdir() if x.is_dir()]:
        updated_files = [f for f in d.glob(pattern) if f.is_file() and
                         f.stat().st_mtime > threshold]
        if len(updated_files) == 0:
            logger.warn("No files in %s updated in the last %d seconds", d,
                        timeout)


if __name__ == '__main__':
    logging.basicConfig()
    # Do stuff
    parser = argparse.ArgumentParser(
        description="Monitor each subdirectory in the given path and display"
                    " a message if a matching file has not been updated"
                    " recently."
    )
    parser.add_argument(
        '-d', '--debug',
        action='store_true',
        help="Turn on debugging output",
    )
    parser.add_argument(
        '-p', '--path',
        default=Path('~/python/scripts/output').expanduser(),
        help="Path to directories to monitor",
    )
    parser.add_argument(
        '--pattern',
        default="*.csv",
        help="Glob pattern for files to be monitored.",
    )
    parser.add_argument(
        '-t', '--timeout',
        type=int,
        default="300",
        help="Print a message if no files have been updated within this many"
             " seconds of the script running",
    )
    args = parser.parse_args()
    if args.debug:
        # Set the level on the root logger
        logging.getLogger().setLevel(logging.DEBUG)
    logger.debug("Path is %s", args.path)
    logger.debug("Pattern is %s", args.pattern)
    logger.debug("Timeout is %d", args.timeout)

    check_output_updates(args.path, args.timeout, args.pattern)