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

changes

Browse Source
This commit is contained in:
gsinghpal
2026-02-23 00:32:20 -05:00
parent d6bac8e623
commit e8e554de95
549 changed files with 1330 additions and 124935 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
fusion_claims/models/__init__.py
View File

@@ -29,6 +29,6 @@ from . import dashboard
from . import res_partner
from . import res_users
from . import technician_task
from . import task_sync
from . import technician_location
from . import push_subscription
from . import pdf_template_inherit
from . import push_subscription

162
fusion_claims/models/pdf_template_inherit.py
View File

@@ -1,162 +0,0 @@
# -*- coding: utf-8 -*-
import base64
import logging
from io import BytesIO
from odoo import models, fields, api
_logger = logging.getLogger(__name__)
class FusionSaSignatureTemplate(models.Model):
_name = 'fusion.sa.signature.template'
_description = 'SA Mobility Signature Position Template'
_order = 'name'
name = fields.Char(string='Template Name', required=True)
active = fields.Boolean(default=True)
notes = fields.Text(string='Notes')
sa_default_sig_page = fields.Integer(string='Default Signature Page', default=2)
# Absolute PDF point coordinates. Y = distance from top of page.
sa_sig_name_x = fields.Integer(string='Name X', default=105)
sa_sig_name_y = fields.Integer(string='Name Y from top', default=97)
sa_sig_date_x = fields.Integer(string='Date X', default=430)
sa_sig_date_y = fields.Integer(string='Date Y from top', default=97)
sa_sig_x = fields.Integer(string='Signature X', default=72)
sa_sig_y = fields.Integer(string='Signature Y from top', default=68)
sa_sig_w = fields.Integer(string='Signature Width', default=190)
sa_sig_h = fields.Integer(string='Signature Height', default=25)
preview_pdf = fields.Binary(
string='Sample PDF',
help='Upload a sample SA Mobility approval form to preview signature placement.',
attachment=True,
)
preview_pdf_filename = fields.Char(string='PDF Filename')
preview_pdf_page = fields.Integer(
string='Preview Page', default=0,
help='Page to render preview for. 0 = use Default Signature Page.',
)
preview_image = fields.Binary(
string='Preview', readonly=True,
compute='_compute_preview_image',
)
@api.depends(
'preview_pdf', 'preview_pdf_page', 'sa_default_sig_page',
'sa_sig_name_x', 'sa_sig_name_y',
'sa_sig_date_x', 'sa_sig_date_y',
'sa_sig_x', 'sa_sig_y', 'sa_sig_w', 'sa_sig_h',
)
def _compute_preview_image(self):
for rec in self:
if not rec.preview_pdf:
rec.preview_image = False
continue
try:
rec.preview_image = rec._render_preview()
except Exception as e:
_logger.warning("SA template preview render failed: %s", e)
rec.preview_image = False
def _render_preview(self):
"""Render the sample PDF page with colored markers for all signature positions."""
self.ensure_one()
from odoo.tools.pdf import PdfFileReader
pdf_bytes = base64.b64decode(self.preview_pdf)
reader = PdfFileReader(BytesIO(pdf_bytes))
num_pages = reader.getNumPages()
page_num = self.preview_pdf_page or self.sa_default_sig_page or 2
page_idx = page_num - 1
if page_idx < 0 or page_idx >= num_pages:
return False
try:
from pdf2image import convert_from_bytes
except ImportError:
_logger.warning("pdf2image not installed")
return False
images = convert_from_bytes(
pdf_bytes, first_page=page_idx + 1, last_page=page_idx + 1, dpi=150,
)
if not images:
return False
from PIL import ImageDraw, ImageFont
img = images[0]
draw = ImageDraw.Draw(img)
page = reader.getPage(page_idx)
page_w_pts = float(page.mediaBox.getWidth())
page_h_pts = float(page.mediaBox.getHeight())
img_w, img_h = img.size
sx = img_w / page_w_pts
sy = img_h / page_h_pts
try:
font_b = ImageFont.truetype("/usr/share/fonts/truetype/dejavu/DejaVuSans-Bold.ttf", 14)
font = ImageFont.truetype("/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf", 12)
font_sm = ImageFont.truetype("/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf", 10)
except Exception:
font_b = font = font_sm = ImageFont.load_default()
def _draw_sample_text(label, x_pts, y_top_pts, color, sample_text):
px_x = int(x_pts * sx)
px_y = int(y_top_pts * sy)
draw.text((px_x, px_y - 16), sample_text, fill=color, font=font_b)
draw.text((px_x, px_y + 2), label, fill=color, font=font_sm)
def _draw_box(label, x_pts, y_top_pts, w_pts, h_pts, color):
px_x = int(x_pts * sx)
px_y = int(y_top_pts * sy)
pw = int(w_pts * sx)
ph = int(h_pts * sy)
for off in range(3):
draw.rectangle(
[px_x - off, px_y - off, px_x + pw + off, px_y + ph + off],
outline=color,
)
draw.text((px_x + 4, px_y + 4), label, fill=color, font=font_sm)
_draw_sample_text(
"Name", self.sa_sig_name_x, self.sa_sig_name_y,
'blue', "John Smith",
)
_draw_sample_text(
"Date", self.sa_sig_date_x, self.sa_sig_date_y,
'purple', "2026-02-17",
)
_draw_box(
"Signature", self.sa_sig_x, self.sa_sig_y,
self.sa_sig_w, self.sa_sig_h, 'red',
)
buf = BytesIO()
img.save(buf, format='PNG')
return base64.b64encode(buf.getvalue())
def get_sa_coordinates(self, page_h=792):
"""Convert to ReportLab bottom-origin coordinates.
Template stores Y as distance from TOP of page.
ReportLab uses Y from BOTTOM.
For text (name/date): baseline Y = page_h - y_from_top
For signature image: drawImage Y is bottom-left corner,
so Y = page_h - y_from_top - height
"""
self.ensure_one()
return {
'name_x': self.sa_sig_name_x,
'name_y': page_h - self.sa_sig_name_y,
'date_x': self.sa_sig_date_x,
'date_y': page_h - self.sa_sig_date_y,
'sig_x': self.sa_sig_x,
'sig_y': page_h - self.sa_sig_y - self.sa_sig_h,
'sig_w': self.sa_sig_w,
'sig_h': self.sa_sig_h,
}

