<commit_before># Copyright 2014 OpenStack Foundation
# Copyright 2014 Hewlett-Packard Development Company, L.P.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.

import collections
import logging
import math
import os
import threading
import urlparse
import json
import time
import Queue
import datetime

import dateutil.parser
try:
    import ordereddict
except:
    pass
import requests
import requests.utils

import gertty.version

HIGH_PRIORITY=0
NORMAL_PRIORITY=1
LOW_PRIORITY=2

class MultiQueue(object):
    def __init__(self, priorities):
        try:
            self.queues = collections.OrderedDict()
        except AttributeError:
            self.queues = ordereddict.OrderedDict()
        for key in priorities:
            self.queues[key] = collections.deque()
        self.condition = threading.Condition()

    def qsize(self):
        count = 0
        for queue in self.queues.values():
            count += len(queue)
        return count

    def put(self, item, priority):
        self.condition.acquire()
        try:
            self.queues[priority].append(item)
            self.condition.notify()
        finally:
            self.condition.release()

    def get(self):
        self.condition.acquire()
        try:
            while True:
                for queue in self.queues.values():
                    try:
                        ret = queue.popleft()
                        return ret
                    except IndexError:
                        pass
                self.condition.wait()
        finally:
            self.condition.release()

class Task(object):
    def __init__(self, priority=NORMAL_PRIORITY):
        self.log = logging.getLogger('gertty.sync')
        self.priority = priority
        self.succeeded = None
        self.event = threading.Event()

    def complete(self, success):
        self.succeeded = success
        self.event.set()

    def wait(self, timeout=None):
        self.event.wait(timeout)
        return self.succeeded

class SyncOwnAccountTask(Task):
    def __repr__(self):
        return '<SyncOwnAccountTask>'

    def run(self, sync):
        app = sync.app
        remote = sync.get('accounts/self')
        with app.db.getSession() as session:
            session.getAccountByID(remote['_account_id'],
                                   remote.get('name'),
                                   remote['username'],
                                   remote.get('email'))

class SyncProjectListTask(Task):
    def __repr__(self):
        return '<SyncProjectListTask>'

    def run(self, sync):
        app = sync.app
        remote = sync.get('projects/?d')
        remote_keys = set(remote.keys())
        with app.db.getSession() as session:
            local = {}
            for p in session.getProjects():
                local[p.name] = p
            local_keys = set(local.keys())

            for name in local_keys-remote_keys:
                session.delete(local[name])

            for name in remote_keys-local_keys:
                p = remote[name]
                session.createProject(name, description=p.get('description', ''))

class SyncSubscribedProjectsTask(Task):
    def __repr__(self):
        return '<SyncSubscribedProjectsTask>'

    def run(self, sync):
        app = sync.app
        with app.db.getSession() as session:
            for p in session.getProjects(subscribed=True):
                sync.submitTask(SyncProjectTask(p.key, self.priority))

class SyncProjectTask(Task):
    _closed_statuses = ['MERGED', 'ABANDONED']

    def __init__(self, project_key, priority=NORMAL_PRIORITY):
        super(SyncProjectTask, self).__init__(priority)
        self.project_key = project_key

    def __repr__(self):
        return '<SyncProjectTask %s>' % (self.project_key,)

    def run(self, sync):
        app = sync.app
        now = datetime.datetime.utcnow()
        with app.db.getSession() as session:
            project = session.getProject(self.project_key)
            query = 'project:%s' % project.name
            if project.updated:
                # Allow 4 seconds for request time, etc.
                query += ' -age:%ss' % (int(math.ceil((now-project.updated).total_seconds())) + 4,)
        changes = sync.get('changes/?q=%s' % query)
        self.log.debug('Query: %s ' % (query,))
        with app.db.getSession() as session:
            for c in changes:
                # For now, just sync open changes or changes already
                # in the db optionally we could sync all changes ever
                change = session.getChangeByID(c['id'])
                if change or (c['status'] not in self._closed_statuses):
                    sync.submitTask(SyncChangeTask(c['id'], priority=self.priority))
                    self.log.debug("Change %s update %s" % (c['id'], c['updated']))
        sync.submitTask(SetProjectUpdatedTask(self.project_key, now, priority=self.priority))

class SetProjectUpdatedTask(Task):
    def __init__(self, project_key, updated, priority=NORMAL_PRIORITY):<commit_msg>Support paging in queries<commit_after># Copyright 2014 OpenStack Foundation
# Copyright 2014 Hewlett-Packard Development Company, L.P.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.

import collections
import logging
import math
import os
import threading
import urlparse
import json
import time
import Queue
import datetime

import dateutil.parser
try:
    import ordereddict
except:
    pass
import requests
import requests.utils

import gertty.version

