284 lines
9.2 KiB
Python
284 lines
9.2 KiB
Python
import t, c
|
||
from stdint import *
|
||
import platmacro
|
||
import memhub
|
||
import string
|
||
|
||
# Windows backend dependencies
|
||
import w32.win32base
|
||
import w32.win32file
|
||
|
||
# POSIX backend dependencies
|
||
import posix
|
||
|
||
|
||
# File copy buffer size
|
||
COPY_BUF_SIZE: t.CDefine = 65536 # 2^16
|
||
|
||
# Module-level global MBuddy pointer (initialized by the user in main)
|
||
_mbuddy: memhub.MemBuddy | t.CPtr
|
||
|
||
|
||
# =============================================================================
|
||
# copyfile: Copy file content (without preserving permissions/metadata)
|
||
# =============================================================================
|
||
|
||
def copyfile(src: str, dst: str) -> int:
|
||
"""Copy file content. Returns 0, otherwise -1."""
|
||
if platmacro.IS_WINDOWS:
|
||
return _copyfile_win32(src, dst)
|
||
return _copyfile_posix(src, dst)
|
||
|
||
|
||
def _copyfile_win32(src: str, dst: str) -> int:
|
||
"""Windows: Use CopyFileA kernel implementation (efficient)."""
|
||
result: BOOL = w32.win32file.CopyFileA(src, dst, 0)
|
||
if result == 0:
|
||
return -1
|
||
return 0
|
||
|
||
|
||
def _copyfile_posix(src: str, dst: str) -> int:
|
||
"""POSIX: Manual file read/write loop copy"""
|
||
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: Copy file and preserve permissions/metadata (POSIX only)
|
||
# =============================================================================
|
||
|
||
def copy(src: str, dst: str) -> int:
|
||
"""Copy file and preserve permissions/metadata (POSIX only). Returns 0, otherwise -1."""
|
||
if platmacro.IS_WINDOWS:
|
||
return _copy_win32(src, dst)
|
||
return _copy_posix(src, dst)
|
||
|
||
|
||
def _copy_win32(src: str, dst: str) -> int:
|
||
"""Windows: Copy file (using CopyFileA kernel implementation (simple, delegate to copyfile)."""
|
||
return _copyfile_win32(src, dst)
|
||
|
||
|
||
def _copy_posix(src: str, dst: str) -> int:
|
||
"""POSIX: Copy file and preserve permissions (stat+chmod). Returns 0, otherwise -1."""
|
||
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: Recursively remove directory tree (POSIX only)
|
||
# =============================================================================
|
||
|
||
def rmtree(path: str) -> int:
|
||
"""Recursively delete a directory tree. Returns 0 on success and -1 on failure."""
|
||
if platmacro.IS_WINDOWS:
|
||
return _rmtree_win32(path)
|
||
return _rmtree_posix(path)
|
||
|
||
|
||
def _rmtree_win32(path: str) -> int:
|
||
"""Windows: Recursively delete a directory tree using FindFirstFileA/FindNextFileA."""
|
||
path_len: t.CSizeT = len(path)
|
||
# Construct search pattern 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 . and ..
|
||
skip: bool = False
|
||
if name[0] == '.':
|
||
if name[1] == 0:
|
||
skip = True
|
||
elif name[1] == '.' and name[2] == 0:
|
||
skip = True
|
||
if not skip:
|
||
# Construct full path 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
|
||
# Check if it's a directory or a file
|
||
if (find_data.dwFileAttributes & w32.win32file.FILE_ATTRIBUTE_DIRECTORY) != 0:
|
||
# Recursively delete subdirectory
|
||
sub_result: int = _rmtree_win32(str(full_path))
|
||
if sub_result != 0:
|
||
result = -1
|
||
else:
|
||
# Delete file
|
||
if w32.win32file.DeleteFileA(str(full_path)) == 0:
|
||
result = -1
|
||
_mbuddy.free(full_path)
|
||
# Continue searching next entry
|
||
if w32.win32file.FindNextFileA(handle, t.CPtr(c.Addr(find_data))) == 0:
|
||
has_more = 0
|
||
|
||
w32.win32file.FindClose(handle)
|
||
# Delete empty directory itself
|
||
if result == 0:
|
||
if w32.win32file.RemoveDirectoryA(path) == 0:
|
||
result = -1
|
||
return result
|
||
|
||
|
||
def _rmtree_posix(path: str) -> int:
|
||
"""POSIX: Recursively delete a directory tree using 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 . and ..
|
||
skip: bool = False
|
||
if name[0] == '.':
|
||
if name[1] == 0:
|
||
skip = True
|
||
elif name[1] == '.' and name[2] == 0:
|
||
skip = True
|
||
if not skip:
|
||
# Construct full path 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
|
||
# Check entry type
|
||
if entry.d_type == posix.DT_DIR:
|
||
# Recursively delete subdirectory
|
||
sub_result: int = _rmtree_posix(str(full_path))
|
||
if sub_result != 0:
|
||
result = -1
|
||
else:
|
||
# Delete file or symbolic link etc.
|
||
if posix.unlink(str(full_path)) != 0:
|
||
result = -1
|
||
_mbuddy.free(full_path)
|
||
entry = posix.readdir(dirp)
|
||
|
||
posix.closedir(dirp)
|
||
# Delete empty directory itself
|
||
if result == 0:
|
||
if posix.rmdir(path) != 0:
|
||
result = -1
|
||
return result
|
||
|
||
|
||
# =============================================================================
|
||
# move：Move file/directory
|
||
# =============================================================================
|
||
|
||
def move(src: str, dst: str) -> int:
|
||
"""Move/rename files or directories. If it's the same partition, just rename;
|
||
if it's across partitions, copy then delete. Returns 0 on success, -1 on failure."""
|
||
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 within the same partition, copy and delete across partitions)"""
|
||
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 failed, try copy or 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 (within the same partition), copy and unlink across partitions"""
|
||
if posix.rename(src, dst) == 0:
|
||
return 0
|
||
# rename failed (possibly across partitions), try copy or remove
|
||
if copyfile(src, dst) != 0:
|
||
return -1
|
||
if posix.unlink(src) != 0:
|
||
return -1
|
||
return 0
|