api_demo.py

# -*- coding:utf-8 -*-
import time
import logging
import string
import random
import hmac
import base64
import hashlib

import requests

ssf_signature_len = 32
ssf_client_api = 'https://api.productai.cn'
ssf_client_version = '1'


logging.basicConfig(level=logging.DEBUG)


def short_uuid(length):
    charset = string.ascii_lowercase + string.digits
    return ''.join([random.choice(charset) for i in range(length)])


def client_headers(access_key_id, method='POST'):
    timestamp = int(time.time())
    headers = {
        'x-ca-accesskeyid': access_key_id,
        'x-ca-version': ssf_client_version,
        'x-ca-timestamp': str(timestamp),
        'x-ca-signaturenonce': short_uuid(ssf_signature_len),
        'x-ca-signature': '',
        'requestmethod': method,
    }
    return headers


def client_signature(headers, form, secret_key):
    assert isinstance(headers, dict)
    calc_params = dict(headers)

    if isinstance(form, dict):
        calc_params.update(form)

    sort_value = []
    for k in sorted(calc_params):
        if k not in ['x-ca-signature']:
            v = str(calc_params.get(k, '')).strip()
            value = '{}={}'.format(k, v)
            sort_value.append(value)

    string_to_signature = '&'.join(sort_value)
    signature = hmac.new(
        secret_key.encode('ascii'),
        string_to_signature.encode('ascii'),
        hashlib.sha1
    )
    return base64.b64encode(signature.digest())


def submit_file_to_search(form, files):
    post_form = {}
    loc = form.get('loc')
    if loc:
        post_form['loc'] = loc
    count = form.get('count')
    if count:
        post_form['count'] = count
    url = form.get('image_url')
    if url:
        post_form['url'] = url
    headers = client_headers(
        form.get('access_key_id', ''),
        form.get('request_method', 'post')
    )
    signature = client_signature(headers, post_form, form.get('secret_key', ''))
    headers['x-ca-signature'] = signature
    headers['x-ca-file-md5'] = hashlib.md5(files['search']).hexdigest()
    reps = requests.request(
        method=form.get('request_method', 'post'),
        url='{}/{}/{}/'.format(ssf_client_api, form.get('service_type'), form.get('service_id')),
        data=post_form,
        files=files,
        headers=headers,
        verify=False,
        timeout=50,
    )
    return reps


def submit_form_to_search(form):
    post_form = {}
    loc = form.get('loc', None)
    if loc:
        post_form.setdefault('loc', loc)
    url = form.get('image_url', None)
    if url:
        post_form.setdefault('url', url)
    headers = client_headers(form.get('access_key_id', ''), form.get('request_method', 'post'))
    signature = client_signature(headers, post_form, form.get('secret_key', ''))
    headers['x-ca-signature'] = signature
    reps = requests.request(
        method=form.get('request_method', 'post'),
        url='{}/{}/{}/'.format(ssf_client_api, form.get('service_type'), form.get('service_id')),
        data=post_form,
        headers=headers,
        verify=False,
        timeout=50,
    )
    return reps


def add_image_to_image_set(form):
    request_method = 'post'
    access_key_id = form.get('access_key_id', '')
    secret_key = form.get('secret_key', '')
    service_type = 'image_sets'
    service_id = '_0000014'
    image_url = form.get('image_url')
    meta = form.get('meta')
    image_set_id = form.get('image_set_id')
    pai_user_id = form.get('pai_user_id')

    post_form = dict(image_url=image_url, meta=meta)
    api_url = '/'.join([ssf_client_api, service_type, service_id, image_set_id])
    headers = client_headers(access_key_id, request_method)
    signature = client_signature(headers, post_form, secret_key)
    headers['x-ca-signature'] = signature
    # only some headers are counted into signature.
    headers['pai_user_id'] = str(pai_user_id)

    reps = requests.request(request_method,
                            api_url,
                            data=post_form,
                            headers=headers,
                            verify=False)
    return reps


def delete_image_by_file(form):
    request_method = 'post'
    access_key_id = form.get('access_key_id', '')
    secret_key = form.get('secret_key', '')
    service_type = 'image_sets'
    service_id = '_0000014'
    image_set_id = form.get('image_set_id')
    pai_user_id = form.get('pai_user_id')

    file_path = form.get('file_path')
    urls_to_delete = {'urls_to_delete': open(file_path, 'rb')}
    api_url = '/'.join([ssf_client_api, service_type, service_id, image_set_id])
    headers = client_headers(access_key_id, request_method)
    signature = client_signature(headers, {}, secret_key)
    headers['x-ca-signature'] = signature
    # only some headers are counted into signature.
    headers['pai_user_id'] = str(pai_user_id)

    print(api_url)
    reps = requests.request(request_method,
                            api_url,
                            headers=headers,
                            files=urls_to_delete,
                            verify=False)
    return reps


if __name__ == '__main__':
    # use url for search
    resp = submit_form_to_search(
        dict(
            request_method='post',                # 默认 的请求方法 post
            service_type='',     # require: 你的服务类型 可在 我的服务-测试服务 页面查看
            access_key_id='',   # require: 你的用户配置  access_key_id
            secret_key='',         # require: 你的用户配置  secret_key
            service_id='',         # require: 你的服务ID
            image_url='',           # require: 图片的url
            #loc='0-0-1-1'                        # option:  (选填) 图片标框的位置信息  log = '-'.join([ix,iy,iw,ih])
        )
    )
    print(resp.json())