#!/usr/bin/python3.6
# 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 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',
level=logging.DEBUG)
log = logging.getLogger()
# or HTTPPasswordMgrWithPriorAuth ?
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)
urllib.request.install_opener(opener)
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.
"""
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']
return info_dict['name'], info_dict['version'], resp.info()['ETag']
except HTTPError as ex:
if ex.getcode() == 404:
log.error(f'Cannot find {name} on PyPI')
else:
raise
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_name, pkg_name, etag_fn, etag_spcf):
# rpm library generates awfull 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_name)
except Exception:
log.error("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
sys.stdout = old_stdout
def package_version(proj: str, pkg_name: str,
etag_fn: str = None, etag_spcf: str = None) \
-> Optional[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 as ex:
if ex.getcode() == 404:
log.error(f'Cannot accquire version of {pkg_name} in {proj}')
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())
return parse_spec_in_dev_null(spec_file_name, pkg_name,
etag_fn, etag_spcf)
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):
for pkg in suse_packages(prj):
log.debug(f'pkg = {pkg}')
if pkg.startswith('python-'):
pypi_name = pkg[CUTCHARS:]
pypi_ver = get_version_from_pypi(pypi_name)
suse_ver = package_version(prj, pkg)
# FIXME We should somehow react to the situation, not just
# ignore it.
if suse_ver is not None:
print(f"{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:numeric',
help='The OpenBuildService project. Defaults '
'to %(default)s')
args = parser.parse_args()
sys.exit(main(args.opensuse_project))
