# -*- coding: utf-8 -*- from odoo import http, _, fields from odoo.http import request from odoo.addons.portal.controllers.portal import CustomerPortal, pager as portal_pager from odoo.exceptions import AccessError, MissingError import base64 import logging import pytz _logger = logging.getLogger(__name__) class AuthorizerPortal(CustomerPortal): """Portal controller for Authorizers (OTs/Therapists)""" @http.route(['/my', '/my/home'], type='http', auth='user', website=True) def home(self, **kw): """Override home to add ADP posting info for Fusion users""" partner = request.env.user.partner_id # Get the standard portal home response response = super().home(**kw) # Add ADP posting info and other data for Fusion users if hasattr(response, 'qcontext') and (partner.is_authorizer or partner.is_sales_rep_portal or partner.is_client_portal or partner.is_technician_portal): posting_info = self._get_adp_posting_info() response.qcontext.update(posting_info) # Add signature count (documents to sign) - only if Sign module is installed sign_count = 0 sign_module_available = 'sign.request.item' in request.env if sign_module_available: sign_count = request.env['sign.request.item'].sudo().search_count([ ('partner_id', '=', partner.id), ('state', '=', 'sent'), ]) response.qcontext['sign_count'] = sign_count response.qcontext['sign_module_available'] = sign_module_available ICP = request.env['ir.config_parameter'].sudo() g_start = ICP.get_param('fusion_claims.portal_gradient_start', '#5ba848') g_mid = ICP.get_param('fusion_claims.portal_gradient_mid', '#3a8fb7') g_end = ICP.get_param('fusion_claims.portal_gradient_end', '#2e7aad') response.qcontext['portal_gradient'] = ( 'linear-gradient(135deg, %s 0%%, %s 60%%, %s 100%%)' % (g_start, g_mid, g_end) ) return response def _prepare_home_portal_values(self, counters): """Add authorizer/sales rep counts to portal home""" values = super()._prepare_home_portal_values(counters) partner = request.env.user.partner_id if 'authorizer_case_count' in counters: if partner.is_authorizer: values['authorizer_case_count'] = request.env['sale.order'].sudo().search_count([ ('x_fc_authorizer_id', '=', partner.id) ]) else: values['authorizer_case_count'] = 0 if 'sales_rep_case_count' in counters: if partner.is_sales_rep_portal: values['sales_rep_case_count'] = request.env['sale.order'].sudo().search_count([ ('user_id', '=', request.env.user.id) ]) else: values['sales_rep_case_count'] = 0 if 'assessment_count' in counters: count = 0 if partner.is_authorizer: count += request.env['fusion.assessment'].sudo().search_count([ ('authorizer_id', '=', partner.id) ]) if partner.is_sales_rep_portal: count += request.env['fusion.assessment'].sudo().search_count([ ('sales_rep_id', '=', request.env.user.id) ]) values['assessment_count'] = count if 'technician_delivery_count' in counters: if partner.is_technician_portal: values['technician_delivery_count'] = request.env['sale.order'].sudo().search_count([ ('x_fc_delivery_technician_ids', 'in', [request.env.user.id]) ]) else: values['technician_delivery_count'] = 0 # Add ADP posting schedule info for portal users if partner.is_authorizer or partner.is_sales_rep_portal or partner.is_client_portal or partner.is_technician_portal: values.update(self._get_adp_posting_info()) return values def _get_adp_posting_info(self): """Get ADP posting schedule information for the portal home.""" from datetime import date, timedelta ICP = request.env['ir.config_parameter'].sudo() # Get base date and frequency from settings base_date_str = ICP.get_param('fusion_claims.adp_posting_base_date', '2026-01-23') frequency = int(ICP.get_param('fusion_claims.adp_posting_frequency_days', '14')) try: base_date = date.fromisoformat(base_date_str) except (ValueError, TypeError): base_date = date(2026, 1, 23) # Get user's timezone for accurate date display user_tz = request.env.user.tz or 'UTC' try: tz = pytz.timezone(user_tz) except pytz.exceptions.UnknownTimeZoneError: tz = pytz.UTC # Get today's date in user's timezone from datetime import datetime now_utc = datetime.now(pytz.UTC) now_local = now_utc.astimezone(tz) today = now_local.date() # Calculate next posting date if today < base_date: next_posting = base_date else: days_since_base = (today - base_date).days cycles_passed = days_since_base // frequency next_posting = base_date + timedelta(days=(cycles_passed + 1) * frequency) # If today is a posting day, return the next one if days_since_base % frequency == 0: next_posting = base_date + timedelta(days=(cycles_passed + 1) * frequency) # Calculate key dates for the posting cycle # Wednesday is submission deadline (posting day - 2 if posting is Friday) days_until_wednesday = (next_posting.weekday() - 2) % 7 if days_until_wednesday == 0 and next_posting.weekday() != 2: days_until_wednesday = 7 submission_deadline = next_posting - timedelta(days=days_until_wednesday) # Get next 3 posting dates for the calendar posting_dates = [] current_posting = next_posting for i in range(6): posting_dates.append({ 'date': current_posting.isoformat(), 'display': current_posting.strftime('%B %d, %Y'), 'day': current_posting.day, 'month': current_posting.strftime('%B'), 'year': current_posting.year, 'weekday': current_posting.strftime('%A'), 'is_next': i == 0, }) current_posting = current_posting + timedelta(days=frequency) # Days until next posting days_until_posting = (next_posting - today).days return { 'next_posting_date': next_posting, 'next_posting_display': next_posting.strftime('%B %d, %Y'), 'next_posting_weekday': next_posting.strftime('%A'), 'submission_deadline': submission_deadline, 'submission_deadline_display': submission_deadline.strftime('%B %d, %Y'), 'days_until_posting': days_until_posting, 'posting_dates': posting_dates, 'current_month': today.strftime('%B %Y'), 'today': today, } # ==================== AUTHORIZER PORTAL ==================== @http.route(['/my/authorizer', '/my/authorizer/dashboard'], type='http', auth='user', website=True) def authorizer_dashboard(self, **kw): """Authorizer dashboard - simplified mobile-first view""" partner = request.env.user.partner_id if not partner.is_authorizer: return request.redirect('/my') SaleOrder = request.env['sale.order'].sudo() Assessment = request.env['fusion.assessment'].sudo() # Base domain for this authorizer base_domain = [('x_fc_authorizer_id', '=', partner.id)] # Total cases total_cases = SaleOrder.search_count(base_domain) # Assessment counts (express + accessibility) express_count = Assessment.search_count([('authorizer_id', '=', partner.id)]) accessibility_count = 0 if 'fusion.accessibility.assessment' in request.env: accessibility_count = request.env['fusion.accessibility.assessment'].sudo().search_count([ ('authorizer_id', '=', partner.id) ]) assessment_count = express_count + accessibility_count # Cases needing authorizer attention (waiting for application) needs_attention = SaleOrder.search( base_domain + [('x_fc_adp_application_status', 'in', [ 'waiting_for_application', 'assessment_completed', ])], order='write_date desc', limit=10, ) # Human-readable status labels status_labels = {} if needs_attention: status_labels = dict(needs_attention[0]._fields['x_fc_adp_application_status'].selection) # Sale type labels sale_type_labels = {} if total_cases: sample = SaleOrder.search(base_domain, limit=1) if sample and 'x_fc_sale_type' in sample._fields: sale_type_labels = dict(sample._fields['x_fc_sale_type'].selection) # Recent cases (last 5 updated) recent_cases = SaleOrder.search( base_domain, order='write_date desc', limit=5, ) # Get status labels from recent cases if not already loaded if not status_labels and recent_cases: status_labels = dict(recent_cases[0]._fields['x_fc_adp_application_status'].selection) # Pending assessments pending_assessments = Assessment.search([ ('authorizer_id', '=', partner.id), ('state', 'in', ['draft', 'pending_signature']) ], limit=5, order='assessment_date desc') company = request.env.company values = { 'partner': partner, 'company': company, 'total_cases': total_cases, 'assessment_count': assessment_count, 'needs_attention': needs_attention, 'recent_cases': recent_cases, 'pending_assessments': pending_assessments, 'status_labels': status_labels, 'sale_type_labels': sale_type_labels, 'page_name': 'authorizer_dashboard', } return request.render('fusion_authorizer_portal.portal_authorizer_dashboard', values) @http.route(['/my/authorizer/cases', '/my/authorizer/cases/page/'], type='http', auth='user', website=True) def authorizer_cases(self, page=1, search='', sortby='date', sale_type='', **kw): """List of cases assigned to the authorizer""" partner = request.env.user.partner_id if not partner.is_authorizer: return request.redirect('/my') SaleOrder = request.env['sale.order'].sudo() # Sale type groupings for filtering sale_type_groups = { 'adp': ['adp', 'adp_odsp'], 'odsp': ['odsp'], 'march_of_dimes': ['march_of_dimes'], 'others': ['wsib', 'direct_private', 'insurance', 'muscular_dystrophy', 'other', 'rental'], } # Build domain from odoo.osv import expression domain = [('x_fc_authorizer_id', '=', partner.id)] # Add sale type filter if sale_type and sale_type in sale_type_groups: domain.append(('x_fc_sale_type', 'in', sale_type_groups[sale_type])) # Add search filter if search: search_domain = [ '|', '|', '|', '|', ('partner_id.name', 'ilike', search), ('name', 'ilike', search), ('x_fc_claim_number', 'ilike', search), ('x_fc_client_ref_1', 'ilike', search), ('x_fc_client_ref_2', 'ilike', search), ] domain = expression.AND([domain, search_domain]) # Sorting sortings = { 'date': {'label': _('Date'), 'order': 'date_order desc'}, 'name': {'label': _('Reference'), 'order': 'name'}, 'client': {'label': _('Client'), 'order': 'partner_id'}, 'state': {'label': _('Status'), 'order': 'state'}, } order = sortings.get(sortby, sortings['date'])['order'] # Pager case_count = SaleOrder.search_count(domain) pager = portal_pager( url='/my/authorizer/cases', url_args={'search': search, 'sortby': sortby, 'sale_type': sale_type}, total=case_count, page=page, step=20, ) # Get cases cases = SaleOrder.search(domain, order=order, limit=20, offset=pager['offset']) values = { 'cases': cases, 'pager': pager, 'search': search, 'sortby': sortby, 'sortings': sortings, 'sale_type': sale_type, 'sale_type_label': { 'adp': 'ADP Cases', 'odsp': 'ODSP Cases', 'march_of_dimes': 'March of Dimes', 'others': 'Other Cases', }.get(sale_type, 'All Cases'), 'page_name': 'authorizer_cases', } return request.render('fusion_authorizer_portal.portal_authorizer_cases', values) @http.route('/my/authorizer/cases/search', type='jsonrpc', auth='user') def authorizer_cases_search(self, query='', **kw): """AJAX search endpoint for real-time search""" partner = request.env.user.partner_id if not partner.is_authorizer: return {'error': 'Access denied', 'results': []} if len(query) < 2: return {'results': []} SaleOrder = request.env['sale.order'].sudo() orders = SaleOrder.get_authorizer_portal_cases(partner.id, search_query=query, limit=50) results = [] for order in orders: results.append({ 'id': order.id, 'name': order.name, 'partner_name': order.partner_id.name if order.partner_id else '', 'date_order': order.date_order.strftime('%Y-%m-%d') if order.date_order else '', 'state': order.state, 'state_display': dict(order._fields['state'].selection).get(order.state, order.state), 'claim_number': getattr(order, 'x_fc_claim_number', '') or '', 'client_ref_1': order.x_fc_client_ref_1 or '', 'client_ref_2': order.x_fc_client_ref_2 or '', 'url': f'/my/authorizer/case/{order.id}', }) return {'results': results} @http.route('/my/authorizer/case/', type='http', auth='user', website=True) def authorizer_case_detail(self, order_id, **kw): """View a specific case""" partner = request.env.user.partner_id if not partner.is_authorizer: return request.redirect('/my') try: order = request.env['sale.order'].sudo().browse(order_id) if not order.exists() or order.x_fc_authorizer_id.id != partner.id: raise AccessError(_('You do not have access to this case.')) except (AccessError, MissingError): return request.redirect('/my/authorizer/cases') # Get documents documents = request.env['fusion.adp.document'].sudo().search([ ('sale_order_id', '=', order_id), ('is_current', '=', True), ]) # Get messages from chatter - only those relevant to this user # (authored by them, sent to them, or mentioning them) all_messages = request.env['mail.message'].sudo().search([ ('model', '=', 'sale.order'), ('res_id', '=', order_id), ('message_type', 'in', ['comment', 'notification']), ('body', '!=', ''), ('body', '!=', '


