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

feat(fusion_tasks): copy from Entech Plating, remove sync system

Browse Source 
Forked fusion_tasks module into fusion-plating repo for EN Tech
delivery dispatch. Removed:
- models/task_sync.py (748 lines — cross-instance sync)
- views/task_sync_views.xml
- __pycache__ directories

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
gsinghpal
2026-04-12 20:22:09 -04:00
parent 234a5b2b9f
commit bc72486808
22 changed files with 6703 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
fusion-plating/fusion_tasks/models/__init__.py Normal file
View File

@@ -0,0 +1,13 @@
# -*- coding: utf-8 -*-
# Copyright 2024-2026 Nexa Systems Inc.
# License OPL-1 (Odoo Proprietary License v1.0)
from . import email_builder_mixin
from . import res_partner
from . import res_company
from . import res_users
from . import res_config_settings
from . import technician_task
from . import task_sync
from . import technician_location
from . import push_subscription

241
fusion-plating/fusion_tasks/models/email_builder_mixin.py Normal file
View File

@@ -0,0 +1,241 @@
# -*- coding: utf-8 -*-
# Fusion Claims - Professional Email Builder Mixin
# Provides consistent, dark/light mode safe email templates across all modules.
from odoo import models
class FusionEmailBuilderMixin(models.AbstractModel):
_name = 'fusion.email.builder.mixin'
_description = 'Fusion Email Builder Mixin'
# ------------------------------------------------------------------
# Color constants
# ------------------------------------------------------------------
_EMAIL_COLORS = {
'info': '#2B6CB0',
'success': '#38a169',
'attention': '#d69e2e',
'urgent': '#c53030',
}
# ------------------------------------------------------------------
# Public API
# ------------------------------------------------------------------
def _email_build(
self,
title,
summary,
sections=None,
note=None,
note_color=None,
email_type='info',
attachments_note=None,
button_url=None,
button_text='View Case Details',
sender_name=None,
extra_html='',
):
"""Build a complete professional email HTML string.
Args:
title: Email heading (e.g. "Application Approved")
summary: One-sentence summary HTML (may contain <strong> tags)
sections: list of (heading, rows) where rows is list of (label, value)
e.g. [('Case Details', [('Client', 'John'), ('Case', 'S30073')])]
note: Optional note/next-steps text (plain or HTML)
note_color: Override left-border color for note (default uses email_type)
email_type: 'info' | 'success' | 'attention' | 'urgent'
attachments_note: Optional string listing attached files
button_url: Optional CTA button URL
button_text: CTA button label
sender_name: Name for sign-off (defaults to current user)
extra_html: Any additional HTML to insert before sign-off
"""
accent = self._EMAIL_COLORS.get(email_type, self._EMAIL_COLORS['info'])
company = self._get_company_info()
parts = []
# -- Wrapper open + accent bar (no forced bg/color so it adapts to dark/light)
parts.append(
f'<div style="font-family:-apple-system,BlinkMacSystemFont,\'Segoe UI\',Roboto,Arial,sans-serif;'
f'max-width:600px;margin:0 auto;">'
f'<div style="height:4px;background-color:{accent};"></div>'
f'<div style="padding:32px 28px;">'
)
# -- Company name (accent color works in both themes)
parts.append(
f'<p style="color:{accent};font-size:13px;font-weight:600;letter-spacing:0.5px;'
f'text-transform:uppercase;margin:0 0 24px 0;">{company["name"]}</p>'
)
# -- Title (inherits text color from container)
parts.append(
f'<h2 style="font-size:22px;font-weight:700;'
f'margin:0 0 6px 0;line-height:1.3;">{title}</h2>'
)
# -- Summary (muted via opacity)
parts.append(
f'<p style="opacity:0.65;font-size:15px;line-height:1.5;'
f'margin:0 0 24px 0;">{summary}</p>'
)
# -- Sections (details tables)
if sections:
for heading, rows in sections:
parts.append(self._email_section(heading, rows))
# -- Note / Next Steps
if note:
nc = note_color or accent
parts.append(self._email_note(note, nc))
# -- Extra HTML
if extra_html:
parts.append(extra_html)
# -- Attachment note
if attachments_note:
parts.append(self._email_attachment_note(attachments_note))
# -- CTA Button
if button_url:
parts.append(self._email_button(button_url, button_text, accent))
# -- Sign-off
signer = sender_name or (self.env.user.name if self.env.user else '')
parts.append(
f'<p style="font-size:14px;line-height:1.6;margin:24px 0 0 0;">'
f'Best regards,<br/>'
f'<strong>{signer}</strong><br/>'
f'<span style="opacity:0.6;">{company["name"]}</span></p>'
)
# -- Close content card
parts.append('</div>')
# -- Footer
footer_parts = [company['name']]
if company['phone']:
footer_parts.append(company['phone'])
if company['email']:
footer_parts.append(company['email'])
footer_text = ' &middot; '.join(footer_parts)
parts.append(
f'<div style="padding:16px 28px;text-align:center;">'
f'<p style="opacity:0.5;font-size:11px;line-height:1.5;margin:0;">'
f'{footer_text}<br/>'
f'This is an automated notification from the ADP Claims Management System.</p>'
f'</div>'
)
# -- Close wrapper
parts.append('</div>')
return ''.join(parts)
# ------------------------------------------------------------------
# Building blocks
# ------------------------------------------------------------------
def _email_section(self, heading, rows):
"""Build a labeled details table section.
Args:
heading: Section title (e.g. "Case Details")
rows: list of (label, value) tuples. Value can be plain text or HTML.
"""
if not rows:
return ''
html = (
'<table style="width:100%;border-collapse:collapse;margin:0 0 24px 0;">'
f'<tr><td colspan="2" style="padding:10px 14px;font-size:12px;font-weight:600;'
f'opacity:0.55;text-transform:uppercase;letter-spacing:0.5px;'
f'border-bottom:2px solid rgba(128,128,128,0.25);">{heading}</td></tr>'
)
for label, value in rows:
if value is None or value == '' or value is False:
continue
html += (
f'<tr>'
f'<td style="padding:10px 14px;opacity:0.6;font-size:14px;'
f'border-bottom:1px solid rgba(128,128,128,0.15);width:35%;">{label}</td>'
f'<td style="padding:10px 14px;font-size:14px;'
f'border-bottom:1px solid rgba(128,128,128,0.15);">{value}</td>'
f'</tr>'
)
html += '</table>'
return html
def _email_note(self, text, color='#2B6CB0'):
"""Build a left-border accent note block."""
return (
f'<div style="border-left:3px solid {color};padding:12px 16px;'
f'margin:0 0 24px 0;">'
f'<p style="margin:0;font-size:14px;line-height:1.5;">{text}</p>'
f'</div>'
)
def _email_button(self, url, text='View Case Details', color='#2B6CB0'):
"""Build a centered CTA button."""
return (
f'<p style="text-align:center;margin:28px 0;">'
f'<a href="{url}" style="display:inline-block;background:{color};color:#ffffff;'
f'padding:12px 28px;text-decoration:none;border-radius:6px;'
f'font-size:14px;font-weight:600;">{text}</a></p>'
)
def _email_attachment_note(self, description):
"""Build a dashed-border attachment callout.
Args:
description: e.g. "ADP Application (PDF), XML Data File"
"""
return (
f'<div style="padding:10px 14px;border:1px dashed rgba(128,128,128,0.35);border-radius:6px;'
f'margin:0 0 24px 0;">'
f'<p style="margin:0;font-size:13px;opacity:0.65;">'
f'<strong style="opacity:1;">Attached:</strong> {description}</p>'
f'</div>'
)
def _email_status_badge(self, label, color='#2B6CB0'):
"""Return an inline status badge/pill HTML snippet."""
bg_map = {
'#38a169': 'rgba(56,161,105,0.12)',
'#2B6CB0': 'rgba(43,108,176,0.12)',
'#d69e2e': 'rgba(214,158,46,0.12)',
'#c53030': 'rgba(197,48,48,0.12)',
}
bg = bg_map.get(color, 'rgba(43,108,176,0.12)')
return (
f'<span style="display:inline-block;background:{bg};color:{color};'
f'padding:2px 10px;border-radius:12px;font-size:12px;font-weight:600;">'
f'{label}</span>'
)
# ------------------------------------------------------------------
# Helpers
# ------------------------------------------------------------------
def _get_company_info(self):
"""Return company name, phone, email for email templates."""
company = getattr(self, 'company_id', None) or self.env.company
return {
'name': company.name or 'Our Company',
'phone': company.phone or '',
'email': company.email or '',
}
def _email_is_enabled(self):
"""Check if email notifications are enabled in settings."""
ICP = self.env['ir.config_parameter'].sudo()
val = ICP.get_param('fusion_claims.enable_email_notifications', 'True')
return val.lower() in ('true', '1', 'yes')

