99 lines
3.6 KiB
Python
99 lines
3.6 KiB
Python
# -*- 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
|
|
ICP = self.env['ir.config_parameter'].sudo()
|
|
addr = address.strip()
|
|
nominatim_url = (ICP.get_param('fusion_tasks.nominatim_url', '') or '').strip()
|
|
if nominatim_url:
|
|
headers = {'User-Agent': 'fusion_tasks/1.0'}
|
|
maps_key = (ICP.get_param('fusion_tasks.maps_api_key', '') or '').strip()
|
|
if maps_key:
|
|
headers['X-API-Key'] = maps_key
|
|
try:
|
|
resp = requests.get(
|
|
f'{nominatim_url.rstrip("/")}/search',
|
|
params={'q': addr, 'format': 'json', 'limit': 1, 'countrycodes': 'ca'},
|
|
timeout=5,
|
|
headers=headers,
|
|
)
|
|
data = resp.json()
|
|
if isinstance(data, list) and data:
|
|
return float(data[0]['lat']), float(data[0]['lon'])
|
|
except Exception as e:
|
|
_logger.warning("Nominatim start-address geocoding failed for '%s': %s", addr, e)
|
|
api_key = ICP.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': addr, '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
|