<commit_before>"""
Connection to the PanDA server

"""
import ssl
try:
    # disable SNI for TLSV1_UNRECOGNIZED_NAME before importing requests
    ssl.HAS_SNI = False
except:
    pass
import sys
import copy
import json
import inspect
import datetime
import requests
import traceback
# TO BE REMOVED for python2.7
import requests.packages.urllib3
try:
    requests.packages.urllib3.disable_warnings()
except:
    pass
import core_utils
from pandaharvester.harvesterconfig import harvester_config

# logger
_logger = core_utils.setup_logger()


# connection class
class Communicator:
    # constructor
    def __init__(self):
        if hasattr(harvester_config.pandacon, 'verbose') and harvester_config.pandacon.verbose:
            self.verbose = True
        else:
            self.verbose = False

    # POST with http
    def post(self, path, data):
        try:
            tmpLog = None
            if self.verbose:
                tmpLog = core_utils.make_logger(_logger)
                tmpExec = inspect.stack()[1][3]
            url = '{0}/{1}'.format(harvester_config.pandacon.pandaURL, path)
            if self.verbose:
                tmpLog.debug('exec={0} URL={1} data={2}'.format(tmpExec, url, str(data)))
            res = requests.post(url,
                                data=data,
                                headers={"Accept": "application/json"},
                                timeout=harvester_config.pandacon.timeout)
            if self.verbose:
                tmpLog.debug('exec={0} code={1} return={2}'.format(tmpExec, res.status_code, res.text))
            if res.status_code == 200:
                return True, res
            else:
                errMsg = 'StatusCode={0} {1}'.format(res.status_code,
                                                     res.text)
        except:
            errType, errValue = sys.exc_info()[:2]
            errMsg = "failed to post with {0}:{1} ".format(errType, errValue)
            errMsg += traceback.format_exc()
        return False, errMsg

    # POST with https
    def post_ssl(self, path, data, cert=None):
        try:
            tmpLog = None
            if self.verbose:
                tmpLog = core_utils.make_logger(_logger)
                tmpExec = inspect.stack()[1][3]
            url = '{0}/{1}'.format(harvester_config.pandacon.pandaURLSSL, path)
            if self.verbose:
                tmpLog.debug('exec={0} URL={1} data={2}'.format(tmpExec, url, str(data)))
            if cert is None:
                cert = (harvester_config.pandacon.cert_file,
                        harvester_config.pandacon.key_file)
            res = requests.post(url,
                                data=data,
                                headers={"Accept": "application/json"},
                                timeout=harvester_config.pandacon.timeout,
                                verify=harvester_config.pandacon.ca_cert,
                                cert=cert)
            if self.verbose:
                tmpLog.debug('exec={0} code={1} return={2}'.format(tmpExec, res.status_code, res.text))
            if res.status_code == 200:
                return True, res
            else:
                errMsg = 'StatusCode={0} {1}'.format(res.status_code,
                                                     res.text)
        except:
            errType, errValue = sys.exc_info()[:2]
            errMsg = "failed to post with {0}:{1} ".format(errType, errValue)
            errMsg += traceback.format_exc()
        return False, errMsg

    # check server
    def check_panda(self):
        tmpStat, tmpRes = self.post_ssl('isAlive', {})
        if tmpStat :
            return tmpStat, tmpRes.status_code, tmpRes.text
        else:
            return tmpStat, tmpRes

    # get jobs
    def get_jobs(self, site_name, node_name, prod_source_label, computing_element, n_jobs,
                 additional_criteria):
        # get logger
        tmpLog = core_utils.make_logger(_logger, 'siteName={0}'.format(site_name))
        tmpLog.debug('try to get {0} jobs'.format(n_jobs))
        data = {}
        data['siteName'] = site_name
        data['node'] = node_name
        data['prodSourceLabel'] = prod_source_label
        data['computingElement'] = computing_element
        data['nJobs'] = n_jobs
        if additional_criteria is not None:
            for tmpKey, tmpVal in additional_criteria:
                data[tmpKey] = tmpVal
        tmpStat, tmpRes = self.post_ssl('getJob', data)
        errStr = 'OK'
        if tmpStat is False:
            errStr = core_utils.dump_error_message(tmpLog, tmpRes)
        else:
            try:
                tmpDict = tmpRes.json()
                tmpLog.debug('StatusCode={0}'.format(tmpDict['StatusCode']))
                if tmpDict['StatusCode'] == 0:
                    tmpLog.debug('got {0} jobs'.format(len(tmpDict['jobs'])))
                    return tmpDict['jobs'], errStr
                else:
                    if 'errorDialog' in tmpDict:
                        errStr = tmpDict['errorDialog']
                    else:
                        errStr = "StatusCode={0}".format(tmpDict['StatusCode'])
                return [], errStr
            except:
                errStr = core_utils.dump_error_message(tmpLog, tmpRes)
        return [], errStr

    # update jobs TOBEFIXED to use bulk method
    def update_jobs(self, jobspec_list):
        retList = []
        for jobSpec in jobspec_list:
            tmpLog = core_utils.make_logger(_logger, 'PandaID={0}'.format(jobSpec.PandaID))
            tmpLog.debug('start')
            # update events
            eventRanges, eventSpecs = jobSpec.to_event_data()
            if eventRanges != []:
                tmpRet = self.update_event_ranges(eventRanges, tmpLog)
                if tmpRet['StatusCode'] == 0:
                    for eventSpec, retVal in zip(eventSpecs, tmpRet['Returns']):
                        if retVal in [True, False] and eventSpec.is_final_status():
                            eventSpec.subStatus = 'done'
            # update job
            data = jobSpec.get_job_attributes_for_panda()
            data['jobId'] = jobSpec.PandaID
            data['state'] = jobSpec.get_status()
            data['attemptNr'] = jobSpec.attemptNr
            data['jobSubStatus'] = jobSpec.subStatus
            if jobSpec.startTime is not None:
                data['startTime'] = jobSpec.startTime.strftime('%Y-%m-%d %H:%M:%S')
            if jobSpec.endTime is not None:
                data['endTime'] = jobSpec.endTime.strftime('%Y-%m-%d %H:%M:%S')
            if jobSpec.nCore is not None:
                data['coreCount'] = jobSpec.nCore
            if jobSpec.is_final_status() and jobSpec.status == jobSpec.get_status():
                if jobSpec.metaData is not None:
                    data['metaData'] = json.dumps(jobSpec.metaData)
                if jobSpec.outputFilesToReport is not None:
                    data['xml'] = jobSpec.outputFilesToReport
            tmpLog.debug('data={0}'.format(str(data)))
            tmpStat, tmpRes = self.post_ssl('updateJob', data)
            retMap = None
            errStr = ''
            if tmpStat is False:
                errStr = core_utils.dump_error_message(tmpLog, tmpRes)
            else:
                try:
                    retMap = tmpRes.json()<commit_msg>Reporting jobStatus=failed and jobSubStatus=cancelled to panda when jobs are cancelled<commit_after>"""
Connection to the PanDA server

"""
import ssl
try:
    # disable SNI for TLSV1_UNRECOGNIZED_NAME before importing requests
    ssl.HAS_SNI = False
