<commit_before>import logging

from flask import current_app as app
from superdesk import get_resource_service
from superdesk.celery_app import celery
from superdesk.metadata.item import CONTENT_TYPE, ITEM_TYPE
from superdesk.notification import push_notification

from .exceptions import APIConnectionError
from settings import SYNDICATION_CELERY_MAX_RETRIES, SYNDICATION_CELERY_COUNTDOWN
from .utils import send_api_request


logger = logging.getLogger('superdesk')


@celery.task(bind=True)
def send_post_to_consumer(self, syndication_out, producer_post, action='created'):
    """Send blog post updates to consumers webhook."""
    from .utils import extract_post_items_data, extract_producer_post_data
    consumers = get_resource_service('consumers')
    items = extract_post_items_data(producer_post)
    post = extract_producer_post_data(producer_post)
    try:
        consumers.send_post(syndication_out, {
            'items': items,
            'post': post
        }, action)
    except APIConnectionError as e:
        raise self.retry(exc=e, max_retries=SYNDICATION_CELERY_MAX_RETRIES, countdown=SYNDICATION_CELERY_COUNTDOWN)


@celery.task(bind=True)
def send_posts_to_consumer(self, syndication_out, action='created', limit=50, post_status='submitted'):
    """Send latest blog post updates to consumers webhook."""
    from .utils import extract_post_items_data, extract_producer_post_data
    consumers = get_resource_service('consumers')
    blog_id = syndication_out['blog_id']
    posts_service = get_resource_service('posts')
    lookup = {'blog': blog_id, ITEM_TYPE: CONTENT_TYPE.COMPOSITE, 'deleted': False, 'post_status': 'open'}

    posts = posts_service.find(lookup)

    # sort by order to get the latest first
    posts = posts.sort('order', -1)

    try:
        array = []
        for element in posts:
            array.append(element)
            if len(array) >= limit:
                break

        array.reverse()
        for producer_post in array:
            # Don't forward syndicated posts
            if 'repeat_syndication' in producer_post.keys():
                continue
            items = extract_post_items_data(producer_post)
            post = extract_producer_post_data(producer_post)
            # Force post_status for old posts
            post['post_status'] = post_status
            consumers.send_post(syndication_out, {
                'items': items,
                'post': post
            }, action)
    except APIConnectionError as e:
        logger.warning('Unable to send posts to consumer: {}'.format(e))
        raise self.retry(exc=e, max_retries=SYNDICATION_CELERY_MAX_RETRIES, countdown=SYNDICATION_CELERY_COUNTDOWN)


@celery.task(bind=True)
def check_webhook_status(self, consumer_id):
    """Check if consumer webhook is enabled by sending a fake http api request."""
    from .utils import send_api_request
    if app.config.get('SUPERDESK_TESTING'):
        return

    consumers = get_resource_service('consumers')
    consumer = consumers._get_consumer(consumer_id) or {}
    if 'webhook_url' in consumer:
        try:
            response = send_api_request(consumer['webhook_url'], None, method='GET', json_loads=False)<commit_msg>Send posts one by task instead of whole batch<commit_after>import logging
from celery import chain

from flask import current_app as app
from superdesk import get_resource_service
from superdesk.celery_app import celery
from superdesk.metadata.item import CONTENT_TYPE, ITEM_TYPE
from superdesk.notification import push_notification
from .exceptions import APIConnectionError
from settings import SYNDICATION_CELERY_MAX_RETRIES, SYNDICATION_CELERY_COUNTDOWN
from .utils import send_api_request


logger = logging.getLogger('superdesk')


@celery.task(bind=True)
def send_post_to_consumer(self, syndication_out, producer_post, action='created'):
    """Send blog post updates to consumers webhook."""
    from .utils import extract_post_items_data, extract_producer_post_data
    consumers = get_resource_service('consumers')
    items = extract_post_items_data(producer_post)
    post = extract_producer_post_data(producer_post)
    try:
        consumers.send_post(syndication_out, {
            'items': items,
            'post': post
        }, action)
    except APIConnectionError as e:
        raise self.retry(exc=e, max_retries=SYNDICATION_CELERY_MAX_RETRIES, countdown=SYNDICATION_CELERY_COUNTDOWN)


