admin/Odoo-Modules
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(fusion_repairs): Phase 2 - service catalogue, visit report, warranty, Poynt

Browse Source 
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>
This commit is contained in:
gsinghpal
2026-05-20 21:57:33 -04:00
parent ad553b1082
commit 7727745b73
14 changed files with 845 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
fusion_repairs/models/__init__.py
View File

@@ -6,6 +6,8 @@ from . import repair_product_category
from . import intake_template
from . import intake_question
from . import intake_answer
from . import service_catalog
from . import repair_warranty
from . import product_template
from . import res_partner
from . import res_users

111
fusion_repairs/models/intake_service.py
View File

@@ -132,6 +132,15 @@ class FusionRepairIntakeService(models.AbstractModel):
# 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:
@@ -146,7 +155,11 @@ class FusionRepairIntakeService(models.AbstractModel):
self._schedule_activities(repair)
# Optional dispatch draft task (urgent / safety).
if repair.x_fc_urgency in ('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).
@@ -178,6 +191,102 @@ class FusionRepairIntakeService(models.AbstractModel):
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
# ------------------------------------------------------------------

41
fusion_repairs/models/repair_order.py
View File

@@ -5,6 +5,7 @@
from datetime import timedelta
from odoo import api, fields, models, _
from odoo.exceptions import UserError
INTAKE_SOURCES = [
@@ -62,6 +63,14 @@ class RepairOrder(models.Model):
'repair_id',
string='Intake Answers',
)
# Catalogue match (Phase 2)
x_fc_service_catalog_id = fields.Many2one(
'fusion.repair.service.catalog',
string='Service Catalogue Match',
index=True,
help='Auto-matched catalogue entry that pre-fills estimated cost and duration.',
)
x_fc_intake_answer_count = fields.Integer(
compute='_compute_intake_answer_count',
)
@@ -279,3 +288,35 @@ class RepairOrder(models.Model):
'view_mode': 'form',
'res_id': self.x_fc_original_sale_order_id.id,
}
# ------------------------------------------------------------------
# WIZARDS / PAYMENT
# ------------------------------------------------------------------
def action_open_visit_report(self):
self.ensure_one()
return {
'type': 'ir.actions.act_window',
'name': _('Visit Report'),
'res_model': 'fusion.repair.visit.report.wizard',
'view_mode': 'form',
'target': 'new',
'context': {
'default_repair_id': self.id,
'default_labour_hours': self.x_fc_estimated_duration or 1.0,
},
}
def action_collect_payment(self):
"""Open the Poynt payment wizard for the linked posted invoice."""
self.ensure_one()
# Resolve the linked invoice via the standard repair -> SO -> invoice chain.
if not self.sale_order_id:
raise UserError(_('Confirm a sale order from this repair first.'))
invoice = self.sale_order_id.invoice_ids.filtered(
lambda m: m.state == 'posted' and m.payment_state in ('not_paid', 'partial')
)[:1]
if not invoice:
raise UserError(_('No posted, unpaid invoice was found for this repair.'))
if hasattr(invoice, 'action_open_poynt_payment_wizard'):
return invoice.action_open_poynt_payment_wizard()
raise UserError(_('Poynt payment is not available - install or configure fusion_poynt.'))

121
fusion_repairs/models/repair_warranty.py Normal file
View File

@@ -0,0 +1,121 @@
# -*- coding: utf-8 -*-
# Copyright 2024-2026 Nexa Systems Inc.
# License OPL-1 (Odoo Proprietary License v1.0)
"""Repair warranty coverage.
Tracks the 30/90-day warranty we offer on completed repair work.
When a new repair is created on the same equipment within the
coverage window, the intake wizard / portal shows a banner:
"This repair may be covered by our warranty - no charge".
Phase 2 ships the model + manual creation from a completed repair.
Phase 4 will add automatic creation when a repair moves to 'done'.
"""
from datetime import timedelta
from odoo import api, fields, models
class FusionRepairWarrantyCoverage(models.Model):
_name = 'fusion.repair.warranty.coverage'
_description = 'Repair Warranty Coverage'
_order = 'expiry_date desc, id desc'
name = fields.Char(string='Reference', compute='_compute_name', store=True)
repair_id = fields.Many2one(
'repair.order',
string='Original Repair',
required=True,
ondelete='cascade',
index=True,
)
partner_id = fields.Many2one(
'res.partner',
string='Client',
related='repair_id.partner_id',
store=True,
index=True,
)
product_id = fields.Many2one(
'product.product',
string='Equipment',
related='repair_id.product_id',
store=True,
index=True,
)
lot_id = fields.Many2one(
'stock.lot',
string='Serial Number',
related='repair_id.lot_id',
store=True,
)
start_date = fields.Date(
string='Start Date',
required=True,
default=fields.Date.context_today,
)
coverage_days = fields.Integer(
string='Coverage Window (days)',
default=30,
required=True,
)
expiry_date = fields.Date(
string='Expires',
compute='_compute_expiry_date',
store=True,
)
is_active = fields.Boolean(
string='Active',
compute='_compute_is_active',
store=True,
)
notes = fields.Text()
company_id = fields.Many2one(
'res.company',
string='Company',
related='repair_id.company_id',
store=True,
)
@api.depends('repair_id.name', 'expiry_date')
def _compute_name(self):
for w in self:
w.name = (
f"Warranty {w.repair_id.name or '?'} (until {w.expiry_date or '?'})"
)
@api.depends('start_date', 'coverage_days')
def _compute_expiry_date(self):
for w in self:
if w.start_date and w.coverage_days:
w.expiry_date = w.start_date + timedelta(days=w.coverage_days)
else:
w.expiry_date = False
@api.depends('expiry_date')
def _compute_is_active(self):
today = fields.Date.context_today(self)
for w in self:
w.is_active = bool(w.expiry_date and w.expiry_date >= today)
# ------------------------------------------------------------------
# LOOKUP
# ------------------------------------------------------------------
@api.model
def find_active_for(self, partner_id, product_id=None, lot_id=None):
"""Return active warranty coverage matching the partner + equipment, if any."""
if not partner_id:
return self.browse()
domain = [
('partner_id', '=', partner_id),
('is_active', '=', True),
]
if lot_id:
domain.append(('lot_id', '=', lot_id))
elif product_id:
domain.append(('product_id', '=', product_id))
return self.search(domain, order='expiry_date desc', limit=1)

