多模态文件操作

# coding=utf-8

import os
import json
import requests

from http import HTTPStatus

# 基础 URL
BASE_URL = "https://dashscope.aliyuncs.com/api/v2/apps/web-search-agent"

# 四个接口 URL
get_upload_url = f"{BASE_URL}/file/upload/apply"
upload_callback = f"{BASE_URL}/file/upload/callback"
get_file_info = f"{BASE_URL}/file/info/get"
delete_file = f"{BASE_URL}/file/delete"

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

# Agent 配置参数
AGENT_OPTIONS = {
    "agent_id": "aid-xxx",
}

def call_get_upload_url(session_id: str, files: list) -> dict:
    """
    获取文件上传 URL

    Args:
        session_id: 会话 ID
        files: 文件列表，每个文件包含 file_name, file_size, content_type, file_ext

    Returns:
        上传任务信息，包含 upload_tasks 列表
    """
    params = {
        "input": {
            "session_id": session_id,
            "files": files
        },
        "parameters": {
            "agent_options": AGENT_OPTIONS
        }
    }

    response = requests.post(get_upload_url, json=params, headers=headers)
    if response.status_code == HTTPStatus.OK:
        result = response.json()
        print("获取上传 URL 成功:")
        print(json.dumps(result, indent=2, ensure_ascii=False))
        return result["output"]["result"]
    else:
        print(f'获取上传 URL 失败: code={response.status_code}')
        print(f'response={response.text}')
        raise Exception(f"获取上传 URL 失败: {response.status_code}")

def call_upload_callback(session_id: str, file_id: str) -> dict:
    """
    文件上传完成后的回调

    Args:
        session_id: 会话 ID
        file_id: 文件 ID

    Returns:
        回调结果，包含 status
    """
    params = {
        "input": {
            "session_id": session_id,
            "file_id": file_id
        },
        "parameters": {
            "agent_options": AGENT_OPTIONS
        }
    }

    response = requests.post(upload_callback, json=params, headers=headers)
    if response.status_code == HTTPStatus.OK:
        result = response.json()
        print("上传回调成功:")
        print(json.dumps(result, indent=2, ensure_ascii=False))
        return result["output"]["result"]
    else:
        print(f'上传回调失败: code={response.status_code}')
        print(f'response={response.text}')
        raise Exception(f"上传回调失败: {response.status_code}")

def call_get_file_info(session_id: str, file_ids: list) -> dict:
    """
    获取文件信息

    Args:
        session_id: 会话 ID
        file_ids: 文件 ID 列表

    Returns:
        文件信息列表，包含 file_name, file_size, content_type, preview_url 等
    """
    params = {
        "input": {
            "session_id": session_id,
            "file_ids": file_ids
        },
        "parameters": {
            "agent_options": AGENT_OPTIONS
        }
    }

    response = requests.post(get_file_info, json=params, headers=headers)
    if response.status_code == HTTPStatus.OK:
        result = response.json()
        print("获取文件信息成功:")
        print(json.dumps(result, indent=2, ensure_ascii=False))
        return result["output"]["result"]
    else:
        print(f'获取文件信息失败: code={response.status_code}')
        print(f'response={response.text}')
        raise Exception(f"获取文件信息失败: {response.status_code}")

def call_delete_file(session_id: str, file_ids: list) -> dict:
    """
    删除文件

    Args:
        session_id: 会话 ID
        file_ids: 待删除的文件 ID 列表

    Returns:
        删除结果，包含 status
    """
    params = {
        "input": {
            "session_id": session_id,
            "file_ids": file_ids
        },
        "parameters": {
            "agent_options": AGENT_OPTIONS
        }
    }

    response = requests.post(delete_file, json=params, headers=headers)
    if response.status_code == HTTPStatus.OK:
        result = response.json()
        print("删除文件成功:")
        print(json.dumps(result, indent=2, ensure_ascii=False))
        return result["output"]["result"]
    else:
        print(f'删除文件失败: code={response.status_code}')
        print(f'response={response.text}')
        raise Exception(f"删除文件失败: {response.status_code}")

if __name__ == "__main__":
    # 示例用法
    SESSION_ID = "session-456"

    # 1. 获取上传 URL
    print("=" * 50)
    print("1. 获取上传 URL")
    print("=" * 50)
    upload_result = call_get_upload_url(SESSION_ID, [
        {
            "file_name": "test_1.jpg",
            "file_size": 1024000,
            "content_type": "image/jpeg",
            "file_ext": "jpg"
        }
    ])

    # 从结果中获取 upload_tasks
    upload_tasks = upload_result.get("upload_tasks", [])
    if upload_tasks:
        task = upload_tasks[0]
        file_id = task["file_id"]
        upload_url = task["upload_url"]
        print(f"\nfile_id: {file_id}")
        print(f"upload_url: {upload_url}")

        # 2. 实际上传文件（使用 PUT 方法上传到 OSS）
        print("\n" + "=" * 50)
        print("2. 上传文件到 OSS")
        print("=" * 50)
        # 这里需要实际的文件数据进行上传
        # with open("test_1.jpg", "rb") as f:
        #     response = requests.put(upload_url, data=f)
        #     if response.status_code == HTTPStatus.OK:
        #         print("文件上传成功")

        # 3. 上传完成后回调
        print("\n" + "=" * 50)
        print("3. 上传回调")
        print("=" * 50)
        callback_result = call_upload_callback(SESSION_ID, file_id)

        # 4. 获取文件信息
        print("\n" + "=" * 50)
        print("4. 获取文件信息")
        print("=" * 50)
        file_info = call_get_file_info(SESSION_ID, [file_id])

        # 5. 删除文件（可选）
        print("\n" + "=" * 50)
        print("5. 删除文件")
        print("=" * 50)
        delete_result = call_delete_file(SESSION_ID, [file_id])