RagFlow_Plugs/utils/ragflow/file_service.py

from typing import Dict, Any, List, Optional

class FileService:
    def __init__(self, http_client):
        self.http_client = http_client
    
    def list_files(self, parent_id: str = None, keywords: str = None,
                  page: int = 1, size: int = 20, orderby: str = "create_time",
                  desc: bool = True) -> List[Dict[str, Any]]:
        endpoint = "/api/v1/file/list"
        
        params = {"page": page, "page_size": size, "orderby": orderby, "desc": int(desc)}
        if parent_id is not None:
            params["parent_id"] = parent_id
        if keywords is not None:
            params["keywords"] = keywords
        
        response = self.http_client.get(endpoint, params=params)
        
        if response.get("code") == 0 and response.get("data"):
            return response["data"]
        else:
            raise Exception(f"列出文件失败: {response.get('message', '未知错误')}")
    
    def get_root_folder(self) -> Dict[str, Any]:
        endpoint = "/api/v1/file/root_folder"
        
        response = self.http_client.get(endpoint)
        
        if response.get("code") == 0 and response.get("data"):
            return response["data"]
        else:
            raise Exception(f"获取根目录失败: {response.get('message', '未知错误')}")
    
    def get_parent_folder(self, file_id: str) -> Dict[str, Any]:
        endpoint = "/api/v1/file/parent_folder"
        
        response = self.http_client.get(endpoint, params={"file_id": file_id})
        
        if response.get("code") == 0 and response.get("data"):
            return response["data"]
        else:
            raise Exception(f"获取父目录失败: {response.get('message', '未知错误')}")
    
    def get_all_parent_folders(self, file_id: str) -> List[Dict[str, Any]]:
        endpoint = "/api/v1/file/all_parent_folder"
        
        response = self.http_client.get(endpoint, params={"file_id": file_id})
        
        if response.get("code") == 0 and response.get("data"):
            return response["data"]
        else:
            raise Exception(f"获取所有父目录失败: {response.get('message', '未知错误')}")
    
    def get_file(self, file_id: str) -> Dict[str, Any]:
        endpoint = f"/api/v1/file/get/{file_id}"
        
        response = self.http_client.get(endpoint)
        
        if response.get("code") == 0 and response.get("data"):
            return response["data"]
        else:
            raise Exception(f"获取文件失败: {response.get('message', '未知错误')}")
    
    def upload_file(self, file_path: str) -> Dict[str, Any]:
        endpoint = "/api/v1/file/upload"
        
        with open(file_path, 'rb') as f:
            files = {'file': (file_path.split('/')[-1], f)}
            headers = {'Content-Type': 'multipart/form-data'}
            
            response = self.http_client.post(endpoint, files=files, headers=headers)
        
        if response.get("code") == 0:
            return response.get("data", {})
        else:
            raise Exception(f"上传文件失败: {response.get('message', '未知错误')}")
    
    def create_file(self, file_id: str, tenant_id: str = None) -> Dict[str, Any]:
        endpoint = "/api/v1/file/create"
        
        data = {"file_id": file_id}
        if tenant_id is not None:
            data["tenant_id"] = tenant_id
        
        response = self.http_client.post(endpoint, json_data=data)
        
        if response.get("code") == 0:
            return response.get("data", {})
        else:
            raise Exception(f"创建文件失败: {response.get('message', '未知错误')}")
    
    def delete_file(self, file_id: str) -> bool:
        endpoint = "/api/v1/file/rm"
        
        response = self.http_client.post(endpoint, json_data={"file_id": file_id})
        
        if response.get("code") == 0:
            return True
        else:
            raise Exception(f"删除文件失败: {response.get('message', '未知错误')}")
    
    def rename_file(self, file_id: str, new_name: str) -> Dict[str, Any]:
        endpoint = "/api/v1/file/rename"
        
        data = {
            "file_id": file_id,
            "new_name": new_name
        }
        
        response = self.http_client.post(endpoint, json=data)
        
        if response.get("code") == 0:
            return response.get("data", {})
        else:
            raise Exception(f"重命名文件失败: {response.get('message', '未知错误')}")
    
    def move_file(self, file_id: str, parent_id: str) -> Dict[str, Any]:
        endpoint = "/api/v1/file/mv"
        
        data = {
            "file_id": file_id,
            "parent_id": parent_id
        }
        
        response = self.http_client.post(endpoint, json=data)
        
        if response.get("code") == 0:
            return response.get("data", {})
        else:
            raise Exception(f"移动文件失败: {response.get('message', '未知错误')}")
    
    def convert_file(self, file_id: str) -> Dict[str, Any]:
        endpoint = "/api/v1/file/convert"
        
        response = self.http_client.post(endpoint, json={"file_id": file_id})
        
        if response.get("code") == 0:
            return response.get("data", {})
        else:
            raise Exception(f"转换文件失败: {response.get('message', '未知错误')}")