<commit_before>"""
API described here: http://apfmon.lancs.ac.uk/help
"""

import requests
import json
import time
import traceback

from pandaharvester.harvesterconfig import harvester_config
from pandaharvester.harvestercore import core_utils
from pandaharvester import panda_pkg_info
from pandaharvester.harvestermisc import generic_utils
from pandaharvester.harvestercore.work_spec import WorkSpec
from pandaharvester.harvestermisc.info_utils import PandaQueuesDict

_base_logger = core_utils.setup_logger('apfmon')
NO_CE = 'noCE'

def apfmon_active(method, *args, **kwargs):
    if cls.__active:
        method(*args, **kwargs)
    else:
        return


def clean_ce(ce):
    return ce.split('.')[0].split('://')[-1]


class Apfmon:

    def __init__(self, queue_config_mapper):

        try:
            self.__active = harvester_config.apfmon.active
        except:
            self.__active = False

        try:
            self.__worker_timeout = harvester_config.apfmon.worker_timeout
        except:
            self.__worker_timeout = 0.2

        try:
            self.__label_timeout = harvester_config.apfmon.worker_timeout
        except:
            self.__label_timeout = 1

        # TODO: make proper exception handling and defaults
        try:
            self.harvester_id = harvester_config.master.harvester_id
        except:
            self.harvester_id = 'DUMMY'

        try:
            self.base_url = harvester_config.apfmon.base_url
        except:
            self.base_url = 'http://apfmon.lancs.ac.uk/api'

        self.queue_config_mapper = queue_config_mapper

    def create_factory(self):
        """
        Creates or updates a harvester instance to APF Mon. Should be done at startup of the instance.
        """

        start_time = time.time()
        tmp_log = core_utils.make_logger(_base_logger, 'harvester_id={0}'.format(self.harvester_id),
                                         method_name='create_factory')

        if not self.__active:
            tmp_log.debug('APFMon reporting not enabled')
            return

        try:
            tmp_log.debug('start')

            url = '{0}/factories/{1}'.format(self.base_url, self.harvester_id)

            f = {'url': 'url_to_logs',
                 'email': '<EMAIL>',
                 'version': panda_pkg_info.release_version}
            payload = json.dumps(f)

            r = requests.put(url, data=payload, timeout=self.__label_timeout)
            tmp_log.debug('registration ended with {0} {1}'.format(r.status_code, r.text))
            end_time = time.time()
            tmp_log.debug('done (took {0})'.format(end_time - start_time))
        except:
            tmp_log.error('Excepted with: {0}'.format(traceback.format_exc()))

    def create_labels(self):
        """
        Creates or updates a collection of labels (=panda queue+CE)
        """
        start_time = time.time()
        tmp_log = core_utils.make_logger(_base_logger, 'harvester_id={0}'.format(self.harvester_id),
                                         method_name='create_labels')

        if not self.__active:
            tmp_log.debug('APFMon reporting not enabled')
            return

        try:
            tmp_log.debug('start')

            url = '{0}/labels'.format(self.base_url)

            # get the active queues from the config mapper
            all_sites = self.queue_config_mapper.get_active_queues().keys()
            panda_queues_dict = PandaQueuesDict()

            # publish the active queues to APF mon in shards
            for sites in generic_utils.create_shards(all_sites, 20):
                labels = []
                for site in sites:
                    try:
                        site_info = panda_queues_dict.get(site, dict())
                        if not site_info:
                            tmp_log.warning('No site info for {0}'.format(site))
                            continue

                        if not site_info['queues']:
                            # when no CEs associated to a queue, e.g. P1, HPCs, etc. Try to see if there is something
                            # in local configuration, otherwise set it to a dummy value
                            try:
                                ce = self.queue_config_mapper.queueConfig[site].submitter['ceEndpoint']
                            except:
                                ce = NO_CE
                            queues = [{'ce_endpoint': ce}]
                        else:
                            queues = site_info['queues']

                        for queue in queues:
                            try:
                                ce = clean_ce(queue['ce_endpoint'])
                            except:
                                ce = ''
                            labels.append({'name': '{0}-{1}'.format(site, ce),
                                           'wmsqueue': site,
                                           'ce_queue_id': queue['ce_queue_id'],
                                           'factory': self.harvester_id})
                    except:
                        tmp_log.error('Excepted for site {0} with: {1}'.format(site, traceback.format_exc()))
                        continue

                payload = json.dumps(labels)

                r = requests.put(url, data=payload, timeout=self.__label_timeout)
                tmp_log.debug('label creation for {0} ended with {1} {2}'.format(sites, r.status_code, r.text))

            end_time = time.time()
            tmp_log.debug('done (took {0})'.format(end_time - start_time))<commit_msg>apfmon: catching missing ce_queue_id exception<commit_after>"""
API described here: http://apfmon.lancs.ac.uk/help
"""

import requests
import json
import time
import traceback

