shopdb-flask/plugins/warranty/services/providers.py

"""Warranty provider abstraction.

A provider looks up coverage for a unit by service tag / serial. Manual entry
needs no provider. Dell is a real integration (Dell TechDirect Warranty API,
OAuth2 client-credentials); Lenovo/HP remain config-shaped stubs until wired.

Per-vendor config lives in settings (warranty_<name>_*), masked where secret,
so nothing is hardcoded and the integration is toggled per site.
"""

import time

import requests

from shopdb.api import db
from shopdb.core.models import Setting

# Process-wide Dell token cache. Dell rate-limits the token endpoint, so a fresh
# request per refresh trips a 401 cooldown. Tokens live ~1h; cache + reuse.
_dell_token_cache = {}

# Dell TechDirect defaults, overridable via settings if Dell changes endpoints.
# The asset-entitlements service lives under sbil/eapi (device.warranty path 404s
# with "Service Not Found" for this account). Verified against the live API.
DELL_TOKEN_URL = 'https://apigtwb2c.us.dell.com/auth/oauth/v2/token'
DELL_API_URL = 'https://apigtwb2c.us.dell.com/PROD/sbil/eapi/v5/asset-entitlements'

HTTP_TIMEOUT = 15  # seconds


class ProviderNotConfigured(Exception):
    """Raised when a provider is asked to look up but has no API config."""
    pass


class WarrantyLookupError(Exception):
    """Raised when a configured provider's lookup fails at runtime."""
    pass


def _setting(key, default=None):
    row = Setting.query.filter_by(key=key).first()
    return row.value if row and row.value not in (None, '') else default


def _flag(key):
    return str(_setting(key, 'false')).lower() == 'true'


class WarrantyProvider:
    """Base provider. name matches Warranty.provider values."""
    name = 'base'

    def lookup(self, servicetag, vendor=None):
        """Return {servicelevel, startdate, enddate} or None.

        Raises ProviderNotConfigured when creds are missing, WarrantyLookupError
        when a configured lookup fails.
        """
        raise NotImplementedError


class ManualProvider(WarrantyProvider):
    """No external lookup - values are entered by hand."""
    name = 'manual'

    def lookup(self, servicetag, vendor=None):
        return None


class DellProvider(WarrantyProvider):
    """Dell TechDirect Warranty API v5 (OAuth2 client-credentials)."""
    name = 'dell'

    def _config(self):
        return {
            'enabled': _flag('warranty_dell_enabled'),
            'clientid': _setting('warranty_dell_clientid'),
            'clientsecret': _setting('warranty_dell_clientsecret'),
            'tokenurl': _setting('warranty_dell_tokenurl', DELL_TOKEN_URL),
            'apiurl': _setting('warranty_dell_apiurl', DELL_API_URL),
        }

    def _get_token(self, config):
        # Reuse a cached token while valid - Dell rate-limits the token endpoint.
        cached = _dell_token_cache.get(config['clientid'])
        if cached and cached['expires_at'] > time.time() + 60:
            return cached['token']

        # Dell expects the client id/secret as HTTP Basic auth, grant type in the
        # body. (Passing them in the body trips the token endpoint's rate limiter.)
        try:
            response = requests.post(
                config['tokenurl'],
                auth=(config['clientid'], config['clientsecret']),
                data={'grant_type': 'client_credentials'},
                timeout=HTTP_TIMEOUT,
            )
        except requests.RequestException as exc:
            raise WarrantyLookupError(f'Dell auth failed: {exc}')
        if response.status_code in (401, 429):
            raise WarrantyLookupError(
                'Dell token endpoint is rate-limiting (cooldown). Wait a few '
                'minutes and try again - it caches once obtained.'
            )
        try:
            response.raise_for_status()
            token = response.json().get('access_token')
            expires_in = int(response.json().get('expires_in', 3600))
        except (requests.RequestException, ValueError) as exc:
            raise WarrantyLookupError(f'Dell auth failed: {exc}')
        if not token:
            raise WarrantyLookupError('Dell auth returned no access_token')
        _dell_token_cache[config['clientid']] = {
            'token': token, 'expires_at': time.time() + expires_in - 60,
        }
        return token

    def _fetch(self, config, token, servicetag):
        try:
            response = requests.get(
                config['apiurl'],
                params={'servicetags': servicetag},
                headers={'Authorization': f'Bearer {token}', 'Accept': 'application/json'},
                timeout=HTTP_TIMEOUT,
            )
            response.raise_for_status()
            return response.json()
        except (requests.RequestException, ValueError) as exc:
            raise WarrantyLookupError(f'Dell warranty lookup failed: {exc}')

    @staticmethod
    def _map(payload):
        """Reduce the Dell response to {servicelevel, startdate, enddate}.

        Dell returns a list of assets, each with an 'entitlements' list. Coverage
        spans the earliest start to the latest end across entitlements.
        """
        assets = payload if isinstance(payload, list) else [payload]
        entitlements = []
        for asset in assets:
            if asset.get('invalid'):
                continue
            entitlements.extend(asset.get('entitlements') or [])
        # Only entitlements with an end date define coverage.
        dated = [e for e in entitlements if e.get('endDate')]
        if not dated:
            return None
        starts = [e['startDate'] for e in entitlements if e.get('startDate')]
        latest = max(dated, key=lambda e: e['endDate'])
        return {
            'servicelevel': latest.get('serviceLevelDescription') or latest.get('serviceLevelCode'),
            'startdate': min(starts)[:10] if starts else None,
            'enddate': max(e['endDate'] for e in dated)[:10],
        }

    def lookup(self, servicetag, vendor=None):
        config = self._config()
        if not (config['enabled'] and config['clientid'] and config['clientsecret']):
            raise ProviderNotConfigured(
                'Dell warranty lookup is not configured. Set warranty_dell_enabled, '
                'clientid and clientsecret in Settings > Integrations.'
            )
        if not servicetag:
            raise WarrantyLookupError('Dell lookup needs a service tag on the warranty.')
        token = self._get_token(config)
        return self._map(self._fetch(config, token, servicetag))


class ApiProvider(WarrantyProvider):
    """Generic config-shaped stub for vendors not yet wired (Lenovo, HP)."""

    def lookup(self, servicetag, vendor=None):
        enabled = _flag(f'warranty_{self.name}_enabled')
        url = _setting(f'warranty_{self.name}_apiurl')
        token = _setting(f'warranty_{self.name}_apitoken')
        if not (enabled and url and token):
            raise ProviderNotConfigured(
                f'{self.name} warranty lookup is not configured.'
            )
        raise ProviderNotConfigured(
            f'{self.name} API lookup not implemented yet (config present).'
        )


class LenovoProvider(ApiProvider):
    name = 'lenovo'


class HpProvider(ApiProvider):
    name = 'hp'


_PROVIDERS = {p.name: p for p in (
    ManualProvider(), DellProvider(), LenovoProvider(), HpProvider()
)}


def get_provider(name):
    """Return a provider instance, defaulting to manual for unknown names."""
    return _PROVIDERS.get((name or 'manual').lower(), _PROVIDERS['manual'])


def provider_names():
    return list(_PROVIDERS.keys())