#!/usr/bin/env python import re import requests import sys import getpass import pkgdb2client import subprocess #PAGE_URL = 'https://fedoraproject.org/w/api.php?format=json&action=query&rvprop=content&prop=revisions&titles=User:Codeblock/RequestsSANDBOX' PAGE_URL = 'https://fedoraproject.org/w/api.php?format=json&action=query&rvprop=content&prop=revisions&titles=EPEL/epel7/Requests' NEW_EPEL_VERSION = '7' NEW_EPEL_SOURCE_BRANCH = 'f19' RHEL_PKGS_PATH = '/var/lib/rhel/rhel' + NEW_EPEL_VERSION # parse_page :: String -> IO (Map String String) # This returns a dictionary of {"pkg_name": "branch"} def parse_page(url): r = requests.get(url).json() text = r['query']['pages'][r['query']['pages'].keys()[0]]['revisions'][0]['*'] lines = text.split("\n") pkgs = filter(lambda y: y.startswith('| '), lines) __pkgs_list__ = map(lambda y: ''.join(y.split())[1:].split('||'), pkgs) pkgs_list = filter(lambda y: y[0] != 'foo', __pkgs_list__) pkgs_dict = dict(pkgs_list) return pkgs_dict # is_in_rhel :: String -> IO Bool def is_in_rhel(pkg): with open(RHEL_PKGS_PATH) as f: pkgs = map(lambda x: x.strip(), f.readlines()) return (pkg in pkgs) # These tuples will be used to substitute one pattern for another. # Every transform will be run on every branch name so be sure the # pattern cannot match if you don't want it to be triggered. transforms = ( (re.compile(r'^devel$'), 'master'), (re.compile(r'-'), ''), (re.compile(r'^fc([0-9])'), r'f\1'), (re.compile(r'^epel([456])$'), r'el\1'), (re.compile(r'^el([789]|[1-9][0-9])'), r'epel\1'), ) branch_replacements = {'devel': (re.compile(r'^devel$'), 'master'),} # generate_collection_cache :: PkgDB -> IO [String] def generate_collection_cache(pkgdb): raw_collections = pkgdb.get_collections(clt_status=( 'Active', 'Under Development')) collection_cache = frozenset(map(lambda y: y['branchname'], raw_collections['collections'])) return collection_cache # normalize_branch :: [String] -> String -> IO (Option String) def normalize_branch(collection_cache, branch): # I originally had this implemented as a foldRight (which it really is). # But Python doesn't eliminate tail calls. It probably would have been fine # because "transforms" above is only 5 elements, but instead I will deal # with the local mutation and wish that I had a type system to reason with. # -rbe norm_branch = branch.lower() for transform in transforms: norm_branch = re.sub(transform[0], transform[1], norm_branch) # Ugh, here we break purity. Where is the option type when you need it? if not (norm_branch in collection_cache): print('Unknown collection specified: {0}'.format(branch)) return None return norm_branch # process_package :: PkgDB -> String -> String -> IO Bool def process_package(pkgdb, pkg, src, dest): print "*** Processing package: " + pkg data = pkgdb.get_package(pkg) pkg_list = data['packages'] maybe_source = [branch for branch in pkg_list if branch['collection']['branchname'] == src] maybe_dest = [branch for branch in pkg_list if branch['collection']['branchname'] == dest] if len(maybe_source) == 0: print "Source branch `" + src + "' not found. Please "\ "branch " + pkg + " manually." return False if len(maybe_dest) != 0: print "Package `" + pkg + "' was already branched for `" + dest + "'."\ " Not overwriting branch." return False if 'acls' not in maybe_source[0].keys(): print "No 'acls' key given to us by pkgdb. Cloning ACLs from a "\ "branch that has no ACLs due to retirement/orphan?" return False acls = [ acl for acl in maybe_source[0]['acls'] if acl['fas_name'] != 'group::provenpackager' ] for acl in acls: pkgdb.update_acl(pkg, dest, acl['acl'], acl['status'], acl['fas_name']) pkgdb.update_package_poc(pkg, dest, maybe_source[0]['point_of_contact']) return True # main :: [String] -> IO Unit def main(args): new_epel_requests = "epel" + NEW_EPEL_VERSION + "-requests" if len(args) < 1 or (len(args) < 3 and args[0] != new_epel_requests) or\ len(args) > 3 or (len(args) > 1 and args[0] == new_epel_requests): print "Usage: pkgdb2-clone " + new_epel_requests print " - OR -" print " pkgdb2-clone " sys.exit(1) pkgdb = pkgdb2client.PkgDB() username = raw_input('Username: ') password = getpass.getpass() pkgdb.login(username, password, True) collection_cache = generate_collection_cache(pkgdb) if args[0] == new_epel_requests: pkgs = parse_page(PAGE_URL) for key in pkgs: if is_in_rhel(key): continue src_branchname = normalize_branch(collection_cache, pkgs[key]) dest_branchname = normalize_branch(collection_cache, 'epel' + NEW_EPEL_VERSION) if not src_branchname or not dest_branchname: print "[" + key + "] Invalid source or destination branch "\ "name, " + src_branchname + " -> " + dest_branchname else: if process_package(pkgdb, key, src_branchname, dest_branchname): subprocess.call(["mkbranch", "-s", NEW_EPEL_SOURCE_BRANCH, "epel" + NEW_EPEL_VERSION, key]) print "[" + key + "] \033[1m\033[32mSUCCESS\033[0m" else: print "[" + key + "] \033[1m\033[31mERROR\033[0m" print "Done." else: src_branchname = normalize_branch(collection_cache, args[0]) dest_branchname = normalize_branch(collection_cache, args[1]) if not src_branchname or not dest_branchname: print "[" + key + "] Invalid source or destination branch "\ "name, " + src_branchname + " -> " + dest_branchname for pkg in args[2:]: if process_package(pkgdb, key, src_branchname, dest_branchname): print "[" + key + "] \033[1m\033[32mSUCCESS\033[0m" else: print "[" + key + "] \033[1m\033[31mERROR\033[0m" if __name__ == '__main__': main(sys.argv[1:])