# /// script
# requires-python = ">=3.10"
# ///
# Copyright (c) 2026 Aayush Sharma · SPDX-License-Identifier: MIT
# https://thedataengineering.com/lab-code/LICENSE.txt
"""Find the false green in your pipeline.

An hourly job loads events into a table and reports success when it exits
cleanly. At hour 24 its upstream starts delivering empty files. Three monitors
watch it: (1) the job-status alert, (2) a freshness check on
now − max(event_time), (3) a volume band check against the same hour last
week. Simulates 7 days of history plus the failure and reports how long each
monitor takes to fire, in minutes after the first empty file.
"""
import argparse, json, random

p = argparse.ArgumentParser(description=__doc__.splitlines()[0])
p.add_argument("--freshness-max-h", type=float, default=2.0, help="Freshness threshold: alert when the newest event is older than this (hours).")
p.add_argument("--band", type=float, default=0.5, help="Volume band: alert when rows < band × the same hour last week.")
p.add_argument("--check-every-min", type=int, default=15, help="How often the freshness check runs (minutes).")
p.add_argument("--seed", type=int, default=0, help="Random seed.")
p.add_argument("--json", help="Also write the results to this JSON file.")
args = p.parse_args()
random.seed(args.seed)

HIST = 7 * 24; FAIL = HIST + 24                                       # first empty file arrives at this hour
rows_by_hour, max_event, status = {}, 0.0, {}
first = {"job-status alert": None, "freshness check": None, "volume band check": None}
for minute in range(0, (FAIL + 24) * 60, args.check_every_min):
    hour, at = minute // 60, minute % 60
    if at == 0:                                                        # the job runs on the hour, loading the previous hour
        n = 0 if hour >= FAIL else int(1000 * (1.5 + (hour % 24 in range(8, 20))) * random.uniform(0.9, 1.1))
        rows_by_hour[hour] = n; status[hour] = "success"              # the job exits cleanly either way
        if n: max_event = hour - random.uniform(0, 0.1)               # newest event a few minutes before the run
        if hour >= FAIL and status[hour] != "success" and first["job-status alert"] is None: first["job-status alert"] = minute
        if hour >= HIST and n < args.band * rows_by_hour[hour - HIST] and first["volume band check"] is None and hour >= FAIL:
            first["volume band check"] = minute
    if hour >= FAIL and minute / 60 - max_event > args.freshness_max_h and first["freshness check"] is None:
        first["freshness check"] = minute
rows = [{"monitor": k, "fired": "yes" if v is not None else "never",
         "minutes_after_failure": (v - FAIL * 60) if v is not None else None} for k, v in first.items()]
for r in rows: print(f"{r['monitor']:<20} " + (f"fired {r['minutes_after_failure']} min after the first empty file" if r["fired"] == "yes" else "never fired (24 h of empty loads)"))

if args.json:
    with open(args.json, "w") as f:
        json.dump({"tables": {"monitors": rows}}, f)