<commit_before>import datetime
import glob
import json
import logging
import os
import re
import shutil
import socket
import subprocess
import sys
import work_queue as wq

from collections import defaultdict, Counter
from hashlib import sha1

from lobster import fs, util
from lobster.cmssw import dash
from lobster.core import unit
from lobster.core import Algo
from lobster.core import MergeTaskHandler

from WMCore.Storage.SiteLocalConfig import loadSiteLocalConfig, SiteConfigError

logger = logging.getLogger('lobster.source')


class ReleaseSummary(object):

    """Summary of returned tasks.

    Prints a user-friendly summary of which tasks returned with what exit code/status.
    """

    flags = {
        wq.WORK_QUEUE_RESULT_INPUT_MISSING: "missing input",                # 1
        wq.WORK_QUEUE_RESULT_OUTPUT_MISSING: "missing output",              # 2
        wq.WORK_QUEUE_RESULT_STDOUT_MISSING: "no stdout",                   # 4
        wq.WORK_QUEUE_RESULT_SIGNAL: "signal received",                     # 8
        wq.WORK_QUEUE_RESULT_RESOURCE_EXHAUSTION: "exhausted resources",    # 16
        wq.WORK_QUEUE_RESULT_TASK_TIMEOUT: "time out",                      # 32
        wq.WORK_QUEUE_RESULT_UNKNOWN: "unclassified error",                 # 64
        wq.WORK_QUEUE_RESULT_FORSAKEN: "unrelated error",                   # 128
        wq.WORK_QUEUE_RESULT_MAX_RETRIES: "exceed # retries",               # 256
        wq.WORK_QUEUE_RESULT_TASK_MAX_RUN_TIME: "exceeded runtime"          # 512
    }

    def __init__(self):
        self.__exe = {}
        self.__wq = {}
        self.__taskdirs = {}
        self.__monitors = []

    def exe(self, status, taskid):
        try:
            self.__exe[status].append(taskid)
        except KeyError:
            self.__exe[status] = [taskid]

    def wq(self, status, taskid):
        for flag in ReleaseSummary.flags.keys():
            if status == flag:
                try:
                    self.__wq[flag].append(taskid)
                except KeyError:
                    self.__wq[flag] = [taskid]

    def dir(self, taskid, taskdir):
        self.__taskdirs[taskid] = taskdir

    def monitor(self, taskid):
        self.__monitors.append(taskid)

    def __str__(self):
        s = "received the following task(s):\n"
        for status in sorted(self.__exe.keys()):
            s += "returned with status {0}: {1}\n".format(status, ", ".join(self.__exe[status]))
            if status != 0:
                s += "parameters and logs in:\n\t{0}\n".format(
                    "\n\t".join([self.__taskdirs[t] for t in self.__exe[status]]))
        for flag in sorted(self.__wq.keys()):
            s += "failed due to {0}: {1}\nparameters and logs in:\n\t{2}\n".format(
                ReleaseSummary.flags[flag],
                ", ".join(self.__wq[flag]),
                "\n\t".join([self.__taskdirs[t] for t in self.__wq[flag]]))
        if self.__monitors:
            s += "resource monitoring unavailable for the following tasks: {0}\n".format(", ".join(self.__monitors))
        # Trim final newline
        return s[:-1]


class TaskProvider(util.Timing):

    def __init__(self, config):
        util.Timing.__init__(self, 'dash', 'handler', 'updates', 'elk', 'transfers', 'cleanup', 'propagate', 'sqlite')

        self.config = config
        self.basedirs = [config.base_directory, config.startup_directory]
        self.workdir = config.workdir
        self._storage = config.storage
        self.statusfile = os.path.join(self.workdir, 'status.json')
        self.siteconf = os.path.join(self.workdir, 'siteconf')

        self.parrot_path = os.path.dirname(util.which('parrot_run'))
        self.parrot_bin = os.path.join(self.workdir, 'bin')
        self.parrot_lib = os.path.join(self.workdir, 'lib')

        self.__algo = Algo(config)
        self.__host = socket.getfqdn()
        try:
            siteconf = loadSiteLocalConfig()
            self.__ce = siteconf.siteName
            self.__se = siteconf.localStageOutPNN()
            self.__frontier_proxy = siteconf.frontierProxies[0]
        except SiteConfigError:
            logger.error("can't load siteconfig, defaulting to hostname")
            self.__ce = socket.getfqdn()
            self.__se = socket.getfqdn()
            try:
                self.__frontier_proxy = os.environ['HTTP_PROXY']
            except KeyError:
                logger.error("can't determine proxy for Frontier via $HTTP_PROXY")
                sys.exit(1)

        try:
            with open('/etc/cvmfs/default.local') as f:
                lines = f.readlines()
        except:
            lines = []
        for l in lines:
            m = re.match('\s*CVMFS_HTTP_PROXY\s*=\s*[\'"]?(.*)[\'"]?', l)
            if m:
                self.__cvmfs_proxy = m.group(1)<commit_msg>Update source.py<commit_after>import datetime
import glob
import json
import logging
import os
import re
import shutil
import socket
import subprocess
import sys
import work_queue as wq

