diff options
| -rw-r--r-- | .gitignore | 1 | ||||
| -rwxr-xr-x | dlpcvp.py | 105 |
2 files changed, 103 insertions, 3 deletions
@@ -1 +1,2 @@ __pycache__/ +*.log @@ -1,22 +1,121 @@ #!/usr/bin/python3 +# Requires: python3-rpm +import configparser import json +import logging +import os +import os.path import sqlite3 +import tempfile import urllib.request +from urllib.request import Request, urlopen +from typing import Optional, Tuple +import xml.etree.ElementTree as ET + +import rpm # PyPI API documentation https://warehouse.readthedocs.io/api-reference/ PyPI_base = "https://pypi.org/pypi/{}/json" +OBS_base = "https://api.opensuse.org" +ConfigRC = os.path.expanduser('~/.config/osc/oscrc') + +config = configparser.ConfigParser() +config.read(ConfigRC) + +logging.basicConfig(format='%(levelname)s:%(funcName)s:%(message)s', + level=logging.DEBUG) +log = logging.getLogger() -def get_version_from_pypi(name, etag=None): - req = urllib.request.Request(url=PyPI_base.format(name)) +password_mgr = urllib.request.HTTPPasswordMgrWithDefaultRealm() +user = config[OBS_base]['user'] +passw = config[OBS_base]['pass'] +password_mgr.add_password(None, OBS_base, user, passw) + +handler = urllib.request.HTTPBasicAuthHandler(password_mgr) +opener = urllib.request.build_opener(handler) + +def get_version_from_pypi(name:str, etag: Optional[str] = None) -> Tuple[str, str, str]: + req = Request(url=PyPI_base.format(name)) if etag is not None: req.add_header('ETag', etag) - with urllib.request.urlopen(req) as resp: + with urlopen(req) as resp: if resp.getcode() == 200: data = json.load(resp) info_dict = data['info'] return info_dict['name'], info_dict['version'], resp.info()['ETag'] else: IOError(resp.info()) + +def list_packages(project:str, etag: Optional[str] = None) -> Tuple[str, ...]: + req = Request(url=OBS_base+'/source/{}'.format(project)) + + if etag is not None: + req.add_header('ETag', etag) + + with opener.open(req) as resp: + if resp.getcode() == 200: + raw_xml_data = ET.parse(resp) + root = raw_xml_data.getroot() + out = [] + for elem in root.iter('entry'): + out.append(elem.get('name')) + + return tuple(out) + +def package_version(project:str, pkg:str, etag: Optional[str] = None) -> Tuple[str, ...]: + # GET https://api.opensuse.org/source/devel:languages:python:singlespec-staging/python-venusian?rev=latest + req_spc_name = Request(url=OBS_base+'/source/{}/{}'.format(project, pkg)) + + if etag is not None: + req_spc_name.add_header('ETag', etag) + + with opener.open(req_spc_name) as resp: + if resp.getcode() == 200: + raw_xml_data = ET.parse(resp) + root = raw_xml_data.getroot() + spec_files = [] + for elem in root.iter('entry'): + name = elem.get('name') + if name.endswith('.spec'): + spec_files.append(name) + + log.debug('spec_files = %s', spec_files) + if spec_files: + spc_fname = sorted(spec_files, key=len) + log.debug('spc_fname = %s', spc_fname) + spc_fname = spc_fname[0] + else: + return None + + req_spec = Request(url=OBS_base+'/source/{}/{}/{}'.format(project, pkg, spc_fname)) + with opener.open(req_spec) as resp: + spec_file_str = resp.read() + spec_file_name = '' + with tempfile.NamedTemporaryFile(mode='w+b', delete=False) as spf: + spec_file_name = spf.name + spf.write(spec_file_str) + spf.flush() + os.fsync(spf.fileno()) + try: + spc = rpm.spec(spec_file_name) + except Exception as ex: + print('Exception: {}\n'.format(ex)) + print("Cannot parse {}".format(pkg)) + else: + try: + return spc.packages[0].header['Version'].decode() + except IndexError: + pass + + +cutchars = len('python-') +project = 'devel:languages:python:singlespec-staging' +for pkg in list_packages(project): + pkg = pkg.strip() + if pkg: + assert pkg.startswith('python-') + pypi_name = pkg[cutchars:] + print("{} {}".format(pkg, package_version(project, pkg))) |