141
fusion_repairs/models/service_catalog.py Normal file
View File

@@ -0,0 +1,141 @@
# -*- coding: utf-8 -*-
# Copyright 2024-2026 Nexa Systems Inc.
# License OPL-1 (Odoo Proprietary License v1.0)
"""Service catalogue.
Each fusion.repair.service.catalog record is a named repair / maintenance
service (e.g. "Stairlift motor replacement", "Bed remote troubleshoot")
with estimated duration, estimated cost, default parts, and symptom
keywords used to auto-match an intake to the right catalogue entry.
The catalogue feeds:
- intake auto-match -> sets x_fc_service_catalog_id +
x_fc_estimated_duration + x_fc_estimated_cost on the repair
- visit report -> default labour line + parts pre-fill
- pricing variance -> compares estimate vs actual
"""
from odoo import api, fields, models
class FusionRepairServiceCatalog(models.Model):
_name = 'fusion.repair.service.catalog'
_description = 'Repair Service Catalogue Entry'
_order = 'sequence, name'
name = fields.Char(string='Service Name', required=True, translate=True)
code = fields.Char(string='Code', help='Stable identifier (lowercase, no spaces).')
sequence = fields.Integer(default=10)
active = fields.Boolean(default=True)
company_id = fields.Many2one(
'res.company', string='Company',
default=lambda self: self.env.company,
)
# Routing & matching
product_category_id = fields.Many2one(
'fusion.repair.product.category',
string='Equipment Category',
required=True,
index=True,
)
symptom_keywords = fields.Char(
string='Symptom Keywords',
help='Comma-separated keywords used to auto-match an intake to this catalogue entry. '
'Matched against the issue summary, issue category, and intake answer text.',
)
# Service product (what actually gets invoiced)
service_product_id = fields.Many2one(
'product.product',
string='Service Product',
domain=[('type', '=', 'service')],
help='Product line added to the repair sale order for the labour portion.',
)
default_parts_product_ids = fields.Many2many(
'product.product',
'fusion_repair_catalog_parts_rel',
'catalog_id', 'product_id',
string='Default Parts',
help='Parts typically used. Pre-loaded onto the visit report wizard for the tech to confirm.',
)
pricelist_id = fields.Many2one(
'product.pricelist',
string='Pricelist Override',
help='Optional pricelist applied to repair SOs from this catalogue entry. '
'Leave blank to use the partner default pricelist.',
)
# Estimates
estimated_hours = fields.Float(
string='Estimated Labour (h)',
default=1.0,
help='Used to size the technician task and the visit report labour default.',
)
estimated_cost = fields.Monetary(
string='Estimated Cost',
currency_field='company_currency_id',
help='Headline estimate shown to the client/CS during intake. Phase 1 is a flat number; '
'Phase 2+ may compute from labour + parts.',
)
# Automation hints
auto_schedule = fields.Boolean(
string='Auto-Create Tech Task',
help='When True, the intake service creates a draft technician task immediately for any '
'repair matched to this catalogue entry (even at normal urgency).',
)
task_type = fields.Selection(
[('delivery', 'Delivery'), ('repair', 'Repair'), ('pickup', 'Pickup'),
('troubleshoot', 'Troubleshoot'), ('assessment', 'Assessment'),
('installation', 'Installation'), ('maintenance', 'Maintenance'),
('other', 'Other')],
string='Default Task Type',
default='repair',
)
company_currency_id = fields.Many2one(
'res.currency',
related='company_id.currency_id',
readonly=True,
)
@api.depends('name', 'code')
def _compute_display_name(self):
for c in self:
c.display_name = c.name or c.code or ''
# ------------------------------------------------------------------
# MATCHING
# ------------------------------------------------------------------
@api.model
def find_best_match(self, product_category_id, text_hints):
"""Return the best-matching catalogue entry, or empty recordset.
:param product_category_id: int id of the equipment category
:param text_hints: list[str] - text snippets to look for symptom keywords in
(typically: issue_summary, issue_category, recent intake answer values)
"""
if not product_category_id:
return self.browse()
haystack = ' '.join(s.lower() for s in (text_hints or []) if s).strip()
candidates = self.search([
('product_category_id', '=', product_category_id),
('active', '=', True),
], order='sequence')
if not candidates:
return self.browse()
if not haystack:
return candidates[:1]
best = None
best_score = 0
for c in candidates:
kws = [k.strip().lower() for k in (c.symptom_keywords or '').split(',') if k.strip()]
score = sum(1 for kw in kws if kw and kw in haystack)
if score > best_score:
best = c
best_score = score
if best:
return best
return candidates[:1]