send slack alert if output file doesn't see any additions for 5 minutes

not-alert.py

import glob
import json
import os
import sys
import time
import typing
import urllib.request

period_in_seconds = 60
watch_time_seconds = 300
slack_webhook = 'https://hooks.slack.com/services/AAA/BBB/CCC'

last_filename: typing.Optional[str] = None
tick_values = []
max_tick_values = (watch_time_seconds // period_in_seconds) + 2


def compute_tick_value(filename: str) -> float:
    return os.stat(filename).st_size


def raise_error(filename: str):
    request = urllib.request.Request(url=slack_webhook, headers={'content-type': 'application/json'}, data=json.dumps({
        'text': f'no progress! {filename} stuck.'
    }).encode())
    response = urllib.request.urlopen(request)
    print(dir(response))
    print(f"alert send, status_code={response.status}")


def process(filename: str):
    global last_filename, tick_values

    if not last_filename:
        last_filename = filename

    current_tick_value = compute_tick_value(filename)
    if filename != last_filename:
        last_filename = filename
        tick_values = [current_tick_value]
    else:
        tick_values.append(current_tick_value)

    while len(tick_values) > max_tick_values:
        tick_values.pop(0)

    computed = [a[1] - a[0] for a in zip(tick_values, tick_values[1:])]
    if sum(computed) == 0 and len(computed) == max_tick_values - 1:
        raise_error(last_filename)

    print(f"filename={last_filename} computed={len(computed)} max_tick_values={max_tick_values}, values={computed}")


def main() -> int:
    while True:
        files = glob.glob('output.[0-9]*.txt')
        files = sorted(files)

        if len(files) > 0:
            process(files[-1])

        time.sleep(period_in_seconds)

    return 0


if __name__ == '__main__':
    sys.exit(main())