except:
    pass
import sys
import copy
import json
import inspect
import datetime
import requests
import traceback
# TO BE REMOVED for python2.7
import requests.packages.urllib3
try:
    requests.packages.urllib3.disable_warnings()
except:
    pass
import core_utils
from pandaharvester.harvesterconfig import harvester_config

# logger
_logger = core_utils.setup_logger()


# connection class
class Communicator:
    # constructor
    def __init__(self):
        if hasattr(harvester_config.pandacon, 'verbose') and harvester_config.pandacon.verbose:
            self.verbose = True
        else:
            self.verbose = False

    # POST with http
    def post(self, path, data):
        try:
            tmpLog = None
            if self.verbose:
                tmpLog = core_utils.make_logger(_logger)
                tmpExec = inspect.stack()[1][3]
            url = '{0}/{1}'.format(harvester_config.pandacon.pandaURL, path)
            if self.verbose:
                tmpLog.debug('exec={0} URL={1} data={2}'.format(tmpExec, url, str(data)))
            res = requests.post(url,
                                data=data,
                                headers={"Accept": "application/json"},
                                timeout=harvester_config.pandacon.timeout)
            if self.verbose:
                tmpLog.debug('exec={0} code={1} return={2}'.format(tmpExec, res.status_code, res.text))
            if res.status_code == 200:
                return True, res
            else:
                errMsg = 'StatusCode={0} {1}'.format(res.status_code,
                                                     res.text)
        except:
            errType, errValue = sys.exc_info()[:2]
            errMsg = "failed to post with {0}:{1} ".format(errType, errValue)
            errMsg += traceback.format_exc()
        return False, errMsg

    # POST with https
    def post_ssl(self, path, data, cert=None):
        try:
            tmpLog = None
            if self.verbose:
                tmpLog = core_utils.make_logger(_logger)
                tmpExec = inspect.stack()[1][3]
            url = '{0}/{1}'.format(harvester_config.pandacon.pandaURLSSL, path)
            if self.verbose:
                tmpLog.debug('exec={0} URL={1} data={2}'.format(tmpExec, url, str(data)))
            if cert is None:
                cert = (harvester_config.pandacon.cert_file,
                        harvester_config.pandacon.key_file)
            res = requests.post(url,
                                data=data,
                                headers={"Accept": "application/json"},
                                timeout=harvester_config.pandacon.timeout,
                                verify=harvester_config.pandacon.ca_cert,
                                cert=cert)
            if self.verbose:
                tmpLog.debug('exec={0} code={1} return={2}'.format(tmpExec, res.status_code, res.text))
            if res.status_code == 200:
                return True, res
            else:
                errMsg = 'StatusCode={0} {1}'.format(res.status_code,
                                                     res.text)
        except:
            errType, errValue = sys.exc_info()[:2]
            errMsg = "failed to post with {0}:{1} ".format(errType, errValue)
            errMsg += traceback.format_exc()
        return False, errMsg

    # check server
    def check_panda(self):
        tmpStat, tmpRes = self.post_ssl('isAlive', {})
        if tmpStat :
            return tmpStat, tmpRes.status_code, tmpRes.text
        else:
            return tmpStat, tmpRes

    # get jobs
    def get_jobs(self, site_name, node_name, prod_source_label, computing_element, n_jobs,
                 additional_criteria):
        # get logger
        tmpLog = core_utils.make_logger(_logger, 'siteName={0}'.format(site_name))
        tmpLog.debug('try to get {0} jobs'.format(n_jobs))
        data = {}
        data['siteName'] = site_name
        data['node'] = node_name
        data['prodSourceLabel'] = prod_source_label
        data['computingElement'] = computing_element
        data['nJobs'] = n_jobs
        if additional_criteria is not None:
            for tmpKey, tmpVal in additional_criteria:
                data[tmpKey] = tmpVal
        tmpStat, tmpRes = self.post_ssl('getJob', data)
        errStr = 'OK'
        if tmpStat is False:
            errStr = core_utils.dump_error_message(tmpLog, tmpRes)
        else:
            try:
                tmpDict = tmpRes.json()
                tmpLog.debug('StatusCode={0}'.format(tmpDict['StatusCode']))
                if tmpDict['StatusCode'] == 0:
                    tmpLog.debug('got {0} jobs'.format(len(tmpDict['jobs'])))
                    return tmpDict['jobs'], errStr
                else:
                    if 'errorDialog' in tmpDict:
                        errStr = tmpDict['errorDialog']
                    else:
                        errStr = "StatusCode={0}".format(tmpDict['StatusCode'])
                return [], errStr
            except:
                errStr = core_utils.dump_error_message(tmpLog, tmpRes)
        return [], errStr

    # update jobs TOBEFIXED to use bulk method
    def update_jobs(self, jobspec_list):
        retList = []
        for jobSpec in jobspec_list:
            tmpLog = core_utils.make_logger(_logger, 'PandaID={0}'.format(jobSpec.PandaID))
            tmpLog.debug('start')
            # update events
            eventRanges, eventSpecs = jobSpec.to_event_data()
            if eventRanges != []:
                tmpRet = self.update_event_ranges(eventRanges, tmpLog)
                if tmpRet['StatusCode'] == 0:
                    for eventSpec, retVal in zip(eventSpecs, tmpRet['Returns']):
                        if retVal in [True, False] and eventSpec.is_final_status():
                            eventSpec.subStatus = 'done'
            # update job
            data = jobSpec.get_job_attributes_for_panda()
            data['jobId'] = jobSpec.PandaID
            data['state'] = jobSpec.get_status()
            data['attemptNr'] = jobSpec.attemptNr
            data['jobSubStatus'] = jobSpec.subStatus
            # change cancelled to failed to be accepted by panda server
            if data['state'] == 'cancelled':
                data['jobSubStatus'] = data['state']
                data['state'] = 'failed'
            if jobSpec.startTime is not None:
                data['startTime'] = jobSpec.startTime.strftime('%Y-%m-%d %H:%M:%S')
            if jobSpec.endTime is not None:
                data['endTime'] = jobSpec.endTime.strftime('%Y-%m-%d %H:%M:%S')
            if jobSpec.nCore is not None:
                data['coreCount'] = jobSpec.nCore
            if jobSpec.is_final_status() and jobSpec.status == jobSpec.get_status():
                if jobSpec.metaData is not None:
                    data['metaData'] = json.dumps(jobSpec.metaData)
                if jobSpec.outputFilesToReport is not None:
                    data['xml'] = jobSpec.outputFilesToReport
            tmpLog.debug('data={0}'.format(str(data)))
            tmpStat, tmpRes = self.post_ssl('updateJob', data)
            retMap = None
            errStr = ''
            if tmpStat is False:
                errStr = core_utils.dump_error_message(tmpLog, tmpRes)
            else:
                try:
                    retMap = tmpRes.json()
{'fraction_non_alphanumeric': 0.10544534682465717, 'fraction_numerical': 0.006656903208627347, 'mean_word_length': 1.7339399454049136, 'pattern_counts': {'":': 4, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 2}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 8, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3390782', 'n_tokens_mistral': 4313, 'n_tokens_neox': 4148, 'n_words': 905}