<commit_before># Copyright 2015 <NAME>
#
# This file is part of CloudMailing.
#
# CloudMailing is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# CloudMailing is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with CloudMailing.  If not, see <http://www.gnu.org/licenses/>.


import logging
import time
from datetime import datetime, timedelta

from twisted.internet import task, defer
from twisted.internet.threads import deferToThread

from ..common.db_common import get_db
from . import settings_vars
from .models import Mailing, MailingTempQueue, MailingRecipient, MAILING_STATUS, RECIPIENT_STATUS, MAILING_TYPE
from ..common.singletonmixin import Singleton


class MailingManager(Singleton):
    """The CM mailing queue.

    Manage mailing relayers, keeping track of the existing connections,
    each connection's responsibility in term of messages. Create
    more relayers if the need arises.

    Someone should press .checkState periodically

    @ivar factory: A callable which returns a ClientFactory suitable for
    making SMTP connections.
    """

    def __init__(self):
        """
        MailingManager constructor.
        """
        Singleton.__init__(self)
        self.queue = None
        self.log = logging.getLogger("mailing")
        self.startTime = time.time()
        self.nextTime = 0
        self.filling_queue_running = False
        self.next_time_for_check_orphan_recipients = time.time() + 60
        self.tasks = []

    def start_tasks(self):
        for fn, delay, startNow in ((self.checkState, 5, False),
                                    (self.update_status_for_finished_mailings, 60, False),
                                    (self.check_orphan_recipients, 60, False),
                                    (self.purge_temp_queue_from_finished_and_paused_mailings, 60, False),
                                    (self.retrieve_customized_content, 60, False),
                                    ):
            t = task.LoopingCall(fn)
            t.start(delay, now=startNow)
            self.tasks.append(t)
        self.log.info("Mailing manager started")

    def stop_tasks(self):
        for t in self.tasks:
            t.stop()
        self.tasks = []
        self.log.info("Mailing manager stopped")

    def forceToCheck(self):
        self.nextTime = 0

    @staticmethod
    def make_mailings_queryset():
        now = datetime.utcnow()
        filter = {'$and': [
            {'$or': [{'scheduled_start': {'$lte': now}},
                     {'scheduled_start': None}]},
            {'$or': [{'scheduled_end': {'$gt': now}},
                     {'scheduled_end': None}]},
            {'status': {'$in': (MAILING_STATUS.READY, MAILING_STATUS.RUNNING)}},
        ]}
        return filter

    @staticmethod
    def make_recipients_queryset(mailing):
        filter = {
            '$and': [
                {'$or': [{'next_try': {'$lte': datetime.utcnow()}},
                         {'next_try': None}]},
                {'$or': [{'in_progress': False}, {'in_progress': None}]},
            ],
            'send_status': {'$in': (RECIPIENT_STATUS.READY,
                                    RECIPIENT_STATUS.WARNING), },
            'mailing.$id': mailing.id,
        }
        # TODO Missing 'exclude' but it should be useless if 'in_progress' is well filled.
        # queue = MailingRecipient.objects.filter(
        #     Q(smtp_next_time__lte=datetime.utcnow()) | Q(smtp_next_time__isnull=True),
        #     Q(in_progress=False) | Q(in_progress__isnull=True),
        #     send_status__in=(RECIPIENT_STATUS.READY,
        #                      RECIPIENT_STATUS.WARNING),Fill
        #     mailing__in=mailings_qs,
        # ).exclude(
        #     id__in=MailingTempQueue.objects.values_list('recipient__id', flat=True)
        # )
        return filter

    def filling_mailing_queue(self, mailing_filter, rcpt_count, temp_queue_count, mailing_queue_max_size):
        self.log.debug("Starting filling mailing queue...")
        t0 = time.time()
        count = 0
        try:

            for mailing in Mailing.find(mailing_filter):
                    if mailing.status == MAILING_STATUS.READY:
                        mailing.status = MAILING_STATUS.RUNNING
                        mailing.start_time = datetime.utcnow()
                        mailing.save()
                    # print "total_recipient: %d  / mailing_queue_max_size: %d / temp_queue_count: %d / rcpt_count: %d"\
                    # % (mailing.total_recipient, mailing_queue_max_size, temp_queue_count, rcpt_count)
                    nb_max = max(100, (mailing.total_pending or 0)
                                 * (mailing_queue_max_size - temp_queue_count) / rcpt_count)
                    # Quick starter for empty queue
                    if temp_queue_count == 0:
                        nb_max = min(1000, nb_max)
                    elif temp_queue_count <= 1000:
                        nb_max = min(10000, nb_max)
                    # print "nb_max = %d" % nb_max
                    rcpt_ids = []
                    self.log.debug("Filling mailing queue: selecting max %d recipients from mailing [%d]", nb_max,
                                   mailing.id)
                    filter = MailingManager.make_recipients_queryset(mailing)
                    t1 = time.time()
                    selected_recipients = MailingRecipient.find(filter).sort('next_try').limit(nb_max)
                    for rcpt in selected_recipients:
                        MailingTempQueue.add_recipient(mailing=mailing, recipient=rcpt)
                        rcpt_ids.append(rcpt.id)
                        count += 1
                        # if count % 100 == 0:
                        #     print "Added %d..." % count
                    self.log.debug("Filling mailing queue: selected %d recipients from mailing [%d] (in %.1f seconds)",
                                   len(rcpt_ids), mailing.id, time.time() - t1)
                    if rcpt_ids:
                        n = 100
                        for i in range(0, len(rcpt_ids), n):
                            MailingRecipient.update({'_id': {'$in': rcpt_ids[i:i + n]}}, {'$set': {'in_progress': True}}, multi=True)
