Solventación sobre funcionalidad de un botón de acción en odoo 16 enterprise

[Bryan Palomeque]

Buenos días con todos, he encontrado este grupo de ayuda de odoo y quisiera poder solventar una duda que tengo. Lo que quiero lograr hacer es que un botón que se llama "marcar como hecho y enviar informe" ejecute dos métodos python; el primero llamado "action_mark_done_and_send_email" y el segundo es "action_send_email_report".
A continucación este es el xml de la vista heredada para el módulo field service:

xml version="1.0" encoding="utf-8"?>
<odoo>
<record id="view_task_form2_field_service_management" model="ir.ui.view">
<field name="name">project.task.form.inherit</field>
<field name="model">project.task</field>
<field name="inherit_id" ref="project.view_task_form2"/>
<field name="arch" type="xml">
<xpath expr="//field[@name='user_ids']" position="after">
<field name="agency"/>
<field name="city"/>
<field name="agency_address"/>
<field name="helpdesk_ticket_id"/>
</xpath>
<xpath expr="//form/header" position="inside">
<button name="action_mark_done_and_send_email" string="Marcar como hecho y enviar informe" type="object" class="btn-primary"/>
<button name="action_send_email_report" type="object" string="Enviar informe" class="btn-secondary"
attrs="{'invisible': [('display_send_report_secondary', '=', False)]}"/>
</xpath>
</field>
</record>
</odoo>

Y este es el código python con los modelos

from odoo import models, fields, api, _

class TaskFieldService(models.Model):
_inherit = 'project.task'

agency = fields.Char(string="Agency")
city = fields.Char(string="City")
agency_address = fields.Char(string="Agency Address")

# método python para enviar el informe de la tarea al cliente
def action_send_email_report(self):
tasks_with_report = self.filtered(lambda task: task._is_fsm_report_available())
if not tasks_with_report:
return {
'type': 'ir.actions.client',
'tag': 'display_notification',
'params': {
'message': _("There are no reports to send."),
'sticky': False,
'type': 'danger',
}
}

template_id = self.env.ref('industry_fsm.mail_template_data_task_report').id
return {
'name': _("Send report"),
'type': 'ir.actions.act_window',
'view_mode': 'form',
'res_model': 'mail.compose.message',
'views': [(False, 'form')],
'view_id': False,
'target': 'new',
'context': {
'default_composition_mode': 'mass_mail' if len(tasks_with_report.ids) > 1 else 'comment',
'default_model': 'project.task',
'default_res_id': tasks_with_report.ids[0],
'default_use_template': bool(template_id),
'default_template_id': template_id,
'fsm_mark_as_sent': True,
'active_ids': tasks_with_report.ids,
'mailing_document_based': True,
},
}

def _get_report_base_filename(self):
self.ensure_one()
return 'Worksheet %s - %s' % (self.name, self.partner_id.name)

def _is_fsm_report_available(self):
self.ensure_one()
return self.comment or self.timesheet_ids

def has_to_be_signed(self):
self.ensure_one()
return self._is_fsm_report_available() and not self.worksheet_signature

@api.model
def get_views(self, views, options=None):
options['toolbar'] = not self._context.get('task_worksheet_comment') and options.get('toolbar')
res = super().get_views(views, options)
return res

# ---------------------------------------------------------
# Business Methods
# ---------------------------------------------------------

def _message_post_after_hook(self, message, msg_vals):
if self.env.context.get('fsm_mark_as_sent') and not self.fsm_is_sent:
self.fsm_is_sent = True
return super()._message_post_after_hook(message, msg_vals)

#método python para marcar como hechoy pasar a la siguiente etapa de facturación
def action_mark_done_and_send_email(self, stop_running_timers=False):
""" If allow billable on task, timesheet product set on project and user has privileges :
Create SO confirmed with time and material.
"""
res = super().action_fsm_validate(stop_running_timers)
if res is True:
billable_tasks = self.filtered(
lambda task: task.allow_billable and (task.allow_timesheets or task.allow_material))
timesheets_read_group = self.env['account.analytic.line'].sudo().read_group(
[('task_id', 'in', billable_tasks.ids), ('project_id', '!=', False)], ['task_id', 'id'],
['task_id'])
timesheet_count_by_task_dict = {timesheet['task_id'][0]: timesheet['task_id_count'] for timesheet in
timesheets_read_group}
for task in billable_tasks:
timesheet_count = timesheet_count_by_task_dict.get(task.id)
if not task.sale_order_id and not timesheet_count: # Prevent creating/confirming a SO if there are no products and timesheets
continue
task._fsm_ensure_sale_order()
if task.allow_timesheets:
task._fsm_create_sale_order_line()
if task.sudo().sale_order_id.state in ['draft', 'sent']:
task.sudo().sale_order_id.action_confirm()
billable_tasks._prepare_materials_delivery()
self.action_send_email_report()
return res

def _fsm_ensure_sale_order(self):
""" get the SO of the task. If no one, create it and return it """
self.ensure_one()
if not self.sale_order_id:
self._fsm_create_sale_order()
return self.sale_order_id

