from copy import deepcopy import logging import os import tarfile import tempfile import urllib.request from .package import PackageInfo def resolve_url(url_template, repo_name: str, arch: str): result = url_template for template, replacement in {'$repo': repo_name, '$arch': arch}.items(): result = result.replace(template, replacement) return result class RepoInfo: options: dict[str, str] = {} url_template: str def __init__(self, url_template: str, options: dict[str, str] = {}): self.url_template = url_template self.options.update(options) class Repo(RepoInfo): name: str resolved_url: str arch: str packages: dict[str, PackageInfo] remote: bool scanned: bool = False def resolve_url(self) -> str: return resolve_url(self.url_template, repo_name=self.name, arch=self.arch) def scan(self): self.resolved_url = self.resolve_url() self.remote = not self.resolved_url.startswith('file://') uri = f'{self.resolved_url}/{self.name}.db' path = '' if self.remote: logging.info(f'Downloading repo file from {uri}') with urllib.request.urlopen(uri) as request: fd, path = tempfile.mkstemp() with open(fd, 'wb') as writable: writable.write(request.read()) else: path = uri.split('file://')[1] logging.debug(f'Parsing repo file at {path}') with tarfile.open(path) as index: for node in index.getmembers(): if os.path.basename(node.name) == 'desc': logging.debug(f'Parsing desc file for {os.path.dirname(node.name)}') pkg = PackageInfo.parse_desc(index.extractfile(node).read().decode(), self.resolved_url) self.packages[pkg.name] = pkg self.scanned = True def __init__(self, name: str, url_template: str, arch: str, options={}, scan=False): self.packages = {} self.name = name self.url_template = url_template self.arch = arch self.options = deepcopy(options) if scan: self.scan() def __repr__(self): return f'' def config_snippet(self) -> str: options = {'Server': self.url_template} | self.options return ('[%s]\n' % self.name) + '\n'.join([f"{key} = {value}" for key, value in options.items()]) def get_RepoInfo(self): return RepoInfo(url_template=self.url_template, options=self.options)