73
fusion-plating/fusion_tasks/models/push_subscription.py Normal file
View File

@@ -0,0 +1,73 @@
# -*- coding: utf-8 -*-
# Copyright 2024-2026 Nexa Systems Inc.
# License OPL-1 (Odoo Proprietary License v1.0)
"""
Web Push Subscription model for storing browser push notification subscriptions.
"""
from odoo import models, fields, api
import logging
_logger = logging.getLogger(__name__)
class FusionPushSubscription(models.Model):
_name = 'fusion.push.subscription'
_description = 'Web Push Subscription'
_order = 'create_date desc'
user_id = fields.Many2one(
'res.users',
string='User',
required=True,
ondelete='cascade',
index=True,
)
endpoint = fields.Text(
string='Endpoint URL',
required=True,
)
p256dh_key = fields.Text(
string='P256DH Key',
required=True,
)
auth_key = fields.Text(
string='Auth Key',
required=True,
)
browser_info = fields.Char(
string='Browser Info',
help='User agent or browser identification',
)
active = fields.Boolean(
default=True,
)
_constraints = [
models.Constraint(
'unique(endpoint)',
'This push subscription endpoint already exists.',
),
]
@api.model
def register_subscription(self, user_id, endpoint, p256dh_key, auth_key, browser_info=None):
"""Register or update a push subscription."""
existing = self.sudo().search([('endpoint', '=', endpoint)], limit=1)
if existing:
existing.write({
'user_id': user_id,
'p256dh_key': p256dh_key,
'auth_key': auth_key,
'browser_info': browser_info or existing.browser_info,
'active': True,
})
return existing
return self.sudo().create({
'user_id': user_id,
'endpoint': endpoint,
'p256dh_key': p256dh_key,
'auth_key': auth_key,
'browser_info': browser_info,
})

