<commit_before># -*- coding: utf-8; -*-
#
# This file is part of Superdesk.
#
# Copyright 2013, 2014 Sourcefabric z.u. and contributors.
#
# For the full copyright and license information, please see the
# AUTHORS and LICENSE files distributed with this source code, or
# at https://www.sourcefabric.org/superdesk/license

"""
Created on May 29, 2014

@author: ioan
"""

import redis
import arrow
import werkzeug
import superdesk
from bson import ObjectId
from celery import Celery
from kombu.serialization import register
from eve.io.mongo import MongoJSONEncoder
from eve.utils import str_to_date
from flask import json, current_app as app
from superdesk.errors import SuperdeskError
from superdesk.logging import logger


celery = Celery(__name__)
TaskBase = celery.Task


def try_cast(v):
    # False and 0 are getting converted to datetime by arrow
    if v is None or isinstance(v, bool) or v == 0:
        return v

    try:
        str_to_date(v)  # try if it matches format
        return arrow.get(v).datetime  # return timezone aware time
    except Exception:
        try:
            return ObjectId(v)
        except Exception:
            return v


def dumps(o):
    with superdesk.app.app_context():
        return MongoJSONEncoder().encode(o)


def loads(s):
    o = json.loads(s)
    with superdesk.app.app_context():
        return serialize(o)


def serialize(o):
    if isinstance(o, list):
        return [serialize(item) for item in o]
    elif isinstance(o, dict):
        if o.get('kwargs') and not isinstance(o['kwargs'], dict):
            o['kwargs'] = json.loads(o['kwargs'])
        return {k: serialize(v) for k, v in o.items()}
    else:
        return try_cast(o)


register('eve/json', dumps, loads, content_type='application/json')


def handle_exception(exc):
    """Log exception to logger."""
    logger.exception(exc)


class AppContextTask(TaskBase):
    abstract = True
    serializer = 'eve/json'
    app_errors = (
        SuperdeskError,
        werkzeug.exceptions.InternalServerError,  # mongo layer err
    )

    def __call__(self, *args, **kwargs):
        with superdesk.app.app_context():
            try:
                return super().__call__(*args, **kwargs)
            except self.app_errors as e:
                handle_exception(e)

    def on_failure(self, exc, task_id, args, kwargs, einfo):
        with superdesk.app.app_context():
            handle_exception(exc)


celery.Task = AppContextTask


def init_celery(app):
    celery.config_from_object(app.config, namespace='CELERY')
    app.celery = celery
    app.redis = __get_redis(app)


def add_subtask_to_progress(task_id):
    return _update_subtask_progress(task_id, total=True)


def finish_subtask_from_progress(task_id):
    return _update_subtask_progress(task_id, current=True)


def finish_task_for_progress(task_id):
    return _update_subtask_progress(task_id, done=True)


def __get_redis(app_ctx):
    """Constructs Redis Client object.

    :return: Redis Client object
    """

    return redis.from_url(app_ctx.config['REDIS_URL'])


def update_key(key, flag=False, db=None):

    if db is None:
        db = app.redis

    if flag:
        crt_value = db.incr(key)
    else:
        crt_value = db.get(key)

    if crt_value:
        crt_value = int(crt_value)
    else:
        crt_value = 0

    return crt_value


def _update_subtask_progress(task_id, current=None, total=None, done=None):
    redis_db = redis.from_url(app.config['REDIS_URL'])
    try:<commit_msg>[SDESK-4652] (fix): Redis client fails to connect using amqp scheme (#1652)<commit_after># -*- coding: utf-8; -*-
#
# This file is part of Superdesk.
#
# Copyright 2013, 2014 Sourcefabric z.u. and contributors.
#
# For the full copyright and license information, please see the
# AUTHORS and LICENSE files distributed with this source code, or
# at https://www.sourcefabric.org/superdesk/license

"""
Created on May 29, 2014

@author: ioan
"""

import redis
import arrow
import werkzeug
import superdesk
from bson import ObjectId
from celery import Celery
from kombu.serialization import register
from eve.io.mongo import MongoJSONEncoder
from eve.utils import str_to_date
from flask import json, current_app as app
from superdesk.errors import SuperdeskError
from superdesk.logging import logger


celery = Celery(__name__)
TaskBase = celery.Task


def try_cast(v):
    # False and 0 are getting converted to datetime by arrow
    if v is None or isinstance(v, bool) or v == 0:
        return v

    try:
        str_to_date(v)  # try if it matches format
        return arrow.get(v).datetime  # return timezone aware time
    except Exception:
        try:
            return ObjectId(v)
        except Exception:
            return v


def dumps(o):
    with superdesk.app.app_context():
        return MongoJSONEncoder().encode(o)


def loads(s):
    o = json.loads(s)
    with superdesk.app.app_context():
        return serialize(o)


def serialize(o):
    if isinstance(o, list):
        return [serialize(item) for item in o]
    elif isinstance(o, dict):
        if o.get('kwargs') and not isinstance(o['kwargs'], dict):
            o['kwargs'] = json.loads(o['kwargs'])
        return {k: serialize(v) for k, v in o.items()}
    else:
        return try_cast(o)


register('eve/json', dumps, loads, content_type='application/json')


def handle_exception(exc):
    """Log exception to logger."""
    logger.exception(exc)


class AppContextTask(TaskBase):
    abstract = True
    serializer = 'eve/json'
    app_errors = (
        SuperdeskError,
        werkzeug.exceptions.InternalServerError,  # mongo layer err
    )

    def __call__(self, *args, **kwargs):
        with superdesk.app.app_context():
            try:
                return super().__call__(*args, **kwargs)
            except self.app_errors as e:
                handle_exception(e)

    def on_failure(self, exc, task_id, args, kwargs, einfo):
        with superdesk.app.app_context():
            handle_exception(exc)


celery.Task = AppContextTask


def init_celery(app):
    celery.config_from_object(app.config, namespace='CELERY')
    app.celery = celery
    app.redis = __get_redis(app)


def add_subtask_to_progress(task_id):
    return _update_subtask_progress(task_id, total=True)


def finish_subtask_from_progress(task_id):
    return _update_subtask_progress(task_id, current=True)


def finish_task_for_progress(task_id):
    return _update_subtask_progress(task_id, done=True)


def __get_redis(app_ctx):
    """Constructs Redis Client object.

    :return: Redis Client object
    """
    redis_url = app_ctx.config['REDIS_URL']
    try:
        return redis.from_url(redis_url)
    except ValueError as e:
        logger.warn('Failed to connect to redis using a connection string: {}'.format(e))

        # Newer Redis clients will not accept 'amqp' as the scheme
        # Attempt to mock a redis scheme instead
        protocol = redis_url.split('//')[0]
        new_url = redis_url.replace(protocol, 'redis:')
        return redis.from_url(new_url)


def update_key(key, flag=False, db=None):

    if db is None:
        db = app.redis

    if flag:
        crt_value = db.incr(key)
    else:
        crt_value = db.get(key)

    if crt_value:
        crt_value = int(crt_value)
    else:
        crt_value = 0

    return crt_value


def _update_subtask_progress(task_id, current=None, total=None, done=None):
    redis_db = redis.from_url(app.config['REDIS_URL'])
    try:
{'fraction_non_alphanumeric': 0.099171344560278, 'fraction_numerical': 0.006014434643143544, 'mean_word_length': 3.0779291553133517, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 2, 'lorem ipsum': 0, 'www.': 2, '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': '5413939', 'n_tokens_mistral': 2484, 'n_tokens_neox': 2380, 'n_words': 658}