<commit_before>from celery.task import task
from celery.execute import send_task

import collections
import contextlib
import datetime
import logging
import os
import pickle
import tempfile
import thread
import traceback

import pytz

from django.conf import settings

from humfrey.update.models import UpdateDefinition, UpdateLogRecord
from humfrey.update.transform.base import NotChanged, TransformException
from humfrey.update.utils import evaluate_pipeline

logger = logging.getLogger(__name__)

class _SameThreadFilter(logging.Filter):
    def __init__(self):
        self.thread_ident = thread.get_ident()
    def filter(self, record):
        return record.thread == self.thread_ident

class _TransformHandler(logging.Handler):
    ignore_loggers = frozenset(['django.db.backends'])

    def __init__(self, update_log):
        self.ignore = False
        self.update_log = update_log
        logging.Handler.__init__(self)
        self.setLevel(0)

    def emit(self, record):
        if self.ignore or record.name in self.ignore_loggers:
            return
        record = dict(record.__dict__)
        if record.get('exc_info'):
            exc_info = record['exc_info']
            record['exc_info'] = exc_info[:2] + (traceback.format_tb(exc_info[2]),)
        previous = self.update_log.log_level
        if not previous or record['levelno'] > previous:
            self.update_log.log_level = record['levelno']
        record['time'] = pytz.utc.localize(datetime.datetime.utcnow()).astimezone(pytz.timezone(settings.TIME_ZONE))

        try:
            pickle.dumps(record)
        except Exception:
            for key in record.keys():
                try:
                    pickle.dumps(record[key])
                except Exception:
                    del record[key]

        # Ignore all log messages while attempting to save.
        self.ignore = True
        try:
            update_log_record = UpdateLogRecord(update_log = self.update_log)
            update_log_record.record = record
            update_log_record.save()
        finally:
            self.ignore = False

class TransformManager(object):
    def __init__(self, update_log, output_directory, parameters, force, graphs_touched):
        self.update_log = update_log
        self.owner = update_log.update_definition.owner
        self.output_directory = output_directory
        self.parameters = parameters
        self.force = force

        self.counter = 0
        self.transforms = []
        self.graphs_touched = graphs_touched

    def __call__(self, extension=None, name=None):
        if not name:
            name = '%s.%s' % (self.counter, extension)
            self.counter += 1
        filename = os.path.join(self.output_directory, name)
        return filename

    def start(self, transform, inputs, type='generic'):
        self.current = {'transform': transform,
                        'inputs': inputs,
                        'start': datetime.datetime.now(),
                        'type': type}
    def end(self, outputs):
        self.current['end'] = datetime.datetime.now()
        self.current['outputs'] = outputs
        self.transforms.append(self.current)
        del self.current
    def touched_graph(self, store, graph_name):
        self.graphs_touched[store.slug].add(graph_name)
    def not_changed(self):
        if not self.force:
            raise NotChanged()<commit_msg>Bug fix for graphs_touched.<commit_after>from celery.task import task
from celery.execute import send_task

import collections
import contextlib
import datetime
import logging
import os
import pickle
import tempfile
import thread
import traceback

import pytz

from django.conf import settings

from humfrey.update.models import UpdateDefinition, UpdateLogRecord
from humfrey.update.transform.base import NotChanged, TransformException
from humfrey.update.utils import evaluate_pipeline

logger = logging.getLogger(__name__)

class _SameThreadFilter(logging.Filter):
    def __init__(self):
        self.thread_ident = thread.get_ident()
    def filter(self, record):
        return record.thread == self.thread_ident

class _TransformHandler(logging.Handler):
    ignore_loggers = frozenset(['django.db.backends'])

    def __init__(self, update_log):
        self.ignore = False
        self.update_log = update_log
        logging.Handler.__init__(self)
        self.setLevel(0)

    def emit(self, record):
        if self.ignore or record.name in self.ignore_loggers:
            return
        record = dict(record.__dict__)
        if record.get('exc_info'):
            exc_info = record['exc_info']
            record['exc_info'] = exc_info[:2] + (traceback.format_tb(exc_info[2]),)
        previous = self.update_log.log_level
        if not previous or record['levelno'] > previous:
            self.update_log.log_level = record['levelno']
        record['time'] = pytz.utc.localize(datetime.datetime.utcnow()).astimezone(pytz.timezone(settings.TIME_ZONE))

        try:
            pickle.dumps(record)
        except Exception:
            for key in record.keys():
                try:
                    pickle.dumps(record[key])
                except Exception:
                    del record[key]

        # Ignore all log messages while attempting to save.
        self.ignore = True
        try:
            update_log_record = UpdateLogRecord(update_log = self.update_log)
            update_log_record.record = record
            update_log_record.save()
        finally:
            self.ignore = False

class TransformManager(object):
    def __init__(self, update_log, output_directory, parameters, force, graphs_touched):
        self.update_log = update_log
        self.owner = update_log.update_definition.owner
        self.output_directory = output_directory
        self.parameters = parameters
        self.force = force

        self.counter = 0
        self.transforms = []
        self.graphs_touched = graphs_touched

    def __call__(self, extension=None, name=None):
        if not name:
            name = '%s.%s' % (self.counter, extension)
            self.counter += 1
        filename = os.path.join(self.output_directory, name)
        return filename

    def start(self, transform, inputs, type='generic'):
        self.current = {'transform': transform,
                        'inputs': inputs,
                        'start': datetime.datetime.now(),
                        'type': type}
    def end(self, outputs):
        self.current['end'] = datetime.datetime.now()
        self.current['outputs'] = outputs
        self.transforms.append(self.current)
        del self.current
    def touched_graph(self, store, graph_name):
        self.graphs_touched[store].add(graph_name)
    def not_changed(self):
        if not self.force:
            raise NotChanged()
{'fraction_non_alphanumeric': 0.09254649289793528, 'fraction_numerical': 0.001464343242055938, 'mean_word_length': 2.4811416921508664, 'pattern_counts': {'":': 0, '<': 3, '<?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': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4015174', 'n_tokens_mistral': 1891, 'n_tokens_neox': 1837, 'n_words': 434}