path: root/dlpcvp.py

#!/usr/bin/python3.6
# Requires: python3-rpm

import configparser
import json
import logging
import os
import os.path
# import sqlite3
import sys
import tempfile
import urllib.request
from urllib.error import HTTPError
from urllib.request import Request, urlopen
from typing import Iterable, 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()

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: str = None) -> Tuple[str, str, str]:
    """
    For the given name of module return the latest version available on PyPI.
    """
    log.debug('name = %s', name)
    req = Request(url=PyPI_base.format(name))
    log.debug('req URL = %s', req.full_url)

    if etag is not None:
        req.add_header('ETag', etag)

    try:
        with urlopen(req) as resp:
            data = json.load(resp)
            info_dict = data['info']
            return info_dict['name'], info_dict['version'], resp.info()['ETag']
    except HTTPError:
        log.exception(f'Cannot find {name} on PyPI')


def suse_packages(proj:str) -> Iterable[str]:
    """
    Iterator returning names of all packages in the given proj

    ETag management won't work here, because I don't like any way how to
    return it in iterator.
    """
    req = Request(url=OBS_base + f'/source/{proj}')

    try:
        with opener.open(req) as resp:
            raw_xml_data = ET.parse(resp)
            root = raw_xml_data.getroot()
            for elem in root.iter('entry'):
                yield elem.get('name')
    except HTTPError:
        log.exception(f'Cannot find packages for {proj}!')


def package_version(proj:str, pkg_name:str,
                    etag_fn: str = None, etag_spcf: str = None) -> Tuple[str, str, str]:
    """
    Return the version of the given package in the given proj.

    Downloads SPEC file from OBS and parses it.
    """
    req_spc_name = Request(url=OBS_base + f'/source/{proj}/{pkg_name}')
    spec_files = []

    if etag_fn is not None:
        req_spc_name.add_header('ETag', etag_fn)

    try:
        with opener.open(req_spc_name) as resp:
            etag_fn = resp.info()['ETag']
            raw_xml_data = ET.parse(resp)
            root = raw_xml_data.getroot()
            for elem in root.iter('entry'):
                name = elem.get('name')
                if name.endswith('.spec'):
                    spec_files.append(name)
    except HTTPError:
        log.exception(f'Cannot accquire version of {pkg_name} in {proj}')

    if not spec_files:
        IOError(f'Cannot find SPEC file for {pkg_name}')

    spc_fname = sorted(spec_files, key=len)[0]

    req_spec = Request(url=OBS_base + f'/source/{proj}/{pkg_name}/{spc_fname}')

    if etag_spcf is not None:
        req_spc_name.add_header('ETag', etag_spcf)

    try:
        with opener.open(req_spec) as resp:
            etag_spcf = resp.info()['ETag']
            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())

                # rpm library generates awfull lot of nonsensical goo on
                # stderr
                old_stderr = sys.stderr
                sys.stderr = os.devnull
                try:
                    spc = rpm.spec(spec_file_name)
                except Exception as ex:
                    log.exception("Cannot parse {}".format(pkg_name))
                else:
                    try:
                        return spc.packages[0].header['Version'].decode(), \
                            etag_fn, etag_spcf
                    except IndexError:
                        pass
                finally:
                    sys.stderr = old_stderr
    except HTTPError:
        log.exception(f'Cannot parse SPEC file {spc_fname} for {pkg_name}')


if __name__=='__main__':
    cutchars = len('python-')
    project = 'devel:languages:python:singlespec-staging'
    for pkg in suse_packages(project):
        if pkg:
            assert pkg.startswith('python-')
            pypi_name = pkg[cutchars:]
            pypi_ver = get_version_from_pypi(pypi_name)
            suse_ver = package_version(project, pkg)
            print(f"{pkg} {suse_ver} {pypi_ver}")