234 lines
9.2 KiB
Python
234 lines
9.2 KiB
Python
from odoo import models, fields, api, _
|
|
from odoo.exceptions import UserError
|
|
import requests
|
|
import json
|
|
import datetime
|
|
import logging
|
|
|
|
_logger = logging.getLogger(__name__)
|
|
|
|
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='delivery', help="Space-separated list of scopes, e.g., 'eats.deliveries' or 'delivery'. check your Uber Dashboard.")
|
|
|
|
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_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_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 'delivery'
|
|
|
|
# Request new token
|
|
token_url = "https://login.uber.com/oauth/v2/token"
|
|
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 and auto-detect correct scope if 'invalid_scope' error occurs"""
|
|
self.ensure_one()
|
|
|
|
# 1. Try with current configured scope first
|
|
try:
|
|
token = self._get_access_token()
|
|
message = f"Connection Successful! Token retrieved using scope: {self.scope}"
|
|
msg_type = "success"
|
|
return self._return_notification(message, msg_type)
|
|
except UserError as e:
|
|
# Only attempt auto-fix if error is related to scope
|
|
if "invalid_scope" not in str(e) and "scope" not in str(e).lower():
|
|
return self._return_notification(f"Connection Failed: {str(e)}", "danger")
|
|
|
|
# 2. Auto-Discovery: Try known Uber Direct scopes
|
|
potential_scopes = ['delivery', 'eats.deliveries', 'direct.organizations', 'guest.deliveries']
|
|
|
|
# Remove current scope from list to avoid redundant check
|
|
current = self.scope.strip() if self.scope else ''
|
|
if current in potential_scopes:
|
|
potential_scopes.remove(current)
|
|
|
|
working_scope = None
|
|
|
|
for trial_scope in potential_scopes:
|
|
try:
|
|
# Temporarily set scope to test
|
|
self._auth_with_scope(trial_scope)
|
|
working_scope = trial_scope
|
|
break # Found one!
|
|
except Exception:
|
|
continue # Try next
|
|
|
|
# 3. Handle Result
|
|
if working_scope:
|
|
self.write({'scope': working_scope})
|
|
self._get_access_token() # Refresh token storage
|
|
message = f"Success! We found the correct scope '{working_scope}' and updated your settings."
|
|
msg_type = "success"
|
|
else:
|
|
message = "Connection Failed. Your Client ID does not appear to have ANY Uber Direct permissions (eats.deliveries, delivery, etc). Please enabling the 'Uber Direct' product in your Uber Dashboard."
|
|
msg_type = "danger"
|
|
|
|
return self._return_notification(message, msg_type)
|
|
|
|
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 = "https://login.uber.com/oauth/v2/token"
|
|
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 get_uber_quote(self, pickup_address, dropoff_address, items=None):
|
|
"""Get delivery quote from Uber API"""
|
|
self.ensure_one()
|
|
access_token = self._get_access_token()
|
|
customer_id = self.customer_id
|
|
if not customer_id:
|
|
raise UserError(_("Uber Customer ID is missing in configuration."))
|
|
|
|
api_url = f"https://api.uber.com/v1/customers/{customer_id}/delivery_quotes"
|
|
headers = {
|
|
'Authorization': f'Bearer {access_token}',
|
|
'Content-Type': 'application/json'
|
|
}
|
|
|
|
# Ensure at least one dummy item if none provided (Uber Direct sometimes requires this)
|
|
if not items:
|
|
items = [{
|
|
"name": "Food Delivery",
|
|
"quantity": 1,
|
|
"size": "small"
|
|
}]
|
|
|
|
payload = {
|
|
"pickup_address": pickup_address,
|
|
"dropoff_address": dropoff_address,
|
|
"manifest_items": items
|
|
}
|
|
|
|
_logger.info("Uber Direct Payload: %s", json.dumps(payload, indent=2))
|
|
|
|
try:
|
|
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}"
|
|
|
|
return {
|
|
'success': False,
|
|
'error': msg,
|
|
'code': data.get('code', 'unknown'),
|
|
'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', 'USD'),
|
|
'estimated_arrival': data.get('estimated_arrival'),
|
|
'raw': data
|
|
}
|
|
except Exception as e:
|
|
_logger.exception("Uber Quote API Exception")
|
|
return {'success': False, '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,
|
|
}
|
|
}
|