Clarifai API

"""Smart Tools starter: one versioned Clarifai image concept classifier."""
import base64
import json
import math
import os
import sys
from datetime import datetime, timezone
from pathlib import Path
from time import monotonic
from urllib.error import HTTPError, URLError
from urllib.parse import quote
from urllib.request import Request, urlopen


class ProviderFailure(Exception):
    pass


def normalize(raw):
    if not isinstance(raw, dict) or raw.get('status', {}).get('code') != 10000:
        raise ProviderFailure()
    outputs = raw.get('outputs')
    if not isinstance(outputs, list) or len(outputs) != 1:
        raise ValueError('unexpected_schema')
    output = outputs[0]
    if output.get('status', {}).get('code') != 10000:
        raise ProviderFailure()
    concepts = output.get('data', {}).get('concepts')
    if not isinstance(concepts, list):
        raise ValueError('unexpected_schema')
    items = []
    for concept in concepts:
        score = concept.get('value')
        if not isinstance(concept.get('name'), str) or type(score) not in (int, float) or not math.isfinite(score) or not 0 <= score <= 1:
            raise ValueError('unexpected_schema')
        items.append({'label': concept['name'], 'score': score})
    return items


def run():
    start = monotonic()
    result = {'schema_version': '1.0.0', 'provider': 'Clarifai', 'status': 'error',
              'model_version': None, 'data': None, 'error': None,
              'captured_at': datetime.now(timezone.utc).isoformat()}
    try:
        keys = ('CLARIFAI_PAT', 'CLARIFAI_USER_ID', 'CLARIFAI_APP_ID', 'CLARIFAI_MODEL_ID', 'CLARIFAI_MODEL_VERSION', 'IMAGE_PATH')
        cfg = {key: os.environ[key] for key in keys}
        if not all(cfg.values()):
            raise ValueError('configuration')
        content = Path(cfg['IMAGE_PATH']).read_bytes()
        if not content or len(content) > 5 * 1024 * 1024:
            raise ValueError('starter_image_limit')
        # 5 MiB is this starter's local cap, not a quoted provider limit.
        model, revision = quote(cfg['CLARIFAI_MODEL_ID'], safe=''), quote(cfg['CLARIFAI_MODEL_VERSION'], safe='')
        result['model_version'] = cfg['CLARIFAI_MODEL_VERSION']
        payload = {'user_app_id': {'user_id': cfg['CLARIFAI_USER_ID'], 'app_id': cfg['CLARIFAI_APP_ID']},
                   'inputs': [{'data': {'image': {'base64': base64.b64encode(content).decode('ascii')}}}]}
        request = Request(f'https://api.clarifai.com/v2/models/{model}/versions/{revision}/outputs',
                          data=json.dumps(payload).encode(), method='POST',
                          headers={'Authorization': 'Key ' + cfg['CLARIFAI_PAT'], 'Content-Type': 'application/json'})
        with urlopen(request, timeout=30) as response:
            raw = json.load(response)
        result['data'] = normalize(raw)
        result['status'] = 'ok'
    except HTTPError as exc:
        result['error'] = 'http_' + str(exc.code)
    except (URLError, TimeoutError):
        result['error'] = 'transport_failure'
    except ProviderFailure:
        result['error'] = 'provider_status_failure'
    except (KeyError, OSError):
        result['error'] = 'configuration'
    except (ValueError, TypeError, AttributeError):
        result['error'] = 'configuration_or_schema'
    result['elapsed_ms'] = round((monotonic() - start) * 1000)
    return result


if __name__ == '__main__':
    output = run()
    print(json.dumps(output, ensure_ascii=False, allow_nan=False))
    sys.exit(0 if output['status'] == 'ok' else 1)