odoo-chennora-pos/addons/dine360_uber/models/uber_config.py

from odoo import models, fields, api, _
from odoo.exceptions import UserError
import requests
import json
import datetime

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 _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,
            }
        }