""" title: Knowledgebase File Tools author: Wes Caldwell email: musicheardworldwide.com author_url: https://github.com/musicheardworldwide version: 0.2.0 description: A tool for managing and rendering files with expanded features for better usability. """ import os import requests import logging from datetime import datetime from typing import List, Dict, Optional, Callable # Configure logging logging.basicConfig( level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s" ) class Tools: def __init__(self): """ Initializes the Tools class with default settings. """ # Enable or disable specific features self.file_handler = True self.citation = True def get_files(self, __files__: List[dict] = []) -> str: """ Fetch the list of files and generate rendering instructions. """ logging.info("Fetching file list.") return ( """Show the file content directly using: `/api/v1/files/{file_id}/content` If the file is video content render the video directly using the following template: {{VIDEO_FILE_ID_[file_id]}} If the file is HTML file render the HTML directly as iframe using the following template: {{HTML_FILE_ID_[file_id]}}""" + f"\nFiles: {str(__files__)}" ) def upload_file(self, file_path: str) -> Dict[str, str]: """ Upload a file to the server. Parameters: - file_path (str): The path to the file to upload. Returns: - dict: The server response or an error message. """ url = "/api/v1/files/" headers = {"Authorization": "Bearer YOUR_API_KEY"} try: with open(file_path, "rb") as file: files = {"file": (os.path.basename(file_path), file)} response = requests.post(url, headers=headers, files=files) response.raise_for_status() logging.info(f"File uploaded successfully: {file_path}") return response.json() except Exception as e: logging.error(f"Error uploading file: {str(e)}") return {"error": str(e)} def delete_file(self, file_id: str) -> Dict[str, str]: """ Delete a file by its ID. Parameters: - file_id (str): The ID of the file to delete. Returns: - dict: The server response or an error message. """ url = f"/api/v1/files/{file_id}" headers = {"Authorization": "Bearer YOUR_API_KEY"} try: response = requests.delete(url, headers=headers) response.raise_for_status() logging.info(f"File {file_id} deleted successfully.") return { "status": "success", "message": f"File {file_id} deleted successfully", } except Exception as e: logging.error(f"Error deleting file: {str(e)}") return {"error": str(e)} def search_files(self, query: str) -> List[dict]: """ Search files by query. Parameters: - query (str): The search query. Returns: - list: A list of files matching the query or an error message. """ url = "/api/v1/files/search" params = {"q": query} headers = {"Authorization": "Bearer YOUR_API_KEY"} try: response = requests.get(url, headers=headers, params=params) response.raise_for_status() logging.info(f"Search completed for query: {query}") return response.json() except Exception as e: logging.error(f"Error searching files: {str(e)}") return [{"error": str(e)}] def render_file(self, file_id: str, file_type: str) -> str: """ Render files dynamically based on their type. Parameters: - file_id (str): The ID of the file. - file_type (str): The type of the file (e.g., video, html, pdf, text). Returns: - str: The rendering template. """ templates = { "video": f"{{VIDEO_FILE_ID_{file_id}}}", "html": f"{{HTML_FILE_ID_{file_id}}}", "pdf": f"{{PDF_FILE_ID_{file_id}}}", "text": f"{{TEXT_FILE_ID_{file_id}}}", } return templates.get(file_type, f"Unsupported file type: {file_type}") async def upload_with_progress( self, file_path: str, __event_emitter__: Callable[[dict], None] ): """ Upload a file with progress updates. Parameters: - file_path (str): The path to the file to upload. - __event_emitter__ (Callable): An event emitter for sending real-time updates. """ await __event_emitter__({"status": "start", "description": "Uploading file..."}) result = self.upload_file(file_path) if "error" in result: await __event_emitter__({"status": "error", "description": result["error"]}) else: await __event_emitter__( {"status": "success", "description": "File uploaded successfully."} ) def add_to_collection(self, collection_id: str, file_id: str) -> Dict[str, str]: """ Add a file to a collection. Parameters: - collection_id (str): The ID of the collection. - file_id (str): The ID of the file. Returns: - dict: The server response or an error message. """ url = f"/api/v1/knowledge/{collection_id}/file/add" headers = {"Authorization": "Bearer YOUR_API_KEY"} try: response = requests.post(url, headers=headers, json={"file_id": file_id}) response.raise_for_status() logging.info(f"File {file_id} added to collection {collection_id}.") return response.json() except Exception as e: logging.error(f"Error adding file to collection: {str(e)}") return {"error": str(e)} # Example usage if __name__ == "__main__": tools = Tools() files = tools.get_files([{"id": "123", "name": "example.txt"}]) print("Available Files:", files) upload_result = tools.upload_file("example.txt") print("Upload Result:", upload_result) search_result = tools.search_files("example") print("Search Result:", search_result)