import t, c from stdint import * import platmacro import memhub import string # Windows 后端依赖 import w32.win32base import w32.win32file # POSIX 后端依赖 import posix # 文件拷贝缓冲区大小 COPY_BUF_SIZE: t.CDefine = 65536 # 模块级全局 MBuddy 指针(由用户在 main 中赋值初始化) _mbuddy: memhub.MemBuddy | t.CPtr # ============================================================================= # copyfile:拷贝文件内容(不保留权限/元数据) # ============================================================================= def copyfile(src: str, dst: str) -> int: """拷贝文件内容。成功返回 0,失败返回 -1。""" if platmacro.IS_WINDOWS: return _copyfile_win32(src, dst) return _copyfile_posix(src, dst) def _copyfile_win32(src: str, dst: str) -> int: """Windows:使用 CopyFileA 内核实现(高效)""" result: BOOL = w32.win32file.CopyFileA(src, dst, 0) if result == 0: return -1 return 0 def _copyfile_posix(src: str, dst: str) -> int: """POSIX:手动 read/write 循环拷贝""" fd_in: INT = posix.open(src, posix.O_RDONLY, 0) if fd_in < 0: return -1 fd_out: INT = posix.open(dst, posix.O_WRONLY | posix.O_CREAT | posix.O_TRUNC, 0o644) if fd_out < 0: posix.close(fd_in) return -1 buf: bytes = _mbuddy.alloc(COPY_BUF_SIZE) if buf == None: posix.close(fd_in) posix.close(fd_out) return -1 n: ssize_t = 0 result: int = 0 n = posix.read(fd_in, buf, COPY_BUF_SIZE) while n > 0: written: ssize_t = posix.write(fd_out, buf, t.CSizeT(n)) if written < 0: result = -1 break n = posix.read(fd_in, buf, COPY_BUF_SIZE) if n < 0: result = -1 _mbuddy.free(buf) posix.close(fd_in) posix.close(fd_out) return result # ============================================================================= # copy:拷贝文件并保留权限 # ============================================================================= def copy(src: str, dst: str) -> int: """拷贝文件并保留权限(POSIX 上调用 stat+chmod)。成功返回 0,失败返回 -1。""" if platmacro.IS_WINDOWS: return _copy_win32(src, dst) return _copy_posix(src, dst) def _copy_win32(src: str, dst: str) -> int: """Windows:拷贝文件(CopyFileA,Windows 权限模型简单,直接委托 copyfile)""" return _copyfile_win32(src, dst) def _copy_posix(src: str, dst: str) -> int: """POSIX:拷贝文件并保留权限(stat+chmod)""" result: int = _copyfile_posix(src, dst) if result != 0: return -1 st: posix.Stat = posix.Stat() if posix.stat(src, t.CPtr(c.Addr(st))) == 0: posix.chmod(dst, st.st_mode) return 0 # ============================================================================= # rmtree:递归删除目录树 # ============================================================================= def rmtree(path: str) -> int: """递归删除目录树。成功返回 0,失败返回 -1。""" if platmacro.IS_WINDOWS: return _rmtree_win32(path) return _rmtree_posix(path) def _rmtree_win32(path: str) -> int: """Windows:FindFirstFileA/FindNextFileA 递归删除""" path_len: t.CSizeT = len(path) # 构造搜索模式 path\* pattern: bytes = _mbuddy.alloc(path_len + 3) if pattern == None: return -1 i: t.CSizeT = 0 for ch in path: pattern[i] = ch i += 1 if path_len > 0: if path[path_len - 1] != '\\' and path[path_len - 1] != '/': pattern[i] = '\\' i += 1 pattern[i] = '*' i += 1 pattern[i] = 0 find_data: w32.win32file.WIN32_FIND_DATAA = w32.win32file.WIN32_FIND_DATAA() handle: w32.win32base.HANDLE = w32.win32file.FindFirstFileA(pattern, t.CPtr(c.Addr(find_data))) _mbuddy.free(pattern) if handle == w32.win32base.INVALID_HANDLE_VALUE: return -1 result: int = 0 has_more: int = 1 while has_more == 1: name: str = find_data.cFileName # 跳过 . 和 .. skip: bool = False if name[0] == '.': if name[1] == 0: skip = True elif name[1] == '.' and name[2] == 0: skip = True if not skip: # 构造完整路径 path\name name_len: t.CSizeT = len(name) full_path: bytes = _mbuddy.alloc(path_len + name_len + 2) if full_path == None: result = -1 else: j: t.CSizeT = 0 for ch in path: full_path[j] = ch j += 1 if path_len > 0: if path[path_len - 1] != '\\' and path[path_len - 1] != '/': full_path[j] = '\\' j += 1 for ch in name: full_path[j] = ch j += 1 full_path[j] = 0 # 判断是文件还是目录 if (find_data.dwFileAttributes & w32.win32file.FILE_ATTRIBUTE_DIRECTORY) != 0: # 递归删除子目录 sub_result: int = _rmtree_win32(str(full_path)) if sub_result != 0: result = -1 else: # 删除文件 if w32.win32file.DeleteFileA(str(full_path)) == 0: result = -1 _mbuddy.free(full_path) # 继续搜索下一个条目 if w32.win32file.FindNextFileA(handle, t.CPtr(c.Addr(find_data))) == 0: has_more = 0 w32.win32file.FindClose(handle) # 最后删除空目录本身 if result == 0: if w32.win32file.RemoveDirectoryA(path) == 0: result = -1 return result def _rmtree_posix(path: str) -> int: """POSIX:opendir/readdir 递归删除""" dirp: posix.DIR = posix.opendir(path) if dirp == None: return -1 path_len: t.CSizeT = len(path) result: int = 0 entry: posix.Dirent = posix.readdir(dirp) while entry != None: name: str = entry.d_name # 跳过 . 和 .. skip: bool = False if name[0] == '.': if name[1] == 0: skip = True elif name[1] == '.' and name[2] == 0: skip = True if not skip: # 构造完整路径 path/name name_len: t.CSizeT = len(name) full_path: bytes = _mbuddy.alloc(path_len + name_len + 2) if full_path == None: result = -1 else: j: t.CSizeT = 0 for ch in path: full_path[j] = ch j += 1 if path_len > 0: if path[path_len - 1] != '/': full_path[j] = '/' j += 1 for ch in name: full_path[j] = ch j += 1 full_path[j] = 0 # 判断条目类型 if entry.d_type == posix.DT_DIR: # 递归删除子目录 sub_result: int = _rmtree_posix(str(full_path)) if sub_result != 0: result = -1 else: # 删除文件/符号链接等 if posix.unlink(str(full_path)) != 0: result = -1 _mbuddy.free(full_path) entry = posix.readdir(dirp) posix.closedir(dirp) # 最后删除空目录本身 if result == 0: if posix.rmdir(path) != 0: result = -1 return result # ============================================================================= # move:移动文件/目录 # ============================================================================= def move(src: str, dst: str) -> int: """移动/重命名文件或目录。同分区时直接 rename,跨分区时 copy+delete。 成功返回 0,失败返回 -1。""" if platmacro.IS_WINDOWS: return _move_win32(src, dst) return _move_posix(src, dst) def _move_win32(src: str, dst: str) -> int: """Windows:MoveFileExA(同分区 rename,跨分区 copy+delete)""" rename_result: BOOL = w32.win32file.MoveFileExA( src, dst, w32.win32file.MOVEFILE_REPLACE_EXISTING | w32.win32file.MOVEFILE_COPY_ALLOWED ) if rename_result != 0: return 0 # rename 失败,尝试 copy + remove if copyfile(src, dst) != 0: return -1 if w32.win32file.DeleteFileA(src) == 0: return -1 return 0 def _move_posix(src: str, dst: str) -> int: """POSIX:rename(同分区),跨分区 copy+unlink""" if posix.rename(src, dst) == 0: return 0 # rename 失败(可能跨分区),尝试 copy + unlink if copyfile(src, dst) != 0: return -1 if posix.unlink(src) != 0: return -1 return 0