<commit_before>#!/usr/bin/python

"""
Flask app to run data retrieval tasks for Open Humans
"""

import json
import logging
import os

from celery.signals import (after_setup_logger, after_task_publish,
                            task_postrun, task_prerun)

from flask import Flask, request
from flask_sslify import SSLify

import requests

from data_retrieval.american_gut import create_amgut_ohdatasets
from data_retrieval.pgp_harvard import create_pgpharvard_ohdatasets
from data_retrieval.twenty_three_and_me import create_23andme_ohdataset
from data_retrieval.go_viral import create_go_viral_ohdataset

from celery_worker import make_worker

ohdata_app = Flask(__name__)

DEBUG = os.getenv('DEBUG', False)
PORT = os.getenv('PORT', 5000)

logging.basicConfig(level=logging.DEBUG if DEBUG else logging.INFO)
logging.info('Starting data-processing')

ohdata_app.config.update(
    DEBUG=DEBUG,
    CELERY_BROKER_URL=os.environ.get('CLOUDAMQP_URL', 'amqp://'),
    CELERY_ACCEPT_CONTENT=['json'],
    CELERY_TASK_SERIALIZER='json',
    CELERY_RESULT_SERIALIZER='json',
    CELERYD_LOG_COLOR=True,
    BROKER_POOL_LIMIT=0)

sslify = SSLify(ohdata_app)

celery_worker = make_worker(ohdata_app)


@after_setup_logger.connect
def after_setup_logger_cb(logger, **kwargs):
    """
    Update the Celery logger's level.
    """
    if DEBUG:
        logger.setLevel(logging.DEBUG)


def debug_json(value):
    """
    Return a human-readable representation of JSON data.
    """
    return json.dumps(value, sort_keys=True, indent=2, separators=(',', ': '))


def make_task_data(task_id, task_state):
    """
    Format task data for the Open Humans update endpoint.
    """
    return json.dumps({
        'task_data': {
            'task_id': task_id,
            'task_state': task_state,
        }
    })


@celery_worker.task
def send_queued_update(update_url, task_data):
    """
    The 'after_task_publish' signal runs synchronously so we use celery itself
    to run it asynchronously.
    """
    logging.info('Sending after_task_publish update')

    requests.post(update_url, data=task_data)


@after_task_publish.connect
def task_sent_handler_cb(sender=None, body=None, **other_kwargs):<commit_msg>Fix how task_data is sent<commit_after>#!/usr/bin/python

"""
Flask app to run data retrieval tasks for Open Humans
"""

import json
import logging
import os

from celery.signals import (after_setup_logger, after_task_publish,
                            task_postrun, task_prerun)

from flask import Flask, request
from flask_sslify import SSLify

import requests

from data_retrieval.american_gut import create_amgut_ohdatasets
from data_retrieval.pgp_harvard import create_pgpharvard_ohdatasets
from data_retrieval.twenty_three_and_me import create_23andme_ohdataset
from data_retrieval.go_viral import create_go_viral_ohdataset

from celery_worker import make_worker

ohdata_app = Flask(__name__)

DEBUG = os.getenv('DEBUG', False)
PORT = os.getenv('PORT', 5000)

logging.basicConfig(level=logging.DEBUG if DEBUG else logging.INFO)
logging.info('Starting data-processing')

ohdata_app.config.update(
    DEBUG=DEBUG,
    CELERY_BROKER_URL=os.environ.get('CLOUDAMQP_URL', 'amqp://'),
    CELERY_ACCEPT_CONTENT=['json'],
    CELERY_TASK_SERIALIZER='json',
    CELERY_RESULT_SERIALIZER='json',
    CELERYD_LOG_COLOR=True,
    BROKER_POOL_LIMIT=0)

sslify = SSLify(ohdata_app)

celery_worker = make_worker(ohdata_app)


@after_setup_logger.connect
def after_setup_logger_cb(logger, **kwargs):
    """
    Update the Celery logger's level.
    """
    if DEBUG:
        logger.setLevel(logging.DEBUG)


def debug_json(value):
    """
    Return a human-readable representation of JSON data.
    """
    return json.dumps(value, sort_keys=True, indent=2, separators=(',', ': '))


def make_task_data(task_id, task_state):
    """
    Format task data for the Open Humans update endpoint.
    """
    return {
        'task_data': json.dumps({
            'task_id': task_id,
            'task_state': task_state,
        })
    }


@celery_worker.task
def send_queued_update(update_url, task_data):
    """
    The 'after_task_publish' signal runs synchronously so we use celery itself
    to run it asynchronously.
    """
    logging.info('Sending after_task_publish update')

    requests.post(update_url, data=task_data)


@after_task_publish.connect
def task_sent_handler_cb(sender=None, body=None, **other_kwargs):
{'fraction_non_alphanumeric': 0.09584086799276673, 'fraction_numerical': 0.003616636528028933, 'mean_word_length': 4.312124849939976, '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': '1133412', 'n_tokens_mistral': 1525, 'n_tokens_neox': 1466, 'n_words': 315}