c86f1bbbe5
Critical - C1: _sql_constraints -> models.Constraint (Odoo 19 deprecation rule violation) - C2: variance threshold no longer uses abs() - under-cost is good news, must not block invoicing. Now only OVER-cost triggers requires_requote. - C3: roll_next_due_date() was dead code - now wired from fusion.technician.task.write() when a maintenance task transitions to 'completed', so the whole maintenance lifecycle actually advances. - C4: warranty.is_active was store=True but time-dependent (became stale). Dropped store=True; find_active_for() now filters by expiry_date directly. High - H1: added x_fc_maintenance_contract_id back-link on repair.order and populated it from create_repair_from_booking(). - H2: find_active_for() returns empty when neither lot nor product is supplied - prevents cross-product false warranty matches. - H3: visit-report wizard now creates stock.move records of repair_line_type 'add' for each part line, so Odoo's native action_create_sale_order() chain has lines to invoice and stock gets consumed properly. - H4: office intake email template now carries a fallback email_to header computed from res.company.x_fc_office_notification_ids (or company email), so it does not silently send with no recipient. - H5: maintenance reminder cron nextcall now always rolls to tomorrow at 07:00 local time, so installing/upgrading after 07:00 does not immediately fire all the day's reminders. - H6: public portal no longer hardcodes UID 1 as the intake user fallback (which in Odoo 19 is OdooBot). Prefers base.user_admin, else the lowest-id non-share user, else SUPERUSER_ID. - H7: public portal validates client_email via tools.email_normalize before partner creation; malformed addresses redirect with error=email. - H8: find_best_match() returns empty when no symptom keywords match (no silent first-catalog guess) and uses word-boundary regex to avoid matching 'battery' inside 'no battery problem'. Medium - M1: _inherit moved next to _name on maintenance_contract (cosmetic but brittle if Odoo refactors model class detection) - M2: relativedelta(months=N) instead of timedelta(days=N*30) for warranty and maintenance intervals (correct month boundaries) - M3: unique constraint on fusion.repair.maintenance.contract.booking_token - M6: dispatch task fallback now searches for an actual x_fc_is_field_staff user; gracefully skips and logs if no field staff exists (instead of silently failing the constraint check) - M7: maintenance contract list view date decoration uses context_today() (date) instead of strftime(string) - the str comparison would TypeError - M9: Visit Report button hidden on draft repairs and when no technician task is linked yet Low - L2: portal-created partners get default lang + company_id so mail templates render in the right language - L3: dropped unused exception variable in sales rep portal controller - L4: visit-report wizard 'found another issue' now redirects to the spawned stub repair so the tech can fill it in immediately - L5: dropped unrecognized data-string from <app> in settings view Public portal also: rate-limit check moved BEFORE the counter increment so blocked attempts do not keep inflating the bucket. All fixes verified end-to-end on local westin-v19: - variance one-sided: 0.5h labour vs $500 est -> requires_requote=False; 2h x $250 + $200 parts vs $100 est -> requires_requote=True - maintenance roll-forward: created MC/00006 due 2026-05-31, completed linked maintenance task -> contract rolled to 2026-11-21 with last_reminder_band reset - warranty find_active_for(partner only) -> empty recordset - service catalog find_best_match with unrelated text -> empty recordset - pg_constraint shows fusion_repair_maintenance_contract_booking_token_unique - /repair landing still 200 after restart Co-authored-by: Cursor <cursoragent@cursor.com>
479 lines
20 KiB
Python
479 lines
20 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
|
|
# 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):
|
|
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 AND constrained to x_fc_is_field_staff.
|
|
# Use the intake user if they qualify, otherwise the lowest-id active
|
|
# field-staff user as a placeholder for the dispatcher to reassign.
|
|
if repair.user_id and repair.user_id.x_fc_is_field_staff:
|
|
vals['technician_id'] = repair.user_id.id
|
|
else:
|
|
fallback = self.env['res.users'].sudo().search([
|
|
('x_fc_is_field_staff', '=', True),
|
|
('active', '=', True),
|
|
], order='id', limit=1)
|
|
if not fallback:
|
|
_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'] = fallback.id
|
|
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.
|
|
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
|