84 lines
		
	
	
	
		
			2.2 KiB
		
	
	
	
		
			Python
		
	
	
		
			Executable file
		
	
	
	
	
			
		
		
	
	
			84 lines
		
	
	
	
		
			2.2 KiB
		
	
	
	
		
			Python
		
	
	
		
			Executable file
		
	
	
	
	
#!/usr/bin/python3
 | 
						|
 | 
						|
# SPDX-FileCopyrightText: 2021 Luca Beltrame <lbeltrame@kde.org>
 | 
						|
# SPDX-License-Identifier: BSD-3-Clause
 | 
						|
 | 
						|
import argparse
 | 
						|
import logging
 | 
						|
 | 
						|
import keyring
 | 
						|
import keyring.backend
 | 
						|
from keyrings.alt.file import PlaintextKeyring
 | 
						|
import lxml.etree as etree
 | 
						|
import requests
 | 
						|
 | 
						|
 | 
						|
keyring.set_keyring(PlaintextKeyring())
 | 
						|
 | 
						|
API_URL = "https://api.opensuse.org/build"
 | 
						|
OBS_URL = "https://api.opensuse.org/trigger/runservice"
 | 
						|
REPOSITORIES = [
 | 
						|
    "KDE:Unstable:Frameworks",
 | 
						|
    "KDE:Unstable:Applications",
 | 
						|
    "KDE:Unstable:Extra"
 | 
						|
    ]
 | 
						|
 | 
						|
 | 
						|
def trigger_update(repository, package_name, token):
 | 
						|
 | 
						|
    header = {"Authorization": f"Token {token}"}
 | 
						|
    parameters = {"project": repository, "package": package_name}
 | 
						|
 | 
						|
    logging.info("Updating package %s", package_name)
 | 
						|
    result = requests.post(OBS_URL, params=parameters, headers=header)
 | 
						|
 | 
						|
    if not result:
 | 
						|
        logging.error("Error during service run, package %s, error code %s",
 | 
						|
                      package_name, result.status_code, result.url)
 | 
						|
        return False
 | 
						|
 | 
						|
    logging.debug("Package %s complete", package_name)
 | 
						|
    return result
 | 
						|
 | 
						|
 | 
						|
def check_status(project, username):
 | 
						|
 | 
						|
    password = keyring.get_password("obs", username)
 | 
						|
    token = keyring.get_password("obstoken", "token")
 | 
						|
    url = "{0}/{1}/_result".format(API_URL, project)
 | 
						|
    result = requests.get(url, auth=(username, password))
 | 
						|
 | 
						|
    logging.debug("Checking broken packages in %s", project)
 | 
						|
 | 
						|
    if not result:
 | 
						|
        return
 | 
						|
 | 
						|
    contents = etree.fromstring(result.content)
 | 
						|
    broken_packages = set(contents.xpath(
 | 
						|
        '//result[@arch="x86_64"]/status[@code="broken"]/@package'))
 | 
						|
 | 
						|
    if not broken_packages:
 | 
						|
        logging.debug("No broken packages found in %s", project)
 | 
						|
        return
 | 
						|
 | 
						|
    logging.info("Found %s broken packages", len(broken_packages))
 | 
						|
    for package in broken_packages:
 | 
						|
        logging.info("Triggering package %s", package)
 | 
						|
        trigger_update(project, package, token)
 | 
						|
 | 
						|
 | 
						|
def main():
 | 
						|
 | 
						|
    parser = argparse.ArgumentParser()
 | 
						|
    parser.add_argument("username", help="OBS username")
 | 
						|
 | 
						|
    options = parser.parse_args()
 | 
						|
    logging.basicConfig(format='%(levelname)s - %(message)s',
 | 
						|
                        level=logging.INFO)
 | 
						|
 | 
						|
    for project in REPOSITORIES:
 | 
						|
        check_status(project, options.username)
 | 
						|
 | 
						|
 | 
						|
if __name__ == "__main__":
 | 
						|
    main()
 |