| import asyncio |
| import cloudinary |
| import cloudinary.uploader |
| import cloudinary.api |
| from phi.tools import Toolkit |
| from phi.utils.log import logger |
|
|
| class CloudinaryTool(Toolkit): |
| def __init__( |
| self, |
| cloud_name: str, |
| api_key: str, |
| api_secret: str, |
| ): |
| super().__init__(name="cloudinary_tool") |
|
|
| |
| cloudinary.config( |
| cloud_name=cloud_name, |
| api_key=api_key, |
| api_secret=api_secret |
| ) |
|
|
| |
| self.register(self.upload_file) |
|
|
| def upload_file(self, file_path: str) -> str: |
| """ |
| Uploads a file to Cloudinary. |
| |
| Args: |
| file_path (str): The path to the file to upload. |
| |
| Returns: |
| dict: A dictionary containing the upload response details. |
| |
| Example: |
| >>> upload_file('/path/to/file.pdf') |
| """ |
| logger.info(f"Uploading file to Cloudinary: {file_path}") |
| try: |
| response = cloudinary.uploader.upload(file_path, resource_type="raw") |
| logger.info(f"Upload response: {response}") |
| return f"{response}" |
| except Exception as e: |
| logger.warning(f"Failed to upload file: {e}") |
| return f"Error: {e}" |
|
|
| |
| |
| |
| |
| |
| |
| |
| |
| |
|
|
