from odoo import models, fields, api, _ from odoo.exceptions import UserError import requests import json import datetime import logging import math _logger = logging.getLogger(__name__) _GEOCODE_CACHE = {} def _haversine_distance(lat1, lon1, lat2, lon2): """Calculate great circle distance between two lat/lon coordinates in kilometers""" R = 6371.0 # Earth radius in km dlat = math.radians(lat2 - lat1) dlon = math.radians(lon2 - lon1) a = (math.sin(dlat / 2.0) ** 2 + math.cos(math.radians(lat1)) * math.cos(math.radians(lat2)) * math.sin(dlon / 2.0) ** 2) c = 2.0 * math.atan2(math.sqrt(a), math.sqrt(1.0 - a)) return R * c class UberConfig(models.Model): _name = 'uber.config' _description = 'Uber Integration Configuration' name = fields.Char(string='Config Name', required=True, default='Uber Eats / Direct') client_id = fields.Char(string='Client ID', required=True) client_secret = fields.Char(string='Client Secret', required=True) customer_id = fields.Char(string='Customer ID (Uber Direct)') environment = fields.Selection([ ('sandbox', 'Sandbox / Testing'), ('production', 'Production / Live') ], string='Environment', default='sandbox', required=True) scope = fields.Char(string='OAuth Scope', default='eats.deliveries', help="OAuth scope for Uber Direct, e.g., 'eats.deliveries'.") # Delivery Pricing Source Selection - Default to Live Uber Direct API delivery_pricing_source = fields.Selection([ ('uber', 'Live Uber Direct API Only'), ('distance', 'Manual Distance-Based Calculation (Optional Fallback)'), ('uber_fallback', 'Live Uber API with Distance Fallback') ], string='Delivery Fee Source', default='uber', required=True, help="Delivery charges and delivery availability are fetched strictly from Uber Direct API.") # Dynamic Distance and Radius Settings (Optional backup) restaurant_latitude = fields.Float(string='Restaurant Latitude', digits=(10, 7), default=0.0, help="Latitude of the restaurant (leave 0.0 to automatically fetch from Company Settings).") restaurant_longitude = fields.Float(string='Restaurant Longitude', digits=(10, 7), default=0.0, help="Longitude of the restaurant (leave 0.0 to automatically fetch from Company Settings).") max_delivery_radius = fields.Float(string='Max Delivery Radius (km)', default=25.0, help="Maximum allowed delivery distance from restaurant in kilometers.") base_delivery_fee = fields.Float(string='Base Delivery Fee ($)', default=4.99, help="Delivery fee applied for orders within base distance.") base_distance_km = fields.Float(string='Base Distance (km)', default=3.0, help="Distance included in the base delivery fee.") per_km_fee = fields.Float(string='Per KM Fee ($/km)', default=1.00, help="Additional fee per kilometer beyond the base distance.") fallback_delivery_fee = fields.Float(string='Fallback Delivery Fee ($)', default=4.99, help="Standard delivery fee applied if Uber API is unavailable and distance calculation is used.") enable_fallback_on_error = fields.Boolean(string='Enable Fallback on API Error', default=False, help="If enabled, distance-based calculation will be used when Uber API returns an error.") timeout_minutes = fields.Integer(string='Driver Assignment Alert Timeout (min)', default=15) delivery_product_id = fields.Many2one('product.product', string='Uber Delivery Fee Product', help="Service product used to add Uber charges to the bill.") access_token = fields.Char(string='Current Access Token') token_expiry = fields.Datetime(string='Token Expiry') active = fields.Boolean(default=True) def _get_or_create_delivery_product(self): """Find or create the default Uber Delivery Fee service product""" self.ensure_one() if self.delivery_product_id: return self.delivery_product_id Product = self.env['product.product'].sudo() product = Product.search([('name', '=', 'Uber Delivery Fee')], limit=1) if not product: product = Product.search([('name', 'ilike', 'Delivery Fee'), ('type', '=', 'service')], limit=1) if not product: product = Product.search([('name', 'ilike', 'Delivery'), ('type', '=', 'service')], limit=1) if not product: product = Product.create({ 'name': 'Uber Delivery Fee', 'type': 'service', 'list_price': 0.0, 'available_in_pos': True, 'invoice_policy': 'order', }) self.sudo().write({'delivery_product_id': product.id}) return product def write(self, vals): # Invalidate cached token whenever credentials or scope change if any(k in vals for k in ['client_id', 'client_secret', 'scope', 'environment']): vals['access_token'] = False vals['token_expiry'] = False return super().write(vals) def _get_api_base_url(self): """Return the API base URL based on environment""" self.ensure_one() # Uber Direct API v1 return "https://api.uber.com/v1" def _get_token_url(self): """Return OAuth token URL (login.uber.com supports both Sandbox and Production)""" self.ensure_one() return "https://login.uber.com/oauth/v2/token" def _get_access_token(self): """Get or refresh OAuth 2.0 access token""" self.ensure_one() now = fields.Datetime.now() # Return existing valid token if self.access_token and self.token_expiry and self.token_expiry > now: return self.access_token # Clean credentials client_id = self.client_id.strip() if self.client_id else '' client_secret = self.client_secret.strip() if self.client_secret else '' scope = self.scope.strip() if self.scope else 'eats.deliveries' # Request new token token_url = self._get_token_url() payload = { 'client_id': client_id, 'client_secret': client_secret, 'grant_type': 'client_credentials', 'scope': scope # Required scope for Uber Direct } try: response = requests.post(token_url, data=payload) response.raise_for_status() data = response.json() access_token = data.get('access_token') expires_in = data.get('expires_in', 2592000) # Default 30 days # Save token self.write({ 'access_token': access_token, 'token_expiry': now + datetime.timedelta(seconds=expires_in - 60) # Buffer }) return access_token except requests.exceptions.RequestException as e: error_msg = str(e) if e.response is not None: try: error_data = e.response.json() if 'error' in error_data: error_msg = f"{error_data.get('error')}: {error_data.get('error_description', '')}" except ValueError: error_msg = e.response.text raise UserError(_("Authentication Failed: %s") % error_msg) def action_test_connection(self): """Test connection with eats.deliveries scope for live Uber Direct delivery""" self.ensure_one() current_scope = (self.scope or '').strip() if not current_scope or current_scope != 'eats.deliveries': self.write({'scope': 'eats.deliveries'}) try: token = self._get_access_token() message = "Connection Successful! Token retrieved successfully using scope 'eats.deliveries'. Uber Direct live delivery quotes and couriers are fully active." return self._return_notification(message, "success") except UserError as e: err_str = str(e) client_id = (self.client_id or '').strip() client_secret = (self.client_secret or '').strip() # Diagnostic 1: Check if credentials work on Production login vs Sandbox login org_token = None detected_env = None for env_name, token_url in [ ('production', 'https://login.uber.com/oauth/v2/token'), ('sandbox', 'https://sandbox-login.uber.com/oauth/v2/token') ]: try: r = requests.post(token_url, data={ 'client_id': client_id, 'client_secret': client_secret, 'grant_type': 'client_credentials', 'scope': 'direct.organizations' }, timeout=10) if r.status_code == 200: detected_env = env_name org_token = r.json().get('access_token') break except Exception: pass if detected_env: env_mismatch_note = "" if self.environment != detected_env: env_mismatch_note = f"\n\nNote: Your credentials matched the {detected_env.upper()} environment, but Odoo is currently set to '{self.environment.capitalize()}'. Change Environment to '{detected_env.capitalize()}'." message = ( f"Credentials Authenticated Successfully!\n\n" f"Your Client ID and Secret are valid and connected to your Uber Direct organization (scope: 'direct.organizations').{env_mismatch_note}\n\n" f"To enable 'eats.deliveries' (Courier Dispatch & Live Quotes):\n" f"1. For Sandbox Testing: In direct.uber.com, click 'Switch to testing' (top right of Developer page), then copy the testing Client ID & Secret into Odoo and set Environment to 'Sandbox / Testing'.\n" f"2. For Live Production: In direct.uber.com, click 'Set up' under Billing (left menu) to add a payment card/method and add your store address under Locations. Uber activates 'eats.deliveries' once billing is attached." ) return self._return_notification(message, "warning") if "invalid_scope" in err_str.lower() or "scope" in err_str.lower(): message = ( "Uber Direct Permission Required: Your Uber Client ID requires the 'eats.deliveries' scope.\n\n" "How to resolve:\n" "1. For Sandbox: In direct.uber.com, click 'Switch to testing' to use testing credentials.\n" "2. For Production: Set up Billing and Location in direct.uber.com to activate live courier dispatch.\n" "3. Ensure Environment in Odoo matches (Sandbox vs Production)." ) return self._return_notification(message, "warning") return self._return_notification(f"Connection Failed: {err_str}", "danger") def _auth_with_scope(self, scope_to_test): """Helper to test a specific scope without saving""" client_id = self.client_id.strip() if self.client_id else '' client_secret = self.client_secret.strip() if self.client_secret else '' token_url = self._get_token_url() payload = { 'client_id': client_id, 'client_secret': client_secret, 'grant_type': 'client_credentials', 'scope': scope_to_test } response = requests.post(token_url, data=payload) response.raise_for_status() # Will raise error if scope invalid return True def _geocode_address(self, street, city, state, zip_code, country): """Geocode an address to (lat, lon) using OpenStreetMap Nominatim with memory cache""" cache_key = f"{zip_code}_{city}_{street}_{country}".strip().lower() if cache_key in _GEOCODE_CACHE: return _GEOCODE_CACHE[cache_key] headers = {'User-Agent': 'Dine360-Restaurant-Delivery/1.0 (delivery@dine360.com)'} queries = [] # 1. Full address (street, city, state, zip, country) full_addr = ", ".join(filter(None, [street, city, state, zip_code, country])) if full_addr: queries.append(full_addr) # 2. Street + City + State + Country if street and city and (state or country): queries.append(", ".join(filter(None, [street, city, state, country]))) # 3. Street + City + Country if street and city and country: queries.append(f"{street}, {city}, {country}") # 4. Postal code + City + Country if zip_code and city and country: queries.append(f"{zip_code}, {city}, {country}") # 5. Postal code + Country if zip_code and country: queries.append(f"{zip_code}, {country}") # 6. City + State + Country if city and country: queries.append(f"{city}, {state or ''}, {country}".strip()) # Resolve country ISO code to restrict search boundaries country_code_map = { 'canada': 'ca', 'ca': 'ca', 'united states': 'us', 'usa': 'us', 'us': 'us', 'united kingdom': 'gb', 'uk': 'gb', 'gb': 'gb', 'india': 'in', 'in': 'in' } cc = country_code_map.get((country or '').strip().lower()) for q in queries: try: url = "https://nominatim.openstreetmap.org/search" params = {'q': q, 'format': 'json', 'limit': 1} if cc: params['countrycodes'] = cc resp = requests.get(url, params=params, headers=headers, timeout=5) if resp.status_code == 200: data = resp.json() if data and len(data) > 0: lat = float(data[0]['lat']) lon = float(data[0]['lon']) _GEOCODE_CACHE[cache_key] = (lat, lon) return (lat, lon) except Exception as e: _logger.warning("Geocoding query '%s' error: %s", q, e) return None def _get_company_coordinates(self, company): """Dynamically get coordinates of the restaurant company from Settings without static defaults""" if not company: return None # 1. Check if configured explicitly on uber.config if self.restaurant_latitude and self.restaurant_longitude: return (self.restaurant_latitude, self.restaurant_longitude) # 2. Check if company partner already has coordinates if company.partner_id and company.partner_id.partner_latitude and company.partner_id.partner_longitude: return (company.partner_id.partner_latitude, company.partner_id.partner_longitude) # 3. Dynamically geocode the company's real address from Settings full_street = ", ".join(filter(None, [company.street, company.street2])) coords = self._geocode_address( street=full_street, city=company.city or '', state=company.state_id.name or company.state_id.code or '', zip_code=company.zip or '', country=company.country_id.name or 'Canada' ) if coords and company.partner_id: try: company.partner_id.sudo().write({ 'partner_latitude': coords[0], 'partner_longitude': coords[1] }) except Exception: pass return coords def calculate_distance_quote(self, dropoff_data, company=None): """Calculate distance and quote fee based on geographic coordinates""" self.ensure_one() if not dropoff_data: return {'success': False, 'error': _("No delivery address provided.")} street = (dropoff_data.get('street') or '').strip() street2 = (dropoff_data.get('street2') or '').strip() full_street = f"{street} {street2}".strip() city = (dropoff_data.get('city') or '').strip() state = (dropoff_data.get('state') or '').strip() zip_code = (dropoff_data.get('zip') or '').strip() country = (dropoff_data.get('country') or '').strip() # Check country against restaurant company country if company and company.country_id: company_country = company.country_id if country and country.lower() not in [company_country.name.lower(), company_country.code.lower()]: return { 'success': False, 'error': _("Delivery is not available outside %s. Please select Store Pickup or enter a local delivery address.") % company_country.name } # Restaurant coordinates fetched dynamically from Company in Settings coords_rest = self._get_company_coordinates(company) if not coords_rest: return { 'success': False, 'error': _("Restaurant address in Settings -> Companies is missing or cannot be located.") } rest_lat, rest_lon = coords_rest # Geocode dropoff coords = self._geocode_address(full_street, city, state, zip_code, country) if not coords: _logger.warning("Could not geocode customer address: %s, %s, %s, %s", full_street, city, zip_code, country) if zip_code and zip_code.isdigit() and len(zip_code) == 6: return { 'success': False, 'error': _("The postal code '%s' is not valid for Canadian delivery. Please check your address or select Store Pickup.") % zip_code } return { 'success': False, 'error': _("Unable to verify delivery address location. Please check your street and postal code, or select Store Pickup.") } cust_lat, cust_lon = coords distance_km = _haversine_distance(rest_lat, rest_lon, cust_lat, cust_lon) _logger.info("Delivery distance to %s, %s: %.2f km (Max radius: %.1f km)", city, zip_code, distance_km, self.max_delivery_radius) if distance_km > self.max_delivery_radius: return { 'success': False, 'distance_km': round(distance_km, 1), 'error': _("This address is outside our delivery area (%.1f km away. Maximum delivery radius is %.0f km). Please select Store Pickup.") % (distance_km, self.max_delivery_radius) } # Calculate fee if distance_km <= self.base_distance_km: fee = self.base_delivery_fee else: fee = self.base_delivery_fee + (distance_km - self.base_distance_km) * self.per_km_fee fee = round(max(fee, 0.0), 2) return { 'success': True, 'fee_amount': fee, 'distance_km': round(distance_km, 1), 'currency': 'CAD', 'is_fallback': True, 'is_distance_based': True, 'quote_id': f"DIST_{int(datetime.datetime.now().timestamp())}" } def get_uber_quote(self, pickup_address, dropoff_address, items=None, dropoff_data=None, company=None): """Get delivery quote from Uber API or selected delivery pricing source""" self.ensure_one() pricing_source = self.delivery_pricing_source or 'uber' # If manually configured to distance calculation only if pricing_source == 'distance': if dropoff_data: return self.calculate_distance_quote(dropoff_data, company) return {'success': False, 'error': _("Delivery address missing.")} customer_id = self.customer_id if not customer_id: if pricing_source == 'uber_fallback' and dropoff_data: return self.calculate_distance_quote(dropoff_data, company) return {'success': False, 'error': _("Uber Customer ID is missing in configuration.")} # Ensure at least one dummy item if none provided if not items: items = [{ "name": "Food Delivery", "quantity": 1, "size": "small" }] payload = { "pickup_address": pickup_address, "dropoff_address": dropoff_address, "manifest_items": items } try: access_token = self._get_access_token() api_url = f"https://api.uber.com/v1/customers/{customer_id}/delivery_quotes" headers = { 'Authorization': f'Bearer {access_token}', 'Content-Type': 'application/json' } _logger.info("Uber Direct Payload: %s", json.dumps(payload, indent=2)) response = requests.post(api_url, headers=headers, json=payload) _logger.info("Uber Direct Raw Response (%s): %s", response.status_code, response.text) if response.status_code != 200: # Log detailed error for debugging _logger.error("Uber Quote Error: %s - %s", response.status_code, response.text) data = {} try: data = response.json() except: pass # Construct descriptive error message msg = data.get('message', 'Uber API Error') if data.get('errors'): details = " ".join([e.get('message', '') for e in data['errors']]) if details: msg = f"{msg} {details}" code = data.get('code', '') # 1. Scope missing error if response.status_code == 401 or 'eats.deliveries' in msg.lower() or 'unauthorized' in code.lower(): scope_err = _( "Uber Direct Scope Required: Your Uber Client ID requires the 'eats.deliveries' permission. " "Please go to your Uber Developer Dashboard (https://developer.uber.com), " "open your application, and enable the 'Uber Direct' product to activate live quotes." ) if pricing_source == 'uber_fallback' and dropoff_data: dist_result = self.calculate_distance_quote(dropoff_data, company) dist_result['warning'] = scope_err return dist_result return { 'success': False, 'error': scope_err, 'code': 'unauthorized', 'raw_error': data } # 2. Tax form / Billing profile required if 'tax_form' in msg.lower() or 'customer_blocked' in code.lower(): tax_err = _( "Uber Direct Account Setup Notice: Please complete your tax/billing details at " "https://direct.uber.com/accounts/%s/billing to enable live deliveries." ) % (customer_id or '') if pricing_source == 'uber_fallback' and dropoff_data: dist_result = self.calculate_distance_quote(dropoff_data, company) dist_result['warning'] = tax_err return dist_result return { 'success': False, 'error': tax_err, 'code': code, 'raw_error': data } # 3. Out of range / undeliverable error from Uber (Live coverage check) if any(x in code.lower() or x in msg.lower() for x in ['out_of_range', 'out of range', 'outside', 'undeliverable', 'address_undeliverable', 'coverage', 'unserviceable']): details = data.get('metadata', {}).get('details') if isinstance(data.get('metadata'), dict) else '' if details: coverage_err = _("Uber Direct: %s Please select an address closer to the restaurant or choose Store Pickup.") % details else: coverage_err = _("Uber Direct: This address is outside Uber's delivery coverage area. Please choose Store Pickup.") return { 'success': False, 'error': coverage_err, 'code': code, 'raw_error': data } # 3. Fallback only if explicitly configured if pricing_source == 'uber_fallback' and dropoff_data: _logger.warning("Uber API failed (%s). Falling back to distance-based quote.", msg) dist_result = self.calculate_distance_quote(dropoff_data, company) if dist_result.get('warning') is None: dist_result['warning'] = msg return dist_result return { 'success': False, 'error': f"Uber API Error: {msg}", 'code': code, 'raw_error': data } data = response.json() # Standard fee is in cents fee_cents = data.get('fee', 0) return { 'success': True, 'quote_id': data.get('id'), 'fee_amount': float(fee_cents) / 100.0, 'currency': data.get('currency_code', 'CAD'), 'estimated_arrival': data.get('estimated_arrival'), 'is_fallback': False, 'is_uber_direct': True, 'raw': data } except Exception as e: _logger.exception("Uber Quote API Exception") if pricing_source == 'uber_fallback' and dropoff_data: dist_result = self.calculate_distance_quote(dropoff_data, company) if dist_result.get('warning') is None: dist_result['warning'] = str(e) return dist_result return {'success': False, 'error': f"Uber Connection Error: {str(e)}"} def _return_notification(self, message, msg_type): return { 'type': 'ir.actions.client', 'tag': 'display_notification', 'params': { 'title': 'Connection Test', 'message': message, 'type': msg_type, 'sticky': False if msg_type == 'success' else True, } }