<commit_before># -*- test-case-name: tests.test_ttc -*-
import sys
import traceback
import re

from twisted.internet.defer import inlineCallbacks, Deferred, returnValue
from twisted.internet import task, reactor

from pymongo import MongoClient
from bson.objectid import ObjectId

from redis import Redis

from datetime import datetime, time, date, timedelta
import pytz

from vumi.application import ApplicationWorker
from vumi.message import Message, TransportUserMessage, TransportEvent
from vumi.application import SessionManager
from vumi import log
from vumi.errors import VumiError

from vusion.utils import (
    time_to_vusion_format, get_local_time, get_local_time_as_timestamp,
    time_from_vusion_format, get_shortcode_value, get_offset_date_time,
    split_keywords, add_char_to_pattern, dynamic_content_notation_to_string,
    clean_phone)
from vusion.error import (
    MissingData, SendingDatePassed, VusionError, MissingTemplate,
    MissingProperty)
from vusion.message import DispatcherControl, WorkerControl
from vusion.context import Context
from vusion.component import (
    DialogueWorkerPropertyHelper, CreditManager, RedisLogger)

from vusion.persist.action import (
    Actions, action_generator, FeedbackAction, EnrollingAction, OptinAction,
    OptoutAction, RemoveRemindersAction)
from vusion.persist import (
    FeedbackSchedule, HistoryManager, ContentVariableManager, DialogueManager,
    RequestManager, ParticipantManager, ScheduleManager,
    ProgramCreditLogManager, ShortcodeManager, UnattachedMessageManager,
    Interaction)

from vusion.connectors import (
    ReceiveWorkerControlConnector, SendControlConnector)


