-
Notifications
You must be signed in to change notification settings - Fork 0
/
remove-old-packages.py
137 lines (116 loc) · 4.43 KB
/
remove-old-packages.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
#!/usr/bin/env python3
# From https://github.com/aptly-dev/aptly/issues/291#issuecomment-276404030
import argparse
import re
import sys
from time import sleep
from typing import List
from apt_pkg import version_compare, init_system
from subprocess import check_output, CalledProcessError
from functools import cmp_to_key
from tqdm import tqdm
class PurgeOldVersions:
def __init__(self):
self.args = self.parse_arguments()
if self.args.dry_run:
print("Run in dry mode, without actually deleting the packages.")
if not self.args.repo:
sys.exit("You must declare a repository with: --repo")
@staticmethod
def parse_arguments():
parser = argparse.ArgumentParser(formatter_class=argparse.RawTextHelpFormatter)
parser.add_argument(
"--dry-run",
dest="dry_run",
help="List packages to remove without removing " "them.",
action="store_true",
)
parser.add_argument(
"--repo", dest="repo", help="Which repository should be searched?", type=str, default="defaultrepo"
)
parser.add_argument(
"--package-query",
dest="package_query",
help="Which packages should be removed?\n"
"e.g.\n"
" - Single package: ros-indigo-rbdl.\n"
" - Query: 'Name (%% ros-indigo-*)' "
"to match all ros-indigo packages. See \n"
"https://www.aptly.info/doc/feature/query/",
type=str,
default="",
)
parser.add_argument(
"-n",
"--retain-how-many",
dest="retain_how_many",
help="How many package versions should be kept?",
type=int,
default=1,
)
return parser.parse_args()
def get_packages(self) -> List[str]:
init_system()
packages = []
try:
output: bytes = check_output(
["aptly", "repo", "search", self.args.repo, self.args.package_query]
if self.args.package_query
else ["aptly", "repo", "search", self.args.repo]
)
packages = sorted(
set(
map(
lambda p: re.sub(
"[_](\d{1,}[:])?\d{1,}[.]\d{1,}[.]\d{1,}[-](.*)", "", p
),
[line for line in output.decode().split("\n") if len(line) > 0],
)
)
)
except CalledProcessError as e:
print(e)
finally:
return packages
def purge(self) -> None:
packages = self.get_packages()
if not packages:
sys.exit("No packages to remove.")
else:
print(f"{len(packages)} package names to look at: {','.join(packages)}")
packages_to_remove: List[str] = []
for package in packages:
try:
output = check_output(
["aptly", "repo", "search", self.args.repo, package]
)
def sort_by_version_cmp(name1, name2):
version_and_build_1 = name1.split("_")[1]
version_and_build_2 = name2.split("_")[1]
return version_compare(version_and_build_1, version_and_build_2)
packages_to_remove.extend(
sorted(
[line for line in output.decode().split("\n") if len(line) > 0],
key=cmp_to_key(sort_by_version_cmp),
)[: -self.args.retain_how_many]
)
except CalledProcessError as e:
print(e)
if len(packages_to_remove) > 0:
print(f"Removing {len(packages_to_remove)} packages in total")
with tqdm(total=len(packages_to_remove)) as pbar:
for package in packages_to_remove:
pbar.set_description(package)
if not self.args.dry_run:
check_output(
["aptly", "repo", "remove", self.args.repo, package]
)
else:
sleep(0.1)
pbar.update(1)
check_output(["aptly", "db", "cleanup"])
else:
print("No packages to remove")
if __name__ == "__main__":
purge_old_versions = PurgeOldVersions()
purge_old_versions.purge()