<commit_before>"""
Celery setup and wraper tasks to periodically update the database.
"""
import requests
import logging
import traceback
from celery import task
import time
import json
import os
from queue import Full
import shutil
from multiprocessing import Queue
from datetime import datetime, timedelta
import pytz

from meerkat_abacus import config, data_management
import meerkat_libs as libs
from meerkat_abacus import data_import
from meerkat_abacus import util
from meerkat_abacus.pipeline import process_chunk
from meerkat_abacus.util import create_fake_data


sqs_client = None
sqs_queue_url = None

worker_buffer = Queue(maxsize=1000)

@task
def set_up_db():
    data_management.set_up_database(leave_if_data=False,
                                    drop_db=True)

@task
def initial_data_setup():
    logging.info("Starting initial setup")
    while not worker_buffer.empty():
        worker_buffer.get()
    engine, session = util.get_db_engine()
    if config.initial_data == "S3":
        data_import.download_data_from_s3(config)
        get_function = util.read_csv_filename
    elif config.initial_data == "CSV":
        get_function = util.read_csv_filename
        data_management.add_fake_data(session)
    
        data_import.read_stationary_data(get_function, worker_buffer,
                                         config, process_buffer, session, engine)
        process_buffer(internal_buffer=worker_buffer, start=False)
    process_buffer(internal_buffer=worker_buffer, start=False)
    session.close()
    engine.dispose()


@task
def process_buffer(start=True, internal_buffer=None):
    if internal_buffer is None:
        internal_buffer = worker_buffer
    engine, session = util.get_db_engine()
    process_chunk(internal_buffer, session, engine)
    if start:
        process_buffer.apply_async(countdown=30,
                                   kwargs={"start": True})
    session.close()
    engine.dispose()


@task(bind=True, default_retry_delay=300, max_retries=5)
def poll_queue(self, sqs_queue_name, sqs_endpoint, start=True):
    """ Get's messages from SQS queue"""
    logging.info("Running Poll Queue")

    global sqs_client
    global sqs_queue_url
    if sqs_client is None:
        sqs_client, sqs_queue_url = util.subscribe_to_sqs(sqs_endpoint,
                                                          sqs_queue_name)
    try:
        messages = sqs_client.receive_message(QueueUrl=sqs_queue_url,
                                              WaitTimeSeconds=19)
    except:
        self.retry()
    if "Messages" in messages:
        for message in messages["Messages"]:
            logging.info("Message %s", message)
            receipt_handle = message["ReceiptHandle"]
            logging.info("Deleting message %s", receipt_handle)
            try:
                message_body = json.loads(message["Body"])
                form = message_body["formId"]
                form_data = message_body["data"]
                uuid = message_body["uuid"]
                try:
                    worker_buffer.put_nowait(
                        {"form": form,
                         "uuid": uuid,
                         "data": form_data}
                    )
                except Full:
                    process_buffer(start=False)
                    worker_buffer.put(
                        {"form": form,
                         "uuid": uuid,
                         "data": form_data}
                    )
                logging.warning(worker_buffer)
                logging.info(worker_buffer.qsize())
                sqs_client.delete_message(QueueUrl=sqs_queue_url,
                                          ReceiptHandle=receipt_handle)
            except Exception as e:
                logging.exception("Error in reading message", exc_info=True)
                                    
    if start:
        poll_queue.delay(sqs_queue_name, sqs_endpoint, start=True)


@task
def add_fake_data(N=10, interval_next=None, dates_is_now=False, internal_fake_data=True, aggregate_url=None):
    logging.info("Adding fake data")
    engine, session = util.get_db_engine()
    for form in config.country_config["tables"]:
        logging.info("Generating fake data for form:" + form)
        new_data = create_fake_data.get_new_fake_data(form, session, N, config,
                                                      dates_is_now=dates_is_now)
        for row, uuid in new_data:
            if internal_fake_data:
                try:
                    worker_buffer.put_nowait(
                        {"form": form,
                         "uuid": uuid,
                         "data": row}
                    )
                except Full:
                    process_buffer(start=False)
                    worker_buffer.put(
                        {"form": form,
                         "uuid": uuid,
                         "data": row})
            elif aggregate_url:
                logging.info("Submitting fake data for form {0} to Aggregate".format(form))
                # util.submit_data_to_aggregate(row, form, config)
    if interval_next:
        add_fake_data.apply_async(countdown=interval_next,
                                  kwargs={"interval_next": interval_next,
                                          "N": N,
                                          "dates_is_now": dates_is_now,
                                          "internal_fake_data": internal_fake_data,
                                          "aggregate_url": aggregate_url})
                    
    session.close()
    engine.dispose()
        

@task
def cleanup_downloads():
    folder = '/var/www/meerkat_api/api_background/api_background/exported_data'
    downloads = os.listdir(folder)
    oldest = time.time() - 3600
    for download in downloads:
        path = '{}/{}'.format(folder, download)
        if os.stat(path).st_mtime < oldest:
            try:
                shutil.rmtree(path)
            except NotADirectoryError:
                pass

<commit_msg>reactivate Aggragete submissions<commit_after>"""
Celery setup and wraper tasks to periodically update the database.
"""
import requests
import logging
import traceback
from celery import task
import time
import json
import os
from queue import Full
import shutil
from multiprocessing import Queue
from datetime import datetime, timedelta
import pytz