14
fusion-plating/fusion_tasks/models/res_company.py Normal file
View File

@@ -0,0 +1,14 @@
# -*- coding: utf-8 -*-
# Copyright 2024-2026 Nexa Systems Inc.
# License OPL-1 (Odoo Proprietary License v1.0)
from odoo import models, fields
class ResCompany(models.Model):
_inherit = 'res.company'
x_fc_google_review_url = fields.Char(
string='Google Review URL',
help='Google Business Profile review link sent to clients after service completion',
)

73
fusion-plating/fusion_tasks/models/res_config_settings.py Normal file
View File

@@ -0,0 +1,73 @@
# -*- coding: utf-8 -*-
# Copyright 2024-2026 Nexa Systems Inc.
# License OPL-1 (Odoo Proprietary License v1.0)
from odoo import models, fields
class ResConfigSettings(models.TransientModel):
_inherit = 'res.config.settings'
# Google Maps API Settings
fc_google_maps_api_key = fields.Char(
string='Google Maps API Key',
config_parameter='fusion_claims.google_maps_api_key',
help='API key for Google Maps Places autocomplete in address fields',
)
fc_google_review_url = fields.Char(
related='company_id.x_fc_google_review_url',
readonly=False,
string='Google Review URL',
)
# Technician Management
fc_store_open_hour = fields.Float(
string='Store Open Time',
config_parameter='fusion_claims.store_open_hour',
help='Store opening time for technician scheduling (e.g. 9.0 = 9:00 AM)',
)
fc_store_close_hour = fields.Float(
string='Store Close Time',
config_parameter='fusion_claims.store_close_hour',
help='Store closing time for technician scheduling (e.g. 18.0 = 6:00 PM)',
)
fc_google_distance_matrix_enabled = fields.Boolean(
string='Enable Distance Matrix',
config_parameter='fusion_claims.google_distance_matrix_enabled',
help='Enable Google Distance Matrix API for travel time calculations between technician tasks',
)
fc_technician_start_address = fields.Char(
string='Technician Start Address',
config_parameter='fusion_claims.technician_start_address',
help='Default start location for technician travel calculations (e.g. warehouse/office address)',
)
fc_location_retention_days = fields.Char(
string='Location History Retention (Days)',
config_parameter='fusion_claims.location_retention_days',
help='How many days to keep technician location history. '
'Leave empty = 30 days (1 month). '
'0 = delete at end of each day. '
'1+ = keep for that many days.',
)
# Web Push Notifications
fc_push_enabled = fields.Boolean(
string='Enable Push Notifications',
config_parameter='fusion_claims.push_enabled',
help='Enable web push notifications for technician tasks',
)
fc_vapid_public_key = fields.Char(
string='VAPID Public Key',
config_parameter='fusion_claims.vapid_public_key',
help='Public key for Web Push VAPID authentication (auto-generated)',
)
fc_vapid_private_key = fields.Char(
string='VAPID Private Key',
config_parameter='fusion_claims.vapid_private_key',
help='Private key for Web Push VAPID authentication (auto-generated)',
)
fc_push_advance_minutes = fields.Integer(
string='Notification Advance (min)',
config_parameter='fusion_claims.push_advance_minutes',
help='Send push notifications this many minutes before a scheduled task',
)