def _fsm_create_sale_order(self):
""" Create the SO from the task, with the 'service product' sales line and link all timesheet to that line it """
self.ensure_one()
if not self.partner_id:
raise UserError(_('A customer should be set on the task to generate a worksheet.'))

SaleOrder = self.env['sale.order']
if self.user_has_groups('project.group_project_user'):
SaleOrder = SaleOrder.sudo()

user_id = self.user_ids[0] if self.user_ids else self.env['res.users']
team = self.env['crm.team'].sudo()._get_default_team_id(user_id=user_id.id, domain=None)
sale_order = SaleOrder.create({
'partner_id': self.partner_id.id,
'company_id': self.company_id.id,
'analytic_account_id': self._get_task_analytic_account_id().id,
'team_id': team.id if team else False,
'origin': _('%s - %s') % (self.project_id.name, self.name),
})
# update after creation since onchange_partner_id sets the current user
sale_order.user_id = user_id.id

self.sale_order_id = sale_order

def _fsm_create_sale_order_line(self):
""" Generate sales order item based on the pricing_type on the project and the timesheets in the current task

When the pricing_type = 'employee_rate', we need to search the employee mappings for the employee who timesheeted
in the current task to retrieve the product in each mapping and generate an SOL for this product with the total
hours of the related timesheet(s) as the ordered quantity. Some SOLs can be already generated if the user manually
adds the SOL in the task or when he adds some materials in the tasks, a SO is generated.
If the user manually adds in the SO some service products, we must check in these before generating new one.
When no SOL is linked to the task before marking this task as done and no existing SOLs correspond to the default
product in the project, we take the first SOL generated if no generated SOL contain the default product of the project.
Here are the steps realized for this case:
1) Get all timesheets in the tasks
2) Classify this timesheets by employee
3) Search the employee mappings (project.sale.line.employee.map model or the sale_line_employee_ids field in the
project model) for the employee who timesheets to have the product linked to the employee.
4) Use the dict created in the second step to classify the timesheets in another dict in which the key is the id
and the price_unit of the product and the id uom. This information is important for the generation of the SOL.
5) if no SOL is linked in the task then we add the default service project defined in the project into the dict
created in the previous step and value is the remaining timesheets.
That is, the ones are no impacted in the employee mappings (sale_line_employee_ids field) defined in the project.
6) Classify the existing SOLs of the SO linked to the task, because the SO can be generated before the user clicks
on 'mark as done' button, for instance, when the user adds materials for this task. A dict is created containing
the id and price_unit of the product as key and the SOL(s) containing this product.
6.1) If no SOL is linked, then we check in the existing SOLs if there is a SOL with the default product defined
in the product, if it is the case then the SOL will be linked to the task.
This step can be useless if the user doesn't manually add a service product in the SO. In fact, this step
searchs in the SOLs of the SO, if there is an SOL with the default service product defined in the project.
If it is the case then the SOL will be linked to the task.
7) foreach in the dict created in the step 4, in this loop, first of all, we search in the dict containing the
existing SOLs if the id of the product is containing in an existing SOL. If yes then, we don't generate an SOL
and link it to the timesheets linked to this product. Otherwise, we generate the SOL with the information containing
in the key and the timesheets containing in the value of the dict for this key.

When the pricing_type = 'task_rate', we generate a sales order item with product_uom_qty is equal to the total hours of timesheets in the task.
Once the SOL is generated we link this one to the task and its timesheets.
"""
self.ensure_one()
# Get all timesheets in the current task (step 1)
not_billed_timesheets = self.env['account.analytic.line'].sudo().search(
[('task_id', '=', self.id), ('project_id', '!=', False), ('is_so_line_edited', '=', False)]).filtered(
lambda t: t._is_not_billed())
if self.pricing_type == 'employee_rate':
# classify these timesheets by employee (step 2)
timesheets_by_employee_dict = defaultdict(
lambda: self.env['account.analytic.line']) # key: employee_id, value: timesheets
for timesheet in not_billed_timesheets:
timesheets_by_employee_dict[timesheet.employee_id.id] |= timesheet

# Search the employee mappings for the employees whose timesheets in the task (step 3)
employee_mappings = self.env['project.sale.line.employee.map'].search([
('employee_id', 'in', list(timesheets_by_employee_dict.keys())),
('timesheet_product_id', '!=', False),
('project_id', '=', self.project_id.id)])

# Classify the timesheets by product (step 4)
product_timesheets_dict = defaultdict(lambda: self.env[
'account.analytic.line']) # key: (timesheet_product_id.id, price_unit, uom_id.id), value: list of timesheets
for mapping in employee_mappings:
employee_timesheets = timesheets_by_employee_dict[mapping.employee_id.id]
product_timesheets_dict[
mapping.timesheet_product_id.id, mapping.price_unit, mapping.timesheet_product_id.uom_id.id] |= employee_timesheets
not_billed_timesheets -= employee_timesheets # we remove the timesheets because are linked to the mapping

