3a15164605
H1 Float -> Monetary for outstanding_balance
Added currency_id companion field on the wizard so widget="monetary"
renders properly. Currency defaults to env.company.currency_id.
H2 Maps URL address duplication
fusion_tasks address_street often contains the full Google-Places-
formatted address. Concatenating address_street + address_city + zip
was producing "15 Fisherman Dr, Brampton, ON L7A 1B7, Canada, Brampton,
L7A 1B7". Now uses the existing address_display field (fusion_tasks
computes it correctly for both Google Places and manual entries), with
a partner-based fallback that includes street, street2, city,
state_id.name, zip, country_id.name.
H3 Banner copy hardcoded "14 days"
Added duplicate_window_days compute field; banner now reads
"in last <N> days" from the ir.config_parameter.
H4 Outstanding-balance multi-company + child_of direction
- Dropped .sudo() (CS users already have access to their own company's
invoices via standard groups + the Repairs Office rule)
- Replaced child_of (which only walks descendants) with
commercial_partner_id (the canonical Odoo "billed-to root" - covers
child contacts AND walks up from a child if the caller IS a child)
- Added ('company_id', 'in', env.companies.ids) filter to both the
invoice search AND the duplicate-repair search so a CS rep in
Westin Healthcare doesn't see NEXA Systems balances
H5 duplicate_count capped at 5 (false reassurance)
Now uses search_count for the true total + search(limit=5) for the
display list. Earlier verification showed count=5 was actually
capped; running again shows 15 for the same partner.
M1 Function-level imports
Moved urllib.parse.quote_plus and odoo.exceptions.UserError to module
top in technician_task.py.
M2 Many2many 'in' with scalar
Changed ('x_fc_repair_skills', 'in', category.id) to
('x_fc_repair_skills', 'in', [category.id]) - safer against future
ORM tightening.
M4 C6 - added x_fc_is_quote_only field + filter + form indicator
Boolean tracked field on repair.order (was previously discoverable
only via chatter text). Indexed. Visible on the form's intake metadata
row and filterable on the dashboard search view as "Quote Only".
M5 Account-move read perf
Replaced Move.search() + Python sum with _read_group(
aggregates=['amount_residual:sum', '__count']) - pushes the SUM to
Postgres; O(1) record load vs O(N).
M6 Hide Maps button when no address
Added invisible="not address_display and not partner_id" on the
Open in Maps button so it doesn't appear on in-store tasks.
Plus the dispatch-task cutoff is now a datetime (was a date) so the
create_date >= cutoff comparison is type-correct.
Verified end-to-end on local westin-v19 after fixes:
C1 count: 15 (was capped at 5) window_days: 14
C5 balance: 0.0 currency: CAD warning: False (correct)
C6 x_fc_is_quote_only: True tech_tasks: 0 (urgent intake, NOT dispatched)
T1 URL: https://www.google.com/maps?q=15+Fisherman+Dr%2C+Brampton%2C+ON+L7A+1B7%2C+Canada%2C+Unit+7
(no duplicated city/zip)
Bumped to 19.0.1.1.1.
Co-authored-by: Cursor <cursoragent@cursor.com>
531 lines
22 KiB
Python
531 lines
22 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 markupsafe import Markup
|
|
|
|
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)
|
|
- quote_only: bool (optional, C6 - skips dispatch task creation)
|
|
- 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 [{}]
|
|
quote_only = bool(payload.get('quote_only'))
|
|
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,
|
|
quote_only=quote_only,
|
|
)
|
|
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
|
|
# Sensible defaults for partners created via public portals so mail
|
|
# templates pick up the right language / company.
|
|
partner_vals.setdefault('lang', self.env.user.lang or 'en_CA')
|
|
partner_vals.setdefault('company_id', self.env.company.id)
|
|
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, quote_only=False):
|
|
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,
|
|
'x_fc_is_quote_only': bool(quote_only),
|
|
'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, quote_only=quote_only)
|
|
|
|
# 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.
|
|
# Skip entirely if intake is quote-only (C6).
|
|
if (
|
|
not quote_only
|
|
and repair.x_fc_urgency in ('urgent', 'safety')
|
|
and not repair.x_fc_technician_task_ids
|
|
):
|
|
self._create_dispatch_task(repair)
|
|
elif quote_only:
|
|
repair.message_post(body=Markup(_(
|
|
'Created in <b>Quote Only</b> mode - no technician dispatched.'
|
|
)))
|
|
|
|
# Emails (client + office).
|
|
self._send_intake_emails(repair)
|
|
|
|
# Audit message in chatter.
|
|
repair.message_post(
|
|
body=Markup(_(
|
|
'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) or '',
|
|
'user': intake_user.name or '',
|
|
'ref': session_ref or '',
|
|
},
|
|
)
|
|
|
|
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, quote_only=False):
|
|
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).
|
|
# Quote-only intakes skip this too.
|
|
if (
|
|
catalog.auto_schedule
|
|
and repair.x_fc_technician_task_count == 0
|
|
and not quote_only
|
|
):
|
|
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=Markup(_(
|
|
'This repair MAY be covered by our active warranty <b>%(ref)s</b> '
|
|
'(expires %(exp)s). Manager review recommended before invoicing.'
|
|
)) % {
|
|
'ref': warranty.name or '',
|
|
'exp': warranty.expiry_date and str(warranty.expiry_date) or '',
|
|
},
|
|
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 AND constrained to x_fc_is_field_staff.
|
|
# D2: prefer a tech whose x_fc_repair_skills covers this repair's
|
|
# category. Falls back to ANY active field-staff user if no skilled
|
|
# tech exists, then to the lowest-id field-staff user as a placeholder.
|
|
tech_id = self._pick_dispatch_technician(repair)
|
|
if not tech_id:
|
|
_logger.warning(
|
|
'No field-staff user available - skipping auto-dispatch '
|
|
'task for repair %s (mark a user as Field Staff under '
|
|
'Settings > Users).',
|
|
repair.name,
|
|
)
|
|
return
|
|
vals['technician_id'] = tech_id
|
|
Task.create(vals)
|
|
except Exception as e:
|
|
_logger.warning('Failed to auto-create dispatch task for repair %s: %s',
|
|
repair.name, e)
|
|
|
|
@api.model
|
|
def _pick_dispatch_technician(self, repair):
|
|
"""D2: pick the best technician for the initial dispatch task.
|
|
|
|
Preference order:
|
|
1. The intake user IF they are field staff AND have the skill
|
|
2. Any active field-staff user with x_fc_repair_skills covering
|
|
the repair's product category
|
|
3. Any active field-staff user (no skills filter)
|
|
|
|
Returns the chosen user id, or False if none found.
|
|
"""
|
|
Users = self.env['res.users'].sudo()
|
|
category = repair.x_fc_repair_category_id
|
|
|
|
# Try intake user first if they qualify.
|
|
if repair.user_id and repair.user_id.x_fc_is_field_staff:
|
|
if not category or category in repair.user_id.x_fc_repair_skills:
|
|
return repair.user_id.id
|
|
|
|
# Skills-filtered candidates.
|
|
if category:
|
|
skilled = Users.search([
|
|
('x_fc_is_field_staff', '=', True),
|
|
('active', '=', True),
|
|
('x_fc_repair_skills', 'in', [category.id]),
|
|
], order='id', limit=1)
|
|
if skilled:
|
|
return skilled.id
|
|
|
|
# Any active field staff.
|
|
fallback = Users.search([
|
|
('x_fc_is_field_staff', '=', True),
|
|
('active', '=', True),
|
|
], order='id', limit=1)
|
|
return fallback.id if fallback else False
|
|
|
|
# ------------------------------------------------------------------
|
|
# 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.
|
|
company_sudo = company.sudo()
|
|
recipients = getattr(company_sudo, 'x_fc_office_notification_ids', False)
|
|
emails = [p.email for p in (recipients or []) if p.email]
|
|
if not emails:
|
|
_logger.info(
|
|
'No office notification recipients configured on company %s - '
|
|
'skipping office intake email.',
|
|
company.name,
|
|
)
|
|
return emails
|