'), ], order='date desc', limit=100) # Filter to only show messages relevant to this partner: # 1. Messages authored by this partner # 2. Messages where this partner is in notified_partner_ids # 3. Messages where this partner is mentioned (partner_ids) def is_relevant_message(msg): if not msg.body or len(msg.body.strip()) == 0 or '


' in msg.body: return False # Authored by current partner if msg.author_id.id == partner.id: return True # Partner is in notified partners if partner.id in msg.notified_partner_ids.ids: return True # Partner is mentioned in partner_ids if partner.id in msg.partner_ids.ids: return True return False filtered_messages = all_messages.filtered(is_relevant_message) values = { 'order': order, 'documents': documents, 'messages': filtered_messages, 'page_name': 'authorizer_case_detail', } return request.render('fusion_authorizer_portal.portal_authorizer_case_detail', values) @http.route('/my/authorizer/case//comment', type='http', auth='user', website=True, methods=['POST']) def authorizer_add_comment(self, order_id, comment='', **kw): """Add a comment to a case - posts to sale order chatter and emails salesperson""" partner = request.env.user.partner_id if not partner.is_authorizer: return request.redirect('/my') if not comment.strip(): return request.redirect(f'/my/authorizer/case/{order_id}') try: order = request.env['sale.order'].sudo().browse(order_id) if not order.exists() or order.x_fc_authorizer_id.id != partner.id: raise AccessError(_('You do not have access to this case.')) # Post message to sale order chatter (internal note, not to all followers) message = order.message_post( body=comment.strip(), message_type='comment', subtype_xmlid='mail.mt_note', # Internal note - doesn't notify followers author_id=partner.id, ) # Send email notification to the salesperson if order.user_id and order.user_id.partner_id: from markupsafe import Markup salesperson_partner = order.user_id.partner_id order.message_notify( partner_ids=[salesperson_partner.id], body=Markup(f"

New message from Authorizer {partner.name}:

{comment.strip()}

