Files
Odoo-Modules/fusion_claims/models/technician_task.py
Nexa Admin 431052920e feat: separate fusion field service and LTC into standalone modules, update core modules
- fusion_claims: separated field service logic, updated controllers/views
- fusion_tasks: updated task views and map integration
- fusion_authorizer_portal: added page 11 signing, schedule booking, migrations
- fusion_shipping: new standalone shipping module (Canada Post, FedEx, DHL, Purolator)
- fusion_ltc_management: new standalone LTC management module
2026-03-11 16:19:52 +00:00

675 lines
27 KiB
Python

# -*- coding: utf-8 -*-
# Copyright 2024-2026 Nexa Systems Inc.
# License OPL-1 (Odoo Proprietary License v1.0)
"""
Fusion Technician Task - Claims Extension
Adds sale order, purchase order, and rental inspection
features to the base fusion.technician.task model.
"""
from odoo import models, fields, api, _
from odoo.exceptions import UserError, ValidationError
from markupsafe import Markup
import logging
_logger = logging.getLogger(__name__)
class FusionTechnicianTaskClaims(models.Model):
_inherit = 'fusion.technician.task'
# ------------------------------------------------------------------
# LINKED ORDER FIELDS
# ------------------------------------------------------------------
sale_order_id = fields.Many2one(
'sale.order',
string='Related SO',
tracking=True,
ondelete='restrict',
help='Sale order / case linked to this task',
)
sale_order_name = fields.Char(
related='sale_order_id.name',
string='Case Reference',
store=True,
)
purchase_order_id = fields.Many2one(
'purchase.order',
string='Related PO',
tracking=True,
ondelete='restrict',
help='Purchase order linked to this task (e.g. manufacturer pickup)',
)
purchase_order_name = fields.Char(
related='purchase_order_id.name',
string='PO Reference',
store=True,
)
# ------------------------------------------------------------------
# RENTAL INSPECTION
# ------------------------------------------------------------------
rental_inspection_condition = fields.Selection([
('excellent', 'Excellent'),
('good', 'Good'),
('fair', 'Fair'),
('damaged', 'Damaged'),
], string='Inspection Condition')
rental_inspection_notes = fields.Text(
string='Inspection Notes',
)
rental_inspection_photo_ids = fields.Many2many(
'ir.attachment',
'technician_task_inspection_photo_rel',
'task_id',
'attachment_id',
string='Inspection Photos',
)
rental_inspection_completed = fields.Boolean(
string='Inspection Completed',
default=False,
)
# ------------------------------------------------------------------
# ONCHANGES
# ------------------------------------------------------------------
@api.onchange('sale_order_id')
def _onchange_sale_order_id(self):
"""Auto-fill client and address from the sale order's shipping address."""
if self.sale_order_id:
self.purchase_order_id = False
order = self.sale_order_id
if not self.partner_id:
self.partner_id = order.partner_id
addr = order.partner_shipping_id or order.partner_id
self._fill_address_from_partner(addr)
@api.onchange('purchase_order_id')
def _onchange_purchase_order_id(self):
"""Auto-fill client and address from the purchase order's vendor."""
if self.purchase_order_id:
self.sale_order_id = False
order = self.purchase_order_id
if not self.partner_id:
self.partner_id = order.partner_id
addr = order.dest_address_id or order.partner_id
self._fill_address_from_partner(addr)
# ------------------------------------------------------------------
# CONSTRAINTS
# ------------------------------------------------------------------
@api.constrains('sale_order_id', 'purchase_order_id')
def _check_order_link(self):
for task in self:
if task.x_fc_sync_source:
continue
if task.task_type == 'ltc_visit':
continue
if not task.sale_order_id and not task.purchase_order_id:
raise ValidationError(_(
"A task must be linked to either a Sale Order (Case) or a Purchase Order."
))
# ------------------------------------------------------------------
# HOOK OVERRIDES
# ------------------------------------------------------------------
def _get_linked_order(self):
"""Return the linked sale or purchase order."""
return self.sale_order_id or self.purchase_order_id or False
def _create_vals_fill(self, vals):
"""Fill address from sale order or purchase order during create."""
if vals.get('sale_order_id') and not vals.get('address_street'):
order = self.env['sale.order'].browse(vals['sale_order_id'])
addr = order.partner_shipping_id or order.partner_id
if addr:
self._fill_address_vals(vals, addr)
if not vals.get('partner_id'):
vals['partner_id'] = order.partner_id.id
elif vals.get('purchase_order_id') and not vals.get('address_street'):
po = self.env['purchase.order'].browse(vals['purchase_order_id'])
addr = po.dest_address_id or po.partner_id
if addr:
self._fill_address_vals(vals, addr)
if not vals.get('partner_id'):
vals['partner_id'] = po.partner_id.id
else:
super()._create_vals_fill(vals)
def _on_create_post_actions(self):
"""Post-create actions: chatter notices, delivery marking, ODSP."""
for rec in self:
rec._post_task_created_to_linked_order()
if self.env.context.get('mark_ready_for_delivery'):
self._mark_sale_order_ready_for_delivery()
if self.env.context.get('mark_odsp_ready_for_delivery'):
for rec in self:
order = rec.sale_order_id
if order and order.x_fc_is_odsp_sale and order._get_odsp_status() != 'ready_delivery':
order._odsp_advance_status('ready_delivery',
"Order is ready for delivery. Delivery task scheduled.")
def _check_completion_requirements(self):
"""Check rental inspection requirement before completing pickup tasks."""
if self._is_rental_pickup_task() and not self.rental_inspection_completed:
raise UserError(_(
"Rental pickup tasks require a security inspection before "
"completion. Please complete the inspection from the "
"technician portal first."
))
def _on_complete_extra(self):
"""ODSP advancement and rental inspection on task completion."""
if (self.task_type == 'delivery'
and self.sale_order_id
and self.sale_order_id.x_fc_is_odsp_sale
and self.sale_order_id._get_odsp_status() == 'ready_delivery'):
self.sale_order_id._odsp_advance_status(
'delivered',
"Delivery task completed by technician. Order marked as delivered.",
)
if self._is_rental_pickup_task():
self._apply_rental_inspection_results()
def _on_cancel_extra(self):
"""Revert sale order on delivery cancellation, send email otherwise."""
if self.task_type == 'delivery':
self._revert_sale_order_on_cancel()
else:
self._send_task_cancelled_email()
# ------------------------------------------------------------------
# ORDER LINKING METHODS
# ------------------------------------------------------------------
def _post_task_created_to_linked_order(self):
"""Post a brief task creation notice to the linked order's chatter."""
self.ensure_one()
order = self.sale_order_id or self.purchase_order_id
if not order:
return
task_type_label = dict(self._fields['task_type'].selection).get(self.task_type, self.task_type)
date_str = self.scheduled_date.strftime('%B %d, %Y') if self.scheduled_date else 'TBD'
time_str = self._float_to_time_str(self.time_start)
task_url = f'/web#id={self.id}&model=fusion.technician.task&view_type=form'
body = Markup(
f'<div style="background:#e8f4fd;border-left:4px solid #17a2b8;padding:10px;border-radius:4px;">'
f'<strong><i class="fa fa-wrench"></i> Technician Task Scheduled</strong><br/>'
f'<strong>{self.name}</strong> ({task_type_label}) - {date_str} at {time_str}<br/>'
f'Technician(s): {self.all_technician_names or self.technician_id.name}<br/>'
f'<a href="{task_url}">View Task</a>'
f'</div>'
)
order.message_post(
body=body, message_type='notification', subtype_xmlid='mail.mt_note',
)
def _mark_sale_order_ready_for_delivery(self):
"""Mark linked sale orders as Ready for Delivery."""
for task in self:
order = task.sale_order_id
if not order:
continue
if order.x_fc_adp_application_status == 'ready_delivery':
continue
user_name = self.env.user.name
tech_name = task.technician_id.name or ''
previous_status = order.x_fc_adp_application_status
all_tech_ids = (task.technician_id | task.additional_technician_ids).ids
order.with_context(skip_status_validation=True).write({
'x_fc_adp_application_status': 'ready_delivery',
'x_fc_status_before_delivery': previous_status,
'x_fc_delivery_technician_ids': [(4, tid) for tid in all_tech_ids],
'x_fc_ready_for_delivery_date': fields.Datetime.now(),
'x_fc_scheduled_delivery_datetime': task.datetime_start,
})
early_badge = ''
if order.x_fc_early_delivery:
early_badge = ' <span class="badge bg-warning text-dark">Early Delivery</span>'
scheduled_str = ''
if task.scheduled_date:
time_str = task._float_to_time_str(task.time_start) if task.time_start else ''
date_str = task.scheduled_date.strftime('%B %d, %Y')
scheduled_str = f'<li><strong>Scheduled:</strong> {date_str} at {time_str}</li>'
notes_str = ''
if task.description:
notes_str = f'<hr/><p class="mb-0"><strong>Delivery Notes:</strong> {task.description}</p>'
chatter_body = Markup(
f'<div class="alert alert-success" role="alert">'
f'<h5 class="alert-heading"><i class="fa fa-truck"></i> Ready for Delivery{early_badge}</h5>'
f'<ul>'
f'<li><strong>Marked By:</strong> {user_name}</li>'
f'<li><strong>Technician(s):</strong> {task.all_technician_names or tech_name}</li>'
f'{scheduled_str}'
f'<li><strong>Delivery Address:</strong> {task.address_display or "N/A"}</li>'
f'</ul>'
f'{notes_str}'
f'</div>'
)
order.message_post(
body=chatter_body,
message_type='notification',
subtype_xmlid='mail.mt_note',
)
try:
order._send_ready_for_delivery_email(
technicians=task.technician_id | task.additional_technician_ids,
scheduled_datetime=task.datetime_start,
notes=task.description,
)
except Exception as e:
_logger.warning("Ready for delivery email failed for %s: %s", order.name, e)
def _post_completion_to_linked_order(self):
"""Post the completion notes to the linked order's chatter."""
self.ensure_one()
order = self.sale_order_id or self.purchase_order_id
if not order or not self.completion_notes:
return
task_type_label = dict(self._fields['task_type'].selection).get(self.task_type, self.task_type)
body = Markup(
f'<div class="alert alert-info">'
f'<h5><i class="fa fa-wrench"></i> Technician Task Completed</h5>'
f'<ul>'
f'<li><strong>Task:</strong> {self.name} ({task_type_label})</li>'
f'<li><strong>Technician(s):</strong> {self.all_technician_names or self.technician_id.name}</li>'
f'<li><strong>Completed:</strong> {self._utc_to_local(self.completion_datetime).strftime("%B %d, %Y at %I:%M %p") if self.completion_datetime else "N/A"}</li>'
f'</ul>'
f'<hr/>'
f'{self.completion_notes}'
f'</div>'
)
order.message_post(
body=body,
message_type='notification',
subtype_xmlid='mail.mt_note',
)
def _revert_sale_order_on_cancel(self):
"""When a delivery task is cancelled, revert the sale order status."""
self.ensure_one()
if self.task_type != 'delivery' or not self.sale_order_id:
return
order = self.sale_order_id
if order.x_fc_adp_application_status != 'ready_delivery':
return
other_delivery_tasks = self.sudo().search([
('sale_order_id', '=', order.id),
('task_type', '=', 'delivery'),
('status', 'not in', ['cancelled']),
('id', '!=', self.id),
], limit=1)
if other_delivery_tasks:
return
prev_status = order.x_fc_status_before_delivery or 'approved'
status_labels = dict(order._fields['x_fc_adp_application_status'].selection)
prev_label = status_labels.get(prev_status, prev_status)
order.with_context(
skip_status_validation=True,
skip_status_emails=True,
).write({
'x_fc_adp_application_status': prev_status,
'x_fc_status_before_delivery': False,
})
body = Markup(
f'<div class="alert alert-warning" role="alert">'
f'<h5><i class="fa fa-undo"></i> Delivery Task Cancelled</h5>'
f'<p>Delivery task <strong>{self.name}</strong> was cancelled by '
f'{self.env.user.name}.</p>'
f'<p>Order status reverted to <strong>{prev_label}</strong>.</p>'
f'</div>'
)
order.message_post(
body=body,
message_type='notification',
subtype_xmlid='mail.mt_note',
)
self._send_task_cancelled_email()
def _is_rental_pickup_task(self):
"""Check if this is a pickup task for a rental order."""
self.ensure_one()
return (
self.task_type == 'pickup'
and self.sale_order_id
and self.sale_order_id.is_rental_order
)
def _apply_rental_inspection_results(self):
"""Write inspection results from the task back to the rental order."""
self.ensure_one()
order = self.sale_order_id
if not order or not order.is_rental_order:
return
inspection_status = 'passed'
if self.rental_inspection_condition in ('fair', 'damaged'):
inspection_status = 'flagged'
vals = {
'rental_inspection_status': inspection_status,
'rental_inspection_notes': self.rental_inspection_notes or '',
}
if self.rental_inspection_photo_ids:
vals['rental_inspection_photo_ids'] = [(6, 0, self.rental_inspection_photo_ids.ids)]
order.write(vals)
if inspection_status == 'passed':
order._refund_security_deposit()
elif inspection_status == 'flagged':
order.activity_schedule(
'mail.mail_activity_data_todo',
date_deadline=fields.Date.today(),
summary=_("Review rental inspection: %s", order.name),
note=_(
"Technician flagged rental pickup for %s. "
"Condition: %s. Please review inspection photos and notes.",
order.partner_id.name,
self.rental_inspection_condition or 'Unknown',
),
user_id=order.user_id.id or self.env.uid,
)
try:
order._send_damage_notification_email()
except Exception as e:
_logger.error(
"Failed to send damage notification for %s: %s",
order.name, e,
)
# ------------------------------------------------------------------
# VIEW ACTIONS
# ------------------------------------------------------------------
def action_view_sale_order(self):
"""Open the linked sale order / case."""
self.ensure_one()
if not self.sale_order_id:
return
return {
'name': self.sale_order_id.name,
'type': 'ir.actions.act_window',
'res_model': 'sale.order',
'view_mode': 'form',
'res_id': self.sale_order_id.id,
}
def action_view_purchase_order(self):
"""Open the linked purchase order."""
self.ensure_one()
if not self.purchase_order_id:
return
return {
'name': self.purchase_order_id.name,
'type': 'ir.actions.act_window',
'res_model': 'purchase.order',
'view_mode': 'form',
'res_id': self.purchase_order_id.id,
}
# ------------------------------------------------------------------
# EMAIL OVERRIDES
# ------------------------------------------------------------------
def _get_email_builder(self):
"""Prefer the linked sale order for email building."""
if self.sale_order_id:
return self.sale_order_id
return super()._get_email_builder()
def _is_email_notifications_enabled(self):
"""Check linked sale order's notification settings."""
if self.sale_order_id:
try:
return self.sale_order_id._is_email_notifications_enabled()
except Exception:
return True
return super()._is_email_notifications_enabled()
def _get_task_email_details(self):
"""Add SO/PO reference rows to email details."""
rows = super()._get_task_email_details()
# Insert after Client row (index 1)
insert_idx = 2
if self.sale_order_id:
rows.insert(insert_idx, ('Case', self.sale_order_id.name))
insert_idx += 1
if self.purchase_order_id:
rows.insert(insert_idx, ('Purchase Order', self.purchase_order_id.name))
return rows
def _get_task_email_recipients(self):
"""Add sales rep and office CC from linked sale order."""
result = super()._get_task_email_recipients()
if self.sale_order_id and self.sale_order_id.user_id and \
self.sale_order_id.user_id.email:
result['cc'].append(self.sale_order_id.user_id.email)
if self.sale_order_id:
try:
office_cc = self.sale_order_id._get_email_recipients(
include_client=False).get('office_cc', [])
result['cc'].extend(office_cc)
except Exception:
pass
result['cc'] = list(set(result['cc']))
return result
def _send_task_cancelled_email(self):
"""Send cancellation email using linked sale order's email builder."""
self.ensure_one()
if self.x_fc_sync_source:
return False
order = self.sale_order_id
if not order:
return False
try:
if not order._is_email_notifications_enabled():
return False
except Exception:
return False
recipients = self._get_task_email_recipients()
to_emails = recipients.get('to', [])
cc_emails = recipients.get('cc', [])
if not to_emails and not cc_emails:
return False
client_name = self.partner_id.name or 'Client'
type_label = dict(self._fields['task_type'].selection).get(
self.task_type, self.task_type or 'Task')
sender_name = self.env.user.name
detail_rows = self._get_task_email_details()
detail_rows.append(('Cancelled By', sender_name))
body_html = order._email_build(
title=f'{type_label.title()} Cancelled',
summary=(
f'The scheduled <strong>{type_label.lower()}</strong> for '
f'<strong>{client_name}</strong> has been <strong>cancelled</strong>.'
),
email_type='urgent',
sections=[('Cancellation Details', detail_rows)],
note=(
'<strong>What happens next:</strong> If you need to reschedule, '
'please contact our office and we will arrange a new appointment.'
),
note_color='#e53e3e',
button_url=f'{order.get_base_url()}/web#id={order.id}&model=sale.order&view_type=form',
sender_name=sender_name,
)
email_to = ', '.join(to_emails) if to_emails else ', '.join(cc_emails[:1])
email_cc = ', '.join(cc_emails) if to_emails else ', '.join(cc_emails[1:])
try:
self.env['mail.mail'].sudo().create({
'subject': f'{type_label.title()} Cancelled - {client_name} - {order.name}',
'body_html': body_html,
'email_to': email_to,
'email_cc': email_cc,
'model': 'sale.order',
'res_id': order.id,
}).send()
order._email_chatter_log(
f'{type_label.title()} Cancelled email sent', email_to, email_cc)
return True
except Exception as e:
_logger.error("Failed to send task cancelled email for %s: %s", self.name, e)
return False
def _send_task_scheduled_email(self):
"""Send appointment scheduled email using linked sale order."""
self.ensure_one()
if self.x_fc_sync_source:
return False
order = self.sale_order_id
if not order:
return False
try:
if not order._is_email_notifications_enabled():
return False
except Exception:
return False
recipients = self._get_task_email_recipients()
to_emails = recipients.get('to', [])
cc_emails = recipients.get('cc', [])
if not to_emails and not cc_emails:
return False
client_name = self.partner_id.name or 'Client'
type_label = dict(self._fields['task_type'].selection).get(
self.task_type, self.task_type or 'Task')
sender_name = self.env.user.name
detail_rows = self._get_task_email_details()
if self.description:
detail_rows.append(('Notes', self.description))
body_html = order._email_build(
title=f'{type_label.title()} Scheduled',
summary=(
f'A <strong>{type_label.lower()}</strong> has been scheduled for '
f'<strong>{client_name}</strong>.'
),
email_type='success',
sections=[('Appointment Details', detail_rows)],
note=(
'<strong>Please note:</strong> If you need to change this appointment, '
'please contact our office as soon as possible so we can accommodate '
'the change.'
),
note_color='#38a169',
button_url=f'{order.get_base_url()}/web#id={order.id}&model=sale.order&view_type=form',
sender_name=sender_name,
)
email_to = ', '.join(to_emails) if to_emails else ', '.join(cc_emails[:1])
email_cc = ', '.join(cc_emails) if to_emails else ', '.join(cc_emails[1:])
try:
self.env['mail.mail'].sudo().create({
'subject': f'{type_label.title()} Scheduled - {client_name} - {order.name}',
'body_html': body_html,
'email_to': email_to,
'email_cc': email_cc,
'model': 'sale.order',
'res_id': order.id,
}).send()
order._email_chatter_log(
f'{type_label.title()} Scheduled email sent', email_to, email_cc)
return True
except Exception as e:
_logger.error("Failed to send task scheduled email for %s: %s", self.name, e)
return False
def _send_task_rescheduled_email(self, old_date=None, old_start=None, old_end=None):
"""Send reschedule email using linked sale order."""
self.ensure_one()
if self.x_fc_sync_source:
return False
order = self.sale_order_id
if not order:
return False
try:
if not order._is_email_notifications_enabled():
return False
except Exception:
return False
recipients = self._get_task_email_recipients()
to_emails = recipients.get('to', [])
cc_emails = recipients.get('cc', [])
if not to_emails and not cc_emails:
return False
client_name = self.partner_id.name or 'Client'
type_label = dict(self._fields['task_type'].selection).get(
self.task_type, self.task_type or 'Task')
sender_name = self.env.user.name
detail_rows = self._get_task_email_details()
if old_date or old_start is not None:
old_parts = []
if old_date:
old_parts.append(old_date.strftime('%B %d, %Y'))
if old_start is not None:
old_parts.append(
f'{self._float_to_time_str(old_start)} - '
f'{self._float_to_time_str(old_end or old_start + 1.0)}')
detail_rows.insert(3, ('Previous Schedule', ', '.join(old_parts)))
body_html = order._email_build(
title=f'{type_label.title()} Rescheduled',
summary=(
f'The <strong>{type_label.lower()}</strong> for '
f'<strong>{client_name}</strong> has been <strong>rescheduled</strong>.'
),
email_type='attention',
sections=[('Updated Appointment Details', detail_rows)],
note=(
'<strong>Please note:</strong> The appointment has been updated '
'to the new date and time shown above. If you have any questions, '
'please contact our office.'
),
note_color='#d69e2e',
button_url=f'{order.get_base_url()}/web#id={order.id}&model=sale.order&view_type=form',
sender_name=sender_name,
)
email_to = ', '.join(to_emails) if to_emails else ', '.join(cc_emails[:1])
email_cc = ', '.join(cc_emails) if to_emails else ', '.join(cc_emails[1:])
try:
self.env['mail.mail'].sudo().create({
'subject': f'{type_label.title()} Rescheduled - {client_name} - {order.name}',
'body_html': body_html,
'email_to': email_to,
'email_cc': email_cc,
'model': 'sale.order',
'res_id': order.id,
}).send()
order._email_chatter_log(
f'{type_label.title()} Rescheduled email sent', email_to, email_cc)
return True
except Exception as e:
_logger.error("Failed to send rescheduled email for %s: %s", self.name, e)
return False