6
fusion_claims/models/res_users.py
View File

@@ -18,3 +18,9 @@ class ResUsers(models.Model):
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,
)

167
fusion_claims/models/sale_order.py
View File

@@ -1141,22 +1141,6 @@ class SaleOrder(models.Model):
default=2,
help='Page number in approval form where signature should be placed (1-indexed)',
)
x_fc_sa_signature_position = fields.Selection([
('default', 'Default (SA Mobility Standard)'),
('middle', 'Middle of Page'),
('bottom', 'Bottom of Page'),
('custom', 'Custom Position'),
], string='Signature Position', default='default')
x_fc_sa_signature_offset_x = fields.Integer(
string='Signature X Offset',
default=0,
help='Horizontal offset in points (positive = right)',
)
x_fc_sa_signature_offset_y = fields.Integer(
string='Signature Y Offset',
default=0,
help='Vertical offset in points (positive = up)',
)
# --- Ontario Works document fields ---
x_fc_ow_discretionary_form = fields.Binary(
@@ -1593,48 +1577,12 @@ class SaleOrder(models.Model):
return {'type': 'ir.actions.act_window_close'}
def _get_sa_pdf_template(self):
"""Find the active SA Mobility signature template."""
return self.env['fusion.sa.signature.template'].search([
('active', '=', True),
], limit=1)
def _get_sa_signature_coordinates(self, page_h=792):
"""Get signature overlay coordinates.
Reads the global template from Configuration > PDF Templates,
then applies any per-case offsets stored on this sale order.
"""
self.ensure_one()
tpl = self._get_sa_pdf_template()
ox = self.x_fc_sa_signature_offset_x or 0
oy = self.x_fc_sa_signature_offset_y or 0
if tpl:
coords = tpl.get_sa_coordinates(page_h)
else:
coords = {
'name_x': 105, 'name_y': page_h - 97,
'date_x': 430, 'date_y': page_h - 97,
'sig_x': 72, 'sig_y': page_h - 72 - 25,
'sig_w': 190, 'sig_h': 25,
}
if ox or oy:
coords = {
'name_x': coords['name_x'] + ox,
'name_y': coords['name_y'] + oy,
'date_x': coords['date_x'] + ox,
'date_y': coords['date_y'] + oy,
'sig_x': coords['sig_x'] + ox,
'sig_y': coords['sig_y'] + oy,
'sig_w': coords['sig_w'],
'sig_h': coords['sig_h'],
}
return coords
def _apply_pod_signature_to_approval_form(self):
"""Auto-overlay POD signature onto the ODSP approval form at the configured page/position."""
"""Auto-overlay POD signature onto the ODSP approval form.
Uses the ODSP PDF Template (fusion.pdf.template, category=odsp) for
field positions, and the per-case signature page number.
"""
self.ensure_one()
if not all([
self.x_fc_odsp_division == 'sa_mobility',
@@ -1645,71 +1593,54 @@ class SaleOrder(models.Model):
return
import base64
from io import BytesIO
try:
from reportlab.pdfgen import canvas as rl_canvas
from reportlab.lib.utils import ImageReader
from odoo.tools.pdf import PdfFileReader, PdfFileWriter
except ImportError:
_logger.warning("PDF libraries not available for approval form signing.")
from odoo.addons.fusion_authorizer_portal.utils.pdf_filler import PDFTemplateFiller
tpl = self.env['fusion.pdf.template'].search([
('category', '=', 'odsp'), ('state', '=', 'active'),
], limit=1)
if not tpl:
_logger.warning("No active ODSP PDF template found for signing %s", self.name)
return
sig_page = self.x_fc_sa_signature_page or 2
fields_by_page = {}
for field in tpl.field_ids.filtered(lambda f: f.is_active):
page = sig_page
if page not in fields_by_page:
fields_by_page[page] = []
fields_by_page[page].append({
'field_name': field.name,
'field_key': field.field_key or field.name,
'pos_x': field.pos_x,
'pos_y': field.pos_y,
'width': field.width,
'height': field.height,
'field_type': field.field_type,
'font_size': field.font_size,
'font_name': field.font_name or 'Helvetica',
'text_align': field.text_align or 'left',
})
client_name = self.x_fc_pod_client_name or self.x_fc_sa_client_name or self.partner_id.name or ''
sign_date = self.x_fc_pod_signature_date or self.x_fc_sa_client_signed_date
context_data = {
'sa_client_name': client_name,
'sa_sign_date': sign_date.strftime('%b %d, %Y') if sign_date else '',
}
signatures = {
'sa_signature': base64.b64decode(self.x_fc_pod_signature),
}
pdf_bytes = base64.b64decode(self.x_fc_sa_approval_form)
original = PdfFileReader(BytesIO(pdf_bytes))
output = PdfFileWriter()
num_pages = original.getNumPages()
target_page = (self.x_fc_sa_signature_page or 2) - 1
if target_page < 0 or target_page >= num_pages:
_logger.warning(
"Signature page %s out of range (total %s) for %s",
self.x_fc_sa_signature_page, num_pages, self.name
try:
signed_pdf = PDFTemplateFiller.fill_template(
pdf_bytes, fields_by_page, context_data, signatures,
)
except Exception as e:
_logger.error("Failed to apply signature to approval form for %s: %s", self.name, e)
return
for page_idx in range(num_pages):
page = original.getPage(page_idx)
if page_idx == target_page:
page_w = float(page.mediaBox.getWidth())
page_h = float(page.mediaBox.getHeight())
coords = self._get_sa_signature_coordinates(page_h)
overlay_buf = BytesIO()
c = rl_canvas.Canvas(overlay_buf, pagesize=(page_w, page_h))
client_name = self.x_fc_pod_client_name or self.x_fc_sa_client_name or self.partner_id.name or ''
if client_name:
c.setFont('Helvetica', 11)
c.drawString(coords['name_x'], coords['name_y'], client_name)
sign_date = self.x_fc_pod_signature_date or self.x_fc_sa_client_signed_date
if sign_date:
date_str = sign_date.strftime('%b %d, %Y')
c.setFont('Helvetica', 11)
c.drawString(coords['date_x'], coords['date_y'], date_str)
if self.x_fc_pod_signature:
sig_data = base64.b64decode(self.x_fc_pod_signature)
sig_image = ImageReader(BytesIO(sig_data))
c.drawImage(
sig_image,
coords['sig_x'], coords['sig_y'],
width=coords['sig_w'], height=coords['sig_h'],
preserveAspectRatio=True, mask='auto',
)
c.save()
overlay_buf.seek(0)
overlay_pdf = PdfFileReader(overlay_buf)
page.mergePage(overlay_pdf.getPage(0))
output.addPage(page)
result_buf = BytesIO()
output.write(result_buf)
signed_pdf = result_buf.getvalue()
filename = f'SA_Approval_Signed_{self.name}.pdf'
self.with_context(skip_pod_signature_hook=True).write({
'x_fc_sa_signed_form': base64.b64encode(signed_pdf),
@@ -1725,7 +1656,7 @@ class SaleOrder(models.Model):
'mimetype': 'application/pdf',
})
self.message_post(
body="POD signature applied to ODSP approval form (page %s)." % self.x_fc_sa_signature_page,
body="POD signature applied to ODSP approval form (page %s)." % sig_page,
message_type='comment',
attachment_ids=[att.id],
)
@@ -6546,7 +6477,7 @@ class SaleOrder(models.Model):
'default_model': 'sale.order',
'default_res_ids': self.ids,
'default_template_id': template.id,
'default_email_layout_xmlid': 'mail.mail_notification_layout_with_responsible_signature',
'default_email_layout_xmlid': 'mail.mail_notification_layout',
'default_composition_mode': 'comment',
'mark_so_as_sent': True,
'force_email': True,

409
fusion_claims/models/task_sync.py Normal file
View File

@@ -0,0 +1,409 @@
# -*- coding: utf-8 -*-
# Copyright 2024-2026 Nexa Systems Inc.
# License OPL-1 (Odoo Proprietary License v1.0)
"""
Cross-instance technician task sync.
Enables two Odoo instances (e.g. Westin and Mobility) that share the same
field technicians to see each other's delivery tasks, preventing double-booking.
Remote tasks appear as read-only "shadow" records in the local calendar.
The existing _find_next_available_slot() automatically sees shadow tasks,
so collision detection works without changes to the scheduling algorithm.
Technicians are matched across instances using the x_fc_tech_sync_id field
on res.users. Set the same value (e.g. "gordy") on both instances for the
same person -- no mapping table needed.
"""
from odoo import models, fields, api, _
from odoo.exceptions import UserError
import logging
import requests
from datetime import timedelta
_logger = logging.getLogger(__name__)
SYNC_TASK_FIELDS = [
'x_fc_sync_uuid', 'name', 'technician_id', 'task_type', 'status',
'scheduled_date', 'time_start', 'time_end', 'duration_hours',
'address_street', 'address_street2', 'address_city', 'address_zip',
'address_lat', 'address_lng', 'priority', 'partner_id',
]
class FusionTaskSyncConfig(models.Model):
_name = 'fusion.task.sync.config'
_description = 'Task Sync Remote Instance'
name = fields.Char('Instance Name', required=True,
help='e.g. Westin Healthcare, Mobility Specialties')
instance_id = fields.Char('Instance ID', required=True,
help='Short identifier, e.g. westin or mobility')
url = fields.Char('Odoo URL', required=True,
help='e.g. http://192.168.1.40:8069')
database = fields.Char('Database', required=True)
username = fields.Char('API Username', required=True)
api_key = fields.Char('API Key', required=True)
active = fields.Boolean(default=True)
last_sync = fields.Datetime('Last Successful Sync', readonly=True)
last_sync_error = fields.Text('Last Error', readonly=True)
# ------------------------------------------------------------------
# JSON-RPC helpers
# ------------------------------------------------------------------
def _jsonrpc(self, service, method, args):
"""Execute a JSON-RPC call against the remote Odoo instance."""
self.ensure_one()
url = f"{self.url.rstrip('/')}/jsonrpc"
payload = {
'jsonrpc': '2.0',
'method': 'call',
'id': 1,
'params': {
'service': service,
'method': method,
'args': args,
},
}
try:
resp = requests.post(url, json=payload, timeout=15)
resp.raise_for_status()
result = resp.json()
if result.get('error'):
err = result['error'].get('data', {}).get('message', str(result['error']))
raise UserError(f"Remote error: {err}")
return result.get('result')
except requests.exceptions.ConnectionError:
_logger.warning("Task sync: cannot connect to %s", self.url)
return None
except requests.exceptions.Timeout:
_logger.warning("Task sync: timeout connecting to %s", self.url)
return None
def _authenticate(self):
"""Authenticate with the remote instance and return the uid."""
self.ensure_one()
uid = self._jsonrpc('common', 'authenticate',
[self.database, self.username, self.api_key, {}])
if not uid:
_logger.error("Task sync: authentication failed for %s", self.name)
return uid
def _rpc(self, model, method, args, kwargs=None):
"""Execute a method on the remote instance via execute_kw.
execute_kw(db, uid, password, model, method, [args], {kwargs})
"""
self.ensure_one()
uid = self._authenticate()
if not uid:
return None
call_args = [self.database, uid, self.api_key, model, method, args]
if kwargs:
call_args.append(kwargs)
return self._jsonrpc('object', 'execute_kw', call_args)
# ------------------------------------------------------------------
# Tech sync ID helpers
# ------------------------------------------------------------------
def _get_local_tech_map(self):
"""Build {local_user_id: x_fc_tech_sync_id} for all local field staff."""
techs = self.env['res.users'].sudo().search([
('x_fc_is_field_staff', '=', True),
('x_fc_tech_sync_id', '!=', False),
('active', '=', True),
])
return {u.id: u.x_fc_tech_sync_id for u in techs}
def _get_remote_tech_map(self):
"""Build {x_fc_tech_sync_id: remote_user_id} from the remote instance."""
self.ensure_one()
remote_users = self._rpc('res.users', 'search_read', [
[('x_fc_is_field_staff', '=', True),
('x_fc_tech_sync_id', '!=', False),
('active', '=', True)],
], {'fields': ['id', 'x_fc_tech_sync_id']})
if not remote_users:
return {}
return {
ru['x_fc_tech_sync_id']: ru['id']
for ru in remote_users
if ru.get('x_fc_tech_sync_id')
}
def _get_local_syncid_to_uid(self):
"""Build {x_fc_tech_sync_id: local_user_id} for local field staff."""
techs = self.env['res.users'].sudo().search([
('x_fc_is_field_staff', '=', True),
('x_fc_tech_sync_id', '!=', False),
('active', '=', True),
])
return {u.x_fc_tech_sync_id: u.id for u in techs}
# ------------------------------------------------------------------
# Connection test
# ------------------------------------------------------------------
def action_test_connection(self):
"""Test the connection to the remote instance."""
self.ensure_one()
uid = self._authenticate()
if uid:
remote_map = self._get_remote_tech_map()
local_map = self._get_local_tech_map()
matched = set(local_map.values()) & set(remote_map.keys())
return {
'type': 'ir.actions.client',
'tag': 'display_notification',
'params': {
'title': 'Connection Successful',
'message': f'Connected to {self.name}. '
f'{len(matched)} technician(s) matched by sync ID.',
'type': 'success',
'sticky': False,
},
}
raise UserError(f"Cannot connect to {self.name}. Check URL, database, and API key.")
# ------------------------------------------------------------------
# PUSH: send local task changes to remote instance
# ------------------------------------------------------------------
def _get_local_instance_id(self):
"""Return this instance's own ID from config parameters."""
return self.env['ir.config_parameter'].sudo().get_param(
'fusion_claims.sync_instance_id', '')
@api.model
def _push_tasks(self, tasks, operation='create'):
"""Push local task changes to all active remote instances.
Called from technician_task create/write overrides.
Non-blocking: errors are logged, not raised.
"""
configs = self.sudo().search([('active', '=', True)])
if not configs:
return
local_id = configs[0]._get_local_instance_id()
if not local_id:
return
for config in configs:
try:
config._push_tasks_to_remote(tasks, operation, local_id)
except Exception:
_logger.exception("Task sync push to %s failed", config.name)
def _push_tasks_to_remote(self, tasks, operation, local_instance_id):
"""Push task data to a single remote instance."""
self.ensure_one()
local_map = self._get_local_tech_map()
remote_map = self._get_remote_tech_map()
if not local_map or not remote_map:
return
ctx = {'context': {'skip_task_sync': True, 'skip_travel_recalc': True}}
for task in tasks:
sync_id = local_map.get(task.technician_id.id)
if not sync_id:
continue
remote_tech_uid = remote_map.get(sync_id)
if not remote_tech_uid:
continue
task_data = {
'x_fc_sync_uuid': task.x_fc_sync_uuid,
'x_fc_sync_source': local_instance_id,
'x_fc_sync_remote_id': task.id,
'name': f"[{local_instance_id.upper()}] {task.name}",
'technician_id': remote_tech_uid,
'task_type': task.task_type,
'status': task.status,
'scheduled_date': str(task.scheduled_date) if task.scheduled_date else False,
'time_start': task.time_start,
'time_end': task.time_end,
'duration_hours': task.duration_hours,
'address_street': task.address_street or '',
'address_street2': task.address_street2 or '',
'address_city': task.address_city or '',
'address_zip': task.address_zip or '',
'address_lat': float(task.address_lat or 0),
'address_lng': float(task.address_lng or 0),
'priority': task.priority or 'normal',
'x_fc_sync_client_name': task.partner_id.name if task.partner_id else '',
}
existing = self._rpc(
'fusion.technician.task', 'search',
[[('x_fc_sync_uuid', '=', task.x_fc_sync_uuid)]],
{'limit': 1})
if operation in ('create', 'write'):
if existing:
self._rpc('fusion.technician.task', 'write',
[existing, task_data], ctx)
elif operation == 'create':
task_data['sale_order_id'] = False
self._rpc('fusion.technician.task', 'create',
[[task_data]], ctx)
elif operation == 'unlink' and existing:
self._rpc('fusion.technician.task', 'write',
[existing, {'status': 'cancelled', 'active': False}], ctx)
# ------------------------------------------------------------------
# PULL: cron-based full reconciliation
# ------------------------------------------------------------------
@api.model
def _cron_pull_remote_tasks(self):
"""Cron job: pull tasks from all active remote instances."""
configs = self.sudo().search([('active', '=', True)])
for config in configs:
try:
config._pull_tasks_from_remote()
config.sudo().write({
'last_sync': fields.Datetime.now(),
'last_sync_error': False,
})
except Exception as e:
_logger.exception("Task sync pull from %s failed", config.name)
config.sudo().write({'last_sync_error': str(e)})
def _pull_tasks_from_remote(self):
"""Pull all active tasks for matched technicians from the remote instance."""
self.ensure_one()
local_syncid_to_uid = self._get_local_syncid_to_uid()
if not local_syncid_to_uid:
return
remote_map = self._get_remote_tech_map()
if not remote_map:
return
matched_sync_ids = set(local_syncid_to_uid.keys()) & set(remote_map.keys())
if not matched_sync_ids:
_logger.info("Task sync: no matched technicians between local and %s", self.name)
return
remote_tech_ids = [remote_map[sid] for sid in matched_sync_ids]
remote_syncid_by_uid = {v: k for k, v in remote_map.items()}
cutoff = fields.Date.today() - timedelta(days=7)
remote_tasks = self._rpc(
'fusion.technician.task', 'search_read',
[[
('technician_id', 'in', remote_tech_ids),
('scheduled_date', '>=', str(cutoff)),
('x_fc_sync_source', '=', False),
]],
{'fields': SYNC_TASK_FIELDS + ['id']})
if remote_tasks is None:
return
Task = self.env['fusion.technician.task'].sudo().with_context(
skip_task_sync=True, skip_travel_recalc=True)
remote_uuids = set()
for rt in remote_tasks:
sync_uuid = rt.get('x_fc_sync_uuid')
if not sync_uuid:
continue
remote_uuids.add(sync_uuid)
remote_tech_raw = rt['technician_id']
remote_uid = remote_tech_raw[0] if isinstance(remote_tech_raw, (list, tuple)) else remote_tech_raw
tech_sync_id = remote_syncid_by_uid.get(remote_uid)
local_uid = local_syncid_to_uid.get(tech_sync_id) if tech_sync_id else None
if not local_uid:
continue
partner_raw = rt.get('partner_id')
client_name = partner_raw[1] if isinstance(partner_raw, (list, tuple)) and len(partner_raw) > 1 else ''
vals = {
'x_fc_sync_uuid': sync_uuid,
'x_fc_sync_source': self.instance_id,
'x_fc_sync_remote_id': rt['id'],
'name': f"[{self.instance_id.upper()}] {rt.get('name', '')}",
'technician_id': local_uid,
'task_type': rt.get('task_type', 'delivery'),
'status': rt.get('status', 'scheduled'),
'scheduled_date': rt.get('scheduled_date'),
'time_start': rt.get('time_start', 9.0),
'time_end': rt.get('time_end', 10.0),
'duration_hours': rt.get('duration_hours', 1.0),
'address_street': rt.get('address_street', ''),
'address_street2': rt.get('address_street2', ''),
'address_city': rt.get('address_city', ''),
'address_zip': rt.get('address_zip', ''),
'address_lat': rt.get('address_lat', 0),
'address_lng': rt.get('address_lng', 0),
'priority': rt.get('priority', 'normal'),
'x_fc_sync_client_name': client_name,
}
existing = Task.search([('x_fc_sync_uuid', '=', sync_uuid)], limit=1)
if existing:
existing.write(vals)
else:
vals['sale_order_id'] = False
Task.create([vals])
stale_shadows = Task.search([
('x_fc_sync_source', '=', self.instance_id),
('x_fc_sync_uuid', 'not in', list(remote_uuids)),
('scheduled_date', '>=', str(cutoff)),
('active', '=', True),
])
if stale_shadows:
stale_shadows.write({'active': False, 'status': 'cancelled'})
_logger.info("Deactivated %d stale shadow tasks from %s",
len(stale_shadows), self.instance_id)
# ------------------------------------------------------------------
# CLEANUP
# ------------------------------------------------------------------
@api.model
def _cron_cleanup_old_shadows(self):
"""Remove shadow tasks older than 30 days (completed/cancelled)."""
cutoff = fields.Date.today() - timedelta(days=30)
old_shadows = self.env['fusion.technician.task'].sudo().search([
('x_fc_sync_source', '!=', False),
('scheduled_date', '<', str(cutoff)),
('status', 'in', ['completed', 'cancelled']),
])
if old_shadows:
count = len(old_shadows)
old_shadows.unlink()
_logger.info("Cleaned up %d old shadow tasks", count)
# ------------------------------------------------------------------
# Manual trigger
# ------------------------------------------------------------------
def action_sync_now(self):
"""Manually trigger a full sync for this config."""
self.ensure_one()
self._pull_tasks_from_remote()
self.sudo().write({
'last_sync': fields.Datetime.now(),
'last_sync_error': False,
})
shadow_count = self.env['fusion.technician.task'].sudo().search_count([
('x_fc_sync_source', '=', self.instance_id),
])
return {
'type': 'ir.actions.client',
'tag': 'display_notification',
'params': {
'title': 'Sync Complete',
'message': f'Synced from {self.name}. {shadow_count} shadow task(s) now visible.',
'type': 'success',
'sticky': False,
},
}

190
fusion_claims/models/technician_task.py
View File

@@ -14,6 +14,7 @@ from odoo.osv import expression
from markupsafe import Markup
import logging
import json
import uuid
import requests
from datetime import datetime as dt_datetime, timedelta
import urllib.parse
@@ -32,7 +33,7 @@ class FusionTechnicianTask(models.Model):
"""Richer display name: Client - Type | 9:00 AM - 10:00 AM."""
type_labels = dict(self._fields['task_type'].selection)
for task in self:
client = task.partner_id.name or ''
client = task.x_fc_sync_client_name if task.x_fc_sync_source else (task.partner_id.name or '')
ttype = type_labels.get(task.task_type, task.task_type or '')
start = self._float_to_time_str(task.time_start)
end = self._float_to_time_str(task.time_end)
@@ -70,6 +71,40 @@ class FusionTechnicianTask(models.Model):
)
active = fields.Boolean(default=True)
# Cross-instance sync fields
x_fc_sync_source = fields.Char(
'Source Instance', readonly=True, index=True,
help='Origin instance ID if this is a synced shadow task (e.g. westin, mobility)',
)
x_fc_sync_remote_id = fields.Integer(
'Remote Task ID', readonly=True,
help='ID of the task on the remote instance',
)
x_fc_sync_uuid = fields.Char(
'Sync UUID', readonly=True, index=True, copy=False,
help='Unique ID for cross-instance deduplication',
)
x_fc_is_shadow = fields.Boolean(
'Shadow Task', compute='_compute_is_shadow', store=True,
help='True if this task was synced from another instance',
)
x_fc_sync_client_name = fields.Char(
'Synced Client Name', readonly=True,
help='Client name from the remote instance (shadow tasks only)',
)
x_fc_source_label = fields.Char(
'Source', compute='_compute_is_shadow', store=True,
)
@api.depends('x_fc_sync_source')
def _compute_is_shadow(self):
local_id = self.env['ir.config_parameter'].sudo().get_param(
'fusion_claims.sync_instance_id', '')
for task in self:
task.x_fc_is_shadow = bool(task.x_fc_sync_source)
task.x_fc_source_label = task.x_fc_sync_source or local_id
technician_id = fields.Many2one(
'res.users',
string='Technician',
@@ -87,10 +122,9 @@ class FusionTechnicianTask(models.Model):
sale_order_id = fields.Many2one(
'sale.order',
string='Related Case',
required=True,
tracking=True,
ondelete='restrict',
help='Sale order / case linked to this task (required)',
help='Sale order / case linked to this task',
)
sale_order_name = fields.Char(
related='sale_order_id.name',
@@ -98,6 +132,19 @@ class FusionTechnicianTask(models.Model):
store=True,
)
purchase_order_id = fields.Many2one(
'purchase.order',
string='Related Purchase Order',
tracking=True,
ondelete='restrict',
help='Purchase order linked to this task (e.g. manufacturer pickup)',
)
purchase_order_name = fields.Char(
related='purchase_order_id.name',
string='PO Reference',
store=True,
)
task_type = fields.Selection([
('delivery', 'Delivery'),
('repair', 'Repair'),
@@ -854,30 +901,60 @@ class FusionTechnicianTask(models.Model):
def _onchange_sale_order_id(self):
"""Auto-fill client and address from the sale order's shipping address."""
if self.sale_order_id:
self.purchase_order_id = False
order = self.sale_order_id
if not self.partner_id:
self.partner_id = order.partner_id
# Use shipping address if different
addr = order.partner_shipping_id or order.partner_id
self.address_partner_id = addr.id
self.address_street = addr.street or ''
self.address_street2 = addr.street2 or ''
self.address_city = addr.city or ''
self.address_state_id = addr.state_id.id if addr.state_id else False
self.address_zip = addr.zip or ''
self.address_lat = addr.x_fc_latitude if hasattr(addr, 'x_fc_latitude') and addr.x_fc_latitude else 0
self.address_lng = addr.x_fc_longitude if hasattr(addr, 'x_fc_longitude') and addr.x_fc_longitude else 0
self._fill_address_from_partner(addr)
@api.onchange('purchase_order_id')
def _onchange_purchase_order_id(self):
"""Auto-fill client and address from the purchase order's vendor."""
if self.purchase_order_id:
self.sale_order_id = False
order = self.purchase_order_id
if not self.partner_id:
self.partner_id = order.partner_id
addr = order.dest_address_id or order.partner_id
self._fill_address_from_partner(addr)
def _fill_address_from_partner(self, addr):
"""Populate address fields from a partner record."""
if not addr:
return
self.address_partner_id = addr.id
self.address_street = addr.street or ''
self.address_street2 = addr.street2 or ''
self.address_city = addr.city or ''
self.address_state_id = addr.state_id.id if addr.state_id else False
self.address_zip = addr.zip or ''
self.address_lat = addr.x_fc_latitude if hasattr(addr, 'x_fc_latitude') and addr.x_fc_latitude else 0
self.address_lng = addr.x_fc_longitude if hasattr(addr, 'x_fc_longitude') and addr.x_fc_longitude else 0
# ------------------------------------------------------------------
# CONSTRAINTS + VALIDATION
# ------------------------------------------------------------------
@api.constrains('sale_order_id', 'purchase_order_id')
def _check_order_link(self):
for task in self:
if task.x_fc_sync_source:
continue
if not task.sale_order_id and not task.purchase_order_id:
raise ValidationError(_(
"A task must be linked to either a Sale Order (Case) or a Purchase Order."
))
@api.constrains('technician_id', 'scheduled_date', 'time_start', 'time_end')
def _check_no_overlap(self):
"""Prevent overlapping bookings for the same technician on the same date."""
for task in self:
if task.status == 'cancelled':
continue
if task.x_fc_sync_source:
continue
# Validate time range
if task.time_start >= task.time_end:
raise ValidationError(_("Start time must be before end time."))
@@ -889,8 +966,8 @@ class FusionTechnicianTask(models.Model):
raise ValidationError(_(
"Tasks must be scheduled within store hours (%s - %s)."
) % (open_str, close_str))
# Validate not in the past (only for new/scheduled tasks)
if task.status == 'scheduled' and task.scheduled_date:
# Validate not in the past (only for new/scheduled local tasks)
if task.status == 'scheduled' and task.scheduled_date and not task.x_fc_sync_source:
today = fields.Date.context_today(self)
if task.scheduled_date < today:
raise ValidationError(_("Cannot schedule tasks in the past."))
@@ -1138,6 +1215,8 @@ class FusionTechnicianTask(models.Model):
for vals in vals_list:
if vals.get('name', _('New')) == _('New'):
vals['name'] = self.env['ir.sequence'].next_by_code('fusion.technician.task') or _('New')
if not vals.get('x_fc_sync_uuid') and not vals.get('x_fc_sync_source'):
vals['x_fc_sync_uuid'] = str(uuid.uuid4())
# Auto-populate address from sale order if not provided
if vals.get('sale_order_id') and not vals.get('address_street'):
order = self.env['sale.order'].browse(vals['sale_order_id'])
@@ -1146,15 +1225,23 @@ class FusionTechnicianTask(models.Model):
self._fill_address_vals(vals, addr)
if not vals.get('partner_id'):
vals['partner_id'] = order.partner_id.id
# Auto-populate address from partner if sale order not set
# Auto-populate address from purchase order if not provided
elif vals.get('purchase_order_id') and not vals.get('address_street'):
po = self.env['purchase.order'].browse(vals['purchase_order_id'])
addr = po.dest_address_id or po.partner_id
if addr:
self._fill_address_vals(vals, addr)
if not vals.get('partner_id'):
vals['partner_id'] = po.partner_id.id
# Auto-populate address from partner if no order set
elif vals.get('partner_id') and not vals.get('address_street'):
partner = self.env['res.partner'].browse(vals['partner_id'])
if partner.street:
self._fill_address_vals(vals, partner)
records = super().create(vals_list)
# Post creation notice to linked sale order chatter
# Post creation notice to linked order chatter
for rec in records:
rec._post_task_created_to_sale_order()
rec._post_task_created_to_linked_order()
# If created from "Ready for Delivery" flow, mark the sale order
if self.env.context.get('mark_ready_for_delivery'):
records._mark_sale_order_ready_for_delivery()
@@ -1170,6 +1257,10 @@ class FusionTechnicianTask(models.Model):
# Send "Appointment Scheduled" email
for rec in records:
rec._send_task_scheduled_email()
# Push new local tasks to remote instances
local_records = records.filtered(lambda r: not r.x_fc_sync_source)
if local_records and not self.env.context.get('skip_task_sync'):
self.env['fusion.task.sync.config']._push_tasks(local_records, 'create')
return records
def write(self, vals):
@@ -1226,6 +1317,15 @@ class FusionTechnicianTask(models.Model):
old_start=old['time_start'],
old_end=old['time_end'],
)
# Push updates to remote instances for local tasks
sync_fields = {'technician_id', 'scheduled_date', 'time_start', 'time_end',
'duration_hours', 'status', 'task_type', 'address_street',
'address_city', 'address_zip', 'address_lat', 'address_lng',
'partner_id'}
if sync_fields & set(vals.keys()) and not self.env.context.get('skip_task_sync'):
local_records = self.filtered(lambda r: not r.x_fc_sync_source)
if local_records:
self.env['fusion.task.sync.config']._push_tasks(local_records, 'write')
return res
@api.model
@@ -1242,10 +1342,11 @@ class FusionTechnicianTask(models.Model):
'address_lng': partner.x_fc_longitude if hasattr(partner, 'x_fc_longitude') else 0,
})
def _post_task_created_to_sale_order(self):
"""Post a brief task creation notice to the linked sale order's chatter."""
def _post_task_created_to_linked_order(self):
"""Post a brief task creation notice to the linked order's chatter."""
self.ensure_one()
if not self.sale_order_id:
order = self.sale_order_id or self.purchase_order_id
if not order:
return
task_type_label = dict(self._fields['task_type'].selection).get(self.task_type, self.task_type)
date_str = self.scheduled_date.strftime('%B %d, %Y') if self.scheduled_date else 'TBD'
@@ -1259,7 +1360,7 @@ class FusionTechnicianTask(models.Model):
f'<a href="{task_url}">View Task</a>'
f'</div>'
)
self.sale_order_id.message_post(
order.message_post(
body=body, message_type='notification', subtype_xmlid='mail.mt_note',
)
@@ -1462,6 +1563,19 @@ class FusionTechnicianTask(models.Model):
'res_id': self.sale_order_id.id,
}
def action_view_purchase_order(self):
"""Open the linked purchase order."""
self.ensure_one()
if not self.purchase_order_id:
return
return {
'name': self.purchase_order_id.name,
'type': 'ir.actions.act_window',
'res_model': 'purchase.order',
'view_mode': 'form',
'res_id': self.purchase_order_id.id,
}
def action_complete_task(self):
"""Mark task as Completed."""
for task in self:
@@ -1472,9 +1586,9 @@ class FusionTechnicianTask(models.Model):
'completion_datetime': fields.Datetime.now(),
})
task._post_status_message('completed')
# Post completion notes to sale order chatter if linked
if task.sale_order_id and task.completion_notes:
task._post_completion_to_sale_order()
# Post completion notes to linked order chatter
if task.completion_notes and (task.sale_order_id or task.purchase_order_id):
task._post_completion_to_linked_order()
# Notify the person who scheduled the task
task._notify_scheduler_on_completion()
# Auto-advance ODSP status for delivery tasks
@@ -1607,10 +1721,11 @@ class FusionTechnicianTask(models.Model):
)
self.message_post(body=body, message_type='notification', subtype_xmlid='mail.mt_note')
def _post_completion_to_sale_order(self):
"""Post the completion notes to the linked sale order's chatter."""
def _post_completion_to_linked_order(self):
"""Post the completion notes to the linked order's chatter."""
self.ensure_one()
if not self.sale_order_id or not self.completion_notes:
order = self.sale_order_id or self.purchase_order_id
if not order or not self.completion_notes:
return
task_type_label = dict(self._fields['task_type'].selection).get(self.task_type, self.task_type)
body = Markup(
@@ -1625,7 +1740,7 @@ class FusionTechnicianTask(models.Model):
f'{self.completion_notes}'
f'</div>'
)
self.sale_order_id.message_post(
order.message_post(
body=body,
message_type='notification',
subtype_xmlid='mail.mt_note',
@@ -1639,7 +1754,8 @@ class FusionTechnicianTask(models.Model):
task_type_label = dict(self._fields['task_type'].selection).get(self.task_type, self.task_type)
task_url = f'/web#id={self.id}&model=fusion.technician.task&view_type=form'
client_name = self.partner_id.name or 'N/A'
case_ref = self.sale_order_id.name if self.sale_order_id else ''
order = self.sale_order_id or self.purchase_order_id
case_ref = order.name if order else ''
# Build address string
addr_parts = [p for p in [
self.address_street,
@@ -1694,6 +1810,8 @@ class FusionTechnicianTask(models.Model):
]
if self.sale_order_id:
rows.append(('Case', self.sale_order_id.name))
if self.purchase_order_id:
rows.append(('Purchase Order', self.purchase_order_id.name))
if self.scheduled_date:
date_str = self.scheduled_date.strftime('%B %d, %Y')
start_str = self._float_to_time_str(self.time_start)
@@ -1963,9 +2081,10 @@ class FusionTechnicianTask(models.Model):
"""
api_key = self.env['ir.config_parameter'].sudo().get_param(
'fusion_claims.google_maps_api_key', '')
local_instance = self.env['ir.config_parameter'].sudo().get_param(
'fusion_claims.sync_instance_id', '')
base_domain = [
('status', 'not in', ['cancelled']),
('address_lat', '!=', 0), ('address_lng', '!=', 0),
]
if domain:
base_domain = expression.AND([base_domain, domain])
@@ -1974,13 +2093,18 @@ class FusionTechnicianTask(models.Model):
['name', 'partner_id', 'technician_id', 'task_type',
'address_lat', 'address_lng', 'address_display',
'time_start', 'time_start_display', 'time_end_display',
'status', 'scheduled_date', 'travel_time_minutes'],
'status', 'scheduled_date', 'travel_time_minutes',
'x_fc_sync_client_name', 'x_fc_is_shadow', 'x_fc_sync_source'],
order='scheduled_date asc, time_start asc',
limit=500,
)
# Also include live technician locations
locations = self.env['fusion.technician.location'].get_latest_locations()
return {'api_key': api_key, 'tasks': tasks, 'locations': locations}
return {
'api_key': api_key,
'tasks': tasks,
'locations': locations,
'local_instance_id': local_instance,
}
def _geocode_address(self):
"""Geocode the task address using Google Geocoding API."""