demo/pycurl_demo.py

import pycurl
from io import BytesIO
import json

def embeddings_run(input, url="https://sanbo1200-jina-embeddings-v3.hf.space/api/v1/embeddings", model="jinaai/jina-embeddings-v3"):
    # 准备数据
    data = json.dumps({
        "input": input,
        "model": model
    })
    
    # 创建缓冲区存储响应
    buffer = BytesIO()
    
    # 初始化 pycurl
    c = pycurl.Curl()
    
    # 设置请求参数
    c.setopt(c.URL, url)
    c.setopt(c.WRITEDATA, buffer)
    c.setopt(c.POST, 1)
    c.setopt(c.POSTFIELDS, data)
    c.setopt(c.HTTPHEADER, [
        'Content-Type: application/json',
        f'Content-Length: {len(data)}'
    ])
    
    try:
        # 执行请求
        c.perform()
        
        # 检查状态码
        status_code = c.getinfo(pycurl.HTTP_CODE)
        if status_code == 200:
            # 获取响应数据
            response_data = buffer.getvalue().decode('utf-8')
            return json.loads(response_data)
        else:
            raise Exception(f"Request failed with status code: {status_code}")
    finally:
        c.close()
        buffer.close()

if __name__ == "__main__":
    input_text = "Your text string goes here"
    print(f"---{embeddings_run(input_text)}")