HIGH_PRIORITY=0
NORMAL_PRIORITY=1
LOW_PRIORITY=2

class MultiQueue(object):
    def __init__(self, priorities):
        try:
            self.queues = collections.OrderedDict()
        except AttributeError:
            self.queues = ordereddict.OrderedDict()
        for key in priorities:
            self.queues[key] = collections.deque()
        self.condition = threading.Condition()

    def qsize(self):
        count = 0
        for queue in self.queues.values():
            count += len(queue)
        return count

    def put(self, item, priority):
        self.condition.acquire()
        try:
            self.queues[priority].append(item)
            self.condition.notify()
        finally:
            self.condition.release()

    def get(self):
        self.condition.acquire()
        try:
            while True:
                for queue in self.queues.values():
                    try:
                        ret = queue.popleft()
                        return ret
                    except IndexError:
                        pass
                self.condition.wait()
        finally:
            self.condition.release()

class Task(object):
    def __init__(self, priority=NORMAL_PRIORITY):
        self.log = logging.getLogger('gertty.sync')
        self.priority = priority
        self.succeeded = None
        self.event = threading.Event()

    def complete(self, success):
        self.succeeded = success
        self.event.set()

    def wait(self, timeout=None):
        self.event.wait(timeout)
        return self.succeeded

class SyncOwnAccountTask(Task):
    def __repr__(self):
        return '<SyncOwnAccountTask>'

    def run(self, sync):
        app = sync.app
        remote = sync.get('accounts/self')
        with app.db.getSession() as session:
            session.getAccountByID(remote['_account_id'],
                                   remote.get('name'),
                                   remote['username'],
                                   remote.get('email'))

class SyncProjectListTask(Task):
    def __repr__(self):
        return '<SyncProjectListTask>'

    def run(self, sync):
        app = sync.app
        remote = sync.get('projects/?d')
        remote_keys = set(remote.keys())
        with app.db.getSession() as session:
            local = {}
            for p in session.getProjects():
                local[p.name] = p
            local_keys = set(local.keys())

            for name in local_keys-remote_keys:
                session.delete(local[name])

            for name in remote_keys-local_keys:
                p = remote[name]
                session.createProject(name, description=p.get('description', ''))

class SyncSubscribedProjectsTask(Task):
    def __repr__(self):
        return '<SyncSubscribedProjectsTask>'

    def run(self, sync):
        app = sync.app
        with app.db.getSession() as session:
            for p in session.getProjects(subscribed=True):
                sync.submitTask(SyncProjectTask(p.key, self.priority))

class SyncProjectTask(Task):
    _closed_statuses = ['MERGED', 'ABANDONED']

    def __init__(self, project_key, priority=NORMAL_PRIORITY):
        super(SyncProjectTask, self).__init__(priority)
        self.project_key = project_key

    def __repr__(self):
        return '<SyncProjectTask %s>' % (self.project_key,)

    def run(self, sync):
        app = sync.app
        now = datetime.datetime.utcnow()
        with app.db.getSession() as session:
            project = session.getProject(self.project_key)
            query = 'project:%s' % project.name
            if project.updated:
                # Allow 4 seconds for request time, etc.
                query += ' -age:%ss' % (int(math.ceil((now-project.updated).total_seconds())) + 4,)
            else:
                query += ' status:open'
        changes = []
        sortkey = ''
        while True:
            # We don't actually want to limit to 500, but that's the server-side default, and
            # if we don't specify this, we won't get a _more_changes flag.
            q = 'changes/?n=500%s&q=%s' % (sortkey, query,)
            self.log.debug('Query: %s ' % (q,))
            batch = sync.get(q)
            changes += batch
            if batch and '_more_changes' in batch[-1]:
                sortkey = '&N=%s' % (batch[-1]['_sortkey'],)
            else:
                break
        with app.db.getSession() as session:
            for c in changes:
                # For now, just sync open changes or changes already
                # in the db optionally we could sync all changes ever
                change = session.getChangeByID(c['id'])
                if change or (c['status'] not in self._closed_statuses):
                    sync.submitTask(SyncChangeTask(c['id'], priority=self.priority))
                    self.log.debug("Change %s update %s" % (c['id'], c['updated']))
        sync.submitTask(SetProjectUpdatedTask(self.project_key, now, priority=self.priority))

class SetProjectUpdatedTask(Task):
    def __init__(self, project_key, updated, priority=NORMAL_PRIORITY):
{'fraction_non_alphanumeric': 0.10062026188835287, 'fraction_numerical': 0.003790489317711923, 'mean_word_length': 2.12911051212938, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 11, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 8, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6836071', 'n_tokens_mistral': 3223, 'n_tokens_neox': 3105, 'n_words': 858}