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
|
from __future__ import print_function
import argparse
import bz2
import email
import json
import logging
from itertools import product
from operator import itemgetter
import attr
import pkg_resources
from pathlib2 import Path
from requests import Session
from six.moves.urllib_parse import urljoin
@attr.s
class ReleaseElement(object):
sha256 = attr.ib(repr=False)
size = attr.ib(convert=int)
path = attr.ib()
log = logging.getLogger('enpass.updater')
parser = argparse.ArgumentParser()
parser.add_argument('--repo')
parser.add_argument('--target', type=Path)
session = Session()
def parse_bz2_msg(msg):
msg = bz2.decompress(msg)
if '\n\n' in msg:
parts = msg.split('\n\n')
return list(map(email.message_from_string, parts))
return email.message_from_string(msg)
def fetch_meta(repo, name, parse=email.message_from_string, split=False):
url = urljoin(repo, 'dists/stable', name)
response = session.get("{repo}/dists/stable/{name}".format(**locals()))
return parse(response.content)
def fetch_filehashes(repo, path):
meta = fetch_meta(repo, path, parse=parse_bz2_msg)
for item in meta:
yield {
'version': pkg_resources.parse_version(str(item['Version'])),
'path': item['Filename'],
'sha256': item['sha256'],
}
def fetch_archs(repo):
m = fetch_meta(repo, 'Release')
architectures = m['Architectures'].split()
elements = [ReleaseElement(*x.split()) for x in m['SHA256'].splitlines()]
elements = [x for x in elements if x.path.endswith('bz2')]
for arch, elem in product(architectures, elements):
if arch in elem.path:
yield arch, max(fetch_filehashes(repo, elem.path),
key=itemgetter('version'))
class OurVersionEncoder(json.JSONEncoder):
def default(self, obj):
# the other way around to avoid issues with
# newer setuptools having strict/legacy versions
if not isinstance(obj, (dict, str)):
return str(obj)
return json.JSONEncoder.default(self, obj)
def main(repo, target):
logging.basicConfig(level=logging.DEBUG)
with target.open(mode='wb') as fp:
json.dump(
dict(fetch_archs(repo)), fp,
cls=OurVersionEncoder,
indent=2,
sort_keys=True)
opts = parser.parse_args()
main(opts.repo, opts.target)
|