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 hashlib
import json
import os
import time

import requests
from flask import current_app

from shopdb.api import Setting

# Two-level Dell token cache. Dell rate-limits the token endpoint, so a fresh
# request per refresh (or per app restart) trips a 401 cooldown. Tokens live ~1h.
# L1 = process memory (fast); L2 = a file in the instance dir so the token
# survives restarts and is shared across workers - mirrors warranty_sync.py.
_dell_token_cache = {}


def _token_file(clientid):
    # Per-client file, name keyed by a hash so the id never lands on disk.
    tag = hashlib.sha256((clientid or '').encode()).hexdigest()[:16]
    return os.path.join(current_app.instance_path, f'.dell_token_{tag}.json')


def _read_file_token(clientid):
    try:
        with open(_token_file(clientid)) as handle:
            cached = json.load(handle)
        if cached.get('expires_at', 0) > time.time() + 60:
            return cached['token']
    except (OSError, ValueError, KeyError):
        pass
    return None


def _write_file_token(clientid, token, expires_at):
    try:
        os.makedirs(current_app.instance_path, exist_ok=True)
        with open(_token_file(clientid), 'w') as handle:
            json.dump({'token': token, 'expires_at': expires_at}, handle)
    except OSError:
        pass

# 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):
        clientid = config['clientid']
        # L1: process memory.
        cached = _dell_token_cache.get(clientid)
        if cached and cached['expires_at'] > time.time() + 60:
            return cached['token']
        # L2: file cache (survives restarts / shared across workers).
        file_token = _read_file_token(clientid)
        if file_token:
            return file_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')
        expires_at = time.time() + expires_in - 60
        _dell_token_cache[clientid] = {'token': token, 'expires_at': expires_at}
        _write_file_token(clientid, token, expires_at)
        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))

    def bulk_lookup(self, servicetags):
        """Look up many tags at once. Dell takes up to 100 tags per call.

        Returns {SERVICETAG_UPPER: {servicelevel, startdate, enddate}} for tags
        Dell recognizes; unknown/non-Dell tags are simply absent.
        """
        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.'
            )
        tags = [t.strip() for t in servicetags if t and t.strip()]
        if not tags:
            return {}
        token = self._get_token(config)
        results = {}
        for start in range(0, len(tags), 100):
            chunk = tags[start:start + 100]
            payload = self._fetch(config, token, ','.join(chunk))
            assets = payload if isinstance(payload, list) else [payload]
            for asset in assets:
                tag = (asset.get('serviceTag') or '').upper()
                mapped = self._map([asset])
                if tag and mapped:
                    results[tag] = mapped
        return results


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())