class DialogueWorker(ApplicationWorker):

    @inlineCallbacks
    def setup_application(self):   
        self.sender = None
        self.r_prefix = None
        self.r_config = {}
        self.control_name = None
        self.transport_name = None
        self.program_name = None
        
        #Store basic configuration data
        self.transport_name = self.config['transport_name']
        self.control_name = self.config['control_name']
        self.r_config = self.config.get('redis_config', {})
        self.r_prefix = "%(control_name)s:" % self.config

        #Initializing
        self.program_name = None
        self.last_script_used = None
        self.r_key = 'vusion:programs:' + self.config['database_name']
        self.r_server = Redis(**self.r_config)

        # Component / Manager initialization
        self.logger = RedisLogger(
           self.config['database_name'], 
           self.r_key,
           self.r_server)

        self.collections = {}
        self.init_program_db(
           self.config['database_name'],
           self.config['vusion_database_name'])

        self.properties = DialogueWorkerPropertyHelper(
           self.collections['program_settings'],
           self.collections['shortcodes'])

        self.logger.startup(self.properties)

        #TODO replace by a loop
        for collection in ['history', 'dialogues', 'requests', 'participants', 
                           'content_variables', 'schedules', 'credit_logs',
                           'shortcodes', 'unattached_messages']:
            self.collections[collection].set_property_helper(self.properties)
            self.collections[collection].set_log_helper(self.logger)

        self.credit_manager = CreditManager(
           self.r_key, self.r_server,
           self.collections['credit_logs'],
           self.collections['history'], 
           self.collections['schedules'],
           self.properties, 
           self.logger)

        self.logger.log("Dialogue Worker is starting")
        yield self.setup_dc_connector(self.config['dispatcher_name'])
        #Will need to register the keywords
        self.load_properties(if_needed_register_keywords=True)
        self.sender = reactor.callLater(2, self.daemon_process)
        
    def teardown_application(self):
        self.log("Worker is stopped.")
        self.logger.stop()
        if self.is_ready():
            self.unregister_from_dispatcher()
        if (self.sender.active()):
            self.sender.cancel()

    def setup_connectors(self):
        d = super(DialogueWorker, self).setup_connectors()

        def cb2(connector):
            connector.set_control_handler(self.dispatch_control)
            return connector
        
        return d.addCallback(cb2)

    @inlineCallbacks
    def dispatch_control(self, control):
        yield self.consume_control(control)

    def setup_ri_connector(self, connector_name):
        return self.setup_connector(
            ReceiveWorkerControlConnector,
            self.transport_name)

    def setup_dc_connector(self, connector_name):
        return self.setup_connector(SendControlConnector, connector_name)

    def save_history(self, **kwargs):
        return self.collections['history'].save_history(**kwargs)
  
    def init_program_db(self, database_name, vusion_database_name):
        self.log("Initialization of the program")
        self.database_name = database_name
        self.vusion_database_name = vusion_database_name
        self.log("Connecting to database: %s" % self.database_name)

        #Initilization of the database
        mongo_client = MongoClient(
            self.config['mongodb_host'],
            self.config['mongodb_port'],
            w=1) #write in safe mode by default

        ## Program specific
        program_db = mongo_client[self.database_name]
        self.setup_collections(program_db, {'program_settings': None,
                                            'unattached_messages': None})
        self.collections['history'] = HistoryManager(
            program_db, 'history', self.r_prefix, self.r_server)
        self.collections['content_variables'] = ContentVariableManager(
            program_db, 'content_variables', 'content_variable_tables')
        self.collections['dialogues'] = DialogueManager(
            program_db, 'dialogues')
        self.collections['requests'] = RequestManager(
            program_db, 'requests')
        self.collections['participants'] = ParticipantManager(
            program_db, 'participants')
        self.collections['schedules'] = ScheduleManager(
            program_db, 'schedules')
        self.collections['unattached_messages'] = UnattachedMessageManager(
            program_db, 'unattached_messages')

        ## Vusion 
        vusion_db = mongo_client[self.vusion_database_name]
        self.setup_collections(vusion_db, {'templates': None})
        self.collections['shortcodes'] = ShortcodeManager(
            vusion_db, 'shortcodes')
        self.collections['credit_logs'] = ProgramCreditLogManager(
            vusion_db, 'credit_logs', self.database_name)

    def setup_collections(self, db, names):
        for name, index in names.items():
            self.setup_collection(db, name, index)

    def setup_collection(self, db, name, index):
        if name in db.collection_names():
            self.collections[name] = db[name]
        else:
            self.collections[name] = db.create_collection(name)
        if index is not None:
            self.collections[name].ensure_index(index, background=True)
        self.log("Collection initialised: %s" % name)

    @inlineCallbacks
    def consume_control(self, message):
        try:
            self.log("Control message received to %r" % (message,))
            message = WorkerControl(**message.payload)
            if message['action'] == 'reload_program_settings':
                self.load_properties()
                return
            if (not self.is_ready()):
                self.log("Worker is not ready, cannot performe the action.")
                return
            if message['action'] == 'update_schedule':
                if message['schedule_type'] == 'dialogue':
                    self.collections['dialogues'].load_dialogue(message['object_id'])
                    self.schedule_dialogue(message['object_id'])
                    self.register_keywords_in_dispatcher()
                elif message['schedule_type'] == 'unattach':
                    yield self.schedule_unattach(message['object_id'])
                elif message['schedule_type'] == 'participant':
                    yield self.schedule_participant(message['object_id'])
                self.update_time_next_daemon_iteration()
            elif message['action'] == 'mass_tag':
                yield self.schedule_mass_tag(message['tag'], message['selector'])
                self.update_time_next_daemon_iteration()
            elif message['action'] == 'mass_untag':
                yield self.schedule_mass_untag(message['tag'])
            elif message['action'] == 'reload_request':
                self.collections['requests'].load_request(message['object_id'])
                self.register_keywords_in_dispatcher()
            elif message['action'] == 'test_send_all_messages':
                dialogue = self.collections['dialogues'].get_dialogue_obj(message['dialogue_obj_id'])
                self.send_all_messages(dialogue, message['phone_number'])
            elif message['action'] == 'update_registered_keywords':
                self.register_keywords_in_dispatcher()
            elif message['action'] == 'run_actions':
                actions = self.collections['dialogues'].get_actions(
                    message['dialogue_id'],
                    message['interaction_id'],
                    message['answer'])
                for action in actions.items():
                    yield self.run_action(message['participant_phone'], action)
        except (VusionError, VumiError) as e:
            self.log('ERROR: %s(%s)' % (e.__class__.__name__, e.message), level='error')
        except:
            exc_type, exc_value, exc_traceback = sys.exc_info()
            self.log(
                "UNKNOWN ERROR during consume control message: %r" %
                traceback.format_exception(exc_type, exc_value, exc_traceback))

    def dispatch_event(self, message):
        self.log("Event message received %s" % (message,))
        new_status, old_status, credits = self.collections['history'].update_status_from_event(message)
        if new_status is None:
            return
        self.collections['credit_logs'].increment_event_counter(old_status, new_status, credits)

    def update_participant_transport_metadata(self, message):
        if message['transport_metadata'] is not {}:
            self.collections['participants'].save_transport_metadata(
                message['from_addr'], message['transport_metadata'])

    @inlineCallbacks
    def run_action(self, participant_phone, action, context=Context(),
                   participant_session_id=None):
        if action.has_condition():
            query = action.get_condition_mongodb_for(participant_phone, participant_session_id)
            if not self.collections['participants'].is_matching(query):
                self.log(("Participant %s doesn't satify the condition for action for %s" % (participant_phone, action,)))
                return
        self.log(("Run action for %s action %s" % (participant_phone, action,)))
        if (action.get_type() == 'optin'):
            if self.collections['participants'].opting_in(participant_phone):
                self.schedule_participant(participant_phone)
            else:
                ## The participant is still optin and opting in again
                if self.properties['double-optin-error-feedback'] is not None:
                    self.run_action(
                        participant_phone, 
                        FeedbackAction(**{'content': self.properties['double-optin-error-feedback']}),
                        context,
                        participant_session_id)
        elif (action.get_type() == 'optout'):
            self.collections['participants'].opting_out(participant_phone)
            self.collections['schedules'].remove_participant_schedules(participant_phone)
        elif (action.get_type() == 'feedback'):
            schedule = FeedbackSchedule(**{
                'participant-phone': participant_phone,
                'participant-session-id': participant_session_id,
                'date-time': time_to_vusion_format(self.get_local_time()),
                'content': self.customize_message(action['content'], participant_phone, context, False),
                'context': context.get_dict_for_history()})
            self.send_schedule(schedule)
        elif (action.get_type() == 'unmatching-answer'):
            setting = self.collections['program_settings'].find_one({
                'key': 'default-template-unmatching-answer'})
            if setting is None:
                return
            template = self.collections['templates'].find_one({
                '_id': ObjectId(setting['value'])})
            if template is None:
                return
            regex_ANSWER = re.compile('ANSWER')
            error_message = re.sub(regex_ANSWER,
                                   action['answer'],
                                   template['template'])
            schedule = FeedbackSchedule(**{
                'participant-phone': participant_phone,
                'participant-session-id': participant_session_id,
                'date-time': time_to_vusion_format(self.get_local_time()),
                'content': error_message,
                'context': context.get_dict_for_history()})
            self.send_schedule(schedule)
        elif (action.get_type() == 'tagging'):
            self.collections['participants'].tagging(participant_phone, action['tag'])
            yield self.schedule_participant(participant_phone)
        elif (action.get_type() == 'enrolling'):
            if not self.collections['participants'].is_optin(participant_phone):
                self.run_action(
                    participant_phone, 
                    OptinAction(), 
                    context, 
                    participant_session_id)
            self.collections['participants'].enrolling(
                participant_phone, action['enroll'])
            dialogue = self.collections['dialogues'].get_current_dialogue(action['enroll'])
            if dialogue is None:
                self.log(("Enrolling error: Missing Dialogue %s" % action['enroll']))
                return
            participant = self.collections['participants'].get_participant(participant_phone)
            self.schedule_participant_dialogue(participant, dialogue)
        elif (action.get_type() == 'delayed-enrolling'):
            schedule_time = get_offset_date_time(
                self.get_local_time(), 
                action['offset-days']['days'],
                action['offset-days']['at-time'])
            action = EnrollingAction(**{'enroll': action['enroll']})
            self.collections['schedules'].add_action(
                participant_phone, participant_session_id,
                schedule_time, action, context)
        elif (action.get_type() == 'profiling'):
            self.collections['participants'].labelling(
                participant_phone, action['label'], action['value'], context['message'])
            yield self.schedule_participant(participant_phone)
        elif (action.get_type() == 'offset-conditioning'):
            participant = self.collections['participants'].get_participant(participant_phone, True)
            if participant is None:
                return
            self.schedule_participant_dialogue(
                participant,
                self.collections['dialogues'].get_current_dialogue(action['dialogue-id']))
        elif (action.get_type() == 'remove-question'):
            self.collections['schedules'].remove_participant_interaction(
                participant_phone, action['dialogue-id'], action['interaction-id'])
        elif (action.get_type() == 'remove-reminders'):
            self.collections['schedules'].remove_participant_reminders(
                participant_phone, action['dialogue-id'], action['interaction-id'])
        elif (action.get_type() == 'remove-deadline'):
            self.collections['schedules'].remove_participant_deadline(
                participant_phone, action['dialogue-id'], action['interaction-id'])
        elif (action.get_type() == 'reset'):
            participant = self.collections['participants'].get_participant(participant_phone)
            participant_labels = participant['profile']
            participant_tags = participant['tags']
            self.run_action(participant_phone, OptoutAction())
            self.run_action(participant_phone, OptinAction())
            labels = action.get_keep_labels(participant['profile'])
            tags = action.get_keep_tags(participant['tags'])
            for tag in tags:
                self.collections['participants'].tagging(participant_phone, tag)
            for label in labels:
                self.collections['participants'].labelling(participant_phone, label['label'], label['value'], '')
            yield self.schedule_participant(participant_phone)
        elif (action.get_type() == 'proportional-tagging'):
            yield self.run_action_proportional_tagging(participant_phone, action)
        elif (action.get_type() == 'proportional-labelling'):
            yield self.run_action_proportional_labelling(participant_phone, action)
        elif (action.get_type() == 'url-forwarding'):
            yield self.run_action_url_forwarding(participant_phone, action, context, participant_session_id)
        elif (action.get_type() == 'sms-forwarding'):
            yield self.run_action_sms_forwarding(participant_phone, action, context)
        elif (action.get_type() == 'sms-invite'):
            yield self.run_action_sms_invite(participant_phone, action, context)
        elif (action.get_type() == 'save-content-variable-table'):
            yield self.run_action_save_content_variable(
                participant_phone, action, context)
        else:
            self.log("The action is not supported %s" % action.get_type())

    @inlineCallbacks
    def run_action_url_forwarding(self, participant_phone, action, context, participant_session_id):
        if not self.properties.is_sms_forwarding_allowed():
            self.log('SMS Forwarding not allowed, dump action')
            return
        history = self.collections['history'].get_history(context['history_id'])
        participant = self.collections['participants'].get_participant(participant_phone)
        options = {
           'from_addr': self.transport_name,
           'transport_name': self.transport_name,
           'transport_type': 'http_api',
           'transport_metadata': {
               'program_shortcode': self.properties['shortcode'],
               'participant_phone': participant_phone,
               'participant_profile': participant['profile']}}
        message = yield self.send_to(
            action['forward-url'], history['message-content'], **options)
        self.collections['history'].update_forwarding(
            context['history_id'], message['message_id'], action['forward-url'])

    @inlineCallbacks
    def run_action_proportional_tagging(self, participant_phone, action, context=None):
        if self.collections['participants'].is_tagged(participant_phone, action.get_tags()):
            return
        for tag in action.get_tags():
            count = yield self.collections['participants'].count_tag_async(tag)
            action.set_tag_count(tag, count)
        self.run_action(participant_phone, action.get_tagging_action())

    @inlineCallbacks
    def run_action_proportional_labelling(self, participant_phone, action, context=None):
        if self.collections['participants'].is_labelled(participant_phone, action.get_label_name()):
            return
        for label in action.get_labels():
            count = yield self.collections['participants'].count_label_async(label)
            action.set_count(label['value'], count)
        self.run_action(participant_phone, action.get_labelling_action())

    @inlineCallbacks
    def run_action_sms_forwarding(self, participant_phone, action, context):
        sender = self.collections['participants'].get_participant(participant_phone)
        query = action.get_query_selector(sender, context)
        participants = self.collections['participants'].get_participants(query)
        if participants.count() == 0 and action.has_no_participant_feedback():
            content = self.customize_message(
                action.get_no_participant_feedback(),
                sender['phone'], 
                context)
            schedule = FeedbackSchedule(**{
                'participant-phone': sender['phone'],
                'participant-session-id': sender['session-id'],
                'date-time': self.get_local_time('vusion'),
                'content': content,
                'context': context.payload})
            yield self.send_schedule(schedule)
            return
        for participant in participants:
            content = self.customize_message(
                action['forward-content'],
                participant_phone,
                context)
            schedule = FeedbackSchedule(**{
                'participant-phone': participant['phone'],
                'participant-session-id': participant['session-id'],
                'date-time': self.get_local_time('vusion'),
                'content': content,
                'context': context.payload})
            yield self.send_schedule(schedule)

    @inlineCallbacks
    def run_action_sms_invite(self, participant_phone, action, context):
        sender = self.collections['participants'].get_participant(participant_phone)
        invitee_phone = clean_phone(context.get_message_second_word())

        if invitee_phone is None or not self.collections['participants'].opting_in(invitee_phone):
            schedule = FeedbackSchedule(**{
                'participant-phone': sender['phone'],
                'participant-session-id': sender['session-id'],
                'date-time': self.get_local_time('vusion'),
                'content': action['feedback-inviter'],
                'context': context.payload})
            yield self.send_schedule(schedule)
            return
        else:
            ## The participant is opting
            invitee = self.collections['participants'].get_participant(invitee_phone)
            self.collections['participants'].tagging(
                invitee['phone'], action['invitee-tag'])
            content = self.customize_message(
                action['invite-content'],
                participant_phone,
                context)
            schedule = FeedbackSchedule(**{
                'participant-phone': invitee['phone'],
                'participant-session-id': invitee['session-id'],
                'date-time': self.get_local_time('vusion'),
                'content': content,
                'context': context.payload})
            yield self.send_schedule(schedule)

    def run_action_save_content_variable(self, participant_phone, action, context):
        match = action.get_match()
        self._run_action_save_content_variable(
            participant_phone, action, context, match, context.get_matching_answer())
        extra_matchs = action.get_extra_matchs()
        for (match, value) in extra_matchs:
            self._run_action_save_content_variable(
                participant_phone, action, context, match, value)

    def _run_action_save_content_variable(self, participant_phone, action,
                                          context, match, value):
        for i, key in match.iteritems():
            match[i] = self.customize_message(key, participant_phone, context)
        value = self.customize_message(value, participant_phone, context)
        self.collections['content_variables'].save_content_variable(
            match, value, action.get_table_id())

    def consume_user_message(self, message):
        self.log("User message received from %s '%s'" % (message['from_addr'],
                                                         message['content']))
        try:
            history = {'object-type': 'unmatching-history'}
            context = Context(**{'message': message['content']})
            actions = Actions()
            self.collections['requests'].get_matching_request_actions(
                message['content'], actions, context)
            if context.is_matching():
                history = {'object-type': 'request-history'}
            else:
                self.collections['dialogues'].get_matching_dialogue_actions(
                    message['content'], actions, context)
                if context.is_matching():
                    history = {'object-type': 'dialogue-history'}
            # High priority to run an optin or enrolling action to get sessionId 
            if (not self.collections['participants'].is_optin(message['from_addr']) 
                    and (actions.contains('optin') or actions.contains('enrolling'))):
                self.run_action(message['from_addr'], actions.get_priority_action(), context)
            participant = self.collections['participants'].get_participant(message['from_addr'], only_optin=True)
            message_credits = self.properties.use_credits(message['content'])
            history.update({
                'participant-phone': message['from_addr'],
                'participant-session-id': (participant['session-id'] if participant else None),
                'message-content': message['content'],
                'message-direction': 'incoming',
                'message-credits': message_credits})
            history.update(context.get_dict_for_history())
            history_id = self.save_history(**history)
            context['history_id'] = str(history_id)
            self.credit_manager.received_message(message_credits)
            self.update_participant_transport_metadata(message)
            if (context.is_matching() and participant is not None):
                if ('interaction' in context):
                    if self.collections['history'].has_oneway_marker(
                        participant['phone'], participant['session-id'],
                        context['dialogue-id'], context['interaction-id']):
                        actions.clear_all()
                    else:
                        self.get_program_actions(participant, context, actions)
                        if self.collections['history'].participant_has_max_unmatching_answers(participant, context['dialogue-id'], context['interaction']):
                            self.collections['history'].add_oneway_marker(
                                participant['phone'], participant['session-id'], context)
                            context['interaction'].get_max_unmatching_action(context['dialogue-id'], actions)
                elif ('request-id' in context):
                    self.get_program_actions(participant, context, actions)                    
                self.run_actions(participant, context, actions)
        except:
            exc_type, exc_value, exc_traceback = sys.exc_info()
            self.log(
                "Error during consume user message: %r" %<commit_msg>remove function from dialogue worker<commit_after># -*- test-case-name: tests.test_ttc -*-
import sys
import traceback
import re

from twisted.internet.defer import inlineCallbacks, Deferred, returnValue
from twisted.internet import task, reactor

from pymongo import MongoClient
from bson.objectid import ObjectId

from redis import Redis

from datetime import datetime, time, date, timedelta
import pytz

from vumi.application import ApplicationWorker
from vumi.message import Message, TransportUserMessage, TransportEvent
from vumi.application import SessionManager
from vumi import log
from vumi.errors import VumiError

from vusion.utils import (
    time_to_vusion_format, get_local_time, get_local_time_as_timestamp,
    time_from_vusion_format, get_shortcode_value, get_offset_date_time,
    split_keywords, add_char_to_pattern, dynamic_content_notation_to_string,
    clean_phone)
from vusion.error import (
    MissingData, SendingDatePassed, VusionError, MissingTemplate,
    MissingProperty)
from vusion.message import DispatcherControl, WorkerControl
from vusion.context import Context
from vusion.component import (
    DialogueWorkerPropertyHelper, CreditManager, RedisLogger)

from vusion.persist.action import (
    Actions, action_generator, FeedbackAction, EnrollingAction, OptinAction,
    OptoutAction, RemoveRemindersAction)
from vusion.persist import (
    FeedbackSchedule, HistoryManager, ContentVariableManager, DialogueManager,
    RequestManager, ParticipantManager, ScheduleManager,
    ProgramCreditLogManager, ShortcodeManager, UnattachedMessageManager,
    Interaction)

from vusion.connectors import (
    ReceiveWorkerControlConnector, SendControlConnector)


class DialogueWorker(ApplicationWorker):

    @inlineCallbacks
    def setup_application(self):   
        self.sender = None
        self.r_prefix = None
        self.r_config = {}
        self.control_name = None
        self.transport_name = None
        self.program_name = None
        
        #Store basic configuration data
        self.transport_name = self.config['transport_name']
        self.control_name = self.config['control_name']
        self.r_config = self.config.get('redis_config', {})
        self.r_prefix = "%(control_name)s:" % self.config

        #Initializing
        self.program_name = None
        self.last_script_used = None
        self.r_key = 'vusion:programs:' + self.config['database_name']
        self.r_server = Redis(**self.r_config)

        # Component / Manager initialization
        self.logger = RedisLogger(
           self.config['database_name'], 
           self.r_key,
           self.r_server)

        self.collections = {}
        self.init_program_db(
           self.config['database_name'],
           self.config['vusion_database_name'])

        self.properties = DialogueWorkerPropertyHelper(
           self.collections['program_settings'],
           self.collections['shortcodes'])

        self.logger.startup(self.properties)

        #TODO replace by a loop
        for collection in ['history', 'dialogues', 'requests', 'participants', 
                           'content_variables', 'schedules', 'credit_logs',
                           'shortcodes', 'unattached_messages']:
            self.collections[collection].set_property_helper(self.properties)
            self.collections[collection].set_log_helper(self.logger)

        self.credit_manager = CreditManager(
           self.r_key, self.r_server,
           self.collections['credit_logs'],
           self.collections['history'], 
           self.collections['schedules'],
           self.properties, 
           self.logger)

        self.logger.log("Dialogue Worker is starting")
        yield self.setup_dc_connector(self.config['dispatcher_name'])
        #Will need to register the keywords
        self.load_properties(if_needed_register_keywords=True)
        self.sender = reactor.callLater(2, self.daemon_process)
        
    def teardown_application(self):
        self.log("Worker is stopped.")
        self.logger.stop()
        if self.is_ready():
            self.unregister_from_dispatcher()
        if (self.sender.active()):
            self.sender.cancel()

    def setup_connectors(self):
        d = super(DialogueWorker, self).setup_connectors()

        def cb2(connector):
            connector.set_control_handler(self.dispatch_control)
            return connector
        
        return d.addCallback(cb2)

    @inlineCallbacks
    def dispatch_control(self, control):
        yield self.consume_control(control)

    def setup_ri_connector(self, connector_name):
        return self.setup_connector(
            ReceiveWorkerControlConnector,
            self.transport_name)

    def setup_dc_connector(self, connector_name):
        return self.setup_connector(SendControlConnector, connector_name)

    def save_history(self, **kwargs):
        return self.collections['history'].save_history(**kwargs)
  
    def init_program_db(self, database_name, vusion_database_name):
        self.log("Initialization of the program")
        self.database_name = database_name
        self.vusion_database_name = vusion_database_name
        self.log("Connecting to database: %s" % self.database_name)

        #Initilization of the database
        mongo_client = MongoClient(
            self.config['mongodb_host'],
            self.config['mongodb_port'],
            w=1) #write in safe mode by default

        ## Program specific
        program_db = mongo_client[self.database_name]
        self.setup_collections(program_db, {'program_settings': None,
                                            'unattached_messages': None})
        self.collections['history'] = HistoryManager(
            program_db, 'history', self.r_prefix, self.r_server)
        self.collections['content_variables'] = ContentVariableManager(
            program_db, 'content_variables', 'content_variable_tables')
        self.collections['dialogues'] = DialogueManager(
            program_db, 'dialogues')
        self.collections['requests'] = RequestManager(
            program_db, 'requests')
        self.collections['participants'] = ParticipantManager(
            program_db, 'participants')
        self.collections['schedules'] = ScheduleManager(
            program_db, 'schedules')
        self.collections['unattached_messages'] = UnattachedMessageManager(
            program_db, 'unattached_messages')

        ## Vusion 
        vusion_db = mongo_client[self.vusion_database_name]
        self.setup_collections(vusion_db, {'templates': None})
        self.collections['shortcodes'] = ShortcodeManager(
            vusion_db, 'shortcodes')
        self.collections['credit_logs'] = ProgramCreditLogManager(
            vusion_db, 'credit_logs', self.database_name)

    def setup_collections(self, db, names):
        for name, index in names.items():
            self.setup_collection(db, name, index)

    def setup_collection(self, db, name, index):
        if name in db.collection_names():
            self.collections[name] = db[name]
        else:
            self.collections[name] = db.create_collection(name)
        if index is not None:
            self.collections[name].ensure_index(index, background=True)
        self.log("Collection initialised: %s" % name)

    @inlineCallbacks
    def consume_control(self, message):
        try:
            self.log("Control message received to %r" % (message,))
            message = WorkerControl(**message.payload)
            if message['action'] == 'reload_program_settings':
                self.load_properties()
                return
            if (not self.is_ready()):
                self.log("Worker is not ready, cannot performe the action.")
                return
            if message['action'] == 'update_schedule':
                if message['schedule_type'] == 'dialogue':
                    self.collections['dialogues'].load_dialogue(message['object_id'])
                    self.schedule_dialogue(message['object_id'])
                    self.register_keywords_in_dispatcher()
                elif message['schedule_type'] == 'unattach':
                    yield self.schedule_unattach(message['object_id'])
                elif message['schedule_type'] == 'participant':
                    yield self.schedule_participant(message['object_id'])
                self.update_time_next_daemon_iteration()
            elif message['action'] == 'mass_tag':
                yield self.schedule_mass_tag(message['tag'], message['selector'])
                self.update_time_next_daemon_iteration()
            elif message['action'] == 'mass_untag':
                yield self.schedule_mass_untag(message['tag'])
            elif message['action'] == 'reload_request':
                self.collections['requests'].load_request(message['object_id'])
                self.register_keywords_in_dispatcher()
            elif message['action'] == 'test_send_all_messages':
                dialogue = self.collections['dialogues'].get_dialogue_obj(message['dialogue_obj_id'])
                self.send_all_messages(dialogue, message['phone_number'])
            elif message['action'] == 'update_registered_keywords':
                self.register_keywords_in_dispatcher()
            elif message['action'] == 'run_actions':
                actions = self.collections['dialogues'].get_actions(
                    message['dialogue_id'],
                    message['interaction_id'],
                    message['answer'])
                for action in actions.items():
                    yield self.run_action(message['participant_phone'], action)
        except (VusionError, VumiError) as e:
            self.log('ERROR: %s(%s)' % (e.__class__.__name__, e.message), level='error')
        except:
            exc_type, exc_value, exc_traceback = sys.exc_info()
            self.log(
                "UNKNOWN ERROR during consume control message: %r" %
                traceback.format_exception(exc_type, exc_value, exc_traceback))

    def dispatch_event(self, message):
        self.log("Event message received %s" % (message,))
        new_status, old_status, credits = self.collections['history'].update_status_from_event(message)
        if new_status is None:
            return
        self.collections['credit_logs'].increment_event_counter(old_status, new_status, credits)

    @inlineCallbacks
    def run_action(self, participant_phone, action, context=Context(),
                   participant_session_id=None):
        if action.has_condition():
            query = action.get_condition_mongodb_for(participant_phone, participant_session_id)
            if not self.collections['participants'].is_matching(query):
                self.log(("Participant %s doesn't satify the condition for action for %s" % (participant_phone, action,)))
                return
        self.log(("Run action for %s action %s" % (participant_phone, action,)))
        if (action.get_type() == 'optin'):
            if self.collections['participants'].opting_in(participant_phone):
                self.schedule_participant(participant_phone)
            else:
                ## The participant is still optin and opting in again
                if self.properties['double-optin-error-feedback'] is not None:
                    self.run_action(
                        participant_phone, 
                        FeedbackAction(**{'content': self.properties['double-optin-error-feedback']}),
                        context,
                        participant_session_id)
        elif (action.get_type() == 'optout'):
            self.collections['participants'].opting_out(participant_phone)
            self.collections['schedules'].remove_participant_schedules(participant_phone)
        elif (action.get_type() == 'feedback'):
            schedule = FeedbackSchedule(**{
                'participant-phone': participant_phone,
                'participant-session-id': participant_session_id,
                'date-time': time_to_vusion_format(self.get_local_time()),
                'content': self.customize_message(action['content'], participant_phone, context, False),
                'context': context.get_dict_for_history()})
            self.send_schedule(schedule)
        elif (action.get_type() == 'unmatching-answer'):
            setting = self.collections['program_settings'].find_one({
                'key': 'default-template-unmatching-answer'})
            if setting is None:
                return
            template = self.collections['templates'].find_one({
                '_id': ObjectId(setting['value'])})
            if template is None:
                return
            regex_ANSWER = re.compile('ANSWER')
            error_message = re.sub(regex_ANSWER,
                                   action['answer'],
                                   template['template'])
            schedule = FeedbackSchedule(**{
                'participant-phone': participant_phone,
                'participant-session-id': participant_session_id,
                'date-time': time_to_vusion_format(self.get_local_time()),
                'content': error_message,
                'context': context.get_dict_for_history()})
            self.send_schedule(schedule)
        elif (action.get_type() == 'tagging'):
            self.collections['participants'].tagging(participant_phone, action['tag'])
            yield self.schedule_participant(participant_phone)
        elif (action.get_type() == 'enrolling'):
            if not self.collections['participants'].is_optin(participant_phone):
                self.run_action(
                    participant_phone, 
                    OptinAction(), 
                    context, 
                    participant_session_id)
            self.collections['participants'].enrolling(
                participant_phone, action['enroll'])
            dialogue = self.collections['dialogues'].get_current_dialogue(action['enroll'])
            if dialogue is None:
                self.log(("Enrolling error: Missing Dialogue %s" % action['enroll']))
                return
            participant = self.collections['participants'].get_participant(participant_phone)
            self.schedule_participant_dialogue(participant, dialogue)
        elif (action.get_type() == 'delayed-enrolling'):
            schedule_time = get_offset_date_time(
                self.get_local_time(), 
                action['offset-days']['days'],
                action['offset-days']['at-time'])
            action = EnrollingAction(**{'enroll': action['enroll']})
            self.collections['schedules'].add_action(
                participant_phone, participant_session_id,
                schedule_time, action, context)
        elif (action.get_type() == 'profiling'):
            self.collections['participants'].labelling(
                participant_phone, action['label'], action['value'], context['message'])
            yield self.schedule_participant(participant_phone)
        elif (action.get_type() == 'offset-conditioning'):
            participant = self.collections['participants'].get_participant(participant_phone, True)
            if participant is None:
                return
            self.schedule_participant_dialogue(
                participant,
                self.collections['dialogues'].get_current_dialogue(action['dialogue-id']))
        elif (action.get_type() == 'remove-question'):
            self.collections['schedules'].remove_participant_interaction(
                participant_phone, action['dialogue-id'], action['interaction-id'])
        elif (action.get_type() == 'remove-reminders'):
            self.collections['schedules'].remove_participant_reminders(
                participant_phone, action['dialogue-id'], action['interaction-id'])
        elif (action.get_type() == 'remove-deadline'):
            self.collections['schedules'].remove_participant_deadline(
                participant_phone, action['dialogue-id'], action['interaction-id'])
        elif (action.get_type() == 'reset'):
            participant = self.collections['participants'].get_participant(participant_phone)
            participant_labels = participant['profile']
            participant_tags = participant['tags']
            self.run_action(participant_phone, OptoutAction())
            self.run_action(participant_phone, OptinAction())
            labels = action.get_keep_labels(participant['profile'])
            tags = action.get_keep_tags(participant['tags'])
            for tag in tags:
                self.collections['participants'].tagging(participant_phone, tag)
            for label in labels:
                self.collections['participants'].labelling(participant_phone, label['label'], label['value'], '')
            yield self.schedule_participant(participant_phone)
        elif (action.get_type() == 'proportional-tagging'):
            yield self.run_action_proportional_tagging(participant_phone, action)
        elif (action.get_type() == 'proportional-labelling'):
            yield self.run_action_proportional_labelling(participant_phone, action)
        elif (action.get_type() == 'url-forwarding'):
            yield self.run_action_url_forwarding(participant_phone, action, context, participant_session_id)
        elif (action.get_type() == 'sms-forwarding'):
            yield self.run_action_sms_forwarding(participant_phone, action, context)
        elif (action.get_type() == 'sms-invite'):
            yield self.run_action_sms_invite(participant_phone, action, context)
        elif (action.get_type() == 'save-content-variable-table'):
            yield self.run_action_save_content_variable(
                participant_phone, action, context)
        else:
            self.log("The action is not supported %s" % action.get_type())

    @inlineCallbacks
    def run_action_url_forwarding(self, participant_phone, action, context, participant_session_id):
        if not self.properties.is_sms_forwarding_allowed():
            self.log('SMS Forwarding not allowed, dump action')
            return
        history = self.collections['history'].get_history(context['history_id'])
        participant = self.collections['participants'].get_participant(participant_phone)
        options = {
           'from_addr': self.transport_name,
           'transport_name': self.transport_name,
           'transport_type': 'http_api',
           'transport_metadata': {
               'program_shortcode': self.properties['shortcode'],
               'participant_phone': participant_phone,
               'participant_profile': participant['profile']}}
        message = yield self.send_to(
            action['forward-url'], history['message-content'], **options)
        self.collections['history'].update_forwarding(
            context['history_id'], message['message_id'], action['forward-url'])

    @inlineCallbacks
    def run_action_proportional_tagging(self, participant_phone, action, context=None):
        if self.collections['participants'].is_tagged(participant_phone, action.get_tags()):
            return
        for tag in action.get_tags():
            count = yield self.collections['participants'].count_tag_async(tag)
            action.set_tag_count(tag, count)
        self.run_action(participant_phone, action.get_tagging_action())

    @inlineCallbacks
    def run_action_proportional_labelling(self, participant_phone, action, context=None):
        if self.collections['participants'].is_labelled(participant_phone, action.get_label_name()):
            return
        for label in action.get_labels():
            count = yield self.collections['participants'].count_label_async(label)
            action.set_count(label['value'], count)
        self.run_action(participant_phone, action.get_labelling_action())

    @inlineCallbacks
    def run_action_sms_forwarding(self, participant_phone, action, context):
        sender = self.collections['participants'].get_participant(participant_phone)
        query = action.get_query_selector(sender, context)
        participants = self.collections['participants'].get_participants(query)
        if participants.count() == 0 and action.has_no_participant_feedback():
            content = self.customize_message(
                action.get_no_participant_feedback(),
                sender['phone'], 
                context)
            schedule = FeedbackSchedule(**{
                'participant-phone': sender['phone'],
                'participant-session-id': sender['session-id'],
                'date-time': self.get_local_time('vusion'),
                'content': content,
                'context': context.payload})
            yield self.send_schedule(schedule)
            return
        for participant in participants:
            content = self.customize_message(
                action['forward-content'],
                participant_phone,
                context)
            schedule = FeedbackSchedule(**{
                'participant-phone': participant['phone'],
                'participant-session-id': participant['session-id'],
                'date-time': self.get_local_time('vusion'),
                'content': content,
                'context': context.payload})
            yield self.send_schedule(schedule)

    @inlineCallbacks
    def run_action_sms_invite(self, participant_phone, action, context):
        sender = self.collections['participants'].get_participant(participant_phone)
        invitee_phone = clean_phone(context.get_message_second_word())

        if invitee_phone is None or not self.collections['participants'].opting_in(invitee_phone):
            schedule = FeedbackSchedule(**{
                'participant-phone': sender['phone'],
                'participant-session-id': sender['session-id'],
                'date-time': self.get_local_time('vusion'),
                'content': action['feedback-inviter'],
                'context': context.payload})
            yield self.send_schedule(schedule)
            return
        else:
            ## The participant is opting
            invitee = self.collections['participants'].get_participant(invitee_phone)
            self.collections['participants'].tagging(
                invitee['phone'], action['invitee-tag'])
            content = self.customize_message(
                action['invite-content'],
                participant_phone,
                context)
            schedule = FeedbackSchedule(**{
                'participant-phone': invitee['phone'],
                'participant-session-id': invitee['session-id'],
                'date-time': self.get_local_time('vusion'),
                'content': content,
                'context': context.payload})
            yield self.send_schedule(schedule)

    def run_action_save_content_variable(self, participant_phone, action, context):
        match = action.get_match()
        self._run_action_save_content_variable(
            participant_phone, action, context, match, context.get_matching_answer())
        extra_matchs = action.get_extra_matchs()
        for (match, value) in extra_matchs:
            self._run_action_save_content_variable(
                participant_phone, action, context, match, value)

    def _run_action_save_content_variable(self, participant_phone, action,
                                          context, match, value):
        for i, key in match.iteritems():
            match[i] = self.customize_message(key, participant_phone, context)
        value = self.customize_message(value, participant_phone, context)
        self.collections['content_variables'].save_content_variable(
            match, value, action.get_table_id())

    def consume_user_message(self, message):
        self.log("User message received from %s '%s'" % (message['from_addr'],
                                                         message['content']))
        try:
            history = {'object-type': 'unmatching-history'}
            context = Context(**{'message': message['content']})
            actions = Actions()
            self.collections['requests'].get_matching_request_actions(
                message['content'], actions, context)
            if context.is_matching():
                history = {'object-type': 'request-history'}
            else:
                self.collections['dialogues'].get_matching_dialogue_actions(
                    message['content'], actions, context)
                if context.is_matching():
                    history = {'object-type': 'dialogue-history'}
            # High priority to run an optin or enrolling action to get sessionId 
            if (not self.collections['participants'].is_optin(message['from_addr']) 
                    and (actions.contains('optin') or actions.contains('enrolling'))):
                self.run_action(message['from_addr'], actions.get_priority_action(), context)
            participant = self.collections['participants'].get_participant(message['from_addr'], only_optin=True)
            message_credits = self.properties.use_credits(message['content'])
            history.update({
                'participant-phone': message['from_addr'],
                'participant-session-id': (participant['session-id'] if participant else None),
                'message-content': message['content'],
                'message-direction': 'incoming',
                'message-credits': message_credits})
            history.update(context.get_dict_for_history())
            history_id = self.save_history(**history)
            context['history_id'] = str(history_id)
            self.credit_manager.received_message(message_credits)
            self.collections['participants'].save_transport_metadata(
                message['from_addr'], message['transport_metadata'])
            if (context.is_matching() and participant is not None):
                if ('interaction' in context):
                    if self.collections['history'].has_oneway_marker(
                        participant['phone'], participant['session-id'],
                        context['dialogue-id'], context['interaction-id']):
                        actions.clear_all()
                    else:
                        self.get_program_actions(participant, context, actions)
                        if self.collections['history'].participant_has_max_unmatching_answers(participant, context['dialogue-id'], context['interaction']):
                            self.collections['history'].add_oneway_marker(
                                participant['phone'], participant['session-id'], context)
                            context['interaction'].get_max_unmatching_action(context['dialogue-id'], actions)
                elif ('request-id' in context):
                    self.get_program_actions(participant, context, actions)                    
                self.run_actions(participant, context, actions)
        except:
            exc_type, exc_value, exc_traceback = sys.exc_info()
            self.log(
                "Error during consume user message: %r" %
{'fraction_non_alphanumeric': 0.1058909891723252, 'fraction_numerical': 0.00018351991191044228, 'mean_word_length': 2.508757244043786, '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': 16, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1847967', 'n_tokens_mistral': 14520, 'n_tokens_neox': 13834, 'n_words': 2516}