<commit_before>import time, datetime, logging, threading, simplejson, copy, couchdb
from totalimpact import default_settings
from totalimpact.models import Item, ItemFactory
from totalimpact.pidsupport import StoppableThread
from totalimpact.providers.provider import ProviderFactory
import default_settings

log = logging.getLogger("ti.queue")


# some data useful for testing
# d = {"doi" : ["10.1371/journal.pcbi.1000361", "10.1016/j.meegid.2011.02.004"], "url" : ["http://cottagelabs.com"]}


class QueueMonitor(StoppableThread):
    """ Worker to watch couch for items that need aliases (new or update)
        and place them onto the aliases in-memory queue. 
    """

    def __init__(self, dao):
        self.dao = dao
        StoppableThread.__init__(self)

    def run(self, runonce=False):
        """ runonce is for the test suite """

        while not self.stopped():
            viewname = 'queues/needs_aliases'
            res = self.dao.view(viewname)

            try:
                rows = res.rows
            except couchdb.ResourceNotFound:
                log.warning("%20s can't find database. Sleeping then will try again" 
                    % ("QueueMonitor"))
                time.sleep(0.5)
                continue

            for row in rows:
                item_doc = copy.deepcopy(row["value"])

                item = ItemFactory.get_item_object_from_item_doc(self.dao, 
                        item_doc)

                tiid = item.id
                log.info("%20s detected on request queue: item %s" 
                    % ("QueueMonitor", tiid))

                # remove biblio so it will be overwritten
                #item_doc["biblio"] = {}

                # now save back the updated needs_aliases information
                # do this before putting on queue, so that no one has changed it.
                log.info("%20s UPDATING needs_aliases date in db: item %s" 
                    % ("QueueMonitor", tiid))

                # remove the needs_aliases key from doc, to take off queue
                del item_doc["needs_aliases"]
                delattr(item, "needs_aliases")  #not sure this is needed
                item_doc["id"] = item.id
                self.dao.save(item_doc)

                # Now add the item to the in-memory queue
                Queue.init_queue("aliases")
                Queue.enqueue("aliases", item)

            if runonce:
                break
            self._interruptable_sleep(0.5)
    
class Queue():
    # This is a FIFO queue, add new item ids to the end of this list
    # to queue, remove from the head
    queued_items = {}
    queue_lock = threading.Lock()
    

    def __init__(self, queue_name):
        self.queue_name = queue_name
        self.init_queue(queue_name)

    @classmethod
    def clear(cls):
        """ This is only used from the test suite, normally not needed """
        for queue_name in cls.queued_items.keys():
            cls.queued_items[queue_name] = []
<commit_msg>add providersWithMetricsCount to item when putting it on alias queue for updating<commit_after>import time, datetime, logging, threading, simplejson, copy, couchdb
from totalimpact import default_settings
from totalimpact.models import Item, ItemFactory
from totalimpact.pidsupport import StoppableThread
from totalimpact.providers.provider import ProviderFactory
import default_settings

log = logging.getLogger("ti.queue")


# some data useful for testing
# d = {"doi" : ["10.1371/journal.pcbi.1000361", "10.1016/j.meegid.2011.02.004"], "url" : ["http://cottagelabs.com"]}


class QueueMonitor(StoppableThread):
    """ Worker to watch couch for items that need aliases (new or update)
        and place them onto the aliases in-memory queue. 
    """

    def __init__(self, dao):
        self.dao = dao
        StoppableThread.__init__(self)

    def run(self, runonce=False):
        """ runonce is for the test suite """

        while not self.stopped():
            viewname = 'queues/needs_aliases'
            res = self.dao.view(viewname)

            try:
                rows = res.rows
            except couchdb.ResourceNotFound:
                log.warning("%20s can't find database. Sleeping then will try again" 
                    % ("QueueMonitor"))
                time.sleep(0.5)
                continue

            for row in rows:
                item_doc = copy.deepcopy(row["value"])

                item = ItemFactory.get_item_object_from_item_doc(self.dao, 
                        item_doc)

                tiid = item.id
                log.info("%20s detected on request queue: item %s" 
                    % ("QueueMonitor", tiid))

                # remove biblio so it will be overwritten
                #item_doc["biblio"] = {}

                # now save back the updated needs_aliases information
                # do this before putting on queue, so that no one has changed it.
                log.info("%20s UPDATING needs_aliases date in db: item %s" 
                    % ("QueueMonitor", tiid))

                # remove the needs_aliases key from doc, to take off queue
                del item_doc["needs_aliases"]
                delattr(item, "needs_aliases")  #not sure this is needed

                # add the number of providers with metrics, so can compare it
                #  to estimate progress of reporting metrics
                item_doc["providersWithMetricsCount"] = ProviderFactory.num_providers_with_metrics(default_settings.PROVIDERS)
                item.providersWithMetricsCount = item_doc["providersWithMetricsCount"]

                item_doc["id"] = item.id

                self.dao.save(item_doc)

                # Now add the item to the in-memory queue
                Queue.init_queue("aliases")
                Queue.enqueue("aliases", item)

            if runonce:
                break
            self._interruptable_sleep(0.5)
    
class Queue():
    # This is a FIFO queue, add new item ids to the end of this list
    # to queue, remove from the head
    queued_items = {}
    queue_lock = threading.Lock()
    

    def __init__(self, queue_name):
        self.queue_name = queue_name
        self.init_queue(queue_name)

    @classmethod
    def clear(cls):
        """ This is only used from the test suite, normally not needed """
        for queue_name in cls.queued_items.keys():
            cls.queued_items[queue_name] = []

{'fraction_non_alphanumeric': 0.08655043586550436, 'fraction_numerical': 0.011830635118306352, 'mean_word_length': 1.9855947955390334, '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': '3912839', 'n_tokens_mistral': 1877, 'n_tokens_neox': 1708, 'n_words': 526}