"), subject=f"[{order.name}] New message from Authorizer", author_id=partner.id, ) # Also save to fusion.authorizer.comment for portal display if 'fusion.authorizer.comment' in request.env: request.env['fusion.authorizer.comment'].sudo().create({ 'sale_order_id': order_id, 'author_id': partner.id, 'comment': comment.strip(), 'comment_type': 'general', }) except Exception as e: _logger.error(f"Error adding comment: {e}") return request.redirect(f'/my/authorizer/case/{order_id}') @http.route('/my/authorizer/case//upload', type='http', auth='user', website=True, methods=['POST'], csrf=True) def authorizer_upload_document(self, order_id, document_type='full_application', document_file=None, revision_note='', **kw): """Upload a document for a case""" partner = request.env.user.partner_id if not partner.is_authorizer: return request.redirect('/my') if not document_file or not document_file.filename: return request.redirect(f'/my/authorizer/case/{order_id}') try: order = request.env['sale.order'].sudo().browse(order_id) if not order.exists() or order.x_fc_authorizer_id.id != partner.id: raise AccessError(_('You do not have access to this case.')) # Don't allow authorizers to upload 'submitted_final' if document_type == 'submitted_final': document_type = 'full_application' file_content = document_file.read() file_base64 = base64.b64encode(file_content) request.env['fusion.adp.document'].sudo().create({ 'sale_order_id': order_id, 'document_type': document_type, 'file': file_base64, 'filename': document_file.filename, 'revision_note': revision_note, 'source': 'authorizer', }) except Exception as e: _logger.error(f"Error uploading document: {e}") return request.redirect(f'/my/authorizer/case/{order_id}') @http.route('/my/authorizer/document//download', type='http', auth='user') def authorizer_download_document(self, doc_id, **kw): """Download a document""" partner = request.env.user.partner_id if not partner.is_authorizer and not partner.is_sales_rep_portal: return request.redirect('/my') try: document = request.env['fusion.adp.document'].sudo().browse(doc_id) if not document.exists(): raise MissingError(_('Document not found.')) # Verify access if document.sale_order_id: order = document.sale_order_id has_access = ( (partner.is_authorizer and order.x_fc_authorizer_id.id == partner.id) or (partner.is_sales_rep_portal and order.user_id.id == request.env.user.id) ) if not has_access: raise AccessError(_('You do not have access to this document.')) file_content = base64.b64decode(document.file) # Check if viewing inline or downloading view_inline = kw.get('view', '0') == '1' disposition = 'inline' if view_inline else 'attachment' return request.make_response( file_content, headers=[ ('Content-Type', document.mimetype or 'application/octet-stream'), ('Content-Disposition', f'{disposition}; filename="{document.filename}"'), ('Content-Length', len(file_content)), ] ) except Exception as e: _logger.error(f"Error downloading document: {e}") return request.redirect('/my') @http.route(['/my/authorizer/case//attachment/', '/my/sales/case//attachment/'], type='http', auth='user') def authorizer_download_attachment(self, order_id, attachment_type, **kw): """Download an attachment from sale order (original application, xml, proof of delivery)""" partner = request.env.user.partner_id if not partner.is_authorizer and not partner.is_sales_rep_portal: return request.redirect('/my') try: order = request.env['sale.order'].sudo().browse(order_id) if not order.exists(): raise MissingError(_('Order not found.')) # Verify access has_access = ( (partner.is_authorizer and order.x_fc_authorizer_id.id == partner.id) or (partner.is_sales_rep_portal and order.user_id.id == request.env.user.id) ) if not has_access: raise AccessError(_('You do not have access to this order.')) # Get the attachment based on type attachment_map = { 'original_application': ('x_fc_original_application', 'x_fc_original_application_filename', 'application/pdf', 'original_application.pdf'), 'final_application': ('x_fc_final_submitted_application', 'x_fc_final_application_filename', 'application/pdf', 'final_application.pdf'), 'xml_file': ('x_fc_xml_file', 'x_fc_xml_filename', 'application/xml', 'application.xml'), 'proof_of_delivery': ('x_fc_proof_of_delivery', 'x_fc_proof_of_delivery_filename', 'application/pdf', 'proof_of_delivery.pdf'), } if attachment_type not in attachment_map: raise MissingError(_('Invalid attachment type.')) field_name, filename_field, default_mimetype, default_filename = attachment_map[attachment_type] if not hasattr(order, field_name) or not getattr(order, field_name): raise MissingError(_('Attachment not found.')) file_content = base64.b64decode(getattr(order, field_name)) filename = getattr(order, filename_field, None) or default_filename # Check if viewing inline or downloading view_inline = kw.get('view', '0') == '1' disposition = 'inline' if view_inline else 'attachment' return request.make_response( file_content, headers=[ ('Content-Type', default_mimetype), ('Content-Disposition', f'{disposition}; filename="{filename}"'), ('Content-Length', len(file_content)), ] ) except Exception as e: _logger.error(f"Error downloading attachment: {e}") return request.redirect('/my') @http.route(['/my/authorizer/case//photo/', '/my/sales/case//photo/'], type='http', auth='user') def authorizer_view_photo(self, order_id, photo_id, **kw): """View an approval photo""" partner = request.env.user.partner_id if not partner.is_authorizer and not partner.is_sales_rep_portal: return request.redirect('/my') try: order = request.env['sale.order'].sudo().browse(order_id) if not order.exists(): raise MissingError(_('Order not found.')) # Verify access has_access = ( (partner.is_authorizer and order.x_fc_authorizer_id.id == partner.id) or (partner.is_sales_rep_portal and order.user_id.id == request.env.user.id) ) if not has_access: raise AccessError(_('You do not have access to this order.')) # Find the photo attachment attachment = request.env['ir.attachment'].sudo().browse(photo_id) if not attachment.exists() or attachment.id not in order.x_fc_approval_photo_ids.ids: raise MissingError(_('Photo not found.')) file_content = base64.b64decode(attachment.datas) return request.make_response( file_content, headers=[ ('Content-Type', attachment.mimetype or 'image/png'), ('Content-Disposition', f'inline; filename="{attachment.name}"'), ('Content-Length', len(file_content)), ] ) except Exception as e: _logger.error(f"Error viewing photo: {e}") return request.redirect('/my') # ==================== SALES REP PORTAL ==================== @http.route(['/my/sales', '/my/sales/dashboard'], type='http', auth='user', website=True) def sales_rep_dashboard(self, search='', sale_type='', status='', **kw): """Sales rep dashboard with search and filters""" partner = request.env.user.partner_id user = request.env.user if not partner.is_sales_rep_portal: return request.redirect('/my') SaleOrder = request.env['sale.order'].sudo() Assessment = request.env['fusion.assessment'].sudo() # Get case counts by status (unfiltered for stats) base_domain = [('user_id', '=', user.id)] draft_count = SaleOrder.search_count(base_domain + [('state', '=', 'draft')]) sent_count = SaleOrder.search_count(base_domain + [('state', '=', 'sent')]) sale_count = SaleOrder.search_count(base_domain + [('state', '=', 'sale')]) total_count = SaleOrder.search_count(base_domain) # Build filtered domain for recent cases filtered_domain = base_domain.copy() # Apply search filter if search: search = search.strip() filtered_domain += [ '|', '|', '|', ('name', 'ilike', search), ('partner_id.name', 'ilike', search), ('x_fc_claim_number', 'ilike', search), ('partner_id.email', 'ilike', search), ] # Apply sale type filter if sale_type: filtered_domain += [('x_fc_sale_type', '=', sale_type)] # Apply status filter if status: filtered_domain += [('state', '=', status)] # Recent cases (filtered) recent_cases = SaleOrder.search(filtered_domain, limit=20, order='date_order desc') # Assessments assessment_domain = [('sales_rep_id', '=', user.id)] pending_assessments = Assessment.search( assessment_domain + [('state', 'in', ['draft', 'pending_signature'])], limit=5, order='assessment_date desc' ) completed_assessments_count = Assessment.search_count( assessment_domain + [('state', '=', 'completed')] ) values = { 'partner': partner, 'draft_count': draft_count, 'sent_count': sent_count, 'sale_count': sale_count, 'total_count': total_count, 'recent_cases': recent_cases, 'pending_assessments': pending_assessments, 'completed_assessments_count': completed_assessments_count, 'page_name': 'sales_dashboard', # Search and filter values 'search': search, 'sale_type_filter': sale_type, 'status_filter': status, } return request.render('fusion_authorizer_portal.portal_sales_dashboard', values) @http.route(['/my/sales/cases', '/my/sales/cases/page/'], type='http', auth='user', website=True) def sales_rep_cases(self, page=1, search='', sortby='date', **kw): """List of cases for the sales rep""" partner = request.env.user.partner_id user = request.env.user if not partner.is_sales_rep_portal: return request.redirect('/my') SaleOrder = request.env['sale.order'].sudo() # Build domain from odoo.osv import expression domain = [('user_id', '=', user.id)] # Add search filter if search: search_domain = [ '|', '|', '|', '|', ('partner_id.name', 'ilike', search), ('name', 'ilike', search), ('x_fc_claim_number', 'ilike', search), ('x_fc_client_ref_1', 'ilike', search), ('x_fc_client_ref_2', 'ilike', search), ] domain = expression.AND([domain, search_domain]) # Sorting sortings = { 'date': {'label': _('Date'), 'order': 'date_order desc'}, 'name': {'label': _('Reference'), 'order': 'name'}, 'client': {'label': _('Client'), 'order': 'partner_id'}, 'state': {'label': _('Status'), 'order': 'state'}, } order = sortings.get(sortby, sortings['date'])['order'] # Pager case_count = SaleOrder.search_count(domain) pager = portal_pager( url='/my/sales/cases', url_args={'search': search, 'sortby': sortby}, total=case_count, page=page, step=20, ) # Get cases cases = SaleOrder.search(domain, order=order, limit=20, offset=pager['offset']) values = { 'cases': cases, 'pager': pager, 'search': search, 'sortby': sortby, 'sortings': sortings, 'page_name': 'sales_cases', } return request.render('fusion_authorizer_portal.portal_sales_cases', values) @http.route('/my/sales/cases/search', type='jsonrpc', auth='user') def sales_rep_cases_search(self, query='', **kw): """AJAX search endpoint for sales rep real-time search""" partner = request.env.user.partner_id user = request.env.user if not partner.is_sales_rep_portal: return {'error': 'Access denied', 'results': []} if len(query) < 2: return {'results': []} SaleOrder = request.env['sale.order'].sudo() orders = SaleOrder.get_sales_rep_portal_cases(user.id, search_query=query, limit=50) results = [] for order in orders: results.append({ 'id': order.id, 'name': order.name, 'partner_name': order.partner_id.name if order.partner_id else '', 'date_order': order.date_order.strftime('%Y-%m-%d') if order.date_order else '', 'state': order.state, 'state_display': dict(order._fields['state'].selection).get(order.state, order.state), 'claim_number': getattr(order, 'x_fc_claim_number', '') or '', 'client_ref_1': order.x_fc_client_ref_1 or '', 'client_ref_2': order.x_fc_client_ref_2 or '', 'url': f'/my/sales/case/{order.id}', }) return {'results': results} @http.route('/my/sales/case/', type='http', auth='user', website=True) def sales_rep_case_detail(self, order_id, **kw): """View a specific case for sales rep""" partner = request.env.user.partner_id user = request.env.user if not partner.is_sales_rep_portal: return request.redirect('/my') try: order = request.env['sale.order'].sudo().browse(order_id) if not order.exists() or order.user_id.id != user.id: raise AccessError(_('You do not have access to this case.')) except (AccessError, MissingError): return request.redirect('/my/sales/cases') # Get documents documents = request.env['fusion.adp.document'].sudo().search([ ('sale_order_id', '=', order_id), ('is_current', '=', True), ]) # Get messages from chatter - only those relevant to this user all_messages = request.env['mail.message'].sudo().search([ ('model', '=', 'sale.order'), ('res_id', '=', order_id), ('message_type', 'in', ['comment', 'notification']), ('body', '!=', ''), ('body', '!=', '


'), ], order='date desc', limit=100) # Filter to only show messages relevant to this partner: # 1. Messages authored by this partner # 2. Messages where this partner is in notified_partner_ids # 3. Messages where this partner is mentioned (partner_ids) def is_relevant_message(msg): if not msg.body or len(msg.body.strip()) == 0 or '


' in msg.body: return False # Authored by current partner if msg.author_id.id == partner.id: return True # Partner is in notified partners if partner.id in msg.notified_partner_ids.ids: return True # Partner is mentioned in partner_ids if partner.id in msg.partner_ids.ids: return True return False filtered_messages = all_messages.filtered(is_relevant_message) values = { 'order': order, 'documents': documents, 'messages': filtered_messages, 'page_name': 'sales_case_detail', } return request.render('fusion_authorizer_portal.portal_sales_case_detail', values) @http.route('/my/sales/case//comment', type='http', auth='user', website=True, methods=['POST']) def sales_rep_add_comment(self, order_id, comment='', **kw): """Add a comment to a case (sales rep) - posts to sale order chatter and emails authorizer""" partner = request.env.user.partner_id user = request.env.user if not partner.is_sales_rep_portal: return request.redirect('/my') if not comment.strip(): return request.redirect(f'/my/sales/case/{order_id}') try: order = request.env['sale.order'].sudo().browse(order_id) if not order.exists() or order.user_id.id != user.id: raise AccessError(_('You do not have access to this case.')) # Post message to sale order chatter (internal note, not to all followers) message = order.message_post( body=comment.strip(), message_type='comment', subtype_xmlid='mail.mt_note', # Internal note - doesn't notify followers author_id=partner.id, ) # Send email notification to the authorizer if order.x_fc_authorizer_id: from markupsafe import Markup order.message_notify( partner_ids=[order.x_fc_authorizer_id.id], body=Markup(f"

New message from Sales Rep {partner.name}:

{comment.strip()}

"), subject=f"[{order.name}] New message from Sales Rep", author_id=partner.id, ) # Also save to fusion.authorizer.comment for portal display if 'fusion.authorizer.comment' in request.env: request.env['fusion.authorizer.comment'].sudo().create({ 'sale_order_id': order_id, 'author_id': partner.id, 'comment': comment.strip(), 'comment_type': 'general', }) except Exception as e: _logger.error(f"Error adding comment: {e}") return request.redirect(f'/my/sales/case/{order_id}') # ==================== CLIENT FUNDING CLAIMS PORTAL ==================== def _prepare_home_portal_values(self, counters): """Add client funding claims count to portal home""" values = super()._prepare_home_portal_values(counters) partner = request.env.user.partner_id if 'funding_claims_count' in counters: # Count sale orders where partner is the customer values['funding_claims_count'] = request.env['sale.order'].sudo().search_count([ ('partner_id', '=', partner.id), ('x_fc_sale_type', 'in', ['adp', 'adp_odsp', 'odsp', 'march_of_dimes']), ]) return values @http.route(['/my/funding-claims', '/my/funding-claims/page/'], type='http', auth='user', website=True) def client_funding_claims(self, page=1, sortby='date', **kw): """List of funding claims for the client""" partner = request.env.user.partner_id SaleOrder = request.env['sale.order'].sudo() # Build domain - orders where partner is the customer domain = [ ('partner_id', '=', partner.id), ('x_fc_sale_type', 'in', ['adp', 'adp_odsp', 'odsp', 'march_of_dimes']), ] # Sorting sortings = { 'date': {'label': _('Date'), 'order': 'date_order desc'}, 'name': {'label': _('Reference'), 'order': 'name'}, 'status': {'label': _('Status'), 'order': 'x_fc_adp_application_status'}, } order = sortings.get(sortby, sortings['date'])['order'] # Pager claim_count = SaleOrder.search_count(domain) pager = portal_pager( url='/my/funding-claims', url_args={'sortby': sortby}, total=claim_count, page=page, step=20, ) # Get claims claims = SaleOrder.search(domain, order=order, limit=20, offset=pager['offset']) values = { 'claims': claims, 'pager': pager, 'sortby': sortby, 'sortings': sortings, 'page_name': 'funding_claims', } return request.render('fusion_authorizer_portal.portal_client_claims', values) @http.route('/my/funding-claims/', type='http', auth='user', website=True) def client_funding_claim_detail(self, order_id, **kw): """View a specific funding claim""" partner = request.env.user.partner_id try: order = request.env['sale.order'].sudo().browse(order_id) if not order.exists() or order.partner_id.id != partner.id: raise AccessError(_('You do not have access to this claim.')) except (AccessError, MissingError): return request.redirect('/my/funding-claims') # Check if case is closed - documents only visible after case closed is_case_closed = order.x_fc_adp_application_status == 'case_closed' # Get documents (only if case is closed) documents = [] if is_case_closed: documents = request.env['fusion.adp.document'].sudo().search([ ('sale_order_id', '=', order_id), ('is_current', '=', True), ('document_type', 'in', ['submitted_final', 'pages_11_12']), ]) # Get invoices invoices = order.invoice_ids.filtered(lambda inv: inv.state != 'cancel') values = { 'order': order, 'is_case_closed': is_case_closed, 'documents': documents, 'invoices': invoices, 'page_name': 'funding_claim_detail', } return request.render('fusion_authorizer_portal.portal_client_claim_detail', values) @http.route('/my/funding-claims//document//download', type='http', auth='user') def client_download_document(self, order_id, doc_id, **kw): """Download a document from a funding claim""" partner = request.env.user.partner_id try: order = request.env['sale.order'].sudo().browse(order_id) if not order.exists() or order.partner_id.id != partner.id: raise AccessError(_('You do not have access to this claim.')) # Check if case is closed if order.x_fc_adp_application_status != 'case_closed': raise AccessError(_('Documents are only available after the case is closed.')) document = request.env['fusion.adp.document'].sudo().browse(doc_id) if not document.exists() or document.sale_order_id.id != order_id: raise MissingError(_('Document not found.')) file_content = base64.b64decode(document.file) return request.make_response( file_content, headers=[ ('Content-Type', document.mimetype or 'application/octet-stream'), ('Content-Disposition', f'attachment; filename="{document.filename}"'), ('Content-Length', len(file_content)), ] ) except Exception as e: _logger.error(f"Error downloading document: {e}") return request.redirect('/my/funding-claims') @http.route('/my/funding-claims//proof-of-delivery', type='http', auth='user') def client_download_proof_of_delivery(self, order_id, **kw): """Download proof of delivery from a funding claim""" partner = request.env.user.partner_id try: order = request.env['sale.order'].sudo().browse(order_id) if not order.exists() or order.partner_id.id != partner.id: raise AccessError(_('You do not have access to this claim.')) # Check if case is closed if order.x_fc_adp_application_status != 'case_closed': raise AccessError(_('Documents are only available after the case is closed.')) if not order.x_fc_proof_of_delivery: raise MissingError(_('Proof of delivery not found.')) file_content = base64.b64decode(order.x_fc_proof_of_delivery) filename = order.x_fc_proof_of_delivery_filename or 'proof_of_delivery.pdf' return request.make_response( file_content, headers=[ ('Content-Type', 'application/pdf'), ('Content-Disposition', f'attachment; filename="{filename}"'), ('Content-Length', len(file_content)), ] ) except Exception as e: _logger.error(f"Error downloading proof of delivery: {e}") return request.redirect('/my/funding-claims') # ==================== TECHNICIAN PORTAL ==================== def _check_technician_access(self): """Check if current user is a technician portal user.""" partner = request.env.user.partner_id if not partner.is_technician_portal: return False return True @http.route(['/my/technician', '/my/technician/dashboard'], type='http', auth='user', website=True) def technician_dashboard(self, **kw): """Technician dashboard - today's schedule with timeline.""" if not self._check_technician_access(): return request.redirect('/my') partner = request.env.user.partner_id user = request.env.user Task = request.env['fusion.technician.task'].sudo() SaleOrder = request.env['sale.order'].sudo() today = fields.Date.context_today(request.env['fusion.technician.task']) # Today's tasks today_tasks = Task.search([ ('technician_id', '=', user.id), ('scheduled_date', '=', today), ('status', '!=', 'cancelled'), ], order='sequence, time_start, id') # Current in-progress task current_task = today_tasks.filtered(lambda t: t.status == 'in_progress')[:1] # Next upcoming task (first scheduled/en_route today) next_task = today_tasks.filtered(lambda t: t.status in ('scheduled', 'en_route'))[:1] # Stats completed_today = len(today_tasks.filtered(lambda t: t.status == 'completed')) remaining_today = len(today_tasks.filtered(lambda t: t.status in ('scheduled', 'en_route', 'in_progress'))) total_today = len(today_tasks) # Total travel time for the day total_travel = sum(today_tasks.mapped('travel_time_minutes')) # Legacy: deliveries assigned (for backward compat with existing delivery views) delivery_domain = [('x_fc_delivery_technician_ids', 'in', [user.id])] pending_pod_count = SaleOrder.search_count(delivery_domain + [ ('x_fc_pod_signature', '=', False), ('x_fc_adp_application_status', '=', 'ready_delivery'), ]) # Tomorrow's task count from datetime import timedelta tomorrow = today + timedelta(days=1) tomorrow_count = Task.search_count([ ('technician_id', '=', user.id), ('scheduled_date', '=', tomorrow), ('status', '!=', 'cancelled'), ]) # Technician's personal start address start_address = user.sudo().x_fc_start_address or '' # Google Maps API key for Places autocomplete ICP = request.env['ir.config_parameter'].sudo() google_maps_api_key = ICP.get_param('fusion_claims.google_maps_api_key', '') values = { 'today_tasks': today_tasks, 'current_task': current_task, 'next_task': next_task, 'completed_today': completed_today, 'remaining_today': remaining_today, 'total_today': total_today, 'total_travel': total_travel, 'pending_pod_count': pending_pod_count, 'tomorrow_count': tomorrow_count, 'today_date': today, 'start_address': start_address, 'google_maps_api_key': google_maps_api_key, 'page_name': 'technician_dashboard', } return request.render('fusion_authorizer_portal.portal_technician_dashboard', values) @http.route(['/my/technician/tasks', '/my/technician/tasks/page/'], type='http', auth='user', website=True) def technician_tasks(self, page=1, search='', filter_status='all', filter_date='', **kw): """List of all tasks for the technician.""" if not self._check_technician_access(): return request.redirect('/my') user = request.env.user Task = request.env['fusion.technician.task'].sudo() domain = [('technician_id', '=', user.id)] if filter_status == 'scheduled': domain.append(('status', '=', 'scheduled')) elif filter_status == 'in_progress': domain.append(('status', 'in', ('en_route', 'in_progress'))) elif filter_status == 'completed': domain.append(('status', '=', 'completed')) elif filter_status == 'active': domain.append(('status', 'not in', ('cancelled', 'completed'))) # Default: show all if filter_date: domain.append(('scheduled_date', '=', filter_date)) if search: domain += ['|', '|', '|', ('name', 'ilike', search), ('partner_id.name', 'ilike', search), ('address_city', 'ilike', search), ('sale_order_id.name', 'ilike', search), ] task_count = Task.search_count(domain) pager = portal_pager( url='/my/technician/tasks', url_args={'search': search, 'filter_status': filter_status, 'filter_date': filter_date}, total=task_count, page=page, step=20, ) tasks = Task.search(domain, limit=20, offset=pager['offset'], order='scheduled_date desc, sequence, time_start') values = { 'tasks': tasks, 'pager': pager, 'search': search, 'filter_status': filter_status, 'filter_date': filter_date, 'page_name': 'technician_tasks', } return request.render('fusion_authorizer_portal.portal_technician_tasks', values) @http.route('/my/technician/task/', type='http', auth='user', website=True) def technician_task_detail(self, task_id, **kw): """View a specific technician task.""" if not self._check_technician_access(): return request.redirect('/my') user = request.env.user Task = request.env['fusion.technician.task'].sudo() try: task = Task.browse(task_id) if not task.exists() or task.technician_id.id != user.id: raise AccessError(_('You do not have access to this task.')) except (AccessError, MissingError): return request.redirect('/my/technician/tasks') # Check for earlier uncompleted tasks (sequential enforcement) earlier_incomplete = Task.search([ ('technician_id', '=', user.id), ('scheduled_date', '=', task.scheduled_date), ('time_start', '<', task.time_start), ('status', 'not in', ['completed', 'cancelled']), ('id', '!=', task.id), ], order='time_start', limit=1) # Get order lines if linked to a sale order order_lines = [] if task.sale_order_id: order_lines = task.sale_order_id.order_line.filtered(lambda l: not l.display_type) # Get VAPID public key for push notifications vapid_public = request.env['ir.config_parameter'].sudo().get_param( 'fusion_claims.vapid_public_key', '' ) values = { 'task': task, 'order_lines': order_lines, 'vapid_public_key': vapid_public, 'page_name': 'technician_task_detail', 'earlier_incomplete': earlier_incomplete, } return request.render('fusion_authorizer_portal.portal_technician_task_detail', values) @http.route('/my/technician/task//add-notes', type='json', auth='user', website=True) def technician_task_add_notes(self, task_id, notes, photos=None, **kw): """Add notes (and optional photos) to a completed task. :param notes: text content of the note :param photos: list of dicts with 'data' (base64 data-url) and 'name' """ if not self._check_technician_access(): return {'success': False, 'error': 'Access denied'} user = request.env.user Task = request.env['fusion.technician.task'].sudo() Attachment = request.env['ir.attachment'].sudo() try: task = Task.browse(task_id) if not task.exists() or task.technician_id.id != user.id: return {'success': False, 'error': 'Task not found'} from markupsafe import Markup, escape import re # ---------------------------------------------------------- # Process photos -> create ir.attachment records # ---------------------------------------------------------- attachment_ids = [] if photos: for i, photo in enumerate(photos): photo_data = photo.get('data', '') photo_name = photo.get('name', f'photo_{i+1}.jpg') if not photo_data: continue # Strip data-url prefix (e.g. "data:image/jpeg;base64,...") if ',' in photo_data: photo_data = photo_data.split(',', 1)[1] try: att = Attachment.create({ 'name': photo_name, 'type': 'binary', 'datas': photo_data, 'res_model': 'fusion.technician.task', 'res_id': task.id, 'mimetype': 'image/jpeg', }) attachment_ids.append(att.id) except Exception as e: _logger.warning("Failed to attach photo %s: %s", photo_name, e) # ---------------------------------------------------------- # Sanitize and format the notes text # ---------------------------------------------------------- safe_notes = str(escape(notes or '')) formatted_notes = re.sub(r'\n', '
', safe_notes) timestamp = fields.Datetime.now().strftime("%b %d, %Y %I:%M %p") safe_user = str(escape(user.name)) safe_task = str(escape(task.name)) has_text = bool((notes or '').strip()) photo_count = len(attachment_ids) # Build a small photo summary for inline display photo_html = '' if photo_count: photo_html = '
%d photo(s) attached
' % photo_count # --- 1. Append to the completion_notes field on the task --- note_parts = [] if has_text: note_parts.append( '
%s
' % formatted_notes ) if photo_html: note_parts.append(photo_html) if note_parts: new_note = Markup( '
' '%s - %s' '%s' '
' ) % (Markup(timestamp), Markup(safe_user), Markup(''.join(note_parts))) existing = task.completion_notes or '' task.completion_notes = Markup(existing) + new_note # --- 2. Post to the TASK chatter --- chatter_parts = [] if has_text: chatter_parts.append( '
%s
' % formatted_notes ) if photo_html: chatter_parts.append(photo_html) task_chatter = Markup( '
' ' Note Added' '%s' 'By %s' '
' ) % (Markup(''.join(chatter_parts)), Markup(safe_user)) task.message_post( body=task_chatter, message_type='comment', subtype_xmlid='mail.mt_note', attachment_ids=attachment_ids, ) # --- 3. Post to the SALE ORDER chatter (if linked) --- if task.sale_order_id: so_chatter = Markup( '
' ' Technician Note - %s' '%s' 'By %s on %s' '
' ) % (Markup(safe_task), Markup(''.join(chatter_parts)), Markup(safe_user), Markup(timestamp)) # Duplicate attachments for the sale order so both records show them so_att_ids = [] for att_id in attachment_ids: att = Attachment.browse(att_id) so_att = att.copy({ 'res_model': 'sale.order', 'res_id': task.sale_order_id.id, }) so_att_ids.append(so_att.id) task.sale_order_id.message_post( body=so_chatter, message_type='comment', subtype_xmlid='mail.mt_note', attachment_ids=so_att_ids, ) return {'success': True} except Exception as e: _logger.error(f"Add notes error: {e}") return {'success': False, 'error': str(e)} @http.route('/my/technician/task//action', type='json', auth='user', website=True) def technician_task_action(self, task_id, action, **kw): """Handle task status changes (start, complete, en_route, cancel).""" if not self._check_technician_access(): return {'success': False, 'error': 'Access denied'} user = request.env.user Task = request.env['fusion.technician.task'].sudo() try: task = Task.browse(task_id) if not task.exists() or task.technician_id.id != user.id: return {'success': False, 'error': 'Task not found or not assigned to you'} if action == 'en_route': task.action_start_en_route() elif action == 'start': task.action_start_task() elif action == 'complete': completion_notes = kw.get('completion_notes', '') if completion_notes: task.completion_notes = completion_notes task.action_complete_task() elif action == 'cancel': task.action_cancel_task() else: return {'success': False, 'error': f'Unknown action: {action}'} # For completion, also return next task info result = { 'success': True, 'status': task.status, 'redirect_url': f'/my/technician/task/{task_id}', } if action == 'complete': next_task = task.get_next_task_for_technician() if next_task: result['next_task_id'] = next_task.id result['next_task_url'] = f'/my/technician/task/{next_task.id}' result['next_task_name'] = next_task.name result['next_task_time'] = task._float_to_time_str(next_task.time_start) else: result['next_task_id'] = False result['all_done'] = True return result except Exception as e: _logger.error(f"Task action error: {e}") return {'success': False, 'error': str(e)} @http.route('/my/technician/task//voice-transcribe', type='json', auth='user', website=True) def technician_voice_transcribe(self, task_id, audio_data, mime_type='audio/webm', **kw): """Transcribe voice recording using OpenAI Whisper, translate to English.""" if not self._check_technician_access(): return {'success': False, 'error': 'Access denied'} user = request.env.user Task = request.env['fusion.technician.task'].sudo() ICP = request.env['ir.config_parameter'].sudo() task = Task.browse(task_id) if not task.exists() or task.technician_id.id != user.id: return {'success': False, 'error': 'Task not found'} api_key = ICP.get_param('fusion_notes.openai_api_key') or ICP.get_param('fusion_claims.ai_api_key', '') if not api_key: return {'success': False, 'error': 'OpenAI API key not configured'} import base64 import tempfile import os import requests as http_requests try: # Decode audio audio_bytes = base64.b64decode(audio_data) ext_map = {'audio/webm': '.webm', 'audio/ogg': '.ogg', 'audio/mp4': '.m4a', 'audio/wav': '.wav'} ext = ext_map.get(mime_type, '.webm') with tempfile.NamedTemporaryFile(suffix=ext, delete=False) as tmp: tmp.write(audio_bytes) tmp_path = tmp.name # Call Whisper API - use 'translations' endpoint to auto-translate to English with open(tmp_path, 'rb') as f: resp = http_requests.post( 'https://api.openai.com/v1/audio/translations', headers={'Authorization': f'Bearer {api_key}'}, files={'file': (f'recording{ext}', f, mime_type)}, data={'model': 'whisper-1', 'response_format': 'text'}, timeout=60, ) os.unlink(tmp_path) if resp.status_code != 200: return {'success': False, 'error': f'Whisper API error: {resp.text}'} transcription = resp.text.strip() # Store transcription and audio on task task.write({ 'voice_note_audio': audio_data, 'voice_note_transcription': transcription, }) return {'success': True, 'transcription': transcription} except Exception as e: _logger.error(f"Voice transcription error: {e}") return {'success': False, 'error': str(e)} @http.route('/my/technician/task//ai-format', type='json', auth='user', website=True) def technician_ai_format(self, task_id, text, **kw): """Use GPT to clean up and format raw notes text.""" if not self._check_technician_access(): return {'success': False, 'error': 'Access denied'} user = request.env.user Task = request.env['fusion.technician.task'].sudo() ICP = request.env['ir.config_parameter'].sudo() task = Task.browse(task_id) if not task.exists() or task.technician_id.id != user.id: return {'success': False, 'error': 'Task not found'} api_key = ICP.get_param('fusion_notes.openai_api_key') or ICP.get_param('fusion_claims.ai_api_key', '') if not api_key: return {'success': False, 'error': 'OpenAI API key not configured'} ai_model = ICP.get_param('fusion_claims.ai_model', 'gpt-4o-mini') import requests as http_requests try: task_type_label = dict(Task._fields['task_type'].selection).get(task.task_type, task.task_type) system_prompt = ( "You are formatting a field technician's notes into clear, professional text. " "ALWAYS output in English. If the input is in another language, translate it to English. " "Fix grammar, spelling, and punctuation. Remove filler words. " "Keep all facts from the original. Make it concise and professional. " "If it mentions work done, parts used, issues, or follow-ups, organize them clearly. " "Return plain text only (no HTML)." ) resp = http_requests.post( 'https://api.openai.com/v1/chat/completions', headers={'Authorization': f'Bearer {api_key}', 'Content-Type': 'application/json'}, json={ 'model': ai_model, 'messages': [ {'role': 'system', 'content': system_prompt}, {'role': 'user', 'content': f'Task type: {task_type_label}\nClient: {task.partner_id.name or "N/A"}\nRaw notes:\n{text}'}, ], 'temperature': 0.3, 'max_tokens': 1000, }, timeout=30, ) if resp.status_code == 200: data = resp.json() formatted = data['choices'][0]['message']['content'] return {'success': True, 'formatted_text': formatted} else: return {'success': False, 'error': f'AI service error ({resp.status_code})'} except Exception as e: _logger.error(f"AI format error: {e}") return {'success': False, 'error': str(e)} @http.route('/my/technician/task//voice-complete', type='json', auth='user', website=True) def technician_voice_complete(self, task_id, transcription, **kw): """Format transcription with GPT and complete the task.""" if not self._check_technician_access(): return {'success': False, 'error': 'Access denied'} user = request.env.user Task = request.env['fusion.technician.task'].sudo() ICP = request.env['ir.config_parameter'].sudo() task = Task.browse(task_id) if not task.exists() or task.technician_id.id != user.id: return {'success': False, 'error': 'Task not found'} api_key = ICP.get_param('fusion_notes.openai_api_key') or ICP.get_param('fusion_claims.ai_api_key', '') ai_model = ICP.get_param('fusion_claims.ai_model', 'gpt-4o-mini') formatted_notes = transcription # fallback if api_key: import requests as http_requests try: task_type_label = dict(Task._fields['task_type'].selection).get(task.task_type, task.task_type) system_prompt = ( "You are formatting a technician's voice note into a structured completion report. " "ALWAYS output in English. If the input is in another language, translate it to English. " "The technician recorded this after completing a field task. " "Format it into clear, professional HTML with these sections:\n" "Work Performed: [summary]\n" "Parts Used: [if mentioned, otherwise 'None mentioned']\n" "Issues Found: [if any, otherwise 'None']\n" "Follow-up Required: [yes/no + details]\n" "Client Feedback: [if mentioned, otherwise 'N/A']\n\n" "Keep all facts from the original. Fix grammar. Remove filler words. " "Use
for line breaks, for labels. Keep it concise." ) resp = http_requests.post( 'https://api.openai.com/v1/chat/completions', headers={'Authorization': f'Bearer {api_key}', 'Content-Type': 'application/json'}, json={ 'model': ai_model, 'messages': [ {'role': 'system', 'content': system_prompt}, {'role': 'user', 'content': f'Task type: {task_type_label}\nTechnician: {user.name}\nVoice note:\n{transcription}'}, ], 'temperature': 0.3, 'max_tokens': 1000, }, timeout=30, ) if resp.status_code == 200: data = resp.json() formatted_notes = data['choices'][0]['message']['content'] except Exception as e: _logger.warning(f"GPT formatting failed, using raw transcription: {e}") # Build final HTML completion report from markupsafe import Markup completion_html = Markup( f'
' f'