product = self.env['product.product']
sol_in_task = bool(self.sale_line_id)
if not sol_in_task: # Then, add the default product of the project and remaining timesheets in the dict (step 5)
default_product = self.project_id.timesheet_product_id
if not_billed_timesheets:
# The remaining timesheets must be added in the sol with the default product defined in the fsm project
# if there is not SOL in the task
product = default_product
product_timesheets_dict[
product.id, product.lst_price, product.uom_id.id] |= not_billed_timesheets
elif (default_product.id, default_product.lst_price,
default_product.uom_id.id) in product_timesheets_dict:
product = default_product

# Get all existing service sales order items in the sales order (step 6)
existing_service_sols = self.sudo().sale_order_id.order_line.filtered('is_service')
sols_by_product_and_price_dict = defaultdict(
lambda: self.env['sale.order.line']) # key: (product_id, price_unit), value: sales order items
for sol in existing_service_sols: # classify the SOLs to easily find the ones that we want.
sols_by_product_and_price_dict[sol.product_id.id, sol.price_unit] |= sol

task_values = defaultdict() # values to update the current task
update_timesheet_commands = [] # used to update the so_line field of each timesheet in the current task.

if not sol_in_task and sols_by_product_and_price_dict: # Then check in the existing sol if a SOL has the default product defined in the project to set the SOL of the task (step 6.1)
sol = sols_by_product_and_price_dict.get(
(self.project_id.timesheet_product_id.id, self.project_id.timesheet_product_id.lst_price))
if sol:
task_values['sale_line_id'] = sol.id
sol_in_task = True

for (timesheet_product_id, price_unit, uom_id), timesheets in product_timesheets_dict.items():
sol = sols_by_product_and_price_dict.get((timesheet_product_id,
price_unit)) # get the existing SOL with the product and the correct price unit
if not sol: # Then we create it
sol = self.env['sale.order.line'].sudo().create({
'order_id': self.sale_order_id.id,
'product_id': timesheet_product_id,
'price_unit': price_unit,
# The project and the task are given to prevent the SOL to create a new project or task based on the config of the product.
'project_id': self.project_id.id,
'task_id': self.id,
'product_uom_qty': sum(timesheets.mapped('unit_amount')),
'product_uom': uom_id,
})

# Link the SOL to the timesheets
update_timesheet_commands.extend(
[fields.Command.update(timesheet.id, {'so_line': sol.id}) for timesheet in timesheets if
not timesheet.is_so_line_edited])
if not sol_in_task and (
not product or (product.id == timesheet_product_id and product.lst_price == price_unit)):
# If there is no sol in task and the product variable is empty then we give the first sol in this loop to the task
# However, if the product is not empty then we search the sol with the same product and unit price to give to the current task
task_values['sale_line_id'] = sol.id
sol_in_task = True

if update_timesheet_commands:
task_values['timesheet_ids'] = update_timesheet_commands

self.sudo().write(task_values)
elif not self.sale_line_id:
# Check if there is a SOL containing the default product of the project before to create a new one.
sale_order_line = self.sale_order_id and self.sudo().sale_order_id.order_line.filtered(
lambda sol: sol.product_id == self.project_id.timesheet_product_id)[:1]
if not sale_order_line:
sale_order_line = self.env['sale.order.line'].sudo().create({
'order_id': self.sale_order_id.id,
'product_id': self.timesheet_product_id.id,
# The project and the task are given to prevent the SOL to create a new project or task based on the config of the product.
'project_id': self.project_id.id,
'task_id': self.id,
'product_uom_qty': sum(timesheet_id.unit_amount for timesheet_id in not_billed_timesheets),
})
self.sudo().write({ # We need to sudo in case the user cannot see all timesheets in the current task.
'sale_line_id': sale_order_line.id,
# assign SOL to timesheets
'timesheet_ids': [fields.Command.update(timesheet.id, {'so_line': sale_order_line.id}) for timesheet
in not_billed_timesheets if not timesheet.is_so_line_edited]
})

def _prepare_materials_delivery(self):
# While industry_fsm_stock is not installed then we automatically deliver materials
read_group_timesheets = self.env['account.analytic.line'].sudo().search_read(
[('task_id', 'in', self.ids), ('project_id', '!=', False), ('so_line', '!=', False)], ['so_line'])
timesheet_sol_ids = [timesheet['so_line'][0] for timesheet in read_group_timesheets]
sale_order_lines = self.env['sale.order.line'].sudo().search([
('id', 'not in', timesheet_sol_ids),
('task_id', 'in', self.ids),
('order_id', 'in', self.sale_order_id.sudo().filtered(lambda so: so.state == 'sale').ids),
])
for sol in sale_order_lines:
sol.qty_delivered = sol.product_uom_qty

def has_to_be_signed(self):
return super().has_to_be_signed() or (self.sale_order_id and not self.worksheet_signature)

class ProjectTaskRecurrence(models.Model):
_inherit = 'project.task.recurrence'

def _get_sale_line_id(self, task):
if not task.is_fsm:
return super()._get_sale_line_id(task)
return False

Yo estoy trabajando en odoo16 enterprise