上传文件

# coding=utf-8

import os
import sys
import uuid
import json
import time
import hashlib
import requests

from http import HTTPStatus

apply_lease_url = "https://dashscope.aliyuncs.com/api/v2/apps/deep-search-agent/file/lease/apply"
submit_file_url = "https://dashscope.aliyuncs.com/api/v2/apps/deep-search-agent/file/parse/submit"
parse_status_url = "https://dashscope.aliyuncs.com/api/v2/apps/deep-search-agent/file/parse/status"

headers = {
    'Authorization': f'Bearer {os.getenv("DASHSCOPE_API_KEY", "")}',  # 配置 API KEY
    'Content-Type': 'application/json'
}

def get_file_upload_info(file_path: str) -> dict:
    if not os.path.exists(file_path):
        raise Exception("file not exists")
    file_name = os.path.basename(file_path)
    file_size = os.path.getsize(file_path)
    with open(file_path, 'rb') as f:
        md5 = hashlib.md5()
        while True:
            data = f.read(4096)
            if not data:
                break
            md5.update(data)
    file_md5 = md5.hexdigest()
    return {"file_name": file_name, "file_size": file_size, "md5": file_md5}

if __name__ == "__main__":
    if len(sys.argv) != 2:
        print("usage: python deep_search_demo.py <file_path>")
        exit(1)

    file_path = sys.argv[1]
    file_info = get_file_upload_info(file_path)
    file_name: str = file_info["file_name"]
    file_size: int = file_info["file_size"]
    file_md5: str = file_info["md5"]

    # 1. 申请租约
    apply_lease_params = {
        "request_id": str(uuid.uuid4()),
        "input": {
            "file_name": file_name,
            "file_size": file_size,
            "file_md5": file_md5
        },
        "parameters": {}
    }
    response = requests.post(apply_lease_url, json=apply_lease_params, headers=headers)
    lease_data = {}
    if response.status_code == HTTPStatus.OK:
        print("1. Lease applied successfully.")
        lease_data = response.json()["output"]["result"]
        print(json.dumps(lease_data, indent=2, ensure_ascii=False))
    else:
        print(f'code={response.status_code}')
        exit(1)

    # 2. 上传文件
    upload_headers = lease_data["headers"]
    lease_id = lease_data["lease_id"]
    pre_signed_url = lease_data["pre_signed_url"]
    with open(file_path, 'rb') as file:
        # 下方设置请求方法用于文档上传，需与您在上一步中调用ApplyFileUploadLease接口实际返回的Data.Param中Method字段的值一致
        response = requests.put(pre_signed_url, data=file, headers=upload_headers)
    if response.status_code == HTTPStatus.OK:
        print("2. File uploaded successfully.")
    else:
        print(f"Failed to upload the file. ResponseCode: {response.status_code}")
        exit(1)

    # 3.提交解析
    submit_file_params = {
        "request_id": str(uuid.uuid4()),
        "input": {
            "lease_id": lease_data["lease_id"]
        },
        "parameters": {}
    }
    response = requests.post(submit_file_url, json=submit_file_params, headers=headers)
    if response.status_code == HTTPStatus.OK:
        print("3. File submit parse successfully.")
        submit_data = response.json()["output"]["result"][0]
        print(json.dumps(response.json(), indent=2, ensure_ascii=False))
    else:
        submit_data = response.json()
        print(f'response={response.json()}')
        print(f'code={response.status_code}')
        exit(1)

    # 4.获取解析状态
    parse_status_params = {
        "request_id": str(uuid.uuid4()),
        "input": {
            "file_ids": [
                submit_data["file_id"]
            ]
        },
        "parameters": {}
    }
    _start_time = time.time()
    while True:
        response = requests.post(parse_status_url, json=parse_status_params, headers=headers)
        _process_time = time.time() - _start_time
        if response.status_code == HTTPStatus.OK:
            parse_status = response.json()["output"]["result"]
            print(json.dumps(parse_status, indent=2, ensure_ascii=False))
            file_status = parse_status[0]["status"]
            print(f"4. Check file process status: {file_status}. process time: {_process_time}")
            if file_status == "FILE_IS_READY":
                print("5. File process successfully.")
                exit(0)
        else:
            print(f'response={response.json()}')
            print(f'code={response.status_code}')
        try:
            if _process_time > 1800:
                print(f'file process timeout[>1800s]: {_process_time}')
                exit(1)
            time.sleep(10)
        except Exception as e:
            pass