Technician: {user.name}
' f'Task: {task.name} ({dict(Task._fields["task_type"].selection).get(task.task_type, task.task_type)})

' f'
' f'{formatted_notes}' f'
' ) task.write({ 'completion_notes': completion_html, 'voice_note_transcription': transcription, }) task.action_complete_task() return { 'success': True, 'formatted_notes': formatted_notes, 'redirect_url': f'/my/technician/task/{task_id}', } @http.route('/my/technician/tomorrow', type='http', auth='user', website=True) def technician_tomorrow(self, **kw): """Next day preparation view.""" if not self._check_technician_access(): return request.redirect('/my') user = request.env.user Task = request.env['fusion.technician.task'].sudo() from datetime import timedelta today = fields.Date.context_today(Task) tomorrow = today + timedelta(days=1) tomorrow_tasks = Task.search([ ('technician_id', '=', user.id), ('scheduled_date', '=', tomorrow), ('status', '!=', 'cancelled'), ], order='sequence, time_start, id') total_travel = sum(tomorrow_tasks.mapped('travel_time_minutes')) total_distance = sum(tomorrow_tasks.mapped('travel_distance_km')) # Aggregate equipment needed all_equipment = [] for t in tomorrow_tasks: if t.equipment_needed: all_equipment.append(f"{t.name}: {t.equipment_needed}") values = { 'tomorrow_tasks': tomorrow_tasks, 'tomorrow_date': tomorrow, 'total_travel': total_travel, 'total_distance': total_distance, 'all_equipment': all_equipment, 'page_name': 'technician_tomorrow', } return request.render('fusion_authorizer_portal.portal_technician_tomorrow', values) @http.route('/my/technician/schedule/', type='http', auth='user', website=True) def technician_schedule_date(self, date, **kw): """View schedule for a specific date.""" if not self._check_technician_access(): return request.redirect('/my') user = request.env.user Task = request.env['fusion.technician.task'].sudo() try: schedule_date = fields.Date.from_string(date) except (ValueError, TypeError): return request.redirect('/my/technician') tasks = Task.search([ ('technician_id', '=', user.id), ('scheduled_date', '=', schedule_date), ('status', '!=', 'cancelled'), ], order='sequence, time_start, id') total_travel = sum(tasks.mapped('travel_time_minutes')) values = { 'tasks': tasks, 'schedule_date': schedule_date, 'total_travel': total_travel, 'page_name': 'technician_schedule', } return request.render('fusion_authorizer_portal.portal_technician_schedule_date', values) @http.route('/my/technician/admin/map', type='http', auth='user', website=True) def technician_location_map(self, **kw): """Admin map view showing latest technician locations using Google Maps.""" user = request.env.user if not user.has_group('sales_team.group_sale_manager') and not user.has_group('sales_team.group_sale_salesman'): return request.redirect('/my') LocationModel = request.env['fusion.technician.location'].sudo() locations = LocationModel.get_latest_locations() api_key = request.env['ir.config_parameter'].sudo().get_param( 'fusion_claims.google_maps_api_key', '' ) values = { 'locations': locations, 'google_maps_api_key': api_key, } return request.render('fusion_authorizer_portal.portal_technician_map', values) @http.route('/my/technician/location/log', type='json', auth='user', website=True) def technician_location_log(self, latitude, longitude, accuracy=None, **kw): """Log the technician's current GPS location.""" if not self._check_technician_access(): return {'success': False} try: request.env['fusion.technician.location'].sudo().log_location( latitude=latitude, longitude=longitude, accuracy=accuracy, ) return {'success': True} except Exception as e: _logger.warning(f"Location log error: {e}") return {'success': False} @http.route('/my/technician/settings/start-location', type='json', auth='user', website=True) def technician_save_start_location(self, address='', **kw): """Save the technician's personal start location.""" if not self._check_technician_access(): return {'success': False, 'error': 'Access denied'} try: request.env.user.sudo().write({ 'x_fc_start_address': address.strip() if address else False, }) return {'success': True} except Exception as e: _logger.warning("Error saving start location: %s", e) return {'success': False, 'error': str(e)} @http.route('/my/technician/push/subscribe', type='json', auth='user', website=True) def technician_push_subscribe(self, endpoint, p256dh, auth, **kw): """Register a push notification subscription.""" user = request.env.user PushSub = request.env['fusion.push.subscription'].sudo() browser_info = request.httprequest.headers.get('User-Agent', '')[:200] sub = PushSub.register_subscription(user.id, endpoint, p256dh, auth, browser_info) return {'success': True, 'subscription_id': sub.id} # Keep legacy delivery routes for backward compatibility @http.route(['/my/technician/deliveries', '/my/technician/deliveries/page/'], type='http', auth='user', website=True) def technician_deliveries(self, page=1, search='', filter_status='all', **kw): """Legacy: List of deliveries for the technician (redirects to tasks).""" return request.redirect('/my/technician/tasks?filter_status=all') @http.route('/my/technician/delivery/', type='http', auth='user', website=True) def technician_delivery_detail(self, order_id, **kw): """View a specific delivery for technician (legacy, still works).""" if not self._check_technician_access(): return request.redirect('/my') user = request.env.user try: order = request.env['sale.order'].sudo().browse(order_id) if not order.exists() or user.id not in order.x_fc_delivery_technician_ids.ids: raise AccessError(_('You do not have access to this delivery.')) except (AccessError, MissingError): return request.redirect('/my/technician/tasks') values = { 'order': order, 'page_name': 'technician_delivery_detail', } return request.render('fusion_authorizer_portal.portal_technician_delivery_detail_legacy', values) # ==================== POD SIGNATURE CAPTURE ==================== @http.route('/my/pod/', type='http', auth='user', website=True) def pod_signature_page(self, order_id, **kw): """POD signature capture page - accessible by technicians and sales reps""" partner = request.env.user.partner_id user = request.env.user try: order = request.env['sale.order'].sudo().browse(order_id) if not order.exists(): raise MissingError(_('Order not found.')) # Check access - technician (via delivery list or task assignment), sales rep, or internal staff has_access = False user_role = None # Technician: check delivery technician list on the order if partner.is_technician_portal and user.id in order.x_fc_delivery_technician_ids.ids: has_access = True user_role = 'technician' # Technician: also check if user is assigned to a task linked to this order if not has_access and partner.is_technician_portal: task_count = request.env['fusion.technician.task'].sudo().search_count([ ('sale_order_id', '=', order.id), ('technician_id', '=', user.id), ]) if task_count: has_access = True user_role = 'technician' # Internal users with field staff flag can always collect POD if not has_access and not user.share: has_access = True user_role = 'technician' # Sales rep: own orders if not has_access: if partner.is_sales_rep_portal and order.user_id.id == user.id: has_access = True user_role = 'sales_rep' elif order.user_id.id == user.id: has_access = True user_role = 'sales_rep' if not has_access: raise AccessError(_('You do not have access to this order.')) except (AccessError, MissingError) as e: _logger.warning(f"POD access denied for user {user.id} on order {order_id}: {e}") return request.redirect('/my') # Get delivery address delivery_address = order.partner_shipping_id or order.partner_id values = { 'order': order, 'delivery_address': delivery_address, 'user_role': user_role, 'has_existing_signature': bool(order.x_fc_pod_signature), 'page_name': 'pod_signature', } return request.render('fusion_authorizer_portal.portal_pod_signature', values) @http.route('/my/pod//sign', type='json', auth='user', methods=['POST']) def pod_save_signature(self, order_id, client_name, signature_data, signature_date=None, **kw): """Save POD signature via AJAX""" partner = request.env.user.partner_id user = request.env.user try: order = request.env['sale.order'].sudo().browse(order_id) if not order.exists(): return {'success': False, 'error': 'Order not found'} # Check access - same logic as pod_signature_page has_access = False if partner.is_technician_portal and user.id in order.x_fc_delivery_technician_ids.ids: has_access = True elif partner.is_technician_portal: task_count = request.env['fusion.technician.task'].sudo().search_count([ ('sale_order_id', '=', order.id), ('technician_id', '=', user.id), ]) if task_count: has_access = True if not has_access and not user.share: has_access = True if not has_access: if partner.is_sales_rep_portal and order.user_id.id == user.id: has_access = True elif order.user_id.id == user.id: has_access = True if not has_access: return {'success': False, 'error': 'Access denied'} if not client_name or not client_name.strip(): return {'success': False, 'error': 'Client name is required'} if not signature_data: return {'success': False, 'error': 'Signature is required'} # Process signature data (remove data URL prefix if present) if ',' in signature_data: signature_data = signature_data.split(',')[1] # Parse signature date if provided from datetime import datetime sig_date = None if signature_date: try: sig_date = datetime.strptime(signature_date, '%Y-%m-%d').date() except ValueError: pass # Leave as None if invalid # Determine if this is an ADP sale is_adp = order.x_fc_sale_type in ('adp', 'adp_odsp') # Check if there's already an existing POD signature (for chatter logic) had_existing_signature = bool(order.x_fc_pod_signature) # Save signature data order.write({ 'x_fc_pod_signature': signature_data, 'x_fc_pod_client_name': client_name.strip(), 'x_fc_pod_signature_date': sig_date, 'x_fc_pod_signed_by_user_id': user.id, 'x_fc_pod_signed_datetime': datetime.now(), }) # Generate the signed POD PDF # For ADP: save to x_fc_proof_of_delivery field # For non-ADP: don't save to field, just generate for chatter pdf_content = self._generate_signed_pod_pdf(order, save_to_field=is_adp) # Post to chatter from markupsafe import Markup date_str = sig_date.strftime('%B %d, %Y') if sig_date else 'Not specified' pod_type = "ADP Proof of Delivery" if is_adp else "Proof of Delivery" if had_existing_signature: # Update - post note about the update with new attachment chatter_body = Markup(f'''