79
fusion-plating/fusion_tasks/models/res_partner.py Normal file
View File

@@ -0,0 +1,79 @@
# -*- coding: utf-8 -*-
# Copyright 2024-2026 Nexa Systems Inc.
# License OPL-1 (Odoo Proprietary License v1.0)
import logging
import requests
from odoo import models, fields, api
_logger = logging.getLogger(__name__)
class ResPartner(models.Model):
_inherit = 'res.partner'
x_fc_start_address = fields.Char(
string='Start Location',
help='Technician daily start location (home, warehouse, etc.). '
'Used as origin for first travel time calculation. '
'If empty, the company default HQ address is used.',
)
x_fc_start_address_lat = fields.Float(
string='Start Latitude', digits=(10, 7),
)
x_fc_start_address_lng = fields.Float(
string='Start Longitude', digits=(10, 7),
)
def _geocode_start_address(self, address):
if not address or not address.strip():
return 0.0, 0.0
api_key = self.env['ir.config_parameter'].sudo().get_param(
'fusion_claims.google_maps_api_key', '')
if not api_key:
return 0.0, 0.0
try:
resp = requests.get(
'https://maps.googleapis.com/maps/api/geocode/json',
params={'address': address.strip(), 'key': api_key, 'region': 'ca'},
timeout=10,
)
data = resp.json()
if data.get('status') == 'OK' and data.get('results'):
loc = data['results'][0]['geometry']['location']
return loc['lat'], loc['lng']
except Exception as e:
_logger.warning("Start address geocoding failed for '%s': %s", address, e)
return 0.0, 0.0
@api.model_create_multi
def create(self, vals_list):
records = super().create(vals_list)
for rec, vals in zip(records, vals_list):
addr = vals.get('x_fc_start_address')
if addr:
lat, lng = rec._geocode_start_address(addr)
if lat and lng:
rec.write({
'x_fc_start_address_lat': lat,
'x_fc_start_address_lng': lng,
})
return records
def write(self, vals):
res = super().write(vals)
if 'x_fc_start_address' in vals:
addr = vals['x_fc_start_address']
if addr and addr.strip():
lat, lng = self._geocode_start_address(addr)
if lat and lng:
super().write({
'x_fc_start_address_lat': lat,
'x_fc_start_address_lng': lng,
})
else:
super().write({
'x_fc_start_address_lat': 0.0,
'x_fc_start_address_lng': 0.0,
})
return res

26
fusion-plating/fusion_tasks/models/res_users.py Normal file
View File

@@ -0,0 +1,26 @@
# -*- coding: utf-8 -*-
# Copyright 2024-2026 Nexa Systems Inc.
# License OPL-1 (Odoo Proprietary License v1.0)
from odoo import models, fields
class ResUsers(models.Model):
_inherit = 'res.users'
x_fc_is_field_staff = fields.Boolean(
string='Field Staff',
default=False,
help='Check this to show the user in the Technician/Field Staff dropdown when scheduling tasks.',
)
x_fc_start_address = fields.Char(
related='partner_id.x_fc_start_address',
readonly=False,
string='Start Location',
)
x_fc_tech_sync_id = fields.Char(
string='Tech Sync ID',
help='Shared identifier for this technician across Odoo instances. '
'Must be the same value on all instances for the same person.',
copy=False,
)

