distfile-cleaner: handle nothing to clean and add emoji marks
[freebsd-maintainer-scripts/.git] / distfile-cleaner.py
1 #!/usr/bin/env python3
2
3 """
4 This simple script is an helper to purge old files left into
5 freefall:~/public_distfiles for ages to reclaim disk space. 
6
7 It uses paramiko, /bin/ls and /bin/rm to delete the obsolete files
8
9 --
10 sbz@FreeBSD.org
11 """
12
13 import argparse
14 import datetime
15 import time
16 import sys
17
18 from dataclasses import dataclass
19
20 import paramiko
21
22 # by default keep file for at least 5 years
23 MAX_KEEP_YEAR = 5
24
25 __version__ = "1.0"
26
27 @dataclass
28 class FileEntry:
29     name: str
30     date: str
31
32     def is_older(self) -> bool:
33
34         current_year = datetime.datetime.fromtimestamp(time.time()).year
35         year, month, day = self.date.split("-")
36         if current_year - int(year) >= MAX_KEEP_YEAR:
37             return True
38
39         return False
40
41
42 class Checker(object):
43     """Class to purge old FreeBSD distfiles"""
44
45     def __init__(self, host: str="freefall.freebsd.org", user: str="sbz",
46                  dryrun: bool=False):
47         self.host = host
48         self.user = user
49         self.path = f"/home/{self.user}/public_distfiles/"
50         self.list_cmd = f"ls -ltr -D%F {self.path}"
51         self.dryrun = dryrun
52
53     def remote_exec(self, cmd: str) -> str:
54         client = paramiko.SSHClient()
55         client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
56         client.load_system_host_keys()
57         client.connect(self.host)
58
59         try:
60             stdin, stdout, stderr = client.exec_command(cmd)
61         except paramiko.SSHException as e:
62             print(f"Failed to execute remote {cmd}: (stderr: {stderr.read()}")
63             raise e
64
65         return stdout
66
67     def clean_files(self) -> bool:
68         stdout = self.remote_exec(self.list_cmd)
69         files = self.read_stream(stdout)
70         if len(files) == 0:
71             return False
72
73         if self.dryrun:
74             mode = "Dry run"
75         else:
76             mode = "+"
77
78         for f in files:
79             print(f"[{mode}] Cleaning {f.name} from {f.date}", end=' ')
80             try:
81                 self.clean_file(f.name)
82             except Exception as e:
83                 mark = "\u2757"
84             else:
85                 mark = "\N{cross mark}" if self.dryrun else "\u2705"
86             print(f"{mark}")
87
88         return True
89         
90     def read_stream(self, stream_input: paramiko.Channel) -> list:
91         lines = []
92
93         for line in stream_input.readlines()[1:]:
94             if line == "":
95                 continue
96             col = line.strip().split()
97             f = FileEntry(col[-1], col[-2])
98             if f.is_older():
99                 lines.append(f)
100
101         return lines
102
103     def clean_file(self, file_name: str):
104         clean_cmd = f"rm -vf {self.path}{file_name}"
105         if self.dryrun:
106             clean_cmd = "echo " + clean_cmd
107         try:
108             stdout = self.remote_exec(clean_cmd)
109         except Exception as e:
110             raise Exception(f"Cleaning error for file {file_name}")
111
112 def main() -> int:
113     global MAX_KEEP_YEAR
114
115     parser = argparse.ArgumentParser(
116         description="Purge old FreeBSD files stored in ~/public_distfiles"
117     )
118
119     parser.add_argument(
120         "-n", "--dryrun", action='store_true'
121         help="Dry run mode. Do not execute remote command"
122     )
123     parser.add_argument(
124         "-m", "--max-year", type=int,
125         help="Maximum year to keep the files"
126     )
127
128     args = parser.parse_args()
129     if args.max_year:
130         MAX_KEEP_YEAR = args.max_year
131
132     checker = Checker()
133     if args.dryrun:
134         checker = Checker(dryrun=True)
135
136     print(f"[+] Process files older than {MAX_KEEP_YEAR} year old")
137     if not checker.clean_files():
138         print("Nothing to clean.")
139         return 1
140
141     return 0
142
143 if __name__ == "__main__":
144     try:
145         sys.exit(main())
146     except KeyboardInterrupt as e:
147         sys.exit(0)
148     except SystemExit as sysexit:
149         if sysexit.code != 0:
150             raise
151         else:
152             sys.exit(sysexit.code)
153     except Exception:
154         raise