@celery.task(bind=True)
def send_posts_to_consumer(self, syndication_out, action='created', limit=50, post_status='submitted'):
    """
    Sends latest blog post updates to consumers webhook.

    When a consumer attaches to an existing blog, the producer sends a batch of posts that
    should be created sequentially (in the consumer) to preserve the order of the posts.
    To achieve this, we use a celery `chain` of subtasks. Subtasks in the chain must not
    raise any exception in order to continue the subsequent task and the next post is sent.
    """
    from .utils import extract_post_items_data, extract_producer_post_data

    posts_service = get_resource_service('posts')

    blog_id = syndication_out['blog_id']
    lookup = {'blog': blog_id, ITEM_TYPE: CONTENT_TYPE.COMPOSITE, 'deleted': False, 'post_status': 'open'}

    posts = posts_service.find(lookup)
    posts = posts.sort('order', -1)  # order to latest first

    array = []
    for element in posts:
        array.append(element)

        if len(array) >= limit:
            break

    subtasks = []
    array.reverse()
    for producer_post in array:
        is_repeated_syndication = 'repeat_syndication' in producer_post.keys()
        if is_repeated_syndication:
            continue

        items = extract_post_items_data(producer_post)
        post = extract_producer_post_data(producer_post)

        # Force post_status for old posts
        post['post_status'] = post_status

        post_data = {
            'items': items,
            'post': post
        }
        subtask_args = [syndication_out, post_data, action]

        # Because we are running the subtask in a chained mode, we need to provide
        # the initial param or result of other tasks as None. Read more here:
        # https://docs.celeryq.dev/en/stable/userguide/canvas.html#chains
        # This is required as we need to keep the order of the posts
        if len(subtasks) == 0:
            subtask_args.insert(0, None)

        send_internal = send_single_post_subtask.subtask(args=subtask_args)
        subtasks.append(send_internal)

    job = chain(subtasks)
    job.apply_async()


@celery.task(bind=True)
def send_single_post_subtask(self, chained_result, syndication_out, post_data, action):
    try:
        consumers = get_resource_service('consumers')
        consumers.send_post(syndication_out, post_data, action)
    except Exception as err:
        # once retry limit is reached, let's return instead of raising an exception
        # so the next subtask (post) will be also sent
        task_retries = SYNDICATION_CELERY_MAX_RETRIES + 1
        if task_retries == self.request.retries:
            post_id = post_data['post']['_id']
            consumer_id = syndication_out['consumer_id']
            consumer_blog_id = syndication_out['consumer_blog_id']

            logger.error(
                "Unable to send post '%s' to consumer '%s' and blog '%s'. Exception %s",
                post_id, consumer_id, consumer_blog_id, err)
            return

        raise self.retry(exc=err, max_retries=task_retries, countdown=SYNDICATION_CELERY_COUNTDOWN)


@celery.task(bind=True)
def check_webhook_status(self, consumer_id):
    """Check if consumer webhook is enabled by sending a fake http api request."""
    from .utils import send_api_request
    if app.config.get('SUPERDESK_TESTING'):
        return

    consumers = get_resource_service('consumers')
    consumer = consumers._get_consumer(consumer_id) or {}
    if 'webhook_url' in consumer:
        try:
            response = send_api_request(consumer['webhook_url'], None, method='GET', json_loads=False)
{'fraction_non_alphanumeric': 0.08159980188211986, 'fraction_numerical': 0.001114413075780089, 'mean_word_length': 3.492213570634038, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 5, 'https://': 1, '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': '5508873', 'n_tokens_mistral': 2412, 'n_tokens_neox': 2359, 'n_words': 624}