异步请求

import json
import aiohttp
import asyncio
import base64

class AsyncExtractClient:
    def __init__(self, app_id: str, secret_code: str):
        self.app_id = app_id
        self.secret_code = secret_code

    async def extract(self, file_content: bytes, options: dict, request_body: dict) -> str:
        # 将文件内容转换为base64
        file_base64 = base64.b64encode(file_content).decode('utf-8')

        # 构建请求参数
        params = {}
        for key, value in options.items():
            params[key] = str(value)

        # 设置请求头
        headers = {
            "x-ti-app-id": self.app_id,
            "x-ti-secret-code": self.secret_code,
            "Content-Type": "application/json"
        }

        # 构建完整的请求体，包含base64文件数据
        full_request_body = {
            **request_body,
            "file": file_base64  # 添加base64编码的文件数据
        }

        # 发送异步请求
        async with aiohttp.ClientSession() as session:
            async with session.post(
                "https://api.textin.com/ai/service/v2/entity_extraction",
                params=params,
                headers=headers,
                json=full_request_body
            ) as response:
                # 检查响应状态
                response.raise_for_status()
                return await response.text()

async def main():
    # 创建客户端实例，需替换你的API Key
    client = AsyncExtractClient("你的x-ti-app-id", "你的x-ti-secret-code")

    # 定义文件路径，替换为你的文件
    file_path = "你的文件.pdf"

    # 读取本地文件
    with open(file_path, "rb") as f:
        file_content = f.read()

    # 设置URL参数，可按需设置，详情见URL参数说明
    options = dict(
        parse_mode="scan"
    )

    # 设置请求体参数，prompt模式
    request_body = dict(
        prompt="请抽取这张小票中的实付金额、消费日期、店铺名称、订单号并以字段格式返回，请抽取货号、商品名称、数量、单价，并以表格格式返回"
    )

    try:
        response = await client.extract(file_content, options, request_body)

        # 保存完整的JSON响应到result.json文件
        with open("prompt_extract_result.json", "w", encoding="utf-8") as f:
            f.write(response)

        print(response)
    except Exception as e:
        print(f"Error: {e}")

if __name__ == "__main__":
    # 运行异步主函数
    asyncio.run(main())