<commit_before>from django.core.management.base import BaseCommand
import twisted.web.client
import twisted.internet
import mysite.customs.profile_importers
import logging

import mysite.profile.models

MAX_DIAS = 10

class Command(BaseCommand):
    help = "Call this when you want to run a Twisted reactor."

    def create_tasks_from_dias(self, max = 20):
        print 'For all DIAs we know how to process with Twisted: enqueue them.'
        enqueued_dias_count = 0

        for dia in mysite.profile.models.DataImportAttempt.objects.filter(completed=False):
            # If we have done too much work, just stop enqueuing DIAs.
            if enqueued_dias_count >= max:
                print 'Stopped after enqueuing', max, 'dias.'
                return
            if dia.source in mysite.customs.profile_importers.SOURCE_TO_CLASS:
                cls = mysite.customs.profile_importers.SOURCE_TO_CLASS[dia.source]
                self.add_dia_to_reactor(cls, dia.query, dia.id)
                enqueued_dias_count += 1

    def add_dia_to_reactor(self, cls, query, dia_id):
        ### For now, only the 'db' == Debian == qa.debian.org DIAs are in a format where
        ### they can handle asynchronous operation.
        state_manager = cls(query, dia_id, self)

        ### d is the "deferred" object. We create it using getPage(), and then
        ### we configure its next actions.
        urls_and_callbacks = state_manager.getUrlsAndCallbacks()
        for data_dict in urls_and_callbacks:
            self.call_getPage_on_data_dict(state_manager, data_dict)

    def call_getPage_on_data_dict(self, state_manager, data_dict):
        url = data_dict['url']
        callback = data_dict['callback']
        errback = data_dict.get('errback', None)
        
        logging.debug("Creating getPage for " + url)

        # First, actually create the Deferred.
        d = twisted.web.client.getPage(url)
        
        # Then, keep track of it.
        state_manager.urls_we_are_waiting_on[url] += 1
        self.running_deferreds += 1
        
        # wrap the callback
        wrapped_callback = mysite.customs.profile_importers.ImportActionWrapper(
            url=url,
            pi=state_manager,
            fn=callback)
        d.addCallback(wrapped_callback)

        if errback is not None:
            # wrap the errback
            wrapped_errback = mysite.customs.profile_importers.ImportActionWrapper(
                url=url,
                pi=state_manager,
                fn=errback)
            d.addErrback(wrapped_errback)

        d.addCallback(self.decrement_deferred_count_and_maybe_quit)

    def decrement_deferred_count_and_maybe_quit(self, *args):
        self.running_deferreds -= 1
        if self.running_deferreds == 0:
            self.stop_the_reactor()
        if self.running_deferreds < 0:
            raise ValueError("Uh, number of running deferreds went negative.")

    def stop_the_reactor(self, *args):
        if self.already_enqueued_stop_command:
            return
        else:
            self.already_enqueued_stop_command = True
            twisted.internet.reactor.callWhenRunning(lambda *args: twisted.internet.reactor.stop())

    def handle(self, use_reactor=True, *args, **options):
        self.running_deferreds = 0
        self.already_enqueued_stop_command = False

        print "Creating getPage()-based deferreds..."
        self.create_tasks_from_dias()
        if self.running_deferreds:
            print 'Starting Reactor...'
            assert use_reactor
            twisted.internet.reactor.run()
            print '...reactor finished!'<commit_msg>Never enqueue a dia we are currently working on.<commit_after>from django.core.management.base import BaseCommand
import twisted.web.client
import twisted.internet
import mysite.customs.profile_importers
import logging

import mysite.profile.models

MAX_DIAS = 10

class Command(BaseCommand):
    help = "Call this when you want to run a Twisted reactor."

    def create_tasks_from_dias(self, max = 20):
        print 'For all DIAs we know how to process with Twisted: enqueue them.'
        enqueued_dias_count = 0

        for dia in mysite.profile.models.DataImportAttempt.objects.filter(completed=False):
            # If for some reason this dia is one we are working on, skip past it.
            if dia.pk in self.active_dia_pks:
                continue

            # If we have done too much work, just stop enqueuing DIAs.
            if enqueued_dias_count >= max:
                print 'Stopped after enqueuing', max, 'dias.'
                return
            if dia.source in mysite.customs.profile_importers.SOURCE_TO_CLASS:
                self.active_dia_pks.add(dia.id)
                cls = mysite.customs.profile_importers.SOURCE_TO_CLASS[dia.source]
                self.add_dia_to_reactor(cls, dia.query, dia.id)
                enqueued_dias_count += 1

    def add_dia_to_reactor(self, cls, query, dia_id):
        ### For now, only the 'db' == Debian == qa.debian.org DIAs are in a format where
        ### they can handle asynchronous operation.
        state_manager = cls(query, dia_id, self)

        ### d is the "deferred" object. We create it using getPage(), and then
        ### we configure its next actions.
        urls_and_callbacks = state_manager.getUrlsAndCallbacks()
        for data_dict in urls_and_callbacks:
            self.call_getPage_on_data_dict(state_manager, data_dict)

    def call_getPage_on_data_dict(self, state_manager, data_dict):
        url = data_dict['url']
        callback = data_dict['callback']
        errback = data_dict.get('errback', None)
        
        logging.debug("Creating getPage for " + url)

        # First, actually create the Deferred.
        d = twisted.web.client.getPage(url)
        
        # Then, keep track of it.
        state_manager.urls_we_are_waiting_on[url] += 1
        self.running_deferreds += 1
        
        # wrap the callback
        wrapped_callback = mysite.customs.profile_importers.ImportActionWrapper(
            url=url,
            pi=state_manager,
            fn=callback)
        d.addCallback(wrapped_callback)

        if errback is not None:
            # wrap the errback
            wrapped_errback = mysite.customs.profile_importers.ImportActionWrapper(
                url=url,
                pi=state_manager,
                fn=errback)
            d.addErrback(wrapped_errback)

        d.addCallback(self.decrement_deferred_count_and_maybe_quit)

    def decrement_deferred_count_and_maybe_quit(self, *args):
        self.running_deferreds -= 1
        if self.running_deferreds == 0:
            self.stop_the_reactor()
        if self.running_deferreds < 0:
            raise ValueError("Uh, number of running deferreds went negative.")

    def stop_the_reactor(self, *args):
        if self.already_enqueued_stop_command:
            return
        else:
            self.already_enqueued_stop_command = True
            twisted.internet.reactor.callWhenRunning(lambda *args: twisted.internet.reactor.stop())

    def handle(self, use_reactor=True, *args, **options):
        self.running_deferreds = 0
        self.already_enqueued_stop_command = False
        self.active_dia_pks = set()

        print "Creating getPage()-based deferreds..."
        self.create_tasks_from_dias()
        if self.running_deferreds:
            print 'Starting Reactor...'
            assert use_reactor
            twisted.internet.reactor.run()
            print '...reactor finished!'
{'fraction_non_alphanumeric': 0.08025427095748908, 'fraction_numerical': 0.0031783869686134287, 'mean_word_length': 2.366919304502898, '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': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3865702', 'n_tokens_mistral': 2304, 'n_tokens_neox': 2196, 'n_words': 526}