from pandaharvester.harvesterconfig import harvester_config
from pandaharvester.harvestercore import core_utils
from pandaharvester import panda_pkg_info
from pandaharvester.harvestermisc import generic_utils
from pandaharvester.harvestercore.work_spec import WorkSpec
from pandaharvester.harvestermisc.info_utils import PandaQueuesDict

_base_logger = core_utils.setup_logger('apfmon')
NO_CE = 'noCE'

def apfmon_active(method, *args, **kwargs):
    if cls.__active:
        method(*args, **kwargs)
    else:
        return


def clean_ce(ce):
    return ce.split('.')[0].split('://')[-1]


class Apfmon:

    def __init__(self, queue_config_mapper):

        try:
            self.__active = harvester_config.apfmon.active
        except:
            self.__active = False

        try:
            self.__worker_timeout = harvester_config.apfmon.worker_timeout
        except:
            self.__worker_timeout = 0.2

        try:
            self.__label_timeout = harvester_config.apfmon.worker_timeout
        except:
            self.__label_timeout = 1

        # TODO: make proper exception handling and defaults
        try:
            self.harvester_id = harvester_config.master.harvester_id
        except:
            self.harvester_id = 'DUMMY'

        try:
            self.base_url = harvester_config.apfmon.base_url
        except:
            self.base_url = 'http://apfmon.lancs.ac.uk/api'

        self.queue_config_mapper = queue_config_mapper

    def create_factory(self):
        """
        Creates or updates a harvester instance to APF Mon. Should be done at startup of the instance.
        """

        start_time = time.time()
        tmp_log = core_utils.make_logger(_base_logger, 'harvester_id={0}'.format(self.harvester_id),
                                         method_name='create_factory')

        if not self.__active:
            tmp_log.debug('APFMon reporting not enabled')
            return

        try:
            tmp_log.debug('start')

            url = '{0}/factories/{1}'.format(self.base_url, self.harvester_id)

            f = {'url': 'url_to_logs',
                 'email': '<EMAIL>',
                 'version': panda_pkg_info.release_version}
            payload = json.dumps(f)

            r = requests.put(url, data=payload, timeout=self.__label_timeout)
            tmp_log.debug('registration ended with {0} {1}'.format(r.status_code, r.text))
            end_time = time.time()
            tmp_log.debug('done (took {0})'.format(end_time - start_time))
        except:
            tmp_log.error('Excepted with: {0}'.format(traceback.format_exc()))

    def create_labels(self):
        """
        Creates or updates a collection of labels (=panda queue+CE)
        """
        start_time = time.time()
        tmp_log = core_utils.make_logger(_base_logger, 'harvester_id={0}'.format(self.harvester_id),
                                         method_name='create_labels')

        if not self.__active:
            tmp_log.debug('APFMon reporting not enabled')
            return

        try:
            tmp_log.debug('start')

            url = '{0}/labels'.format(self.base_url)

            # get the active queues from the config mapper
            all_sites = self.queue_config_mapper.get_active_queues().keys()
            panda_queues_dict = PandaQueuesDict()

            # publish the active queues to APF mon in shards
            for sites in generic_utils.create_shards(all_sites, 20):
                labels = []
                for site in sites:
                    try:
                        site_info = panda_queues_dict.get(site, dict())
                        if not site_info:
                            tmp_log.warning('No site info for {0}'.format(site))
                            continue

                        if not site_info['queues']:
                            # when no CEs associated to a queue, e.g. P1, HPCs, etc. Try to see if there is something
                            # in local configuration, otherwise set it to a dummy value
                            try:
                                ce = self.queue_config_mapper.queueConfig[site].submitter['ceEndpoint']
                            except:
                                ce = NO_CE
                            queues = [{'ce_endpoint': ce}]
                        else:
                            queues = site_info['queues']

                        for queue in queues:
                            try:
                                ce = clean_ce(queue['ce_endpoint'])
                            except:
                                ce = ''

                            try:
                                ce_queue_id = queue['ce_queue_id']
                            except KeyError:
                                ce_queue_id = 0

                            labels.append({'name': '{0}-{1}'.format(site, ce),
                                           'wmsqueue': site,
                                           'ce_queue_id': ce_queue_id,
                                           'factory': self.harvester_id})
                    except:
                        tmp_log.error('Excepted for site {0} with: {1}'.format(site, traceback.format_exc()))
                        continue

                payload = json.dumps(labels)

                r = requests.put(url, data=payload, timeout=self.__label_timeout)
                tmp_log.debug('label creation for {0} ended with {1} {2}'.format(sites, r.status_code, r.text))

            end_time = time.time()
            tmp_log.debug('done (took {0})'.format(end_time - start_time))
{'fraction_non_alphanumeric': 0.09334735734946095, 'fraction_numerical': 0.004645455342273644, 'mean_word_length': 1.5826165685830693, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 28, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3209488', 'n_tokens_mistral': 3200, 'n_tokens_neox': 3036, 'n_words': 672}