Multimodal file operations

# coding=utf-8

import os
import json
import requests

from http import HTTPStatus

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

# Four API endpoint URLs
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"

# request header
headers = {
    'Authorization': f'Bearer {os.getenv("DASHSCOPE_API_KEY", "")}',  # Configure your API key
    'Content-Type': 'application/json'
}

# agent options
AGENT_OPTIONS = {
    "agent_id": "aid-xxx",
}

def call_get_upload_url(session_id: str, files: list) -> dict:
    """Gets the file upload URL.

    Args:
        session_id: The session ID.
        files: A list of files. Each file contains file_name, file_size, content_type, and file_ext.

    Returns:
        The upload task information, which contains the upload_tasks list.
    """
    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("Successfully retrieved the upload URL:")
        print(json.dumps(result, indent=2, ensure_ascii=False))
        return result["output"]["result"]
    else:
        print(f'Failed to get the upload URL: code={response.status_code}')
        print(f'response={response.text}')
        raise Exception(f"Failed to get the upload URL: {response.status_code}")

def call_upload_callback(session_id: str, file_id: str) -> dict:
    """Performs the upload callback after a file is uploaded.

    Args:
        session_id: The session ID.
        file_id: The file ID.

    Returns:
        The callback result, which contains the 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("Upload callback successful:")
        print(json.dumps(result, indent=2, ensure_ascii=False))
        return result["output"]["result"]
    else:
        print(f'Upload callback failed: code={response.status_code}')
        print(f'response={response.text}')
        raise Exception(f"Upload callback failed: {response.status_code}")

def call_get_file_info(session_id: str, file_ids: list) -> dict:
    """Gets file info.

    Args:
        session_id: The session ID.
        file_ids: A list of file IDs.

    Returns:
        A list of dictionaries containing file information, including file_name, file_size, content_type, preview_url, and more.
    """
    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("Successfully retrieved file info:")
        print(json.dumps(result, indent=2, ensure_ascii=False))
        return result["output"]["result"]
    else:
        print(f'Failed to get file info: code={response.status_code}')
        print(f'response={response.text}')
        raise Exception(f"Failed to get file info: {response.status_code}")

def call_delete_file(session_id: str, file_ids: list) -> dict:
    """Deletes files.

    Args:
        session_id: The session ID.
        file_ids: A list of file IDs to delete.

    Returns:
        The deletion result, which contains the 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("File deletion successful:")
        print(json.dumps(result, indent=2, ensure_ascii=False))
        return result["output"]["result"]
    else:
        print(f'File deletion failed: code={response.status_code}')
        print(f'response={response.text}')
        raise Exception(f"File deletion failed: {response.status_code}")

if __name__ == "__main__":
    # Example usage
    SESSION_ID = "session-456"

    # 1. Get File Upload URL
    print("=" * 50)
    print("1. Get File Upload 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"
        }
    ])

    # Get upload_tasks from the result
    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. Upload file to OSS (use the PUT method)
        print("\n" + "=" * 50)
        print("2. Upload File to OSS")
        print("=" * 50)
        # You must use an actual file for the upload.
        # with open("test_1.jpg", "rb") as f:
        #     response = requests.put(upload_url, data=f)
        #     if response.status_code == HTTPStatus.OK:
        #         print("File uploaded successfully")

        # 3. Upload callback
        print("\n" + "=" * 50)
        print("3. Upload callback")
        print("=" * 50)
        callback_result = call_upload_callback(SESSION_ID, file_id)

        # 4. Get file info
        print("\n" + "=" * 50)
        print("4. Get file info")
        print("=" * 50)
        file_info = call_get_file_info(SESSION_ID, [file_id])

        # 5. Delete file (optional)
        print("\n" + "=" * 50)
        print("5. Delete file (optional)")
        print("=" * 50)
        delete_result = call_delete_file(SESSION_ID, [file_id])