pthbs_genpkgpy

Template engine for producing packages for pthbs written using Python and Jinja
git clone https://ccx.te2000.cz/git/pthbs_genpkgpy
Log | Files | Refs | README

genpkg.py (7520B)


      1 #!/usr/bin/env python3
      2 import argparse
      3 import hashlib
      4 import os.path
      5 import subprocess
      6 from pathlib import Path
      7 
      8 import jinja2
      9 import yaml
     10 
     11 
     12 class SubmoduleInfo:
     13     def __init__(self, cache_dir='cache'):
     14         self._current_commits = None
     15         self._by_commit = Path(cache_dir) / "link" / "git-commit-sha1"
     16 
     17     @property
     18     def current(self):
     19         if self._current_commits is not None:
     20             return self._current_commits
     21         out = subprocess.check_output(
     22             ("git", "submodule", "status", "--cached")
     23         ).decode('utf8')
     24         lines = out.strip('\n').split('\n')
     25         records = [[line[0]] + line[1:].split() for line in lines]
     26         self._current_commits = {
     27             r[2][8:]: r[1] for r in records if r[2].startswith("sources/")
     28         }
     29         for repo, commit in self._current_commits.items():
     30             if not (self._by_commit / commit).exists():
     31                 raise RuntimeError(
     32                     f"commit ID does not seem to be linked:"
     33                     f" {commit} from {repo!r} in {self._by_commit}"
     34                 )
     35         return self._current_commits
     36 
     37     def commit_info(self, commit_id):
     38         assert '/' not in commit_id
     39         assert '.' not in commit_id
     40         assert (self._by_commit / commit_id).exists()
     41         out = subprocess.check_output(
     42             ('git', 'show', '-s', '--pretty=format:%ai by %an'),
     43             cwd=(self._by_commit / commit_id).as_posix(),
     44         ).decode('utf8')
     45         return out
     46 
     47 
     48 class DownloadsInfo:
     49     def __init__(self, downloadlist_path):
     50         assert isinstance(downloadlist_path, Path)
     51         self._basenames = {}
     52         with downloadlist_path.open('rt') as f:
     53             for line in f:
     54                 if line[0] in '#\n':
     55                     continue
     56                 sha256, size, url = line.rstrip().split(maxsplit=2)
     57                 assert len(bytes.fromhex(sha256)) == 32
     58                 assert int(size) >= 0
     59                 basename = os.path.basename(url)
     60                 if basename in self._basenames:
     61                     self._basenames[basename] = ValueError(
     62                         'Duplicate download name: ' + repr(basename)
     63                     )
     64                 else:
     65                     self._basenames[basename] = 'sha256:' + sha256
     66 
     67     def __getitem__(self, key):
     68         value = self._basenames[key]
     69         if isinstance(value, Exception):
     70             raise value
     71         return value
     72 
     73 
     74 def parse_filelist(filelist_path):
     75     assert isinstance(filelist_path, Path)
     76     with filelist_path.open('rt') as f:
     77         return {
     78             os.path.basename(fname): fhash
     79             for fhash, fname in (
     80                 line.rstrip('\n').split('  ', maxsplit=1) for line in f
     81             )
     82         }
     83 
     84 
     85 class Main:
     86     argument_parser = argparse.ArgumentParser()
     87     argument_parser.add_argument('-P', '--package-dir', default='packages')
     88     argument_parser.add_argument('-T', '--template-dir', default='templates')
     89     argument_parser.add_argument('-I', '--index-dir', default='.')
     90     argument_parser.add_argument('-C', '--cache-dir', default='cache')
     91 
     92     def __init__(self, out_dir, template_dir, index_dir, cache_dir):
     93         assert isinstance(out_dir, Path)
     94         assert isinstance(template_dir, Path)
     95         assert isinstance(index_dir, Path)
     96         assert isinstance(cache_dir, Path)
     97         self.out_dir = Path(out_dir)
     98         self.template_dir = Path(template_dir)
     99         self.env = jinja2.Environment(
    100             loader=jinja2.FileSystemLoader(template_dir),
    101             undefined=jinja2.StrictUndefined,
    102             autoescape=False,
    103         )
    104         self.env.globals["pkg_sha256"] = self.pkg_sha256
    105         self.env.globals["pkg_install_name"] = self.pkg_install_name
    106         self.env.globals["pkg_install_dir"] = self.pkg_install_dir
    107         self.env.globals["submodule"] = SubmoduleInfo(cache_dir=cache_dir)
    108         self.env.globals["files"] = parse_filelist(index_dir / 'filelist.sha256')
    109         self.env.globals["downloads"] = DownloadsInfo(index_dir / 'downloadlist.sha256')
    110         self.package_hashes = {}
    111         self.rendering = []
    112         self.deps = {}
    113 
    114     @classmethod
    115     def from_argv(cls, args=None):
    116         args = cls.argument_parser.parse_args(args)
    117         return cls(
    118             out_dir=Path(args.package_dir),
    119             template_dir=Path(args.template_dir),
    120             index_dir=Path(args.index_dir),
    121             cache_dir=Path(args.cache_dir),
    122         )
    123 
    124     def load_vars_yaml(self, fname="vars.yaml"):
    125         with open(fname) as f:
    126             self.env.globals.update(yaml.safe_load(f))
    127 
    128     def pkg_env_sha256(self, name):
    129         current = self.rendering[-1]
    130         if current not in self.deps:
    131             self.deps[current] = set((name,))
    132         else:
    133             self.deps[current].add(name)
    134         self._pkg_sha256(name)
    135         envlist = ''.join(
    136             sorted('%s.%s\n' % (d, self.package_hashes[d]) for d in self.deps[name])
    137         )
    138         return hashlib.sha256(envlist.encode()).hexdigest()
    139 
    140     def pkg_sha256(self, name):
    141         current = self.rendering[-1]
    142         if current not in self.deps:
    143             self.deps[current] = set((name,))
    144         else:
    145             self.deps[current].add(name)
    146         return self._pkg_sha256(name)
    147 
    148     def _pkg_sha256(self, name):
    149         if name in self.package_hashes:
    150             return self.package_hashes[name]
    151         if name in self.rendering:
    152             raise RuntimeError("circular dependency: %r", self.rendering)
    153 
    154         t = self.env.get_template("pkg/" + name)
    155         self.rendering.append(name)
    156         data = bytes(t.render(name=name).encode('utf8'))
    157         self.package_hashes[name] = hashlib.sha256(data).hexdigest()
    158         lastname = self.rendering.pop()
    159         assert name == lastname
    160 
    161         out_path = self.out_dir / name
    162         old_hash = None
    163         if out_path.exists():
    164             with out_path.open('rb') as f:
    165                 old_hash = hashlib.file_digest(f, "sha256").hexdigest()
    166         if old_hash is None or old_hash != self.package_hashes[name]:
    167             tmp_path = out_path.with_suffix('.new')
    168             tmp_path.write_bytes(data)
    169             tmp_path.replace(out_path)
    170 
    171         return self.package_hashes[name]
    172 
    173     def pkg_install_name(self, name):
    174         rootname = name.split(":")[0]
    175         if rootname.endswith('.environment'):
    176             return "env.%s" % (self.pkg_env_sha256(name),)
    177         else:
    178             return "%s.%s" % (name.split(":")[0], self.pkg_sha256(name))
    179 
    180     def pkg_install_dir(self, name):
    181         return os.path.join(
    182             self.env.globals["versions"],
    183             self.pkg_install_name(name),
    184         )
    185 
    186     def list_packages(self):
    187         for tplname in self.env.list_templates():
    188             if not tplname.startswith("pkg/"):
    189                 continue
    190             if "/." in tplname:
    191                 continue
    192             yield tplname[4:]
    193 
    194     def render_all(self):
    195         for pkgname in self.list_packages():
    196             print("%s\t%s" % (pkgname, self._pkg_sha256(pkgname)))
    197             for dep in sorted(self.deps.get(pkgname, ())):
    198                 print(
    199                     "  > %s.%s"
    200                     % (
    201                         dep,
    202                         self.package_hashes[dep],
    203                     )
    204                 )
    205 
    206 
    207 if __name__ == '__main__':
    208     m = Main.from_argv()
    209     m.load_vars_yaml()
    210     m.render_all()
    211 
    212 # pylama:linters=pycodestyle,pyflakes:ignore=D212,D203,D100,D101,D102,D105,D107
    213 # vim: sts=4 ts=4 sw=4 et tw=88 efm=%A%f\:%l%\:%c\ %t%n\ %m