131
fusion-plating/fusion_tasks/models/technician_location.py Normal file
View File

@@ -0,0 +1,131 @@
# -*- coding: utf-8 -*-
# Copyright 2024-2026 Nexa Systems Inc.
# License OPL-1 (Odoo Proprietary License v1.0)
"""
Fusion Technician Location
GPS location logging for field technicians.
"""
from odoo import models, fields, api, _
import logging
_logger = logging.getLogger(__name__)
class FusionTechnicianLocation(models.Model):
_name = 'fusion.technician.location'
_description = 'Technician Location Log'
_order = 'logged_at desc'
user_id = fields.Many2one(
'res.users',
string='Technician',
required=True,
index=True,
ondelete='cascade',
)
latitude = fields.Float(
string='Latitude',
digits=(10, 7),
required=True,
)
longitude = fields.Float(
string='Longitude',
digits=(10, 7),
required=True,
)
accuracy = fields.Float(
string='Accuracy (m)',
help='GPS accuracy in meters',
)
logged_at = fields.Datetime(
string='Logged At',
default=fields.Datetime.now,
required=True,
index=True,
)
source = fields.Selection([
('portal', 'Portal'),
('app', 'Mobile App'),
('sync', 'Synced'),
], string='Source', default='portal')
sync_instance = fields.Char(
'Sync Instance', index=True,
help='Source instance ID if synced (e.g. westin, mobility)',
)
@api.model
def log_location(self, latitude, longitude, accuracy=None):
"""Log the current user's location. Called from portal JS."""
return self.sudo().create({
'user_id': self.env.user.id,
'latitude': latitude,
'longitude': longitude,
'accuracy': accuracy or 0,
'source': 'portal',
})
@api.model
def get_latest_locations(self):
"""Get the most recent location for each technician (for map view).
Includes both local GPS pings and synced locations from remote
instances, so the map shows all shared technicians regardless of
which Odoo instance they are clocked into.
"""
self.env.cr.execute("""
SELECT DISTINCT ON (user_id)
user_id, latitude, longitude, accuracy, logged_at,
COALESCE(sync_instance, '') AS sync_instance
FROM fusion_technician_location
WHERE logged_at > NOW() - INTERVAL '24 hours'
ORDER BY user_id, logged_at DESC
""")
rows = self.env.cr.dictfetchall()
local_id = self.env['ir.config_parameter'].sudo().get_param(
'fusion_claims.sync_instance_id', '')
result = []
for row in rows:
user = self.env['res.users'].sudo().browse(row['user_id'])
src = row.get('sync_instance') or local_id
result.append({
'user_id': row['user_id'],
'name': user.name,
'latitude': row['latitude'],
'longitude': row['longitude'],
'accuracy': row['accuracy'],
'logged_at': str(row['logged_at']),
'sync_instance': src,
})
return result
@api.model
def _cron_cleanup_old_locations(self):
"""Remove location logs based on configurable retention setting.
Setting (fusion_claims.location_retention_days):
- Empty / not set => keep 30 days (default)
- "0" => delete at end of day (keep today only)
- "1" .. "N" => keep for N days
"""
ICP = self.env['ir.config_parameter'].sudo()
raw = (ICP.get_param('fusion_claims.location_retention_days') or '').strip()
if raw == '':
retention_days = 30 # default: 1 month
else:
try:
retention_days = max(int(raw), 0)
except (ValueError, TypeError):
retention_days = 30
cutoff = fields.Datetime.subtract(fields.Datetime.now(), days=retention_days)
old_records = self.search([('logged_at', '<', cutoff)])
count = len(old_records)
if count:
old_records.unlink()
_logger.info(
"Cleaned up %d technician location records (retention=%d days)",
count, retention_days,
)

3028
fusion-plating/fusion_tasks/models/technician_task.py Normal file
View File

File diff suppressed because it is too large Load Diff