Exports the last successful build log for all Jenkins project types

jenkins_export_last_successful_build_log.py

#!/usr/bin/env python3

import code

import os, sys
from re import sub
import argparse
from typing import Any, AnyStr, Dict
import constant
import validators
from sys import stderr, platform
import requests
from requests import Response
from requests.auth import HTTPBasicAuth
from lxml import etree
import logging

logging.basicConfig(level=logging.DEBUG, format='%(asctime)s %(levelname)s %(message)s', datefmt='%H:%M:%S')


def _get(username: AnyStr, password: AnyStr, url: AnyStr, params: Dict, stream: bool):
    """
    :param username:
    :param password:
    :param url:
    :return:
    """

    # Store the initial URL, because if user supplied
    # https scheme we are not going to attempt an
    # unsecure connection if secure connection raises
    # an exception
    initial_url = url

    # removing scheme because domain validator will
    # evaluate to False
    SCHEME_REGEX = '^http([s]?):\/\/'
    url = sub(SCHEME_REGEX, '', url, count=1)

    # If domain is invalid don't bother with HTTP connection
    if not validators.domain(url) and not stream:
        logging.debug(f'{url} failed domain validation')
        return False

    # re-adding scheme for the requests library or else
    # it will raise an Exception
    url = ''.join(('https://', url))

    response = None
    try:
        logging.debug(f'Attempting SSL Connection to {url}')
        response = requests.get(url=url, params=params,
                                auth=HTTPBasicAuth(username, password), stream=stream, verify=True)
    except requests.exceptions.SSLError as e:
        logging.warning(f'SSL Connection to {url} raised exception {e}')
        # If SSLError attempt an http request only if
        # user did not supply a secure scheme. We
        # negate an equality because find returns index
        # of param.  If the initial url starts with https
        # then index is zero and we negate.  But if we don't
        # check for 0 we could be negating a index of greater
        # than 0 which will evaluate incorrectly
        if not initial_url.find('https://') == 0:
            logging.debug(f'Attempting unsecure connection to {url}')
            response = requests.get(url=initial_url, params=params, stream=stream,
                                    auth=HTTPBasicAuth(username, password))
        else:
            logging.error(f'Could not connect to {url}')
            raise requests.exceptions.SSLError

    return response


def _is_connectivity_valid(username: AnyStr, password: AnyStr, url: AnyStr, params: Dict, stream: bool) -> bool:
    response = _get(username, password, url, params=params, stream=stream)

    return response.status_code == requests.codes.ok


def _is_input_length_valid(arguments: Any, size: int) -> bool:
    logging.debug(f'Validating argument length is correct size {str(len(arguments))} {str(size)}')
    return len(arguments) == size


def _is_input_valid(arguments: Any) -> bool:
    logging.debug(f'Starting input validation')
    REQUIRED_ARG_COUNT: int = 3
    # Parameters to endpoint with 2 character response body
    # for light weight verification
    VALIDATION_PARAMS = {'tree': 'job'}
    return _is_input_length_valid(arguments, REQUIRED_ARG_COUNT) and \
           _is_connectivity_valid(username=arguments[0], password=arguments[1], url=arguments[2],
                                  params=VALIDATION_PARAMS, stream=False)


def main(arguments) -> int:
    if not _is_input_valid(arguments):
        return os.EX_SOFTWARE if platform == 'linux' else 70

    username: AnyStr = arguments[0]
    password: AnyStr = arguments[1]
    host: AnyStr = arguments[2]
    # file = "arguments[3]" if len(arguments[3]) else 'log.txt'
    file = 'log.txt'
    
    logging.debug(f'Logging output to {file}')

    api_path = '/jenkins/api/xml'
    url = host + api_path
    params = {'tree': 'jobs[url]', 'exclude': '//*/build/result[text()!="SUCCESS"]/parent::build'}
    last_successful_build_path = 'lastSuccessfulBuild/consoleText'
    response = _get(username=username, password=password, url=url, params=params, stream=False)
    # response = requests.get(url=base, params=params, stream=True, auth=auth)

    console_text = open(file, "a+")  # append mode

    # If endpoint returns anything other than 200
    # we can return and exit
    if response.status_code != requests.codes.ok:
        print(response.status_code, stderr)
        # change to try/catch
        return os.EX_SOFTWARE if platform == 'linux' else -1

    #response.raw.decode_content = True

    #root = etree.fromstring(response.text)
    root = etree.XML(response.content)
    for element in root.getchildren():
        child = element.getchildren()[0]
        url = child.text+last_successful_build_path
        response = _get(username=username, password=password, url=url, params=params, stream=True)
        try:
            console_text.write(response.text)
        except Exception as e:
            logging.error(f'Failed to write console to log for {url} due to {e}')

    console_text.close()

    return os.EX_OK if platform == 'linux' else 0


if __name__ == '__main__':
    sys.exit(main(sys.argv[1:]))