<commit_before>from __future__ import unicode_literals

import ujson as json
import logging
import threading
import redis

from compatibility import urlopen
from config import POLLING_WAIT, REDIS_DB, REDIS_CHARSET, REDIS_PREFIX, REDIS_PORT, REDIS_HOST
from api.vkapi import method
from parallel import realtime, updates
from events.toggle import raise_event
import ssl
import time

logger = logging.getLogger("cyvk")
UPDATE_RESULT = 'lp_result'
LONG_POLLING_KEY = ':'.join([REDIS_PREFIX, 'long_polling_queue'])
START_POLLING_KEY = ':'.join([REDIS_PREFIX, 'long_polling_start_queue'])


def start_polling(jid):
    r = redis.StrictRedis(REDIS_HOST, REDIS_PORT, REDIS_DB, charset=REDIS_CHARSET)
    r.lpush(START_POLLING_KEY, jid)


def _start_polling(jid, attempts=0):
    if realtime.is_polling(jid):
        return logger.debug('%s is already polling' % jid)

    if attempts > 5:
        return logger.error('too many long polling attempts for %s' % jid)

    r = redis.StrictRedis(REDIS_HOST, REDIS_PORT, REDIS_DB, charset=REDIS_CHARSET)

    try:
        args = method('messages.getLongPollServer', jid)
    except ssl.SSLError as e:
        logger.error('SSL error %s' % e)
        time.sleep(3)
        return _start_polling(jid, attempts+1)

    args['wait'] = POLLING_WAIT
    url = 'http://{server}?act=a_check&key={key}&ts={ts}&wait={wait}&mode=2'.format(**args)
    r.lpush(LONG_POLLING_KEY, json.dumps({'jid': jid, 'url': url}))


def handle_url(jid, url):
    realtime.set_polling(jid)
    logger.debug('got url, starting polling')
    data = urlopen(url).read()
    logger.debug('got data from polling server')
    realtime.unset_polling(jid)
    raise_event(UPDATE_RESULT, response=data, jid=jid)


def event_handler(event_body):
    data = None
    try:
        data = json.loads(event_body['response'])
    except ValueError as e:
        logger.error('unable to parse json: %s (%s)' % (data, e))
    jid = event_body['jid']
    is_client = realtime.is_client(jid)

    try:
        if not data['updates']:
            logger.debug('no updates for %s' % jid)

        for update in data['updates']:
            updates.process_data(jid, update)
    except KeyError:
        logger.error('unable to process %s' % event_body)

    if is_client:
        start_polling(jid)
    else:
        logger.debug('ending polling for %s' % jid)


def loop_for_starting():
    logger.debug('starting long polling starter loop')
    r = redis.StrictRedis(REDIS_HOST, REDIS_PORT, REDIS_DB, charset=REDIS_CHARSET)

    while True:
        jid = r.brpop(START_POLLING_KEY)[1]
        jid = unicode(jid)
        t = threading.Thread(target=_start_polling, args=(jid, ))
        t.daemon = True
        t.start()


def loop():
    logger.debug('starting long polling loop')
    r = redis.StrictRedis(REDIS_HOST, REDIS_PORT, REDIS_DB, charset=REDIS_CHARSET)
    while True:
        request_raw = r.brpop(LONG_POLLING_KEY)[1]
        # { jid: 'user_jid', url: 'long_polling_url' }
        request = json.loads(request_raw)
        jid, url = request['jid'], request['url']
        t = threading.Thread(target=handle_url, args=(jid, url))
        t.daemon = True
        t.start()<commit_msg>make some long polling functions private<commit_after>from __future__ import unicode_literals

import ujson as json
import logging
import threading
import redis

from compatibility import urlopen
from config import POLLING_WAIT, REDIS_DB, REDIS_CHARSET, REDIS_PREFIX, REDIS_PORT, REDIS_HOST
from api.vkapi import method
from parallel import realtime, updates
from events.toggle import raise_event
import ssl
import time

logger = logging.getLogger("cyvk")
UPDATE_RESULT = 'lp_result'
LONG_POLLING_KEY = ':'.join([REDIS_PREFIX, 'long_polling_queue'])
START_POLLING_KEY = ':'.join([REDIS_PREFIX, 'long_polling_start_queue'])


def start_polling(jid):
    r = redis.StrictRedis(REDIS_HOST, REDIS_PORT, REDIS_DB, charset=REDIS_CHARSET)
    r.lpush(START_POLLING_KEY, jid)


def _start_polling(jid, attempts=0):
    if realtime.is_polling(jid):
        return logger.debug('%s is already polling' % jid)

    if attempts > 5:
        return logger.error('too many long polling attempts for %s' % jid)

    r = redis.StrictRedis(REDIS_HOST, REDIS_PORT, REDIS_DB, charset=REDIS_CHARSET)

    try:
        args = method('messages.getLongPollServer', jid)
    except ssl.SSLError as e:
        logger.error('SSL error %s' % e)
        time.sleep(3)
        return _start_polling(jid, attempts+1)

    args['wait'] = POLLING_WAIT
    url = 'http://{server}?act=a_check&key={key}&ts={ts}&wait={wait}&mode=2'.format(**args)
    r.lpush(LONG_POLLING_KEY, json.dumps({'jid': jid, 'url': url}))


def _handle_url(jid, url):
    realtime.set_polling(jid)
    logger.debug('got url, starting polling')
    data = urlopen(url).read()
    logger.debug('got data from polling server')
    realtime.unset_polling(jid)
    raise_event(UPDATE_RESULT, response=data, jid=jid)


def event_handler(event_body):
    data = None
    try:
        data = json.loads(event_body['response'])
    except ValueError as e:
        logger.error('unable to parse json: %s (%s)' % (data, e))
    jid = event_body['jid']
    is_client = realtime.is_client(jid)

    try:
        if not data['updates']:
            logger.debug('no updates for %s' % jid)

        for update in data['updates']:
            updates.process_data(jid, update)
    except KeyError:
        logger.error('unable to process %s' % event_body)

    if is_client:
        start_polling(jid)
    else:
        logger.debug('ending polling for %s' % jid)


def loop_for_starting():
    logger.debug('starting long polling starter loop')
    r = redis.StrictRedis(REDIS_HOST, REDIS_PORT, REDIS_DB, charset=REDIS_CHARSET)

    while True:
        jid = r.brpop(START_POLLING_KEY)[1]
        jid = unicode(jid)
        t = threading.Thread(target=_start_polling, args=(jid, ))
        t.daemon = True
        t.start()


def loop():
    logger.debug('starting long polling loop')
    r = redis.StrictRedis(REDIS_HOST, REDIS_PORT, REDIS_DB, charset=REDIS_CHARSET)
    while True:
        request_raw = r.brpop(LONG_POLLING_KEY)[1]
        # { jid: 'user_jid', url: 'long_polling_url' }
        request = json.loads(request_raw)
        jid, url = request['jid'], request['url']
        t = threading.Thread(target=_handle_url, args=(jid, url))
        t.daemon = True
        t.start()
{'fraction_non_alphanumeric': 0.1190438993907202, 'fraction_numerical': 0.0021871582565224186, 'mean_word_length': 3.572857142857143, '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': '4406378', 'n_tokens_mistral': 2320, 'n_tokens_neox': 2221, 'n_words': 494}