<commit_before>#coding: utf-8

from ..models.stats import ActiveTaxis
from ..extensions import redis_store, user_datastore, db, celery
from ..models.taxis import Taxi
from ..models.stats import ActiveTaxis
from itertools import izip, ifilter, imap
from datetime import datetime, timedelta
from time import mktime
from parse import parse as base_parse
from flask import current_app

def scan_as_list(match, redis_store):
    cursor = ''
    while cursor != 0:
        cursor, data = redis_store.scan(cursor=cursor, match=match, count=100)
        yield data

def filter_outoftime_values(operator_values, bound):
    f = lambda o_v: o_v[1] and o_v[1]['timestamp'] >= bound
    return filter(lambda operator_value: f(operator_value), operator_values)

def parse_operator_value(operator, s):
    if type(s) not in (str, unicode):
        return operator, None
    return operator, base_parse(Taxi._FORMAT_OPERATOR, s)

def get_data(taxi_ids, bound, redis_store):
    pipe = redis_store.pipeline()
    map(lambda k: pipe.hgetall(k), taxi_ids)
    parsed_values = map(lambda map_operator_v:\
        map(lambda o_v: parse_operator_value(*o_v), map_operator_v.iteritems()),
            pipe.execute())
    pipe.reset()
    filtered_value = map(lambda v: filter_outoftime_values(v, bound), parsed_values)
    zipped_value = izip(taxi_ids, filtered_value)
    return ifilter(lambda taxi_operator: len(taxi_operator[1]) > 0, zipped_value)

def store_active_taxis():
    current_app.logger.info('store_active_taxis')
    bound_time = datetime.now() - timedelta(minutes=15)
    bound = mktime(bound_time.timetuple())
    map_operateur_zupc_nb_active = dict()
    for l in scan_as_list('taxi:*', redis_store):
        for taxi_id, v in get_data(l, bound, redis_store):
            taxi_id = taxi_id[5:] #We cut the "taxi:" part
            taxi_db = Taxi.get(taxi_id)
            if taxi_db is None:
                current_app.logger.error('Taxi: {}, not found in database'.format(taxi_id))
                continue
            for operator, result in v:
                u = user_datastore.find_user(email=operator)
                if not u:
                    current_app.logger.error('User: {} not found'.format(operator))
                    continue
                if u.id not in map_operateur_zupc_nb_active:
                    map_operateur_zupc_nb_active[u.id] = dict()
                if taxi_db.ads.zupc_id not in map_operateur_zupc_nb_active[u.id]:
                    map_operateur_zupc_nb_active[u.id][taxi_db.ads.zupc_id] = 0
                map_operateur_zupc_nb_active[u.id][taxi_db.ads.zupc_id] += 1

    for operator, zupc_active in map_operateur_zupc_nb_active.iteritems():
        for zupc, active in zupc_active.iteritems():
            a = ActiveTaxis()
            a.nb_taxis = active
            a.operator_id = operator
            a.zupc_id = zupc
            a.timestamp = bound_time
            db.session.add(a)
    db.session.commit()
<commit_msg>Use influxDB<commit_after>#coding: utf-8

from ..extensions import redis_store, user_datastore, db, celery
from ..models.taxis import Taxi
from ..models.stats import ActiveTaxis
from itertools import izip, ifilter, imap
from datetime import datetime, timedelta
from time import mktime
from parse import parse as base_parse
from flask import current_app
from influxdb import InfluxDBClient
from influxdb.exceptions import InfluxDBClientError

def scan_as_list(match, redis_store):
    cursor = ''
    while cursor != 0:
        cursor, data = redis_store.scan(cursor=cursor, match=match, count=100)
        yield data

def filter_outoftime_values(operator_values, bound):
    f = lambda o_v: o_v[1] and o_v[1]['timestamp'] >= bound
    return filter(lambda operator_value: f(operator_value), operator_values)

def parse_operator_value(operator, s):
    if type(s) not in (str, unicode):
        return operator, None
    return operator, base_parse(Taxi._FORMAT_OPERATOR, s)

def get_data(taxi_ids, bound, redis_store):
    pipe = redis_store.pipeline()
    map(lambda k: pipe.hgetall(k), taxi_ids)
    parsed_values = map(lambda map_operator_v:\
        map(lambda o_v: parse_operator_value(*o_v), map_operator_v.iteritems()),
            pipe.execute())
    pipe.reset()
    filtered_value = map(lambda v: filter_outoftime_values(v, bound), parsed_values)
    zipped_value = izip(taxi_ids, filtered_value)
    return ifilter(lambda taxi_operator: len(taxi_operator[1]) > 0, zipped_value)

def store_active_taxis():
    current_app.logger.info('store_active_taxis')
    bound_time = datetime.now() - timedelta(minutes=15)
    bound = mktime(bound_time.timetuple())
    map_operateur_zupc_nb_active = dict()
    for l in scan_as_list('taxi:*', redis_store):
        for taxi_id, v in get_data(l, bound, redis_store):
            taxi_id = taxi_id[5:] #We cut the "taxi:" part
            taxi_db = Taxi.get(taxi_id)
            if taxi_db is None:
                current_app.logger.error('Taxi: {}, not found in database'.format(taxi_id))
                continue
            for operator, result in v:
                u = user_datastore.find_user(email=operator)
                if not u:
                    current_app.logger.error('User: {} not found'.format(operator))
                    continue
                if u.id not in map_operateur_zupc_nb_active:
                    map_operateur_zupc_nb_active[u.id] = dict()
                if taxi_db.ads.zupc_id not in map_operateur_zupc_nb_active[u.id]:
                    map_operateur_zupc_nb_active[u.id][taxi_db.ads.zupc_id] = 0
                map_operateur_zupc_nb_active[u.id][taxi_db.ads.zupc_id] += 1
    client = InfluxDBClient('localhost', 8086, 'root', 'root', 'taxis')
    #try:
    #    client.create_database('taxis')
    #except InfluxDBClientError as e:
    #    if e['content'] != u'database already exists':
    #        raise e
    to_insert = []
    for operator, zupc_active in map_operateur_zupc_nb_active.iteritems():
        for zupc, active in zupc_active.iteritems():
            to_insert.append(
                {
                    "measurement": "nb_taxis",
                    "tags": {
                        "operator": user_datastore.get_user(operator).email,
                        "zupc": zupc
                    },
                    "time": datetime.utcnow().strftime('%Y%m%dT%H:%M:%SZ'),
                    "fields": {
                        "value": active
                    }
                }
            )
    current_app.logger.info('To insert: {}'.format(to_insert))
    client.write_points(to_insert)
{'fraction_non_alphanumeric': 0.10045592705167174, 'fraction_numerical': 0.004863221884498481, 'mean_word_length': 2.703432751828925, 'pattern_counts': {'":': 7, '<': 3, '<?xml version=': 0, '>': 7, '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': '5870677', 'n_tokens_mistral': 2183, 'n_tokens_neox': 2117, 'n_words': 466}