7727745b73
Service catalogue - New fusion.repair.service.catalog model: named service entries per equipment category with symptom keywords, estimated hours / cost, default parts, auto_schedule flag, optional pricelist override - find_best_match() scores candidates by symptom-keyword overlap against intake text hints (issue summary + category + notes) - Intake service wires it in: on submit, the matcher sets x_fc_service_catalog_id + x_fc_estimated_duration + x_fc_estimated_cost and (when auto_schedule=True) creates a draft dispatch task - Double-task guard: if catalogue match already created a task, the urgency-based dispatch skips so we never duplicate Visit report wizard - fusion.repair.visit.report.wizard with labour hours + parts lines + technician notes + 'found another issue' branch - Computes actual cost = (labour x service_product.list_price) + parts - Compares against estimate -> sets requires_requote when variance exceeds configured threshold (% or $); shows warning banner inline - On confirm: writes actuals back to repair, posts notes to chatter, optionally spawns a follow-up repair (T5 'found another issue') Repair warranty - New fusion.repair.warranty.coverage model (start/expiry, partner, product, lot, active flag) - find_active_for(partner, product, lot) returns the most-recent active coverage - Intake service auto-checks: when a new repair lands on an equipment that has active warranty coverage, posts a chatter banner so the office knows the work may be free under our 30/90-day re-do policy (manager review still required; never auto-zeros pricing) Repair form - Header: Visit Report + Collect Payment buttons (gated by group) - action_collect_payment looks up the linked posted unpaid invoice on the repair SO and opens the Poynt wizard (action_open_poynt_payment_wizard) AI intake summary - _generate_ai_summary calls self.env['fusion.api.service'].call_openai with consumer='fusion_repairs', feature='intake_triage' - Strict system prompt: no medical advice, no diagnoses, no recommending stop equipment use; ~80 words; plain English - Try/fallback per fusion-api-integration.mdc: if fusion_api not installed or call fails -> silently skip; intake never blocked Verified end-to-end on local westin-v19: - Stairlift motor intake -> catalogue match -> estimated $500/2h -> auto dispatch task (count=1, not duplicated) - Visit report: 2.5h x $250 + $100 parts = $725 actual vs $500 estimated = 45% variance -> requires_requote=True - Warranty: 30-day coverage on the completed repair; second repair on same partner triggers warranty banner in chatter Co-authored-by: Cursor <cursoragent@cursor.com>
457 lines
19 KiB
Python
457 lines
19 KiB
Python
# -*- coding: utf-8 -*-
|
|
# Copyright 2024-2026 Nexa Systems Inc.
|
|
# License OPL-1 (Odoo Proprietary License v1.0)
|
|
|
|
"""Shared intake service.
|
|
|
|
This AbstractModel is the SINGLE entry point for creating repair orders from
|
|
any intake surface: the backend wizard (Phase 1), the sales rep portal
|
|
(Phase 1+), and the public client self-service portal (Phase 1+).
|
|
|
|
All three surfaces call `create_repair_orders(payload, source='...')` so that
|
|
business logic - activities, emails, warranty determination, AI summary,
|
|
catalogue match, third-party flag, dispatch task creation - lives in one
|
|
place and the surfaces never drift apart.
|
|
"""
|
|
|
|
import logging
|
|
from datetime import timedelta
|
|
|
|
from odoo import _, api, fields, models
|
|
from odoo.exceptions import UserError
|
|
|
|
_logger = logging.getLogger(__name__)
|
|
|
|
|
|
class FusionRepairIntakeService(models.AbstractModel):
|
|
_name = 'fusion.repair.intake.service'
|
|
_description = 'Repair Intake Service (shared by backend / sales rep / client)'
|
|
|
|
# ------------------------------------------------------------------
|
|
# PUBLIC API
|
|
# ------------------------------------------------------------------
|
|
@api.model
|
|
def create_repair_orders(self, payload, source='backend_wizard'):
|
|
"""Create one repair.order per equipment item in the payload.
|
|
|
|
:param payload: dict with keys:
|
|
- partner_id: int (required) or partner_vals: dict to create new partner
|
|
- intake_user_id: int (optional, defaults to env.user)
|
|
- equipment_items: list of dicts, each with:
|
|
- product_id: int (optional)
|
|
- lot_id: int (optional)
|
|
- repair_category_id: int (optional)
|
|
- intake_template_id: int (optional)
|
|
- third_party: bool (optional)
|
|
- urgency: str (optional, default 'normal')
|
|
- issue_summary: str (optional)
|
|
- internal_notes: str (optional)
|
|
- photo_attachment_ids: list[int] (optional)
|
|
- answers: list of dicts with keys
|
|
(question_id, value_char|value_text|value_selection|
|
|
value_boolean|value_integer|value_date)
|
|
:param source: str, one of repair_order.INTAKE_SOURCES values.
|
|
:return: recordset of repair.order records created.
|
|
"""
|
|
partner_id = self._resolve_partner(payload)
|
|
if not partner_id:
|
|
raise UserError(_('A client is required to create a repair request.'))
|
|
|
|
intake_user = self.env['res.users'].browse(
|
|
payload.get('intake_user_id') or self.env.uid
|
|
)
|
|
session_ref = (
|
|
self.env['ir.sequence'].next_by_code('fusion.repair.intake.session')
|
|
or 'RIS/NEW'
|
|
)
|
|
|
|
equipment = payload.get('equipment_items') or [{}]
|
|
repairs = self.env['repair.order']
|
|
for item in equipment:
|
|
repair = self._create_single_repair(
|
|
partner_id=partner_id,
|
|
intake_user=intake_user,
|
|
session_ref=session_ref,
|
|
source=source,
|
|
item=item,
|
|
)
|
|
repairs |= repair
|
|
|
|
return repairs
|
|
|
|
# ------------------------------------------------------------------
|
|
# PARTNER RESOLUTION
|
|
# ------------------------------------------------------------------
|
|
@api.model
|
|
def _resolve_partner(self, payload):
|
|
partner_id = payload.get('partner_id')
|
|
if partner_id:
|
|
return partner_id
|
|
partner_vals = payload.get('partner_vals')
|
|
if not partner_vals:
|
|
return False
|
|
partner = self.env['res.partner'].sudo().create(partner_vals)
|
|
return partner.id
|
|
|
|
# ------------------------------------------------------------------
|
|
# CORE CREATION
|
|
# ------------------------------------------------------------------
|
|
@api.model
|
|
def _create_single_repair(self, partner_id, intake_user, session_ref, source, item):
|
|
Repair = self.env['repair.order']
|
|
product_id = item.get('product_id')
|
|
|
|
vals = {
|
|
'partner_id': partner_id,
|
|
'user_id': intake_user.id,
|
|
'x_fc_intake_user_id': intake_user.id,
|
|
'x_fc_intake_session_id': session_ref,
|
|
'x_fc_intake_source': source,
|
|
'x_fc_repair_category_id': item.get('repair_category_id') or False,
|
|
'x_fc_intake_template_id': item.get('intake_template_id') or False,
|
|
'x_fc_third_party_equipment': bool(item.get('third_party')),
|
|
'x_fc_urgency': item.get('urgency') or 'normal',
|
|
'x_fc_issue_category': item.get('issue_category') or False,
|
|
'internal_notes': self._wrap_internal_notes(item),
|
|
}
|
|
if product_id:
|
|
vals['product_id'] = product_id
|
|
if item.get('lot_id'):
|
|
vals['lot_id'] = item['lot_id']
|
|
if item.get('schedule_date'):
|
|
vals['schedule_date'] = item['schedule_date']
|
|
|
|
repair = Repair.create(vals)
|
|
|
|
# Determine warranty AFTER creation (needs product on record).
|
|
if not repair.x_fc_third_party_equipment:
|
|
self._auto_link_original_sale_order(repair)
|
|
if repair._fc_compute_warranty_status():
|
|
repair.under_warranty = True
|
|
|
|
# Persist intake answers.
|
|
self._create_answers(repair, item.get('answers') or [])
|
|
|
|
# Service catalogue auto-match.
|
|
self._match_service_catalog(repair, item)
|
|
|
|
# Check our own repair-warranty (30/90 day re-do free).
|
|
self._check_repair_warranty(repair)
|
|
|
|
# Optional AI brief generation - never blocks intake.
|
|
self._generate_ai_summary(repair, item)
|
|
|
|
# Attach photos.
|
|
photo_ids = item.get('photo_attachment_ids') or []
|
|
if photo_ids:
|
|
attachments = self.env['ir.attachment'].sudo().browse(photo_ids).exists()
|
|
attachments.write({
|
|
'res_model': 'repair.order',
|
|
'res_id': repair.id,
|
|
})
|
|
repair.write({'x_fc_photo_ids': [(6, 0, attachments.ids)]})
|
|
|
|
# Activities.
|
|
self._schedule_activities(repair)
|
|
|
|
# Optional dispatch draft task (urgent / safety).
|
|
# Skip if the catalogue match already auto-created one.
|
|
if (
|
|
repair.x_fc_urgency in ('urgent', 'safety')
|
|
and not repair.x_fc_technician_task_ids
|
|
):
|
|
self._create_dispatch_task(repair)
|
|
|
|
# Emails (client + office).
|
|
self._send_intake_emails(repair)
|
|
|
|
# Audit message in chatter.
|
|
repair.message_post(
|
|
body=_(
|
|
'Service call submitted via <b>%(source)s</b> by %(user)s. '
|
|
'Session reference: %(ref)s.',
|
|
source=dict(repair._fields['x_fc_intake_source'].selection).get(source),
|
|
user=intake_user.name,
|
|
ref=session_ref,
|
|
),
|
|
)
|
|
|
|
return repair
|
|
|
|
@api.model
|
|
def _wrap_internal_notes(self, item):
|
|
notes = item.get('internal_notes') or ''
|
|
summary = item.get('issue_summary') or ''
|
|
if not (notes or summary):
|
|
return False
|
|
parts = []
|
|
if summary:
|
|
parts.append('<p><strong>Issue summary:</strong> %s</p>' % summary)
|
|
if notes:
|
|
parts.append('<p><strong>Notes:</strong> %s</p>' % notes)
|
|
return ''.join(parts)
|
|
|
|
# ------------------------------------------------------------------
|
|
# SERVICE CATALOGUE MATCH
|
|
# ------------------------------------------------------------------
|
|
@api.model
|
|
def _match_service_catalog(self, repair, item):
|
|
category = repair.x_fc_repair_category_id
|
|
if not category:
|
|
return
|
|
text_hints = [
|
|
(item.get('issue_summary') or ''),
|
|
(item.get('issue_category') or ''),
|
|
(item.get('internal_notes') or ''),
|
|
]
|
|
catalog = self.env['fusion.repair.service.catalog'].sudo().find_best_match(
|
|
category.id, text_hints,
|
|
)
|
|
if not catalog:
|
|
return
|
|
repair.write({
|
|
'x_fc_service_catalog_id': catalog.id,
|
|
'x_fc_estimated_duration': catalog.estimated_hours,
|
|
'x_fc_estimated_cost': catalog.estimated_cost,
|
|
})
|
|
# Auto-create dispatch task if catalogue says so (in addition to urgency rule).
|
|
if catalog.auto_schedule and repair.x_fc_technician_task_count == 0:
|
|
self._create_dispatch_task(repair)
|
|
|
|
# ------------------------------------------------------------------
|
|
# REPAIR WARRANTY (our 30/90-day re-do free)
|
|
# ------------------------------------------------------------------
|
|
@api.model
|
|
def _check_repair_warranty(self, repair):
|
|
if not repair.partner_id:
|
|
return
|
|
warranty = self.env['fusion.repair.warranty.coverage'].sudo() \
|
|
.find_active_for(repair.partner_id.id, repair.product_id.id or None,
|
|
repair.lot_id.id or None)
|
|
if not warranty:
|
|
return
|
|
repair.message_post(
|
|
body=_(
|
|
'This repair MAY be covered by our active warranty <b>%(ref)s</b> '
|
|
'(expires %(exp)s). Manager review recommended before invoicing.',
|
|
ref=warranty.name,
|
|
exp=warranty.expiry_date,
|
|
),
|
|
message_type='comment',
|
|
)
|
|
|
|
# ------------------------------------------------------------------
|
|
# AI SUMMARY (try/fallback per fusion-api-integration rule)
|
|
# ------------------------------------------------------------------
|
|
@api.model
|
|
def _generate_ai_summary(self, repair, item):
|
|
try:
|
|
ApiService = self.env.get('fusion.api.service')
|
|
if not ApiService:
|
|
return
|
|
issue = (item.get('issue_summary') or '').strip()
|
|
if not issue:
|
|
return
|
|
category = repair.x_fc_repair_category_id.name or 'medical equipment'
|
|
urgency = repair.x_fc_urgency or 'normal'
|
|
messages = [
|
|
{
|
|
'role': 'system',
|
|
'content': (
|
|
'You are an assistant for a medical equipment repair service. '
|
|
'Given an intake note, output ONE short paragraph (under 80 words) '
|
|
'briefing the technician about: likely cause, what to bring, and '
|
|
'any safety considerations. NEVER provide medical advice. NEVER '
|
|
'recommend stopping equipment use. NEVER claim a definitive cause. '
|
|
'Plain English, no jargon.'
|
|
),
|
|
},
|
|
{
|
|
'role': 'user',
|
|
'content': (
|
|
f'Equipment category: {category}\n'
|
|
f'Urgency: {urgency}\n'
|
|
f'Issue: {issue}\n'
|
|
f'Notes: {(item.get("internal_notes") or "").strip()}'
|
|
),
|
|
},
|
|
]
|
|
summary = ApiService.call_openai(
|
|
consumer='fusion_repairs',
|
|
feature='intake_triage',
|
|
messages=messages,
|
|
max_tokens=200,
|
|
)
|
|
if summary:
|
|
repair.x_fc_ai_summary = summary.strip()
|
|
except Exception as e:
|
|
_logger.info('AI intake summary skipped: %s', e)
|
|
|
|
# ------------------------------------------------------------------
|
|
# ORIGINAL SO AUTO-LINK
|
|
# ------------------------------------------------------------------
|
|
@api.model
|
|
def _auto_link_original_sale_order(self, repair):
|
|
if not repair.partner_id or not repair.product_id:
|
|
return
|
|
SaleOrder = self.env['sale.order'].sudo()
|
|
domain = [
|
|
('partner_id', '=', repair.partner_id.id),
|
|
('state', 'in', ('sale', 'done')),
|
|
('order_line.product_id', '=', repair.product_id.id),
|
|
]
|
|
if repair.lot_id:
|
|
domain.append(('order_line.lot_ids', 'in', repair.lot_id.id))
|
|
candidate = SaleOrder.search(domain, order='date_order desc', limit=1)
|
|
if candidate:
|
|
repair.x_fc_original_sale_order_id = candidate
|
|
|
|
# ------------------------------------------------------------------
|
|
# ANSWERS
|
|
# ------------------------------------------------------------------
|
|
@api.model
|
|
def _create_answers(self, repair, answers):
|
|
if not answers:
|
|
return
|
|
Answer = self.env['fusion.repair.intake.answer']
|
|
for ans in answers:
|
|
qid = ans.get('question_id')
|
|
if not qid:
|
|
continue
|
|
Answer.create({
|
|
'repair_id': repair.id,
|
|
'question_id': qid,
|
|
'value_char': ans.get('value_char'),
|
|
'value_text': ans.get('value_text'),
|
|
'value_selection': ans.get('value_selection'),
|
|
'value_boolean': bool(ans.get('value_boolean')),
|
|
'value_integer': int(ans.get('value_integer') or 0),
|
|
'value_date': ans.get('value_date') or False,
|
|
})
|
|
|
|
# ------------------------------------------------------------------
|
|
# ACTIVITIES
|
|
# ------------------------------------------------------------------
|
|
@api.model
|
|
def _schedule_activities(self, repair):
|
|
"""Create the 4 intake activities described in the spec."""
|
|
try:
|
|
cs_callback_type = self.env.ref('fusion_repairs.mail_activity_type_cs_callback')
|
|
tech_dispatch_type = self.env.ref('fusion_repairs.mail_activity_type_tech_dispatch')
|
|
manager_review_type = self.env.ref('fusion_repairs.mail_activity_type_manager_review')
|
|
except ValueError:
|
|
_logger.warning('Repair activity types missing - skipping')
|
|
return
|
|
|
|
# CS callback - always, intake user
|
|
repair.activity_schedule(
|
|
activity_type_id=cs_callback_type.id,
|
|
summary=_('Call client back if any intake info was missing'),
|
|
user_id=repair.x_fc_intake_user_id.id or self.env.uid,
|
|
)
|
|
|
|
# Tech dispatch - assigned to responsible user, urgency-adjusted deadline
|
|
deadline_days = {'safety': 0, 'urgent': 1, 'normal': 2}.get(repair.x_fc_urgency, 2)
|
|
repair.activity_schedule(
|
|
activity_type_id=tech_dispatch_type.id,
|
|
summary=_('Assign a technician (urgency: %s)', repair.x_fc_urgency),
|
|
user_id=repair.user_id.id or self.env.uid,
|
|
date_deadline=fields.Date.context_today(self) + timedelta(days=deadline_days),
|
|
)
|
|
|
|
# Manager review - only for third-party equipment
|
|
if repair.x_fc_third_party_equipment:
|
|
manager_group = self.env.ref(
|
|
'fusion_repairs.group_fusion_repairs_manager',
|
|
raise_if_not_found=False,
|
|
)
|
|
manager_user = self.env.user
|
|
if manager_group:
|
|
# res.groups has no .users field in Odoo 19;
|
|
# query via res.users.all_group_ids (Odoo 19 renamed groups_id).
|
|
candidate = self.env['res.users'].sudo().search(
|
|
[('all_group_ids', 'in', manager_group.ids), ('active', '=', True)],
|
|
limit=1,
|
|
)
|
|
if candidate:
|
|
manager_user = candidate
|
|
repair.activity_schedule(
|
|
activity_type_id=manager_review_type.id,
|
|
summary=_('Third-party equipment - manager awareness'),
|
|
user_id=manager_user.id,
|
|
)
|
|
|
|
# ------------------------------------------------------------------
|
|
# DISPATCH TASK
|
|
# ------------------------------------------------------------------
|
|
@api.model
|
|
def _create_dispatch_task(self, repair):
|
|
"""Create a draft fusion.technician.task for urgent / safety repairs.
|
|
|
|
Phase 1 simple approach: no date/technician assigned, dispatcher confirms.
|
|
"""
|
|
Task = self.env['fusion.technician.task'].sudo()
|
|
try:
|
|
vals = {
|
|
'partner_id': repair.partner_id.id,
|
|
'task_type': 'repair',
|
|
'status': 'pending',
|
|
'scheduled_date': fields.Date.context_today(self),
|
|
'duration_hours': repair.x_fc_estimated_duration or 1.0,
|
|
'x_fc_repair_order_id': repair.id,
|
|
'description': repair.internal_notes or repair.name,
|
|
}
|
|
# technician_id is required on fusion.technician.task; we fall back to
|
|
# the intake user. Dispatcher will reassign.
|
|
vals['technician_id'] = (
|
|
repair.user_id.id if repair.user_id and repair.user_id.x_fc_is_field_staff
|
|
else self.env.uid
|
|
)
|
|
Task.create(vals)
|
|
except Exception as e:
|
|
_logger.warning('Failed to auto-create dispatch task for repair %s: %s',
|
|
repair.name, e)
|
|
|
|
# ------------------------------------------------------------------
|
|
# EMAILS
|
|
# ------------------------------------------------------------------
|
|
@api.model
|
|
def _send_intake_emails(self, repair):
|
|
if not self._notifications_enabled():
|
|
return
|
|
# Client confirmation
|
|
if repair.partner_id and repair.partner_id.email:
|
|
try:
|
|
self.env.ref('fusion_repairs.email_template_intake_received_client') \
|
|
.send_mail(repair.id, force_send=False)
|
|
except Exception as e:
|
|
_logger.warning('Failed to send client intake email for %s: %s',
|
|
repair.name, e)
|
|
|
|
# Office notification
|
|
office_emails = self._office_emails(repair.company_id)
|
|
if office_emails:
|
|
try:
|
|
tpl = self.env.ref('fusion_repairs.email_template_intake_received_office')
|
|
tpl.with_context(default_email_to=','.join(office_emails)) \
|
|
.send_mail(repair.id, force_send=False, email_values={
|
|
'email_to': ','.join(office_emails),
|
|
})
|
|
except Exception as e:
|
|
_logger.warning('Failed to send office intake email for %s: %s',
|
|
repair.name, e)
|
|
|
|
@api.model
|
|
def _notifications_enabled(self):
|
|
ICP = self.env['ir.config_parameter'].sudo()
|
|
return ICP.get_param('fusion_repairs.enable_email_notifications', 'True') == 'True'
|
|
|
|
@api.model
|
|
def _office_emails(self, company):
|
|
# Reuse the office notification recipients defined by fusion_claims.
|
|
partners = company.sudo()
|
|
recipients = getattr(partners, 'x_fc_office_notification_ids', False)
|
|
if recipients:
|
|
return [p.email for p in recipients if p.email]
|
|
return []
|