<commit_msg>Fix bad (and negative) recipients count (Fix #4)<commit_after># Copyright 2015 <NAME>
#
# This file is part of CloudMailing.
#
# CloudMailing is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# CloudMailing is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with CloudMailing.  If not, see <http://www.gnu.org/licenses/>.


import logging
import time
from datetime import datetime, timedelta

from twisted.internet import task, defer
from twisted.internet.threads import deferToThread

from ..common.db_common import get_db
from . import settings_vars
from .models import Mailing, MailingTempQueue, MailingRecipient, MAILING_STATUS, RECIPIENT_STATUS, MAILING_TYPE
from ..common.singletonmixin import Singleton


class MailingManager(Singleton):
    """The CM mailing queue.

    Manage mailing relayers, keeping track of the existing connections,
    each connection's responsibility in term of messages. Create
    more relayers if the need arises.

    Someone should press .checkState periodically

    @ivar factory: A callable which returns a ClientFactory suitable for
    making SMTP connections.
    """

    def __init__(self):
        """
        MailingManager constructor.
        """
        Singleton.__init__(self)
        self.queue = None
        self.log = logging.getLogger("mailing")
        self.startTime = time.time()
        self.nextTime = 0
        self.filling_queue_running = False
        self.next_time_for_check_orphan_recipients = time.time() + 60
        self.tasks = []

    def start_tasks(self):
        for fn, delay, startNow in ((self.checkState, 5, False),
                                    (self.update_status_for_finished_mailings, 60, False),
                                    (self.check_orphan_recipients, 60, False),
                                    (self.purge_temp_queue_from_finished_and_paused_mailings, 60, False),
                                    (self.retrieve_customized_content, 60, False),
                                    ):
            t = task.LoopingCall(fn)
            t.start(delay, now=startNow)
            self.tasks.append(t)
        self.log.info("Mailing manager started")

    def stop_tasks(self):
        for t in self.tasks:
            t.stop()
        self.tasks = []
        self.log.info("Mailing manager stopped")

    def forceToCheck(self):
        self.nextTime = 0

    @staticmethod
    def make_mailings_queryset():
        now = datetime.utcnow()
        filter = {'$and': [
            {'$or': [{'scheduled_start': {'$lte': now}},
                     {'scheduled_start': None}]},
            {'$or': [{'scheduled_end': {'$gt': now}},
                     {'scheduled_end': None}]},
            {'status': {'$in': (MAILING_STATUS.READY, MAILING_STATUS.RUNNING)}},
        ]}
        return filter

    @staticmethod
    def make_recipients_queryset(mailing):
        filter = {
            '$and': [
                {'$or': [{'next_try': {'$lte': datetime.utcnow()}},
                         {'next_try': None}]},
                {'$or': [{'in_progress': False}, {'in_progress': None}]},
            ],
            'send_status': {'$in': (RECIPIENT_STATUS.READY,
                                    RECIPIENT_STATUS.WARNING), },
            'mailing.$id': mailing.id,
        }
        # TODO Missing 'exclude' but it should be useless if 'in_progress' is well filled.
        # queue = MailingRecipient.objects.filter(
        #     Q(smtp_next_time__lte=datetime.utcnow()) | Q(smtp_next_time__isnull=True),
        #     Q(in_progress=False) | Q(in_progress__isnull=True),
        #     send_status__in=(RECIPIENT_STATUS.READY,
        #                      RECIPIENT_STATUS.WARNING),Fill
        #     mailing__in=mailings_qs,
        # ).exclude(
        #     id__in=MailingTempQueue.objects.values_list('recipient__id', flat=True)
        # )
        return filter

    def filling_mailing_queue(self, mailing_filter, rcpt_count, temp_queue_count, mailing_queue_max_size):
        self.log.debug("Starting filling mailing queue...")
        t0 = time.time()
        count = 0
        try:
            db = get_db()

            for mailing in Mailing.find(mailing_filter):
                    if mailing.status == MAILING_STATUS.READY:
                        db.mailing.update({'_id': mailing.id}, {'$set': {
                            'status': MAILING_STATUS.RUNNING,
                            'start_time': datetime.utcnow(),
                        }})
                    # print "total_recipient: %d  / mailing_queue_max_size: %d / temp_queue_count: %d / rcpt_count: %d"\
                    # % (mailing.total_recipient, mailing_queue_max_size, temp_queue_count, rcpt_count)
                    nb_max = max(100, (mailing.total_pending or 0)
                                 * (mailing_queue_max_size - temp_queue_count) / rcpt_count)
                    # Quick starter for empty queue
                    if temp_queue_count == 0:
                        nb_max = min(1000, nb_max)
                    elif temp_queue_count <= 1000:
                        nb_max = min(10000, nb_max)
                    # print "nb_max = %d" % nb_max
                    rcpt_ids = []
                    self.log.debug("Filling mailing queue: selecting max %d recipients from mailing [%d]", nb_max,
                                   mailing.id)
                    filter = MailingManager.make_recipients_queryset(mailing)
                    t1 = time.time()
                    selected_recipients = MailingRecipient.find(filter).sort('next_try').limit(nb_max)
                    for rcpt in selected_recipients:
                        MailingTempQueue.add_recipient(mailing=mailing, recipient=rcpt)
                        rcpt_ids.append(rcpt.id)
                        count += 1
                        # if count % 100 == 0:
                        #     print "Added %d..." % count
                    self.log.debug("Filling mailing queue: selected %d recipients from mailing [%d] (in %.1f seconds)",
                                   len(rcpt_ids), mailing.id, time.time() - t1)
                    if rcpt_ids:
                        n = 100
                        for i in range(0, len(rcpt_ids), n):
                            MailingRecipient.update({'_id': {'$in': rcpt_ids[i:i + n]}}, {'$set': {'in_progress': True}}, multi=True)

{'fraction_non_alphanumeric': 0.1, 'fraction_numerical': 0.007470414201183432, 'mean_word_length': 1.8810995099083743, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 18, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6069004', 'n_tokens_mistral': 3961, 'n_tokens_neox': 3667, 'n_words': 953}