{pod_type} Updated

  • Client Name: {client_name.strip()}
  • Signature Date: {date_str}
  • Updated By: {user.name}

The POD document has been replaced with a new signed version.

''') # For non-ADP updates, still attach the new PDF if not is_adp and pdf_content: attachment = request.env['ir.attachment'].sudo().create({ 'name': f'POD_{order.name.replace("/", "_")}.pdf', 'type': 'binary', 'datas': base64.b64encode(pdf_content), 'res_model': 'sale.order', 'res_id': order.id, 'mimetype': 'application/pdf', }) order.message_post( body=chatter_body, message_type='notification', subtype_xmlid='mail.mt_note', attachment_ids=[attachment.id], ) else: order.message_post(body=chatter_body, message_type='notification', subtype_xmlid='mail.mt_note') else: # New POD - post with attachment chatter_body = Markup(f'''

{pod_type} Signed

  • Client Name: {client_name.strip()}
  • Signature Date: {date_str}
  • Collected By: {user.name}
''') # Create attachment for the chatter (always for new POD) if pdf_content: attachment = request.env['ir.attachment'].sudo().create({ 'name': f'POD_{order.name.replace("/", "_")}.pdf', 'type': 'binary', 'datas': base64.b64encode(pdf_content), 'res_model': 'sale.order', 'res_id': order.id, 'mimetype': 'application/pdf', }) order.message_post( body=chatter_body, message_type='notification', subtype_xmlid='mail.mt_note', attachment_ids=[attachment.id], ) else: order.message_post(body=chatter_body, message_type='notification', subtype_xmlid='mail.mt_note') return { 'success': True, 'message': 'Signature saved successfully', 'redirect_url': f'/my/technician/delivery/{order_id}' if partner.is_technician_portal else f'/my/sales/case/{order_id}', } except Exception as e: _logger.error(f"Error saving POD signature: {e}") return {'success': False, 'error': str(e)} def _generate_signed_pod_pdf(self, order, save_to_field=True): """Generate a signed POD PDF with the signature embedded. Args: order: The sale.order record save_to_field: If True, save to x_fc_proof_of_delivery field (for ADP orders) Returns: bytes: The PDF content, or None if generation failed """ try: # Determine which report to use based on sale type is_adp = order.x_fc_sale_type in ('adp', 'adp_odsp') if is_adp: report = request.env.ref('fusion_claims.action_report_proof_of_delivery') else: report = request.env.ref('fusion_claims.action_report_proof_of_delivery_standard') # Render the POD report (signature is now embedded in the template) pdf_content, _ = report.sudo()._render_qweb_pdf( report.id, [order.id] ) # For ADP orders, save to the x_fc_proof_of_delivery field if save_to_field and is_adp: order.write({ 'x_fc_proof_of_delivery': base64.b64encode(pdf_content), 'x_fc_proof_of_delivery_filename': f'POD_{order.name.replace("/", "_")}.pdf', }) _logger.info(f"Generated signed POD PDF for order {order.name} (ADP: {is_adp})") return pdf_content except Exception as e: _logger.error(f"Error generating POD PDF for {order.name}: {e}") return None # ========================================================================= # ACCESSIBILITY ASSESSMENT ROUTES # ========================================================================= @http.route('/my/accessibility', type='http', auth='user', website=True) def accessibility_assessment_selector(self, **kw): """Show the accessibility assessment type selector""" partner = request.env.user.partner_id if not partner.is_sales_rep_portal and not partner.is_authorizer: return request.redirect('/my') # Get Google Maps API key ICP = request.env['ir.config_parameter'].sudo() google_maps_api_key = ICP.get_param('fusion_claims.google_maps_api_key', '') values = { 'page_name': 'accessibility_selector', 'google_maps_api_key': google_maps_api_key, } return request.render('fusion_authorizer_portal.portal_accessibility_selector', values) @http.route('/my/accessibility/list', type='http', auth='user', website=True) def accessibility_assessment_list(self, page=1, **kw): """List all accessibility assessments for the current user (sales rep or authorizer)""" partner = request.env.user.partner_id if not partner.is_sales_rep_portal and not partner.is_authorizer: return request.redirect('/my') Assessment = request.env['fusion.accessibility.assessment'].sudo() # Build domain based on role if partner.is_authorizer and partner.is_sales_rep_portal: domain = ['|', ('authorizer_id', '=', partner.id), ('sales_rep_id', '=', request.env.user.id)] elif partner.is_authorizer: domain = [('authorizer_id', '=', partner.id)] else: domain = [('sales_rep_id', '=', request.env.user.id)] # Pagination assessment_count = Assessment.search_count(domain) pager = portal_pager( url='/my/accessibility/list', total=assessment_count, page=page, step=20, ) assessments = Assessment.search( domain, order='assessment_date desc, id desc', limit=20, offset=pager['offset'], ) values = { 'page_name': 'accessibility_list', 'assessments': assessments, 'pager': pager, } return request.render('fusion_authorizer_portal.portal_accessibility_list', values) @http.route('/my/accessibility/stairlift/straight', type='http', auth='user', website=True) def accessibility_stairlift_straight(self, **kw): """Straight stair lift assessment form""" return self._render_accessibility_form('stairlift_straight', 'Straight Stair Lift') @http.route('/my/accessibility/stairlift/curved', type='http', auth='user', website=True) def accessibility_stairlift_curved(self, **kw): """Curved stair lift assessment form""" return self._render_accessibility_form('stairlift_curved', 'Curved Stair Lift') @http.route('/my/accessibility/vpl', type='http', auth='user', website=True) def accessibility_vpl(self, **kw): """Vertical Platform Lift assessment form""" return self._render_accessibility_form('vpl', 'Vertical Platform Lift') @http.route('/my/accessibility/ceiling-lift', type='http', auth='user', website=True) def accessibility_ceiling_lift(self, **kw): """Ceiling Lift assessment form""" return self._render_accessibility_form('ceiling_lift', 'Ceiling Lift') @http.route('/my/accessibility/ramp', type='http', auth='user', website=True) def accessibility_ramp(self, **kw): """Custom Ramp assessment form""" return self._render_accessibility_form('ramp', 'Custom Ramp') @http.route('/my/accessibility/bathroom', type='http', auth='user', website=True) def accessibility_bathroom(self, **kw): """Bathroom Modification assessment form""" return self._render_accessibility_form('bathroom', 'Bathroom Modification') @http.route('/my/accessibility/tub-cutout', type='http', auth='user', website=True) def accessibility_tub_cutout(self, **kw): """Tub Cutout assessment form""" return self._render_accessibility_form('tub_cutout', 'Tub Cutout') def _render_accessibility_form(self, assessment_type, title): """Render an accessibility assessment form""" partner = request.env.user.partner_id if not partner.is_sales_rep_portal and not partner.is_authorizer: return request.redirect('/my') # Get Google Maps API key ICP = request.env['ir.config_parameter'].sudo() google_maps_api_key = ICP.get_param('fusion_claims.google_maps_api_key', '') from datetime import date values = { 'page_name': f'accessibility_{assessment_type}', 'assessment_type': assessment_type, 'title': title, 'google_maps_api_key': google_maps_api_key, 'today': date.today().isoformat(), } # Route to specific template based on type template_map = { 'stairlift_straight': 'fusion_authorizer_portal.portal_accessibility_stairlift_straight', 'stairlift_curved': 'fusion_authorizer_portal.portal_accessibility_stairlift_curved', 'vpl': 'fusion_authorizer_portal.portal_accessibility_vpl', 'ceiling_lift': 'fusion_authorizer_portal.portal_accessibility_ceiling', 'ramp': 'fusion_authorizer_portal.portal_accessibility_ramp', 'bathroom': 'fusion_authorizer_portal.portal_accessibility_bathroom', 'tub_cutout': 'fusion_authorizer_portal.portal_accessibility_tub_cutout', } template = template_map.get(assessment_type, 'fusion_authorizer_portal.portal_accessibility_selector') return request.render(template, values) @http.route('/my/accessibility/save', type='json', auth='user', methods=['POST'], csrf=True) def accessibility_assessment_save(self, **post): """Save an accessibility assessment and optionally create a Sale Order""" partner = request.env.user.partner_id if not partner.is_sales_rep_portal and not partner.is_authorizer: return {'success': False, 'error': 'Access denied'} try: Assessment = request.env['fusion.accessibility.assessment'].sudo() assessment_type = post.get('assessment_type') if not assessment_type: return {'success': False, 'error': 'Assessment type is required'} # Build assessment values vals = { 'assessment_type': assessment_type, 'sales_rep_id': request.env.user.id, 'client_name': post.get('client_name', '').strip(), 'client_address': post.get('client_address', '').strip(), 'client_unit': post.get('client_unit', '').strip(), 'client_address_street': post.get('client_address_street', '').strip(), 'client_address_city': post.get('client_address_city', '').strip(), 'client_address_province': post.get('client_address_province', '').strip(), 'client_address_postal': post.get('client_address_postal', '').strip(), 'client_phone': post.get('client_phone', '').strip(), 'client_email': post.get('client_email', '').strip(), 'notes': post.get('notes', '').strip(), } # Parse assessment date assessment_date = post.get('assessment_date') if assessment_date: from datetime import date as dt_date try: vals['assessment_date'] = dt_date.fromisoformat(assessment_date) except ValueError: vals['assessment_date'] = dt_date.today() # Add type-specific fields if assessment_type == 'stairlift_straight': vals.update(self._parse_stairlift_straight_fields(post)) elif assessment_type == 'stairlift_curved': vals.update(self._parse_stairlift_curved_fields(post)) elif assessment_type == 'vpl': vals.update(self._parse_vpl_fields(post)) elif assessment_type == 'ceiling_lift': vals.update(self._parse_ceiling_lift_fields(post)) elif assessment_type == 'ramp': vals.update(self._parse_ramp_fields(post)) elif assessment_type == 'bathroom': vals.update(self._parse_bathroom_fields(post)) elif assessment_type == 'tub_cutout': vals.update(self._parse_tub_cutout_fields(post)) # Set authorizer if the current user is an authorizer, or from the linked sale order if partner.is_authorizer: vals['authorizer_id'] = partner.id # Create the assessment assessment = Assessment.create(vals) _logger.info(f"Created accessibility assessment {assessment.reference} by {request.env.user.name}") # Handle photo attachments - General photos photos = post.get('photos', []) if photos: self._attach_accessibility_photos(assessment, photos, category='general') # Handle top landing photos (for curved stair lifts) top_landing_photos = post.get('top_landing_photos', []) if top_landing_photos: self._attach_accessibility_photos(assessment, top_landing_photos, category='top_landing') # Handle bottom landing photos (for curved stair lifts) bottom_landing_photos = post.get('bottom_landing_photos', []) if bottom_landing_photos: self._attach_accessibility_photos(assessment, bottom_landing_photos, category='bottom_landing') # Handle video attachment video_data = post.get('assessment_video') video_filename = post.get('assessment_video_filename') if video_data: self._attach_accessibility_video(assessment, video_data, video_filename) # Complete the assessment and create Sale Order if requested create_sale_order = post.get('create_sale_order', True) if create_sale_order: sale_order = assessment.action_complete() return { 'success': True, 'assessment_id': assessment.id, 'assessment_ref': assessment.reference, 'sale_order_id': sale_order.id, 'sale_order_name': sale_order.name, 'message': f'Assessment {assessment.reference} completed. Sale Order {sale_order.name} created.', 'redirect_url': f'/my/sales/case/{sale_order.id}', } else: return { 'success': True, 'assessment_id': assessment.id, 'assessment_ref': assessment.reference, 'message': f'Assessment {assessment.reference} saved as draft.', 'redirect_url': '/my/accessibility/list', } except Exception as e: _logger.error(f"Error saving accessibility assessment: {e}") return {'success': False, 'error': str(e)} def _parse_stairlift_straight_fields(self, post): """Parse straight stair lift specific fields""" return { 'stair_steps': int(post.get('stair_steps', 0) or 0), 'stair_nose_to_nose': float(post.get('stair_nose_to_nose', 0) or 0), 'stair_side': post.get('stair_side') or None, 'stair_style': post.get('stair_style') or None, 'stair_power_swivel_upstairs': post.get('stair_power_swivel_upstairs') == 'true', 'stair_power_folding_footrest': post.get('stair_power_folding_footrest') == 'true', 'stair_manual_length_override': float(post.get('stair_manual_length_override', 0) or 0), } def _parse_stairlift_curved_fields(self, post): """Parse curved stair lift specific fields""" return { 'stair_curved_steps': int(post.get('stair_curved_steps', 0) or 0), 'stair_curves_count': int(post.get('stair_curves_count', 0) or 0), 'stair_top_landing_type': post.get('stair_top_landing_type') or 'none', 'stair_bottom_landing_type': post.get('stair_bottom_landing_type') or 'none', 'top_overrun_custom_length': float(post.get('top_overrun_custom_length', 0) or 0), 'bottom_overrun_custom_length': float(post.get('bottom_overrun_custom_length', 0) or 0), 'stair_power_swivel_upstairs': post.get('stair_power_swivel_upstairs') == 'true', 'stair_power_swivel_downstairs': post.get('stair_power_swivel_downstairs') == 'true', 'stair_auto_folding_footrest': post.get('stair_auto_folding_footrest') == 'true', 'stair_auto_folding_hinge': post.get('stair_auto_folding_hinge') == 'true', 'stair_auto_folding_seat': post.get('stair_auto_folding_seat') == 'true', 'stair_custom_color': post.get('stair_custom_color') == 'true', 'stair_additional_charging': post.get('stair_additional_charging') == 'true', 'stair_charging_with_remote': post.get('stair_charging_with_remote') == 'true', 'stair_curved_manual_override': float(post.get('stair_curved_manual_override', 0) or 0), } def _parse_vpl_fields(self, post): """Parse VPL specific fields""" return { 'vpl_room_width': float(post.get('vpl_room_width', 0) or 0), 'vpl_room_depth': float(post.get('vpl_room_depth', 0) or 0), 'vpl_rise_height': float(post.get('vpl_rise_height', 0) or 0), 'vpl_has_existing_platform': post.get('vpl_has_existing_platform') == 'true', 'vpl_concrete_depth': float(post.get('vpl_concrete_depth', 0) or 0), 'vpl_model_type': post.get('vpl_model_type') or None, 'vpl_has_nearby_plug': post.get('vpl_has_nearby_plug') == 'true', 'vpl_needs_plug_install': post.get('vpl_needs_plug_install') == 'true', 'vpl_needs_certification': post.get('vpl_needs_certification') == 'true', 'vpl_certification_notes': post.get('vpl_certification_notes', '').strip(), } def _parse_ceiling_lift_fields(self, post): """Parse ceiling lift specific fields""" return { 'ceiling_track_length': float(post.get('ceiling_track_length', 0) or 0), 'ceiling_movement_type': post.get('ceiling_movement_type') or None, 'ceiling_charging_throughout': post.get('ceiling_charging_throughout') == 'true', 'ceiling_carry_bar': post.get('ceiling_carry_bar') == 'true', 'ceiling_additional_slings': int(post.get('ceiling_additional_slings', 0) or 0), } def _parse_ramp_fields(self, post): """Parse ramp specific fields""" return { 'ramp_height': float(post.get('ramp_height', 0) or 0), 'ramp_ground_incline': float(post.get('ramp_ground_incline', 0) or 0), 'ramp_at_door': post.get('ramp_at_door') == 'true', 'ramp_handrail_height': float(post.get('ramp_handrail_height', 32) or 32), 'ramp_manual_override': float(post.get('ramp_manual_override', 0) or 0), } def _parse_bathroom_fields(self, post): """Parse bathroom modification specific fields""" return { 'bathroom_description': post.get('bathroom_description', '').strip(), } def _parse_tub_cutout_fields(self, post): """Parse tub cutout specific fields""" return { 'tub_internal_height': float(post.get('tub_internal_height', 0) or 0), 'tub_external_height': float(post.get('tub_external_height', 0) or 0), 'tub_additional_supplies': post.get('tub_additional_supplies', '').strip(), } def _attach_accessibility_photos(self, assessment, photos, category='general'): """Attach photos to the accessibility assessment Args: assessment: The assessment record photos: List of base64 encoded photo data category: Photo category (general, top_landing, bottom_landing) """ Attachment = request.env['ir.attachment'].sudo() # Category prefix for file naming category_prefixes = { 'general': 'Photo', 'top_landing': 'TopLanding', 'bottom_landing': 'BottomLanding', } prefix = category_prefixes.get(category, 'Photo') for i, photo_data in enumerate(photos): if not photo_data: continue # Handle base64 data URL format if ',' in photo_data: photo_data = photo_data.split(',')[1] try: attachment = Attachment.create({ 'name': f'{prefix}_{i+1}_{assessment.reference}.jpg', 'type': 'binary', 'datas': photo_data, 'res_model': 'fusion.accessibility.assessment', 'res_id': assessment.id, 'mimetype': 'image/jpeg', 'description': f'{category.replace("_", " ").title()} photo for {assessment.reference}', }) _logger.info(f"Attached {category} photo {i+1} to assessment {assessment.reference}") except Exception as e: _logger.warning(f"Failed to attach {category} photo {i+1}: {e}") def _attach_accessibility_video(self, assessment, video_data, video_filename=None): """Attach a video to the accessibility assessment Args: assessment: The assessment record video_data: Base64 encoded video data video_filename: Original filename (optional) """ if not video_data: return Attachment = request.env['ir.attachment'].sudo() # Handle base64 data URL format mimetype = 'video/mp4' if isinstance(video_data, str) and ',' in video_data: # Extract mimetype from data URL header = video_data.split(',')[0] if 'video/' in header: mimetype = header.split(':')[1].split(';')[0] video_data = video_data.split(',')[1] # Determine file extension from mimetype extension_map = { 'video/mp4': '.mp4', 'video/webm': '.webm', 'video/quicktime': '.mov', 'video/x-msvideo': '.avi', } extension = extension_map.get(mimetype, '.mp4') filename = video_filename or f'Video_{assessment.reference}{extension}' try: attachment = Attachment.create({ 'name': filename, 'type': 'binary', 'datas': video_data, 'res_model': 'fusion.accessibility.assessment', 'res_id': assessment.id, 'mimetype': mimetype, 'description': f'Assessment video for {assessment.reference}', }) _logger.info(f"Attached video to assessment {assessment.reference}") except Exception as e: _logger.warning(f"Failed to attach video to assessment {assessment.reference}: {e}")