From 28a2ab2f86e75c4dd012e22d7ac1e040319544dc Mon Sep 17 00:00:00 2001 From: Emilio Pozuelo Monfort Date: Wed, 10 Nov 2021 12:29:40 +0100 Subject: bin/lts-missing-uploads: drop the .py extension --- bin/lts-missing-uploads | 157 +++++++++++++++++++++++++++++++++++++++++++++ bin/lts-missing-uploads.py | 157 --------------------------------------------- 2 files changed, 157 insertions(+), 157 deletions(-) create mode 100755 bin/lts-missing-uploads delete mode 100755 bin/lts-missing-uploads.py (limited to 'bin') diff --git a/bin/lts-missing-uploads b/bin/lts-missing-uploads new file mode 100755 index 0000000000..eb84a234df --- /dev/null +++ b/bin/lts-missing-uploads @@ -0,0 +1,157 @@ +#!/usr/bin/env python3 +# +# Copyright 2016 Chris Lamb +# +# This file is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# This file is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this file. If not, see . + +import re +import sys +import gzip +import datetime +import eventlet +import requests +import dateutil.relativedelta + +from debian.deb822 import Sources +from debian.debian_support import Version + +class LTSMissingUploads(object): + MONTHS = 6 + SOURCES = ['http://security.debian.org/dists/stretch/updates/{}/source/Sources.gz'.format(component) + for component in ('main', 'contrib', 'non-free')] + + re_line = re.compile( + r'(?Pmsg\d+.html).*\[DLA (?P[\d-]+)\] (?P[^\s]+) security update.*' + ) + re_version = re.compile(r'^Version.*: (?P.*)') + + def __init__(self): + self.pool = eventlet.GreenPool(10) + self.session = requests.session() + + def main(self, *args): + self.info("Getting last {} month(s) of LTS annoucements", self.MONTHS) + + dlas = {} + def download(x): + self.info("{source}: parsing announcement from {url} ...", **x) + x.update(self.get_dla(x['url'])[0]) + dlas[x['source']] = x + + for idx in range(self.MONTHS): + dt = datetime.datetime.utcnow().replace(day=1) - \ + dateutil.relativedelta.relativedelta(months=idx) + + self.info( + "Getting announcements for {}/{:02} ...", + dt.year, + dt.month, + ) + + # Prefer later DLAs with reversed(..) + for x in reversed(self.get_dlas(dt.year, dt.month)): + if x['source'] not in dlas: + self.pool.spawn_n(download, x) + self.pool.waitall() + + if not dlas: + return 0 + + sources = self.get_sources() + + for source, dla in sorted(dlas.items()): + try: + dla_version = Version(dla['version']) + except ValueError: + self.warn("{}: DLA-{} announced with invalid version: {} <{}>", + source, + dla['dla'], + dla['version'], + dla['url'], + ) + continue + + archive_version = Version(sources[source]) + + if dla_version <= archive_version: + continue + + self.warn("{}: DLA-{} announced version {} but LTS has {} <{}>", + source, + dla['dla'], + dla_version, + archive_version, + dla['url'], + ) + + return 0 + + def get_dlas(self, year, month): + url = 'https://lists.debian.org/debian-lts-announce/{}/{:02}/'.format( + year, + month, + ) + + result = self.parse(url, self.re_line) + + # Prepend URL as the indices have relative URIs + for x in result: + x['url'] = '{}{}'.format(url, x['suffix']) + + return result + + def get_dla(self, url): + return self.parse(url, self.re_version) + + def get_sources(self): + pkgver = {} + for src in self.SOURCES: + self.info("Downloading Sources from {} ...", src) + + response = self.session.get(src) + response.raise_for_status() + + val = gzip.decompress(response.content).decode('utf-8') + + for x in Sources.iter_paragraphs(val): + pkgver[x['Package']] = x['Version'] + + return pkgver + + def parse(self, url, pattern): + result = [] + + for x in self.session.get(url).content.splitlines(): + m = pattern.search(x.decode('utf8')) + + if m is not None: + result.append(m.groupdict()) + + return result + + ## + + def warn(self, msg, *args, **kwargs): + print("W: " + msg.format(*args, **kwargs), file=sys.stderr) + + def info(self, msg, *args, **kwargs): + print("I: " + msg.format(*args, **kwargs), file=sys.stderr) + +if __name__ == '__main__': + eventlet.monkey_patch(socket=True) + + try: + sys.exit(LTSMissingUploads().main(*sys.argv[1:])) + except KeyboardInterrupt: + sys.exit(1) diff --git a/bin/lts-missing-uploads.py b/bin/lts-missing-uploads.py deleted file mode 100755 index eb84a234df..0000000000 --- a/bin/lts-missing-uploads.py +++ /dev/null @@ -1,157 +0,0 @@ -#!/usr/bin/env python3 -# -# Copyright 2016 Chris Lamb -# -# This file is free software: you can redistribute it and/or modify -# it under the terms of the GNU General Public License as published by -# the Free Software Foundation, either version 3 of the License, or -# (at your option) any later version. -# -# This file is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU General Public License for more details. -# -# You should have received a copy of the GNU General Public License -# along with this file. If not, see . - -import re -import sys -import gzip -import datetime -import eventlet -import requests -import dateutil.relativedelta - -from debian.deb822 import Sources -from debian.debian_support import Version - -class LTSMissingUploads(object): - MONTHS = 6 - SOURCES = ['http://security.debian.org/dists/stretch/updates/{}/source/Sources.gz'.format(component) - for component in ('main', 'contrib', 'non-free')] - - re_line = re.compile( - r'(?Pmsg\d+.html).*\[DLA (?P[\d-]+)\] (?P[^\s]+) security update.*' - ) - re_version = re.compile(r'^Version.*: (?P.*)') - - def __init__(self): - self.pool = eventlet.GreenPool(10) - self.session = requests.session() - - def main(self, *args): - self.info("Getting last {} month(s) of LTS annoucements", self.MONTHS) - - dlas = {} - def download(x): - self.info("{source}: parsing announcement from {url} ...", **x) - x.update(self.get_dla(x['url'])[0]) - dlas[x['source']] = x - - for idx in range(self.MONTHS): - dt = datetime.datetime.utcnow().replace(day=1) - \ - dateutil.relativedelta.relativedelta(months=idx) - - self.info( - "Getting announcements for {}/{:02} ...", - dt.year, - dt.month, - ) - - # Prefer later DLAs with reversed(..) - for x in reversed(self.get_dlas(dt.year, dt.month)): - if x['source'] not in dlas: - self.pool.spawn_n(download, x) - self.pool.waitall() - - if not dlas: - return 0 - - sources = self.get_sources() - - for source, dla in sorted(dlas.items()): - try: - dla_version = Version(dla['version']) - except ValueError: - self.warn("{}: DLA-{} announced with invalid version: {} <{}>", - source, - dla['dla'], - dla['version'], - dla['url'], - ) - continue - - archive_version = Version(sources[source]) - - if dla_version <= archive_version: - continue - - self.warn("{}: DLA-{} announced version {} but LTS has {} <{}>", - source, - dla['dla'], - dla_version, - archive_version, - dla['url'], - ) - - return 0 - - def get_dlas(self, year, month): - url = 'https://lists.debian.org/debian-lts-announce/{}/{:02}/'.format( - year, - month, - ) - - result = self.parse(url, self.re_line) - - # Prepend URL as the indices have relative URIs - for x in result: - x['url'] = '{}{}'.format(url, x['suffix']) - - return result - - def get_dla(self, url): - return self.parse(url, self.re_version) - - def get_sources(self): - pkgver = {} - for src in self.SOURCES: - self.info("Downloading Sources from {} ...", src) - - response = self.session.get(src) - response.raise_for_status() - - val = gzip.decompress(response.content).decode('utf-8') - - for x in Sources.iter_paragraphs(val): - pkgver[x['Package']] = x['Version'] - - return pkgver - - def parse(self, url, pattern): - result = [] - - for x in self.session.get(url).content.splitlines(): - m = pattern.search(x.decode('utf8')) - - if m is not None: - result.append(m.groupdict()) - - return result - - ## - - def warn(self, msg, *args, **kwargs): - print("W: " + msg.format(*args, **kwargs), file=sys.stderr) - - def info(self, msg, *args, **kwargs): - print("I: " + msg.format(*args, **kwargs), file=sys.stderr) - -if __name__ == '__main__': - eventlet.monkey_patch(socket=True) - - try: - sys.exit(LTSMissingUploads().main(*sys.argv[1:])) - except KeyboardInterrupt: - sys.exit(1) -- cgit v1.2.3