model/conan_api_client.py

import os
import hashlib
import random
import string
import time
import requests
import json

class ConanClient:
    def __init__(self, ak, sk, url, timeout=30):
        """
        Initialize the Client

        Args:
            ak: Access Key
            sk: Secret Key
            url: Server URL
            timeout: Request timeout in seconds (default: 30)
        """
        self.ak = ak
        self.sk = sk
        self.url = url
        self.timeout = timeout

    def __random_password(self, size=40, chars=None):
        if chars is None:
            chars = string.ascii_uppercase + string.ascii_lowercase + string.digits
        random_chars = random.SystemRandom().choice
        return "".join(random_chars(chars) for _ in range(size))

    def __signature(self, random_str, time_stamp):
        params_str = "%s:%d:%s:%s" % (self.ak, time_stamp, random_str, self.sk)
        encoded_params_str = params_str.encode("utf-8")
        return hashlib.md5(encoded_params_str).hexdigest()

    def get_signature(self):
        timestamp = int(time.time())
        random_str = self.__random_password(20)
        sig = self.__signature(random_str, timestamp)
        params = {
            "timestamp": timestamp,
            "random": random_str,
            "app_id": self.ak,
            "sign": sig,
        }
        return params

    def embed(self, text):
        """
        Embed text using the server

        Args:
            text: The input text to embed

        Returns:
            requests.Response object
        """
        params = self.get_signature()
        params["body"] = text
        params["content_id"] = f"test_{int(time.time())}"

        headers = {"Content-Type": "application/json"}

        rsp = requests.post(self.url, data=json.dumps(params), timeout=self.timeout, headers=headers)
        result = json.loads(rsp.text)
        return result