From ecaab139483d5c1c1ca927d0d7f7b35bcf3c25b7 Mon Sep 17 00:00:00 2001 From: ninemine <1371605831@qq.com> Date: Mon, 29 Sep 2025 16:22:10 +0800 Subject: [PATCH] EP Interaction --- Convention/Runtime/File.py | 30 +- Convention/Runtime/Interaction.py | 774 ++++++++++++++++++++++++++++++ 2 files changed, 775 insertions(+), 29 deletions(-) create mode 100644 Convention/Runtime/Interaction.py diff --git a/Convention/Runtime/File.py b/Convention/Runtime/File.py index 8e7bcbf..1fa1ebe 100644 --- a/Convention/Runtime/File.py +++ b/Convention/Runtime/File.py @@ -1008,32 +1008,4 @@ class ToolFile(BaseModel): 是否隐藏 """ return self.get_permissions()['hidden'] - -def split_elements( - file: Union[ToolFile, str], - *, - ratios: List[float] = [1,1], - pr: Optional[Callable[[ToolFile], bool]] = None, - shuffler: Optional[Callable[[List[ToolFile]], None]] = None, - output_dirs: Optional[List[ToolFile]] = None, - output_must_exist: bool = True, - output_callback: Optional[Callable[[ToolFile], None]] = None - ) -> List[List[ToolFile]]: - result: List[List[ToolFile]] = tool_split_elements(WrapperFile(file).dir_tool_file_iter(), - ratios=ratios, - pr=pr, - shuffler=shuffler) - if output_dirs is None: - return result - for i in range(min(len(output_dirs), len(result))): - output_dir: ToolFile = output_dirs[i] - if output_dir.IsDir() is False: - raise Exception("Outputs must be directory") - if output_must_exist: - output_dir.must_exists_as_new() - for file in result[i]: - current = output_dirs[i].MakeFileInside(file) - if output_callback: - output_callback(current) - - return result + \ No newline at end of file diff --git a/Convention/Runtime/Interaction.py b/Convention/Runtime/Interaction.py new file mode 100644 index 0000000..29084b0 --- /dev/null +++ b/Convention/Runtime/Interaction.py @@ -0,0 +1,774 @@ +from .Config import * +from .File import ToolFile +from .Web import ToolURL +import json +import urllib.parse +import urllib.request +import urllib.error +import asyncio +import os +import re +from typing import * +from pydantic import BaseModel + +try: + import aiohttp + import aiofiles +except ImportError as e: + ImportingThrow(e, "Interaction", ["aiohttp", "aiofiles"]) + + +class InteractionError(Exception): + """交互操作异常基类""" + pass + + +class PathValidationError(InteractionError): + """路径验证异常""" + pass + + +class LoadError(InteractionError): + """加载异常""" + pass + + +class SaveError(InteractionError): + """保存异常""" + pass + + +class Interaction(BaseModel): + """统一的文件交互类，自适应处理本地文件和网络文件""" + + path: str + _is_url: bool = False + _is_local: bool = False + _tool_file: Optional[ToolFile] = None + _tool_url: Optional[ToolURL] = None + + def __init__(self, path: Union[str, 'Interaction', ToolFile, ToolURL]): + """ + 从路径字符串创建对象，自动识别本地文件或网络URL + + Args: + path: 路径字符串、Interaction对象、ToolFile对象或ToolURL对象 + """ + if isinstance(path, Interaction): + path = path.path + elif isinstance(path, ToolFile): + path = path.GetFullPath() + elif isinstance(path, ToolURL): + path = path.url + + path_str = str(path) + super().__init__(path=path_str) + + # 自动识别路径类型 + self._detect_path_type() + + def _detect_path_type(self): + """自动检测路径类型""" + path = self.path.strip() + + # 检查是否为HTTP/HTTPS URL + if path.startswith(('http://', 'https://')): + self._is_url = True + self._is_local = False + self._tool_url = ToolURL(path) + return + + # 检查是否为localhost URL + if path.startswith('localhost'): + # 转换为完整的HTTP URL + if not path.startswith('localhost:'): + # 默认端口80 + full_url = f"http://{path}" + else: + full_url = f"http://{path}" + self._is_url = True + self._is_local = False + self._tool_url = ToolURL(full_url) + self.path = full_url + return + + # 检查是否为绝对路径或相对路径 + if (os.path.isabs(path) or + path.startswith('./') or + path.startswith('../') or + ':' in path[:3]): # Windows盘符 + self._is_local = True + self._is_url = False + self._tool_file = ToolFile(path) + return + + # 默认作为相对路径处理 + self._is_local = True + self._is_url = False + self._tool_file = ToolFile(path) + + def __str__(self) -> str: + """隐式字符串转换""" + return self.path + + def __bool__(self) -> bool: + """隐式布尔转换，检查路径是否有效""" + return self.IsValid + + @property + def IsValid(self) -> bool: + """检查路径是否有效""" + if self._is_url: + return self._tool_url.IsValid if self._tool_url else False + else: + return self._tool_file.Exists() if self._tool_file else False + + @property + def IsURL(self) -> bool: + """是否为网络URL""" + return self._is_url + + @property + def IsLocal(self) -> bool: + """是否为本地文件""" + return self._is_local + + @property + def IsFile(self) -> bool: + """是否为文件（对于URL，检查是否存在文件名）""" + if self._is_url: + return bool(self._tool_url.GetFilename()) if self._tool_url else False + else: + return self._tool_file.IsFile() if self._tool_file else False + + @property + def IsDir(self) -> bool: + """是否为目录（仅对本地路径有效）""" + if self._is_local: + return self._tool_file.IsDir() if self._tool_file else False + return False + + def GetFilename(self) -> str: + """获取文件名""" + if self._is_url: + return self._tool_url.GetFilename() if self._tool_url else "" + else: + return self._tool_file.GetFilename() if self._tool_file else "" + + def GetExtension(self) -> str: + """获取文件扩展名""" + if self._is_url: + return self._tool_url.GetExtension() if self._tool_url else "" + else: + return self._tool_file.GetExtension() if self._tool_file else "" + + def ExtensionIs(self, *extensions: str) -> bool: + """检查扩展名是否匹配""" + if self._is_url: + return self._tool_url.ExtensionIs(*extensions) if self._tool_url else False + else: + current_ext = self.GetExtension() + return current_ext.lower() in [ext.lower().lstrip('.') for ext in extensions] + + # 文件类型判断属性 + @property + def IsText(self) -> bool: + """是否为文本文件""" + return self.ExtensionIs('txt', 'html', 'htm', 'css', 'js', 'xml', 'csv', 'md', 'py', 'java', 'cpp', 'c', 'h') + + @property + def IsJson(self) -> bool: + """是否为JSON文件""" + return self.ExtensionIs('json') + + @property + def IsImage(self) -> bool: + """是否为图像文件""" + return self.ExtensionIs('jpg', 'jpeg', 'png', 'gif', 'bmp', 'svg', 'webp') + + @property + def IsDocument(self) -> bool: + """是否为文档文件""" + return self.ExtensionIs('pdf', 'doc', 'docx', 'xls', 'xlsx', 'ppt', 'pptx') + + def Open(self, path: str) -> 'Interaction': + """在当前对象上打开新路径""" + new_obj = Interaction(path) + self.path = new_obj.path + self._is_url = new_obj._is_url + self._is_local = new_obj._is_local + self._tool_file = new_obj._tool_file + self._tool_url = new_obj._tool_url + return self + + # 同步加载方法 + def LoadAsText(self) -> str: + """ + 同步加载为文本 + + Returns: + 文本内容 + """ + if self._is_url: + if not self._tool_url or not self._tool_url.IsValid: + raise PathValidationError(f"Invalid URL: {self.path}") + return self._tool_url.LoadAsText() + else: + if not self._tool_file or not self._tool_file.Exists(): + raise PathValidationError(f"File not found: {self.path}") + return self._tool_file.LoadAsText() + + def LoadAsBinary(self) -> bytes: + """ + 同步加载为字节数组 + + Returns: + 二进制内容 + """ + if self._is_url: + if not self._tool_url or not self._tool_url.IsValid: + raise PathValidationError(f"Invalid URL: {self.path}") + return self._tool_url.LoadAsBinary() + else: + if not self._tool_file or not self._tool_file.Exists(): + raise PathValidationError(f"File not found: {self.path}") + return self._tool_file.LoadAsBinary() + + def LoadAsJson(self, model_type: Optional[type] = None) -> Any: + """ + 同步加载并反序列化JSON + + Args: + model_type: 可选的Pydantic模型类型 + + Returns: + JSON数据或模型对象 + """ + if self._is_url: + if not self._tool_url or not self._tool_url.IsValid: + raise PathValidationError(f"Invalid URL: {self.path}") + return self._tool_url.LoadAsJson(model_type) + else: + if not self._tool_file or not self._tool_file.Exists(): + raise PathValidationError(f"File not found: {self.path}") + json_data = self._tool_file.LoadAsJson() + if model_type and issubclass(model_type, BaseModel): + return model_type.model_validate(json_data) + return json_data + + # 异步加载方法 + async def LoadAsTextAsync(self) -> str: + """ + 异步加载为文本 + + Returns: + 文本内容 + """ + if self._is_url: + if not self._tool_url or not self._tool_url.IsValid: + raise PathValidationError(f"Invalid URL: {self.path}") + return await self._tool_url.LoadAsTextAsync() + else: + if not self._tool_file or not self._tool_file.Exists(): + raise PathValidationError(f"File not found: {self.path}") + # 异步读取本地文件 + async with aiofiles.open(self._tool_file.GetFullPath(), 'r', encoding='utf-8') as f: + return await f.read() + + async def LoadAsBinaryAsync(self) -> bytes: + """ + 异步加载为字节数组 + + Returns: + 二进制内容 + """ + if self._is_url: + if not self._tool_url or not self._tool_url.IsValid: + raise PathValidationError(f"Invalid URL: {self.path}") + return await self._tool_url.LoadAsBinaryAsync() + else: + if not self._tool_file or not self._tool_file.Exists(): + raise PathValidationError(f"File not found: {self.path}") + # 异步读取本地文件 + async with aiofiles.open(self._tool_file.GetFullPath(), 'rb') as f: + return await f.read() + + async def LoadAsJsonAsync(self, model_type: Optional[type] = None) -> Any: + """ + 异步加载并反序列化JSON + + Args: + model_type: 可选的Pydantic模型类型 + + Returns: + JSON数据或模型对象 + """ + if self._is_url: + if not self._tool_url or not self._tool_url.IsValid: + raise PathValidationError(f"Invalid URL: {self.path}") + return await self._tool_url.LoadAsJsonAsync(model_type) + else: + if not self._tool_file or not self._tool_file.Exists(): + raise PathValidationError(f"File not found: {self.path}") + # 异步读取本地JSON文件 + text_content = await self.LoadAsTextAsync() + try: + json_data = json.loads(text_content) + if model_type and issubclass(model_type, BaseModel): + return model_type.model_validate(json_data) + return json_data + except json.JSONDecodeError as e: + raise LoadError(f"Failed to parse JSON from {self.path}: {str(e)}") + + # 同步保存方法 + def SaveAsText(self, content: str, local_path: Optional[str] = None) -> Union[ToolFile, 'Interaction']: + """ + 同步保存为文本 + + Args: + content: 文本内容 + local_path: 本地保存路径（仅对URL有效） + + Returns: + 保存的文件对象或Interaction对象 + """ + if self._is_url: + # 对于URL，先下载然后保存到本地 + if local_path is None: + local_path = self.GetFilename() or "downloaded.txt" + file_obj = ToolFile(local_path) + file_obj.TryCreateParentPath() + file_obj.SaveAsText(content) + return file_obj + else: + if not self._tool_file: + raise PathValidationError(f"Invalid file path: {self.path}") + self._tool_file.TryCreateParentPath() + self._tool_file.SaveAsText(content) + return self + + def SaveAsBinary(self, content: bytes, local_path: Optional[str] = None) -> Union[ToolFile, 'Interaction']: + """ + 同步保存为二进制 + + Args: + content: 二进制内容 + local_path: 本地保存路径（仅对URL有效） + + Returns: + 保存的文件对象或Interaction对象 + """ + if self._is_url: + # 对于URL，保存到本地 + if local_path is None: + local_path = self.GetFilename() or "downloaded.bin" + file_obj = ToolFile(local_path) + file_obj.TryCreateParentPath() + file_obj.SaveAsBinary(content) + return file_obj + else: + if not self._tool_file: + raise PathValidationError(f"Invalid file path: {self.path}") + self._tool_file.TryCreateParentPath() + self._tool_file.SaveAsBinary(content) + return self + + def SaveAsJson(self, data: Any, local_path: Optional[str] = None) -> Union[ToolFile, 'Interaction']: + """ + 同步保存为JSON + + Args: + data: JSON数据 + local_path: 本地保存路径（仅对URL有效） + + Returns: + 保存的文件对象或Interaction对象 + """ + if self._is_url: + # 对于URL，保存到本地 + if local_path is None: + local_path = self.GetFilename() or "downloaded.json" + file_obj = ToolFile(local_path) + file_obj.TryCreateParentPath() + file_obj.SaveAsJson(data) + return file_obj + else: + if not self._tool_file: + raise PathValidationError(f"Invalid file path: {self.path}") + self._tool_file.TryCreateParentPath() + self._tool_file.SaveAsJson(data) + return self + + # 异步保存方法 + async def SaveAsTextAsync(self, content: str, local_path: Optional[str] = None) -> Union[ToolFile, 'Interaction']: + """ + 异步保存为文本 + + Args: + content: 文本内容 + local_path: 本地保存路径（仅对URL有效） + + Returns: + 保存的文件对象或Interaction对象 + """ + if self._is_url: + # 对于URL，保存到本地 + if local_path is None: + local_path = self.GetFilename() or "downloaded.txt" + file_obj = ToolFile(local_path) + file_obj.TryCreateParentPath() + async with aiofiles.open(file_obj.GetFullPath(), 'w', encoding='utf-8') as f: + await f.write(content) + return file_obj + else: + if not self._tool_file: + raise PathValidationError(f"Invalid file path: {self.path}") + self._tool_file.TryCreateParentPath() + async with aiofiles.open(self._tool_file.GetFullPath(), 'w', encoding='utf-8') as f: + await f.write(content) + return self + + async def SaveAsBinaryAsync(self, content: bytes, local_path: Optional[str] = None) -> Union[ToolFile, 'Interaction']: + """ + 异步保存为二进制 + + Args: + content: 二进制内容 + local_path: 本地保存路径（仅对URL有效） + + Returns: + 保存的文件对象或Interaction对象 + """ + if self._is_url: + # 对于URL，保存到本地 + if local_path is None: + local_path = self.GetFilename() or "downloaded.bin" + file_obj = ToolFile(local_path) + file_obj.TryCreateParentPath() + async with aiofiles.open(file_obj.GetFullPath(), 'wb') as f: + await f.write(content) + return file_obj + else: + if not self._tool_file: + raise PathValidationError(f"Invalid file path: {self.path}") + self._tool_file.TryCreateParentPath() + async with aiofiles.open(self._tool_file.GetFullPath(), 'wb') as f: + await f.write(content) + return self + + async def SaveAsJsonAsync(self, data: Any, local_path: Optional[str] = None) -> Union[ToolFile, 'Interaction']: + """ + 异步保存为JSON + + Args: + data: JSON数据 + local_path: 本地保存路径（仅对URL有效） + + Returns: + 保存的文件对象或Interaction对象 + """ + # 序列化JSON数据 + try: + from pydantic import BaseModel + if isinstance(data, BaseModel): + json_data = data.model_dump() + json_data["__type"] = f"{data.__class__.__name__}, pydantic.BaseModel" + else: + json_data = data + json_content = json.dumps(json_data, indent=4, ensure_ascii=False) + except Exception as e: + raise SaveError(f"Failed to serialize JSON data: {str(e)}") + + # 保存JSON内容 + return await self.SaveAsTextAsync(json_content, local_path) + + # HTTP请求方法（仅对URL有效） + def Get(self, callback: Callable[[Optional[Any]], None]) -> bool: + """ + 同步GET请求（仅对URL有效） + + Args: + callback: 响应回调函数，成功时接收响应对象，失败时接收None + + Returns: + 是否请求成功 + """ + if not self._is_url: + raise InteractionError("GET method is only available for URLs") + if not self._tool_url: + callback(None) + return False + return self._tool_url.Get(callback) + + def Post(self, callback: Callable[[Optional[Any]], None], form_data: Optional[Dict[str, str]] = None) -> bool: + """ + 同步POST请求（仅对URL有效） + + Args: + callback: 响应回调函数，成功时接收响应对象，失败时接收None + form_data: 表单数据字典 + + Returns: + 是否请求成功 + """ + if not self._is_url: + raise InteractionError("POST method is only available for URLs") + if not self._tool_url: + callback(None) + return False + return self._tool_url.Post(callback, form_data) + + async def GetAsync(self, callback: Callable[[Optional[Any]], None]) -> bool: + """ + 异步GET请求（仅对URL有效） + + Args: + callback: 响应回调函数，成功时接收响应对象，失败时接收None + + Returns: + 是否请求成功 + """ + if not self._is_url: + raise InteractionError("GET method is only available for URLs") + if not self._tool_url: + callback(None) + return False + return await self._tool_url.GetAsync(callback) + + async def PostAsync(self, callback: Callable[[Optional[Any]], None], form_data: Optional[Dict[str, str]] = None) -> bool: + """ + 异步POST请求（仅对URL有效） + + Args: + callback: 响应回调函数，成功时接收响应对象，失败时接收None + form_data: 表单数据字典 + + Returns: + 是否请求成功 + """ + if not self._is_url: + raise InteractionError("POST method is only available for URLs") + if not self._tool_url: + callback(None) + return False + return await self._tool_url.PostAsync(callback, form_data) + + # 便利方法 + def Save(self, local_path: Optional[str] = None) -> Union[ToolFile, 'Interaction']: + """ + 自动选择格式保存 + + Args: + local_path: 本地保存路径 + + Returns: + 保存的文件对象或Interaction对象 + """ + if self._is_url: + # 对于URL，先下载内容再保存 + if not self._tool_url: + raise PathValidationError(f"Invalid URL: {self.path}") + return self._tool_url.Save(local_path) + else: + # 对于本地文件，直接返回自身（已存在） + return self + + async def SaveAsync(self, local_path: Optional[str] = None) -> Union[ToolFile, 'Interaction']: + """ + 异步自动选择格式保存 + + Args: + local_path: 本地保存路径 + + Returns: + 保存的文件对象或Interaction对象 + """ + if self._is_url: + # 对于URL，异步下载内容 + if not self._tool_url: + raise PathValidationError(f"Invalid URL: {self.path}") + + if local_path is None: + local_path = self.GetFilename() or "downloaded_file" + + file_obj = ToolFile(local_path) + file_obj.TryCreateParentPath() + + try: + if self.IsText: + content = await self.LoadAsTextAsync() + await self.SaveAsTextAsync(content, local_path) + elif self.IsJson: + content = await self.LoadAsJsonAsync() + await self.SaveAsJsonAsync(content, local_path) + else: + content = await self.LoadAsBinaryAsync() + await self.SaveAsBinaryAsync(content, local_path) + + return file_obj + except Exception as e: + raise SaveError(f"Failed to save {self.path}: {str(e)}") + else: + # 对于本地文件，直接返回自身 + return self + + def Download(self, local_path: Optional[str] = None) -> ToolFile: + """ + 下载文件（仅对URL有效） + + Args: + local_path: 本地保存路径 + + Returns: + 下载的文件对象 + """ + if not self._is_url: + raise InteractionError("Download method is only available for URLs") + if not self._tool_url: + raise PathValidationError(f"Invalid URL: {self.path}") + return self._tool_url.Download(local_path) + + async def DownloadAsync(self, local_path: Optional[str] = None) -> ToolFile: + """ + 异步下载文件（仅对URL有效） + + Args: + local_path: 本地保存路径 + + Returns: + 下载的文件对象 + """ + if not self._is_url: + raise InteractionError("Download method is only available for URLs") + if not self._tool_url: + raise PathValidationError(f"Invalid URL: {self.path}") + return await self._tool_url.DownloadAsync(local_path) + + def Copy(self, target_path: Optional[Union[str, ToolFile, 'Interaction']] = None) -> 'Interaction': + """ + 复制文件（仅对本地文件有效） + + Args: + target_path: 目标路径 + + Returns: + 新的Interaction对象 + """ + if not self._is_local: + raise InteractionError("Copy method is only available for local files") + if not self._tool_file: + raise PathValidationError(f"Invalid file path: {self.path}") + + if target_path is None: + copied_file = self._tool_file.Copy() + else: + if isinstance(target_path, Interaction): + target_path = target_path.path + elif isinstance(target_path, ToolFile): + target_path = target_path.GetFullPath() + copied_file = self._tool_file.Copy(str(target_path)) + + return Interaction(copied_file.GetFullPath()) + + def Move(self, target_path: Union[str, ToolFile, 'Interaction']) -> 'Interaction': + """ + 移动文件（仅对本地文件有效） + + Args: + target_path: 目标路径 + + Returns: + 更新后的Interaction对象 + """ + if not self._is_local: + raise InteractionError("Move method is only available for local files") + if not self._tool_file: + raise PathValidationError(f"Invalid file path: {self.path}") + + if isinstance(target_path, Interaction): + target_path = target_path.path + elif isinstance(target_path, ToolFile): + target_path = target_path.GetFullPath() + + self._tool_file.Move(str(target_path)) + self.path = self._tool_file.GetFullPath() + return self + + def Remove(self) -> 'Interaction': + """ + 删除文件（仅对本地文件有效） + + Returns: + Interaction对象本身 + """ + if not self._is_local: + raise InteractionError("Remove method is only available for local files") + if not self._tool_file: + raise PathValidationError(f"Invalid file path: {self.path}") + + self._tool_file.Remove() + return self + + def Exists(self) -> bool: + """ + 检查文件是否存在 + + Returns: + 是否存在 + """ + return self.IsValid + + def GetSize(self) -> int: + """ + 获取文件大小（仅对本地文件有效） + + Returns: + 文件大小（字节） + """ + if not self._is_local: + raise InteractionError("GetSize method is only available for local files") + if not self._tool_file or not self._tool_file.Exists(): + raise PathValidationError(f"File not found: {self.path}") + + return self._tool_file.GetSize() + + def GetDir(self) -> str: + """ + 获取目录路径 + + Returns: + 目录路径 + """ + if self._is_local: + return self._tool_file.GetDir() if self._tool_file else "" + else: + # 对于URL，返回基础URL + if self._tool_url: + parsed = urllib.parse.urlparse(self._tool_url.url) + return f"{parsed.scheme}://{parsed.netloc}" + return "" + + def GetParentDir(self) -> 'Interaction': + """ + 获取父目录的Interaction对象 + + Returns: + 父目录的Interaction对象 + """ + if self._is_local: + if not self._tool_file: + raise PathValidationError(f"Invalid file path: {self.path}") + parent_dir = self._tool_file.GetParentDir() + return Interaction(parent_dir.GetFullPath()) + else: + # 对于URL，返回基础URL + base_url = self.GetDir() + return Interaction(base_url) + + def ToString(self) -> str: + """获取完整路径""" + return self.path + + def GetFullPath(self) -> str: + """获取完整路径""" + return self.path