Files
esh-pfi-infrastructure/scripts/nh3-dev-development-backup-retention.py
T
vh 902e16630f feat(dev-backup): add daily and weekly retention (48 hourly + 30 daily + 12 weekly)
Prime's ruling 2026-10-02. retention.py picks the snapshots to delete:
the newest 48, plus the newest of each of the last 30 days and of each of
the last 12 ISO weeks, counting only days and weeks that have snapshots.
Names that are not exactly YYYY-MM-DD_HHMM are never selected, and the
NAS side refuses any path outside that pattern. The unit fails unless
the number kept equals the number expected. Live run: deleted 1, 0
errors, 48 kept as expected.
2026-10-02 07:42:45 -07:00

59 lines
2.4 KiB
Python

#!/usr/bin/env python3
"""dev-backup retention: decide which snapshot dirs to DELETE (Prime, 2026-10-02: add dailies and weeklies).
Reads snapshot paths on stdin, one per line (`/volume1/Backup/nh3-dev-development/YYYY-MM-DD_HHMM`), and
prints the ones to delete. KEEP is the union of:
- the newest HOURLY snapshots (default 48),
- the newest snapshot of each of the most recent DAILY distinct days (default 30),
- the newest snapshot of each of the most recent WEEKLY distinct ISO weeks (default 12).
Days and weeks are counted among those that HAVE snapshots, so an outage does not eat the history.
A line whose basename is not exactly YYYY-MM-DD_HHMM is never printed, so it is never deleted.
python3 retention.py [--hourly 48] [--daily 30] [--weekly 12] [--keep-count] < listing
--keep-count prints only the number of snapshots that will remain (for the post-prune check).
"""
import argparse
import datetime as dt
import os
import re
import sys
NAME = re.compile(r"^(\d{4})-(\d{2})-(\d{2})_(\d{4})$")
def select_delete(paths, hourly, daily, weekly):
snaps = []
for p in paths:
p = p.strip()
m = NAME.match(os.path.basename(p))
if not p or not m:
continue
try:
day = dt.date(int(m[1]), int(m[2]), int(m[3]))
except ValueError: # e.g. month 13: not one of ours, so never delete it
continue
snaps.append((os.path.basename(p), p, day))
snaps.sort(reverse=True) # newest first; the names sort chronologically
keep = {s[1] for s in snaps[:hourly]}
days, weeks = [], []
for name, path, day in snaps:
if day not in days and len(days) < daily:
days.append(day)
keep.add(path) # first seen = newest of that day
wk = day.isocalendar()[:2]
if wk not in weeks and len(weeks) < weekly:
weeks.append(wk)
keep.add(path)
return [s[1] for s in snaps if s[1] not in keep], len(keep)
if __name__ == "__main__":
ap = argparse.ArgumentParser()
ap.add_argument("--hourly", type=int, default=48)
ap.add_argument("--daily", type=int, default=30)
ap.add_argument("--weekly", type=int, default=12)
ap.add_argument("--keep-count", action="store_true")
a = ap.parse_args()
delete, kept = select_delete(sys.stdin.read().splitlines(), a.hourly, a.daily, a.weekly)
print(kept if a.keep_count else "\n".join(delete))