from collections import defaultdict, Counter
from hashlib import sha1

from lobster import fs, util
from lobster.cmssw import dash
from lobster.core import unit
from lobster.core import Algo
from lobster.core import MergeTaskHandler

from WMCore.Storage.SiteLocalConfig import loadSiteLocalConfig, SiteConfigError

logger = logging.getLogger('lobster.source')


class ReleaseSummary(object):

    """Summary of returned tasks.

    Prints a user-friendly summary of which tasks returned with what exit code/status.
    """

    flags = {
        wq.WORK_QUEUE_RESULT_INPUT_MISSING: "missing input",                # 1
        wq.WORK_QUEUE_RESULT_OUTPUT_MISSING: "missing output",              # 2
        wq.WORK_QUEUE_RESULT_STDOUT_MISSING: "no stdout",                   # 4
        wq.WORK_QUEUE_RESULT_SIGNAL: "signal received",                     # 8
        wq.WORK_QUEUE_RESULT_RESOURCE_EXHAUSTION: "exhausted resources",    # 16
        wq.WORK_QUEUE_RESULT_TASK_TIMEOUT: "time out",                      # 32
        wq.WORK_QUEUE_RESULT_UNKNOWN: "unclassified error",                 # 64
        wq.WORK_QUEUE_RESULT_FORSAKEN: "unrelated error",                   # 128
        wq.WORK_QUEUE_RESULT_MAX_RETRIES: "exceed # retries",               # 256
        wq.WORK_QUEUE_RESULT_TASK_MAX_RUN_TIME: "exceeded runtime"          # 512
    }

    def __init__(self):
        self.__exe = {}
        self.__wq = {}
        self.__taskdirs = {}
        self.__monitors = []

    def exe(self, status, taskid):
        try:
            self.__exe[status].append(taskid)
        except KeyError:
            self.__exe[status] = [taskid]

    def wq(self, status, taskid):
        for flag in ReleaseSummary.flags.keys():
            if status == flag:
                try:
                    self.__wq[flag].append(taskid)
                except KeyError:
                    self.__wq[flag] = [taskid]

    def dir(self, taskid, taskdir):
        self.__taskdirs[taskid] = taskdir

    def monitor(self, taskid):
        self.__monitors.append(taskid)

    def __str__(self):
        s = "received the following task(s):\n"
        for status in sorted(self.__exe.keys()):
            s += "returned with status {0}: {1}\n".format(status, ", ".join(self.__exe[status]))
            if status != 0:
                s += "parameters and logs in:\n\t{0}\n".format(
                    "\n\t".join([self.__taskdirs[t] for t in self.__exe[status]]))
        for flag in sorted(self.__wq.keys()):
            s += "failed due to {0}: {1}\nparameters and logs in:\n\t{2}\n".format(
                ReleaseSummary.flags[flag],
                ", ".join(self.__wq[flag]),
                "\n\t".join([self.__taskdirs[t] for t in self.__wq[flag]]))
        if self.__monitors:
            s += "resource monitoring unavailable for the following tasks: {0}\n".format(", ".join(self.__monitors))
        # Trim final newline
        return s[:-1]


class TaskProvider(util.Timing):

    def __init__(self, config):
        util.Timing.__init__(self, 'dash', 'handler', 'updates', 'elk', 'transfers', 'cleanup', 'propagate', 'sqlite')

        self.config = config
        self.basedirs = [config.base_directory, config.startup_directory]
        self.workdir = config.workdir
        self._storage = config.storage
        self.statusfile = os.path.join(self.workdir, 'status.json')
        self.siteconf = os.path.join(self.workdir, 'siteconf')

        self.parrot_path = os.path.dirname(util.which('parrot_run'))
        self.parrot_bin = os.path.join(self.workdir, 'bin')
        self.parrot_lib = os.path.join(self.workdir, 'lib')

        self.__algo = Algo(config)
        self.__host = socket.getfqdn()
        try:
            siteconf = loadSiteLocalConfig()
            self.__ce = siteconf.siteName
            self.__se = siteconf.localStageOutPNN()
            self.__frontier_proxy = siteconf.frontierProxies[0]
        except (SiteConfigError, IndexError):
            logger.error("can't load siteconfig, defaulting to hostname")
            self.__ce = socket.getfqdn()
            self.__se = socket.getfqdn()
            try:
                self.__frontier_proxy = os.environ['HTTP_PROXY']
            except KeyError:
                logger.error("can't determine proxy for Frontier via $HTTP_PROXY")
                sys.exit(1)

        try:
            with open('/etc/cvmfs/default.local') as f:
                lines = f.readlines()
        except:
            lines = []
        for l in lines:
            m = re.match('\s*CVMFS_HTTP_PROXY\s*=\s*[\'"]?(.*)[\'"]?', l)
            if m:
                self.__cvmfs_proxy = m.group(1)
{'fraction_non_alphanumeric': 0.11390180878552972, 'fraction_numerical': 0.006614987080103359, 'mean_word_length': 2.113256113256113, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1422165', 'n_tokens_mistral': 3026, 'n_tokens_neox': 2906, 'n_words': 653}