# -*- coding: utf-8 -*-
# Part of Odoo. See LICENSE file for full copyright and licensing details.
from odoo import _, api, models, fields
from odoo.exceptions import ValidationError
from odoo.fields import Command
from odoo.osv import expression
from odoo.tools import html2plaintext, is_html_empty, email_normalize, plaintext2html
from collections import defaultdict
from markupsafe import Markup
class ChatbotScriptStep(models.Model):
_name = 'chatbot.script.step'
_description = 'Chatbot Script Step'
_order = 'sequence, id'
_rec_name = 'message'
message = fields.Text(string='Message', translate=True)
sequence = fields.Integer(string='Sequence')
chatbot_script_id = fields.Many2one(
'chatbot.script', string='Chatbot', required=True, ondelete='cascade')
step_type = fields.Selection([
('text', 'Text'),
('question_selection', 'Question'),
('question_email', 'Email'),
('question_phone', 'Phone'),
('forward_operator', 'Forward to Operator'),
('free_input_single', 'Free Input'),
('free_input_multi', 'Free Input (Multi-Line)'),
], default='text', required=True)
# answers
answer_ids = fields.One2many(
'chatbot.script.answer', 'script_step_id',
copy=True, string='Answers')
triggering_answer_ids = fields.Many2many(
'chatbot.script.answer', domain="[('script_step_id.sequence', '<', sequence)]",
compute='_compute_triggering_answer_ids', readonly=False, store=True,
copy=False, # copied manually, see chatbot.script#copy
string='Only If', help='Show this step only if all of these answers have been selected.')
# forward-operator specifics
is_forward_operator_child = fields.Boolean(compute='_compute_is_forward_operator_child')
@api.depends('sequence')
def _compute_triggering_answer_ids(self):
for step in self.filtered('triggering_answer_ids'):
update_command = [Command.unlink(answer.id) for answer in step.triggering_answer_ids
if answer.script_step_id.sequence >= step.sequence]
if update_command:
step.triggering_answer_ids = update_command
@api.depends('sequence', 'triggering_answer_ids', 'chatbot_script_id.script_step_ids.triggering_answer_ids',
'chatbot_script_id.script_step_ids.answer_ids', 'chatbot_script_id.script_step_ids.sequence')
def _compute_is_forward_operator_child(self):
parent_steps_by_chatbot = {}
for chatbot in self.chatbot_script_id:
parent_steps_by_chatbot[chatbot.id] = chatbot.script_step_ids.filtered(
lambda step: step.step_type in ['forward_operator', 'question_selection']
).sorted(lambda s: s.sequence, reverse=True)
for step in self:
parent_steps = parent_steps_by_chatbot[step.chatbot_script_id.id].filtered(
lambda s: s.sequence < step.sequence
)
parent = step
while True:
parent = parent._get_parent_step(parent_steps)
if not parent or parent.step_type == 'forward_operator':
break
step.is_forward_operator_child = parent and parent.step_type == 'forward_operator'
@api.model_create_multi
def create(self, vals_list):
""" Ensure we correctly assign sequences when creating steps.
Indeed, sequences are very important within the script, and will break the whole flow if
not correctly defined.
This override will group created steps by chatbot_id and increment the sequence accordingly.
It will also look for an existing step for that chatbot and resume from the highest sequence.
This cannot be done in a default_value for the sequence field as we cannot search by
runbot_id.
It is also safer and more efficient to do it here (we can batch everything).
It is still possible to manually pass the 'sequence' in the values, which will take priority. """
vals_by_chatbot_id = {}
for vals in vals_list:
chatbot_id = vals.get('chatbot_script_id')
if chatbot_id:
step_values = vals_by_chatbot_id.get(chatbot_id, [])
step_values.append(vals)
vals_by_chatbot_id[chatbot_id] = step_values
max_sequence_by_chatbot = {}
if vals_by_chatbot_id:
read_group_results = self.env['chatbot.script.step'].read_group(
[('chatbot_script_id', 'in', list(vals_by_chatbot_id.keys()))],
['sequence:max'],
['chatbot_script_id']
)
max_sequence_by_chatbot = {
read_group_result['chatbot_script_id'][0]: read_group_result['sequence']
for read_group_result in read_group_results
}
for chatbot_id, step_vals in vals_by_chatbot_id.items():
current_sequence = 0
if chatbot_id in max_sequence_by_chatbot:
current_sequence = max_sequence_by_chatbot[chatbot_id] + 1
for vals in step_vals:
if 'sequence' in vals:
current_sequence = vals.get('sequence')
else:
vals['sequence'] = current_sequence
current_sequence += 1
return super().create(vals_list)
# --------------------------
# Business Methods
# --------------------------
def _chatbot_prepare_customer_values(self, mail_channel, create_partner=True, update_partner=True):
""" Common method that allows retreiving default customer values from the mail.channel
following a chatbot.script.
This method will return a dict containing the 'customer' values such as:
{
'partner': The created partner (see 'create_partner') or the partner from the
environment if not public
'email': The email extracted from the mail.channel messages
(see step_type 'question_email')
'phone': The phone extracted from the mail.channel messages
(see step_type 'question_phone')
'description': A default description containing the "Please contact me on" and "Please
call me on" with the related email and phone numbers.
Can be used as a default description to create leads or tickets for example.
}
:param record mail_channel: the mail.channel holding the visitor's conversation with the bot.
:param bool create_partner: whether or not to create a res.partner is the current user is public.
Defaults to True.
:param bool update_partner: whether or not to set update the email and phone on the res.partner
from the environment (if not a public user) if those are not set yet. Defaults to True.
:return dict: a dict containing the customer values."""
partner = False
user_inputs = mail_channel._chatbot_find_customer_values_in_messages({
'question_email': 'email',
'question_phone': 'phone',
})
input_email = user_inputs.get('email', False)
input_phone = user_inputs.get('phone', False)
if self.env.user._is_public() and create_partner:
partner = self.env['res.partner'].create({
'name': input_email,
'email': input_email,
'phone': input_phone,
})
elif not self.env.user._is_public():
partner = self.env.user.partner_id
if update_partner:
# update email/phone value from partner if not set
update_values = {}
if input_email and not partner.email:
update_values['email'] = input_email
if input_phone and not partner.phone:
update_values['phone'] = input_phone
if update_values:
partner.write(update_values)
description = Markup('')
if input_email:
description += Markup('%s<strong>%s</strong><br>') % (_('Please contact me on: '), input_email)
if input_phone:
description += Markup('%s<strong>%s</strong><br>') % (_('Please call me on: '), input_phone)
if description:
description += Markup('<br>')
return {
'partner': partner,
'email': input_email,
'phone': input_phone,
'description': description,
}
def _fetch_next_step(self, selected_answer_ids):
""" Fetch the next step depending on the user's selected answers.
If a step contains multiple triggering answers from the same step the condition between
them must be a 'OR'. If is contains multiple triggering answers from different steps the
condition between them must be a 'AND'.
e.g:
STEP 1 : A B
STEP 2 : C D
STEP 3 : E
STEP 4 ONLY IF A B C E
Scenario 1 (A C E):
A in (A B) -> OK
C in (C) -> OK
E in (E) -> OK
-> OK
Scenario 2 (B D E):
B in (A B) -> OK
D in (C) -> NOK
E in (E) -> OK
-> NOK
"""
self.ensure_one()
domain = [('chatbot_script_id', '=', self.chatbot_script_id.id), ('sequence', '>', self.sequence)]
if selected_answer_ids:
domain = expression.AND([domain, [
'|',
('triggering_answer_ids', '=', False),
('triggering_answer_ids', 'in', selected_answer_ids.ids)]])
steps = self.env['chatbot.script.step'].search(domain)
for step in steps:
if not step.triggering_answer_ids:
return step
answers_by_step = defaultdict(list)
for answer in step.triggering_answer_ids:
answers_by_step[answer.script_step_id.id].append(answer)
if all(any(answer in step_triggering_answers for answer in selected_answer_ids)
for step_triggering_answers in answers_by_step.values()):
return step
return self.env['chatbot.script.step']
def _get_parent_step(self, all_parent_steps):
""" Returns the first preceding step that matches either the triggering answers
or the possible answers the user can select """
self.ensure_one()
if not self.chatbot_script_id.ids:
return self.env['chatbot.script.step']
for step in all_parent_steps:
if step.sequence >= self.sequence:
continue
if self.triggering_answer_ids:
if not (all(answer in self.triggering_answer_ids for answer in step.triggering_answer_ids) or
any(answer in self.triggering_answer_ids for answer in step.answer_ids)):
continue
elif step.triggering_answer_ids:
continue
return step
return self.env['chatbot.script.step']
def _is_last_step(self, mail_channel=False):
self.ensure_one()
mail_channel = mail_channel or self.env['mail.channel']
# if it's not a question and if there is no next step, then we end the script
if self.step_type != 'question_selection' and not self._fetch_next_step(
mail_channel.chatbot_message_ids.user_script_answer_id):
return True
return False
def _process_answer(self, mail_channel, message_body):
""" Method called when the user reacts to the current chatbot.script step.
For most chatbot.script.step#step_types it simply returns the next chatbot.script.step of
the script (see '_fetch_next_step').
Some extra processing is done for steps of type 'question_email' and 'question_phone' where
we store the user raw answer (the mail message HTML body) into the chatbot.message in order
to be able to recover it later (see '_chatbot_prepare_customer_values').
:param mail_channel:
:param message_body:
:return: script step to display next
:rtype: 'chatbot.script.step' """
self.ensure_one()
user_text_answer = html2plaintext(message_body)
if self.step_type == 'question_email' and not email_normalize(user_text_answer):
# if this error is raised, display an error message but do not go to next step
raise ValidationError(_('"%s" is not a valid email.', user_text_answer))
if self.step_type in ['question_email', 'question_phone']:
chatbot_message = self.env['chatbot.message'].search([
('mail_channel_id', '=', mail_channel.id),
('script_step_id', '=', self.id),
], limit=1)
if chatbot_message:
chatbot_message.write({'user_raw_answer': message_body})
self.env.flush_all()
return self._fetch_next_step(mail_channel.chatbot_message_ids.user_script_answer_id)
def _process_step(self, mail_channel):
""" When we reach a chatbot.step in the script we need to do some processing on behalf of
the bot. Which is for most chatbot.script.step#step_types just posting the message field.
Some extra processing may be required for special step types such as 'forward_operator',
'create_lead', 'create_ticket' (in their related bridge modules).
Those will have a dedicated processing method with specific docstrings.
Returns the mail.message posted by the chatbot's operator_partner_id. """
self.ensure_one()
# We change the current step to the new step
mail_channel.chatbot_current_step_id = self.id
if self.step_type == 'forward_operator':
return self._process_step_forward_operator(mail_channel)
return mail_channel._chatbot_post_message(self.chatbot_script_id, plaintext2html(self.message))
def _process_step_forward_operator(self, mail_channel):
""" Special type of step that will add a human operator to the conversation when reached,
which stops the script and allow the visitor to discuss with a real person.
In case we don't find any operator (e.g: no-one is available) we don't post any messages.
The script will continue normally, which allows to add extra steps when it's the case
(e.g: ask for the visitor's email and create a lead). """
human_operator = False
posted_message = False
if mail_channel.livechat_channel_id:
human_operator = mail_channel.livechat_channel_id._get_random_operator()
# handle edge case where we found yourself as available operator -> don't do anything
# it will act as if no-one is available (which is fine)
if human_operator and human_operator != self.env.user:
mail_channel.sudo().add_members(
human_operator.partner_id.ids,
open_chat_window=True,
post_joined_message=False)
# rename the channel to include the operator's name
mail_channel.sudo().name = ' '.join([
self.env.user.display_name if not self.env.user._is_public() else mail_channel.anonymous_name,
human_operator.livechat_username if human_operator.livechat_username else human_operator.name
])
if self.message:
# first post the message of the step (if we have one)
posted_message = mail_channel._chatbot_post_message(self.chatbot_script_id, plaintext2html(self.message))
# then post a small custom 'Operator has joined' notification
mail_channel._chatbot_post_message(
self.chatbot_script_id,
Markup('<div class="o_mail_notification">%s</div>') % _('%s has joined', human_operator.partner_id.name))
mail_channel._broadcast(human_operator.partner_id.ids)
mail_channel.channel_pin(pinned=True)
return posted_message
# --------------------------
# Tooling / Misc
# --------------------------
def _format_for_frontend(self):
""" Small utility method that formats the step into a dict usable by the frontend code. """
self.ensure_one()
return {
'chatbot_script_step_id': self.id,
'chatbot_step_answers': [{
'id': answer.id,
'label': answer.name,
'redirect_link': answer.redirect_link,
} for answer in self.answer_ids],
'chatbot_step_message': plaintext2html(self.message) if not is_html_empty(self.message) else False,
'chatbot_step_is_last': self._is_last_step(),
'chatbot_step_type': self.step_type
}