from meerkat_abacus import config, data_management
import meerkat_libs as libs
from meerkat_abacus import data_import
from meerkat_abacus import util
from meerkat_abacus.pipeline import process_chunk
from meerkat_abacus.util import create_fake_data


sqs_client = None
sqs_queue_url = None

worker_buffer = Queue(maxsize=1000)

@task
def set_up_db():
    data_management.set_up_database(leave_if_data=False,
                                    drop_db=True)

@task
def initial_data_setup():
    logging.info("Starting initial setup")
    while not worker_buffer.empty():
        worker_buffer.get()
    engine, session = util.get_db_engine()
    if config.initial_data == "S3":
        data_import.download_data_from_s3(config)
        get_function = util.read_csv_filename
    elif config.initial_data == "CSV":
        get_function = util.read_csv_filename
        data_management.add_fake_data(session)
    
        data_import.read_stationary_data(get_function, worker_buffer,
                                         config, process_buffer, session, engine)
        process_buffer(internal_buffer=worker_buffer, start=False)
    process_buffer(internal_buffer=worker_buffer, start=False)
    session.close()
    engine.dispose()


@task
def process_buffer(start=True, internal_buffer=None):
    if internal_buffer is None:
        internal_buffer = worker_buffer
    engine, session = util.get_db_engine()
    process_chunk(internal_buffer, session, engine)
    if start:
        process_buffer.apply_async(countdown=30,
                                   kwargs={"start": True})
    session.close()
    engine.dispose()


@task(bind=True, default_retry_delay=300, max_retries=5)
def poll_queue(self, sqs_queue_name, sqs_endpoint, start=True):
    """ Get's messages from SQS queue"""
    logging.info("Running Poll Queue")

    global sqs_client
    global sqs_queue_url
    if sqs_client is None:
        sqs_client, sqs_queue_url = util.subscribe_to_sqs(sqs_endpoint,
                                                          sqs_queue_name)
    try:
        messages = sqs_client.receive_message(QueueUrl=sqs_queue_url,
                                              WaitTimeSeconds=19)
    except:
        self.retry()
    if "Messages" in messages:
        for message in messages["Messages"]:
            logging.info("Message %s", message)
            receipt_handle = message["ReceiptHandle"]
            logging.info("Deleting message %s", receipt_handle)
            try:
                message_body = json.loads(message["Body"])
                form = message_body["formId"]
                form_data = message_body["data"]
                uuid = message_body["uuid"]
                try:
                    worker_buffer.put_nowait(
                        {"form": form,
                         "uuid": uuid,
                         "data": form_data}
                    )
                except Full:
                    process_buffer(start=False)
                    worker_buffer.put(
                        {"form": form,
                         "uuid": uuid,
                         "data": form_data}
                    )
                logging.warning(worker_buffer)
                logging.info(worker_buffer.qsize())
                sqs_client.delete_message(QueueUrl=sqs_queue_url,
                                          ReceiptHandle=receipt_handle)
            except Exception as e:
                logging.exception("Error in reading message", exc_info=True)
                                    
    if start:
        poll_queue.delay(sqs_queue_name, sqs_endpoint, start=True)


@task
def add_fake_data(N=10, interval_next=None, dates_is_now=False, internal_fake_data=True, aggregate_url=None):
    logging.info("Adding fake data")
    engine, session = util.get_db_engine()
    for form in config.country_config["tables"]:
        logging.info("Generating fake data for form:" + form)
        new_data = create_fake_data.get_new_fake_data(form, session, N, config,
                                                      dates_is_now=dates_is_now)
        for row, uuid in new_data:
            if internal_fake_data:
                try:
                    worker_buffer.put_nowait(
                        {"form": form,
                         "uuid": uuid,
                         "data": row}
                    )
                except Full:
                    process_buffer(start=False)
                    worker_buffer.put(
                        {"form": form,
                         "uuid": uuid,
                         "data": row})
            elif aggregate_url:
                logging.info("Submitting fake data for form {0} to Aggregate".format(form))
                util.submit_data_to_aggregate(row, form, config)
    if interval_next:
        add_fake_data.apply_async(countdown=interval_next,
                                  kwargs={"interval_next": interval_next,
                                          "N": N,
                                          "dates_is_now": dates_is_now,
                                          "internal_fake_data": internal_fake_data,
                                          "aggregate_url": aggregate_url})
                    
    session.close()
    engine.dispose()
        

@task
def cleanup_downloads():
    folder = '/var/www/meerkat_api/api_background/api_background/exported_data'
    downloads = os.listdir(folder)
    oldest = time.time() - 3600
    for download in downloads:
        path = '{}/{}'.format(folder, download)
        if os.stat(path).st_mtime < oldest:
            try:
                shutil.rmtree(path)
            except NotADirectoryError:
                pass


{'fraction_non_alphanumeric': 0.07900602658301, 'fraction_numerical': 0.003467349129034921, 'mean_word_length': 1.783547794117647, 'pattern_counts': {'":': 40, '<': 5, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 56, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '776942', 'n_tokens_mistral': 3306, 'n_tokens_neox': 3170, 'n_words': 688}