path: root/dlpcvp.py

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

import argparse
import configparser
import json
import logging
import os
import os.path
import sqlite3
import sys
import tempfile
import urllib.request
from distutils.version import LooseVersion
from urllib.error import HTTPError
from urllib.request import Request, urlopen
from typing import Iterable, 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"
# https://github.com/openSUSE/open-build-service/blob/master/docs/api/api/api.txt
# https://build.opensuse.org/apidocs/index
OBS_base = "https://api.opensuse.org"
ConfigRC = os.path.expanduser('~/.config/osc/oscrc')
CUTCHARS = len('python-')

config = configparser.ConfigParser()
config.read(ConfigRC)

logging.basicConfig(format='%(levelname)s:%(funcName)s:%(message)s',
                    stream=sys.stdout, level=logging.INFO)
log = logging.getLogger()

# or HTTPPasswordMgrWithPriorAuth ?
password_mgr = urllib.request.HTTPPasswordMgrWithDefaultRealm()
user = config[OBS_base]['user']
passw = config[OBS_base]['pass']
# MYPY: Argument 1 to "add_password" of
# "HTTPPasswordMgrWithDefaultRealm" has incompatible type "None";
# expected "str"
password_mgr.add_password(None, OBS_base, user, passw)

handler = urllib.request.HTTPBasicAuthHandler(password_mgr)
opener = urllib.request.build_opener(handler)
urllib.request.install_opener(opener)

TB_EXISTS = """SELECT name FROM sqlite_master
        WHERE type='table' AND name='etags'
        """
TB_CREATE = """CREATE TABLE etags
               (pkg text NOT NULL,
               suse_name_etag text, suse_spec_etag text, pypi_etag text)
            """
TB_CREATE_IDX = "CREATE UNIQUE INDEX etags_names ON etags(pkg)"


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 know about 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 as ex:
        if ex.getcode() == 404:
            log.error(f'Cannot find packages for {proj}!')
        else:
            raise


def parse_spec_in_dev_null(spec_file, pkg):
    # rpm library generates awful lot of nonsensical goo on
    # stderr
    with open(os.devnull, 'wb') as nullf:
        old_stderr = sys.stderr
        old_stdout = sys.stdout
        sys.stderr = nullf
        sys.stdout = nullf
        try:
            spc = rpm.spec(spec_file)
        except Exception:
            log.error("Cannot parse {}".format(pkg))
        else:
            try:
                return pkg, spc.packages[0].header['Version'].decode()
            except IndexError:
                pass
        finally:
            sys.stderr = old_stderr
            sys.stdout = old_stdout


def get_etags(con: sqlite3.Connection, pkg: str) -> \
        Tuple[Optional[str], Optional[str], Optional[str], Optional[str]]:
    # pkg, suse_name_etag, suse_spec_etag, pypi_etag
    cur = con.execute("SELECT * FROM etags WHERE pkg=?", (pkg,))
    ret = cur.fetchone()
    if ret is None:
        return None, None, None, None
    else:
        return ret


def update_etags(con: sqlite3.Connection, pkg: str, e_suse_name: Optional[str], e_suse_spec: Optional[str], e_pypi: Optional[str]):
    res = con.execute("SELECT * FROM etags WHERE pkg=?", (pkg,)).fetchone()
    if res:
        e_suse_name = res[1] if e_suse_name is None else e_suse_name
        e_suse_spec = res[2] if e_suse_spec is None else e_suse_spec
        e_pypi = res[3] if e_pypi is None else e_pypi
        con.execute('''UPDATE etags
            SET suse_name_etag=?, suse_spec_etag=?, pypi_etag=?
            WHERE pkg=?''', (e_suse_name, e_suse_spec, e_pypi, pkg))
    else:
        con.execute('''INSERT INTO etags
            (pkg, suse_name_etag, suse_spec_etag, pypi_etag)
            VALUES (?, ?, ?, ?)''', (pkg, e_suse_name, e_suse_spec, e_pypi))
    con.commit()


def get_version_from_pypi(name: str, con: sqlite3.Connection) \
        -> Optional[Tuple[str, str]]:
    """
    For the given name of module return the latest version available on PyPI.
    """
    # pkg, suse_name_etag, suse_spec_etag, pypi_etag
    _, _, _, etag = get_etags(con, name)
    req = Request(url=PyPI_base.format(name))

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

    try:
        with urlopen(req) as resp:
            data = json.load(resp)
            info_dict = data['info']
            update_etags(con, name, None, None, resp.info()['ETag'])
            return info_dict['name'], info_dict['version']
    except HTTPError as ex:
        if ex.getcode() == 404:
            log.error(f'Cannot find {name} on PyPI')
            return None
        else:
            raise


def package_version(proj: str, pkg_name: str, con: sqlite3.Connection) \
                    -> Optional[Tuple[str, str, str]]:
    """
    Return the version of the given package in the given proj.

    Downloads SPEC file from OBS and parses it.
    """
    # pkg, suse_name_etag, suse_spec_etag, pypi_etag
    _, etag_fn, etag_spcf, _ = get_etags(con, pkg_name)

    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 as ex:
        if ex.getcode() == 404:
            log.error(f'Cannot accquire version of {pkg_name} in {proj}')
            return None
        else:
            raise

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

    try:
        spc_fname = sorted(spec_files, key=len)[0]
    except IndexError:
        log.exception(f'Cannot find correct spec_files: {spec_files}')
        return

    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())

                update_etags(con, pkg_name, etag_fn, etag_spcf, None)
                return parse_spec_in_dev_null(spec_file_name, pkg_name)
    except HTTPError as ex:
        if ex.getcode() == 404:
            log.error(f'Cannot parse SPEC file {spc_fname} for {pkg_name}')
        else:
            raise


def main(prj):
    db_name = os.path.split(os.path.basename(sys.argv[0]))[0] + ".db"
    with sqlite3.connect(db_name) as conn:
        if not conn.execute(TB_EXISTS).fetchone():
                conn.execute(TB_CREATE)
                conn.execute(TB_CREATE_IDX)

        for pkg in suse_packages(prj):
            log.debug(f'pkg = {pkg}')
            if pkg.startswith('python-'):
                pypi_name = pkg[CUTCHARS:]
                try:
                    suse_ver_tup = package_version(prj, pkg, conn)
                    if suse_ver_tup is None:
                        raise RuntimeError('not in OBS')
                    pypi_ver_tup = get_version_from_pypi(pypi_name, conn)
                    if pypi_ver_tup is None:
                        raise RuntimeError('not in PyPI')
                except RuntimeError as ex:
                    log.warning(f'Package {pkg} cannot be found: {ex}')
                    continue

                log.debug(f"{pkg} {suse_ver_tup} {pypi_ver_tup}")

                pypi_ver = LooseVersion(pypi_ver_tup[1])
                suse_ver = LooseVersion(suse_ver_tup[1])

                if pypi_ver > suse_ver:
                    print(
                        f'We need to upgrade {pkg} ({suse_ver} -> {pypi_ver})')
            else:
                print(f'Is {pkg} on PyPI?')


if __name__ == '__main__':
    parser = argparse.ArgumentParser(description='Check available versions '
                                     'of the upstream packages on PyPI')
    parser.add_argument('--opensuse-project',
                        default='devel:languages:python',
                        help='The OpenBuildService project. Defaults '
                        'to %(default)s')

    args = parser.parse_args()
    sys.exit(main(args.opensuse_project))