""" java.io 文件扩展组件 — File扩展, FileSystem, UnixFileSystem, FileDescriptor扩展, FileInputStream扩展, FileChannel扩展, FilePermission扩展, FileVisitor, FileAlreadyExistsException等 方法签名完全对齐 Java SDK """ from __future__ import annotations import errno import io import os import stat import struct import sys import time from typing import (Any, Callable, Dict, IO, Iterator, List, Optional, Set, Tuple, Union) from .._core import Stub, AndroidOnlyError __all__ = [ "FileSystem", "UnixFileSystem", "WinNTFileSystem", "FileStatus", "FileAccess", "FileAccessModes", "FileDescriptor", "FileInputStream", "FileOutputStream", "RandomAccessFile", "FileLock", "FileChannel", "FilePermission", "FileFilter", "FileNameFilter", "FileSystemProvider", "FileTreeWalker", "FileTypeDetector", "FileVisitOption", "FileVisitResult", "FileVisitor", "SimpleFileVisitor", "FileAlreadyExistsException", "FileSystemAlreadyExistsException", "FileSystemException", "FileSystemLoopException", "FileSystemNotFoundException", "AccessDeniedException", "AtomicMoveNotSupportedException", "DirectoryIteratorException", "DirectoryNotEmptyException", "FileSystemOperationException", "InvalidPathException", "ProviderMismatchException", "ProviderNotFoundException", "ReadOnlyFileSystemException", "UserPrincipalLookupService", "UserPrincipal", "GroupPrincipal", "FileAttribute", "FileAttributeView", "FileStore", "FileStoreAttributeView", "FileSystems", ] # ═══════════════════════════════════════════════════════════════ # 文件系统常量 # ═══════════════════════════════════════════════════════════════ ACCESS_READ = 0x01 ACCESS_WRITE = 0x02 ACCESS_EXECUTE = 0x04 CREATE_FILE = 0x01 CREATE_DIRECTORY = 0x02 CREATE_LINK = 0x04 CREATE_SYMBOLIC_LINK = 0x08 DELETE_FILE = 0x01 DELETE_DIRECTORY = 0x02 DELETE_ON_CLOSE = 0x04 OPEN_READ = 0x01 OPEN_WRITE = 0x02 OPEN_APPEND = 0x04 OPEN_TRUNCATE = 0x08 OPEN_CREATE = 0x10 OPEN_CREATE_NEW = 0x20 OPEN_DELETE_ON_CLOSE = 0x40 OPEN_SPARSE = 0x80 OPEN_SYNC = 0x100 OPEN_DSYNC = 0x200 COPY_FILE = 0x01 COPY_DIRECTORY = 0x02 COPY_SYMBOLIC_LINK = 0x04 COPY_ATTRIBUTES = 0x08 COPY_REPLACE_EXISTING = 0x10 COPY_COPY_ATTRIBUTES = 0x08 COPY_NO_FOLLOW_LINKS = 0x04 MOVE_FILE = 0x01 MOVE_DIRECTORY = 0x02 MOVE_REPLACE_EXISTING = 0x04 MOVE_ATOMIC = 0x08 LINK_SYMBOLIC = 0x01 LINK_HARD = 0x02 WALK_FILE_TREE = 0x01 WALK_FOLLOW_LINKS = 0x02 WALK_MAX_DEPTH_UNLIMITED = -1 FILE_TYPE_UNKNOWN = 0 FILE_TYPE_REGULAR = 1 FILE_TYPE_DIRECTORY = 2 FILE_TYPE_SYMBOLIC_LINK = 3 FILE_TYPE_CHARACTER_DEVICE = 4 FILE_TYPE_BLOCK_DEVICE = 5 FILE_TYPE_FIFO = 6 FILE_TYPE_SOCKET = 7 FILE_ATTRIBUTE_READ_ONLY = 0x01 FILE_ATTRIBUTE_HIDDEN = 0x02 FILE_ATTRIBUTE_SYSTEM = 0x04 FILE_ATTRIBUTE_DIRECTORY = 0x08 FILE_ATTRIBUTE_ARCHIVE = 0x10 FILE_ATTRIBUTE_NORMAL = 0x20 FILE_ATTRIBUTE_TEMPORARY = 0x40 FILE_ATTRIBUTE_SPARSE = 0x80 FILE_ATTRIBUTE_REPARSE = 0x100 FILE_ATTRIBUTE_COMPRESSED = 0x200 FILE_ATTRIBUTE_OFFLINE = 0x400 FILE_ATTRIBUTE_NOT_CONTENT_INDEXED = 0x800 FILE_ATTRIBUTE_ENCRYPTED = 0x1000 FILE_ATTRIBUTE_VIRTUAL = 0x2000 FILE_ATTRIBUTE_INTEGRITY_STREAM = 0x4000 FILE_ATTRIBUTE_NO_SCRUB = 0x8000 FILE_ATTRIBUTE_RECALL_ON_OPEN = 0x40000 FILE_ATTRIBUTE_PINNED = 0x80000 FILE_ATTRIBUTE_UNPINNED = 0x100000 FILE_ATTRIBUTE_RECALL_ON_DATA_ACCESS = 0x400000 # ═══════════════════════════════════════════════════════════════ # 异常类 # ═══════════════════════════════════════════════════════════════ class FileSystemException(OSError): """FileSystemException — 文件系统异常""" def __init__(self, file_: str = "", other: str = "", reason: str = ""): self._file = file_ self._other = other self._reason = reason msg = f"{file_}" if other: msg += f" -> {other}" if reason: msg += f": {reason}" super().__init__(msg) def getFile(self) -> str: return self._file def getOtherFile(self) -> str: return self._other def getReason(self) -> str: return self._reason class FileAlreadyExistsException(FileSystemException): """FileAlreadyExistsException — 文件已存在""" pass class FileSystemAlreadyExistsException(FileSystemException): """FileSystemAlreadyExistsException — 文件系统已存在""" pass class FileSystemLoopException(FileSystemException): """FileSystemLoopException — 文件系统循环""" pass class FileSystemNotFoundException(FileSystemException): """FileSystemNotFoundException — 文件系统未找到""" pass class AccessDeniedException(FileSystemException): """AccessDeniedException — 访问被拒绝""" pass class AtomicMoveNotSupportedException(FileSystemException): """AtomicMoveNotSupportedException — 原子移动不支持""" pass class DirectoryIteratorException(FileSystemException): """DirectoryIteratorException — 目录迭代异常""" pass class DirectoryNotEmptyException(FileSystemException): """DirectoryNotEmptyException — 目录非空""" pass class FileSystemOperationException(FileSystemException): """FileSystemOperationException — 文件系统操作异常""" pass class InvalidPathException(FileSystemException): """InvalidPathException — 路径无效""" pass class ProviderMismatchException(FileSystemException): """ProviderMismatchException — 提供器不匹配""" pass class ProviderNotFoundException(FileSystemException): """ProviderNotFoundException — 提供器未找到""" pass class ReadOnlyFileSystemException(FileSystemException): """ReadOnlyFileSystemException — 只读文件系统""" pass # ═══════════════════════════════════════════════════════════════ # FileStatus / FileAccess / FileAccessModes # ═══════════════════════════════════════════════════════════════ class FileStatus(Stub): """FileStatus — 文件状态""" def __init__(self, type_: int = FILE_TYPE_UNKNOWN, size: int = 0, lastModified: int = 0, lastAccess: int = 0, creationTime: int = 0, isSymbolicLink: bool = False, device: int = 0, inode: int = 0, mode: int = 0, uid: int = 0, gid: int = 0): super().__init__() self._type = type_ self._size = size self._last_modified = lastModified self._last_access = lastAccess self._creation_time = creationTime self._is_symlink = isSymbolicLink self._device = device self._inode = inode self._mode = mode self._uid = uid self._gid = gid def getType(self) -> int: return self._type def getSize(self) -> int: return self._size def getLastModified(self) -> int: return self._last_modified def getLastAccess(self) -> int: return self._last_access def getCreationTime(self) -> int: return self._creation_time def isSymbolicLink(self) -> bool: return self._is_symlink def getDevice(self) -> int: return self._device def getInode(self) -> int: return self._inode def getMode(self) -> int: return self._mode def getUid(self) -> int: return self._uid def getGid(self) -> int: return self._gid def toString(self) -> str: return f"FileStatus(type={self._type}, size={self._size})" class FileAccess(Stub): """FileAccess — 文件访问""" def __init__(self, canRead: bool = True, canWrite: bool = True, canExecute: bool = False): super().__init__() self._can_read = canRead self._can_write = canWrite self._can_execute = canExecute def canRead(self) -> bool: return self._can_read def canWrite(self) -> bool: return self._can_write def canExecute(self) -> bool: return self._can_execute def toString(self) -> str: return f"FileAccess(r={self._can_read}, w={self._can_write}, x={self._can_execute})" class FileAccessModes(Stub): """FileAccessModes — 文件访问模式""" def __init__(self, read: bool = False, write: bool = False, append: bool = False, truncate: bool = False): super().__init__() self._read = read self._write = write self._append = append self._truncate = truncate def isRead(self) -> bool: return self._read def isWrite(self) -> bool: return self._write def isAppend(self) -> bool: return self._append def isTruncate(self) -> bool: return self._truncate def toString(self) -> str: return f"FileAccessModes(r={self._read}, w={self._write})" # ═══════════════════════════════════════════════════════════════ # FileDescriptor (扩展) # ═══════════════════════════════════════════════════════════════ class FileDescriptor(Stub): """FileDescriptor — 文件描述符 (完整版)""" _in: Optional["FileDescriptor"] = None _out: Optional["FileDescriptor"] = None _err: Optional["FileDescriptor"] = None @staticmethod def in_() -> "FileDescriptor": if FileDescriptor._in is None: fd = FileDescriptor() fd._fd = 0 FileDescriptor._in = fd return FileDescriptor._in @staticmethod def out() -> "FileDescriptor": if FileDescriptor._out is None: fd = FileDescriptor() fd._fd = 1 FileDescriptor._out = fd return FileDescriptor._out @staticmethod def err() -> "FileDescriptor": if FileDescriptor._err is None: fd = FileDescriptor() fd._fd = 2 FileDescriptor._err = fd return FileDescriptor._err def __init__(self): super().__init__() self._fd: int = -1 self._closed = False self._readonly = False def valid(self) -> bool: return self._fd >= 0 and not self._closed def getInt(self) -> int: return self._fd def setInt(self, fd: int): self._fd = fd self._closed = False def close(self): if self.valid(): try: os.close(self._fd) except Exception: pass self._closed = True self._fd = -1 def isClosed(self) -> bool: return self._closed def sync(self): if self.valid(): try: os.fsync(self._fd) except Exception: pass def toString(self) -> str: return f"FileDescriptor(fd={self._fd})" def equals(self, other: Any) -> bool: if not isinstance(other, FileDescriptor): return False return self._fd == other._fd def hashCode(self) -> int: return self._fd & 0x7FFFFFFF # ═══════════════════════════════════════════════════════════════ # FileInputStream (扩展) # ═══════════════════════════════════════════════════════════════ class FileInputStream(Stub): """FileInputStream — 文件输入流 (完整版)""" def __init__(self, file: Any): super().__init__() if isinstance(file, int): self._fd = FileDescriptor() self._fd.setInt(file) self._native_fd = file elif isinstance(file, FileDescriptor): self._fd = file self._native_fd = file.getInt() elif isinstance(file, str): self._fd = FileDescriptor() self._native_fd = os.open(file, os.O_RDONLY) self._fd.setInt(self._native_fd) else: path = str(file) if hasattr(file, 'getPath') else str(file) self._fd = FileDescriptor() self._native_fd = os.open(path, os.O_RDONLY) self._fd.setInt(self._native_fd) self._closed = False self._pos = 0 def read(self) -> int: if self._closed or self._native_fd < 0: return -1 try: data = os.read(self._native_fd, 1) if not data: return -1 self._pos += 1 return data[0] except Exception: return -1 def readBytes(self, b: bytearray, off: int, len_: int) -> int: if self._closed or self._native_fd < 0: return -1 try: data = os.read(self._native_fd, len_) if not data: return -1 actual = len(data) b[off:off + actual] = data self._pos += actual return actual except Exception: return -1 def available(self) -> int: return 0 def skip(self, n: int) -> int: if self._closed or self._native_fd < 0: return 0 try: os.lseek(self._native_fd, n, os.SEEK_CUR) self._pos += n return n except Exception: return 0 def close(self): if not self._closed: self._closed = True if self._native_fd >= 0: try: os.close(self._native_fd) except Exception: pass self._fd.close() def getFD(self) -> Optional[FileDescriptor]: return self._fd if self._fd.valid() else None def getChannel(self) -> "FileChannel": return FileChannel(fd=self._native_fd, mode="r") def isClosed(self) -> bool: return self._closed def toString(self) -> str: return f"FileInputStream(fd={self._native_fd})" def equals(self, other: Any) -> bool: return self is other def hashCode(self) -> int: return id(self) & 0x7FFFFFFF # ═══════════════════════════════════════════════════════════════ # FileOutputStream (扩展) # ═══════════════════════════════════════════════════════════════ class FileOutputStream(Stub): """FileOutputStream — 文件输出流 (完整版)""" def __init__(self, file: Any, append: bool = False): super().__init__() flags = os.O_WRONLY | os.O_CREAT if append: flags |= os.O_APPEND else: flags |= os.O_TRUNC if isinstance(file, int): self._fd = FileDescriptor() self._fd.setInt(file) self._native_fd = file elif isinstance(file, FileDescriptor): self._fd = file self._native_fd = file.getInt() elif isinstance(file, str): self._fd = FileDescriptor() self._native_fd = os.open(file, flags, 0o644) self._fd.setInt(self._native_fd) else: path = str(file) if hasattr(file, 'getPath') else str(file) self._fd = FileDescriptor() self._native_fd = os.open(path, flags, 0o644) self._fd.setInt(self._native_fd) self._closed = False self._append_mode = append def write(self, b: Union[int, bytes, bytearray]): if self._closed or self._native_fd < 0: return if isinstance(b, int): os.write(self._native_fd, bytes([b])) else: os.write(self._native_fd, bytes(b)) def writeBytes(self, b: bytes, off: int, len_: int): if self._closed or self._native_fd < 0: return os.write(self._native_fd, b[off:off + len_]) def flush(self): pass def close(self): if not self._closed: self._closed = True if self._native_fd >= 0: try: os.close(self._native_fd) except Exception: pass self._fd.close() def getFD(self) -> Optional[FileDescriptor]: return self._fd if self._fd.valid() else None def getChannel(self) -> "FileChannel": return FileChannel(fd=self._native_fd, mode="rw") def isClosed(self) -> bool: return self._closed def toString(self) -> str: return f"FileOutputStream(fd={self._native_fd})" def equals(self, other: Any) -> bool: return self is other def hashCode(self) -> int: return id(self) & 0x7FFFFFFF # ═══════════════════════════════════════════════════════════════ # RandomAccessFile # ═══════════════════════════════════════════════════════════════ class RandomAccessFile(Stub): """RandomAccessFile — 随机访问文件""" def __init__(self, file: Any, mode: str = "r"): super().__init__() self._mode = mode if isinstance(file, str): self._path = file elif hasattr(file, 'getPath'): self._path = file.getPath() else: self._path = str(file) if "r" in mode and "w" in mode: flags = os.O_RDWR | os.O_CREAT elif "w" in mode: flags = os.O_WRONLY | os.O_CREAT | os.O_TRUNC elif "rws" in mode: flags = os.O_RDWR | os.O_SYNC elif "rwd" in mode: flags = os.O_RDWR | os.O_DSYNC else: flags = os.O_RDONLY self._fd = os.open(self._path, flags, 0o644) self._file_descriptor = FileDescriptor() self._file_descriptor.setInt(self._fd) self._closed = False def read(self) -> int: if self._closed: return -1 try: data = os.read(self._fd, 1) return data[0] if data else -1 except Exception: return -1 def readBytes(self, b: bytearray, off: int, len_: int) -> int: if self._closed: return -1 try: data = os.read(self._fd, len_) if not data: return -1 b[off:off + len(data)] = data return len(data) except Exception: return -1 def write(self, b: Union[int, bytes, bytearray]): if self._closed: return if isinstance(b, int): os.write(self._fd, bytes([b])) else: os.write(self._fd, bytes(b)) def writeBytes(self, b: bytes, off: int, len_: int): if not self._closed: os.write(self._fd, b[off:off + len_]) def getFilePointer(self) -> int: return os.lseek(self._fd, 0, os.SEEK_CUR) def seek(self, pos: int): os.lseek(self._fd, pos, os.SEEK_SET) def length(self) -> int: return os.lseek(self._fd, 0, os.SEEK_END) def setLength(self, newLength: int): os.truncate(self._fd, newLength) def skipBytes(self, n: int) -> int: try: current = os.lseek(self._fd, 0, os.SEEK_CUR) file_size = os.lseek(self._fd, 0, os.SEEK_END) os.lseek(self._fd, current, os.SEEK_SET) actual = min(n, file_size - current) os.lseek(self._fd, actual, os.SEEK_CUR) return max(0, actual) except Exception: return 0 def close(self): if not self._closed: self._closed = True try: os.close(self._fd) except Exception: pass def getFD(self) -> FileDescriptor: return self._file_descriptor def getChannel(self) -> "FileChannel": return FileChannel(fd=self._fd, mode=self._mode) def readLine(self) -> Optional[str]: if self._closed: return None result = [] while True: try: b = os.read(self._fd, 1) if not b: break c = b[0] if c == ord('\n'): break if c == ord('\r'): b2 = os.read(self._fd, 1) if b2 and b2[0] != ord('\n'): os.lseek(self._fd, -1, os.SEEK_CUR) break result.append(chr(c)) except Exception: break return "".join(result) if result else None def toString(self) -> str: return f"RandomAccessFile(path={self._path}, mode={self._mode})" # ═══════════════════════════════════════════════════════════════ # FileLock # ═══════════════════════════════════════════════════════════════ class FileLock(Stub): """FileLock — 文件锁""" def __init__(self, channel: "FileChannel", position: int = 0, size: int = 0, shared: bool = False): super().__init__() self._channel = channel self._position = position self._size = size self._shared = shared self._valid = True def position(self) -> int: return self._position def size(self) -> int: return self._size def isShared(self) -> bool: return self._shared def isValid(self) -> bool: return self._valid def release(self): self._valid = False def close(self): self.release() def channel(self) -> "FileChannel": return self._channel def toString(self) -> str: return f"FileLock(pos={self._position}, shared={self._shared})" # ═══════════════════════════════════════════════════════════════ # FileChannel (扩展) # ═══════════════════════════════════════════════════════════════ class FileChannel(Stub): """FileChannel — 文件通道 (完整版)""" def __init__(self, fd: int = -1, mode: str = "r", path: str = ""): super().__init__() self._fd = fd self._mode = mode self._path = path self._open = fd >= 0 self._position = 0 def read(self, dst: bytearray, offset: int = 0, length: int = 0) -> int: if not self._open: return -1 if length <= 0: length = len(dst) - offset try: data = os.read(self._fd, length) if not data: return -1 dst[offset:offset + len(data)] = data self._position += len(data) return len(data) except Exception: return -1 def write(self, src: Union[bytes, bytearray], offset: int = 0, length: int = 0) -> int: if not self._open: return 0 if length <= 0: data = bytes(src) else: data = bytes(src[offset:offset + length]) try: written = os.write(self._fd, data) self._position += written return written except Exception: return 0 def position(self) -> int: return self._position def setPosition(self, newPosition: int) -> "FileChannel": if self._open: os.lseek(self._fd, newPosition, os.SEEK_SET) self._position = newPosition return self def size(self) -> int: if not self._open: return 0 current = os.lseek(self._fd, 0, os.SEEK_CUR) size_ = os.lseek(self._fd, 0, os.SEEK_END) os.lseek(self._fd, current, os.SEEK_SET) return size_ def truncate(self, newSize: int) -> "FileChannel": if self._open: os.truncate(self._fd, newSize) return self def force(self, metaData: bool): if self._open: os.fsync(self._fd) def close(self): if self._open: self._open = False try: os.close(self._fd) except Exception: pass def isOpen(self) -> bool: return self._open def lock(self, position: int = 0, size: int = 0, shared: bool = False) -> FileLock: return FileLock(channel=self, position=position, size=size, shared=shared) def tryLock(self, position: int = 0, size: int = 0, shared: bool = False) -> Optional[FileLock]: return FileLock(channel=self, position=position, size=size, shared=shared) def transferTo(self, position: int, count: int, target: "FileChannel") -> int: return 0 def transferFrom(self, src: "FileChannel", position: int, count: int) -> int: return 0 def map(self, mode: int, position: int, size: int) -> Any: return None def toString(self) -> str: return f"FileChannel(fd={self._fd}, open={self._open})" # ═══════════════════════════════════════════════════════════════ # FilePermission (扩展) # ═══════════════════════════════════════════════════════════════ class FilePermission(Stub): """FilePermission — 文件权限""" def __init__(self, path: str = "", actions: str = ""): super().__init__() self._path = path self._actions = actions self._action_set: Set[str] = set(actions.split(",")) if actions else set() def getPath(self) -> str: return self._path def getActions(self) -> str: return self._actions def implies(self, permission: "FilePermission") -> bool: return self._action_set.issuperset(permission._action_set) def impliesByPath(self, path: str) -> bool: return self._path == path def toString(self) -> str: return f'FilePermission("{self._path}", "{self._actions}")' def equals(self, other: Any) -> bool: if not isinstance(other, FilePermission): return False return self._path == other._path and self._actions == other._actions def hashCode(self) -> int: return hash((self._path, self._actions)) & 0x7FFFFFFF # ═══════════════════════════════════════════════════════════════ # FileFilter / FileNameFilter # ═══════════════════════════════════════════════════════════════ class FileFilter(Stub): """FileFilter — 文件过滤器""" def accept(self, pathname: Any) -> bool: return True def toString(self) -> str: return f"FileFilter@{hex(id(self))}" class FileNameFilter(Stub): """FileNameFilter — 文件名过滤器""" def accept(self, dir_: Any, name: str) -> bool: return True def toString(self) -> str: return f"FileNameFilter@{hex(id(self))}" # ═══════════════════════════════════════════════════════════════ # FileVisitOption / FileVisitResult / FileVisitor / SimpleFileVisitor # ═══════════════════════════════════════════════════════════════ class FileVisitOption(Stub): """FileVisitOption — 文件访问选项""" FOLLOW_LINKS = 0 def __init__(self, value: int = FOLLOW_LINKS): super().__init__() self._value = value def getValue(self) -> int: return self._value class FileVisitResult(Stub): """FileVisitResult — 文件访问结果""" CONTINUE = 0 TERMINATE = 1 SKIP_SUBTREE = 2 SKIP_SIBLINGS = 3 def __init__(self, value: int = CONTINUE): super().__init__() self._value = value def getValue(self) -> int: return self._value class FileVisitor(Stub): """FileVisitor — 文件访问器接口""" def preVisitDirectory(self, dir_: Any, attrs: Any) -> FileVisitResult: return FileVisitResult(FileVisitResult.CONTINUE) def visitFile(self, file_: Any, attrs: Any) -> FileVisitResult: return FileVisitResult(FileVisitResult.CONTINUE) def visitFileFailed(self, file_: Any, exc: Exception) -> FileVisitResult: return FileVisitResult(FileVisitResult.CONTINUE) def postVisitDirectory(self, dir_: Any, exc: Optional[Exception]) -> FileVisitResult: return FileVisitResult(FileVisitResult.CONTINUE) def toString(self) -> str: return f"FileVisitor@{hex(id(self))}" class SimpleFileVisitor(FileVisitor): """SimpleFileVisitor — 简单文件访问器""" def preVisitDirectory(self, dir_: Any, attrs: Any) -> FileVisitResult: return FileVisitResult(FileVisitResult.CONTINUE) def visitFile(self, file_: Any, attrs: Any) -> FileVisitResult: return FileVisitResult(FileVisitResult.CONTINUE) def visitFileFailed(self, file_: Any, exc: Exception) -> FileVisitResult: return FileVisitResult(FileVisitResult.CONTINUE) def postVisitDirectory(self, dir_: Any, exc: Optional[Exception]) -> FileVisitResult: return FileVisitResult(FileVisitResult.CONTINUE) def toString(self) -> str: return f"SimpleFileVisitor@{hex(id(self))}" # ═══════════════════════════════════════════════════════════════ # FileTreeWalker / FileTypeDetector / FileSystemProvider # ═══════════════════════════════════════════════════════════════ class FileTreeWalker(Stub): """FileTreeWalker — 文件树遍历器""" def __init__(self, visitor: Optional[FileVisitor] = None, options: List[FileVisitOption] = None, maxDepth: int = WALK_MAX_DEPTH_UNLIMITED): super().__init__() self._visitor = visitor or SimpleFileVisitor() self._options = options or [] self._max_depth = maxDepth def walk(self, start: Any): self._walk_recursive(str(start), 0) def _walk_recursive(self, path: str, depth: int): if self._max_depth >= 0 and depth > self._max_depth: return if not os.path.exists(path): return if os.path.isdir(path): result = self._visitor.preVisitDirectory(path, None) if result.getValue() == FileVisitResult.TERMINATE: return if result.getValue() != FileVisitResult.SKIP_SUBTREE: try: for entry in sorted(os.listdir(path)): child = os.path.join(path, entry) self._walk_recursive(child, depth + 1) except Exception as e: self._visitor.visitFileFailed(path, e) self._visitor.postVisitDirectory(path, None) else: try: self._visitor.visitFile(path, None) except Exception as e: self._visitor.visitFileFailed(path, e) def toString(self) -> str: return f"FileTreeWalker(maxDepth={self._max_depth})" class FileTypeDetector(Stub): """FileTypeDetector — 文件类型检测器""" def probeContentType(self, path: Any) -> Optional[str]: return None def toString(self) -> str: return f"FileTypeDetector@{hex(id(self))}" class FileSystemProvider(Stub): """FileSystemProvider — 文件系统提供器""" SCHEME_FILE = "file" def __init__(self, scheme: str = SCHEME_FILE): super().__init__() self._scheme = scheme def getScheme(self) -> str: return self._scheme def newFileSystem(self, uri: Any, env: Dict[str, Any] = None) -> Any: return None def getFileSystem(self, uri: Any) -> Any: return None def getPath(self, uri: Any) -> str: return str(uri) def newFileChannel(self, path: Any, options: Set[str] = None, attrs: List[Any] = None) -> Optional[FileChannel]: return None def newInputStream(self, path: Any, options: Set[str] = None) -> Optional[FileInputStream]: return None def newOutputStream(self, path: Any, options: Set[str] = None) -> Optional[FileOutputStream]: return None def toString(self) -> str: return f"FileSystemProvider(scheme={self._scheme})" # ═══════════════════════════════════════════════════════════════ # FileStore / FileStoreAttributeView / FileAttribute / FileAttributeView # ═══════════════════════════════════════════════════════════════ class FileAttribute(Stub): """FileAttribute — 文件属性""" def __init__(self, name: str = "", value: Any = None): super().__init__() self._name = name self._value = value def name(self) -> str: return self._name def value(self) -> Any: return self._value def toString(self) -> str: return f"FileAttribute(name={self._name})" class FileAttributeView(Stub): """FileAttributeView — 文件属性视图基类""" def name(self) -> str: return "basic" def toString(self) -> str: return f"FileAttributeView(name={self.name()})" class FileStoreAttributeView(Stub): """FileStoreAttributeView — 文件存储属性视图基类""" def name(self) -> str: return "basic" def toString(self) -> str: return f"FileStoreAttributeView(name={self.name()})" class FileStore(Stub): """FileStore — 文件存储""" def __init__(self, path: str = "/", type_: str = "ext4", isReadOnly: bool = False): super().__init__() self._path = path self._type = type_ self._is_read_only = isReadOnly def name(self) -> str: return self._path def type(self) -> str: return self._type def isReadOnly(self) -> bool: return self._is_read_only def getTotalSpace(self) -> int: try: st = os.statvfs(self._path) return st.f_frsize * st.f_blocks except Exception: return 0 def getUsableSpace(self) -> int: try: st = os.statvfs(self._path) return st.f_frsize * st.f_bavail except Exception: return 0 def getUnallocatedSpace(self) -> int: try: st = os.statvfs(self._path) return st.f_frsize * st.f_bfree except Exception: return 0 def getFileStoreAttributeView(self, viewType: Any) -> Optional[FileStoreAttributeView]: return None def supportsFileAttributeView(self, viewType: Any) -> bool: return True def toString(self) -> str: return f"FileStore(path={self._path}, type={self._type})" # ═══════════════════════════════════════════════════════════════ # FileSystem (扩展) / UnixFileSystem / WinNTFileSystem # ═══════════════════════════════════════════════════════════════ class FileSystem(Stub): """FileSystem — 文件系统 (扩展版)""" def __init__(self): super().__init__() self._separator = os.sep self._path_separator = os.pathsep self._normalized_separator = "/" if os.sep == "/" else "\\" def getSeparator(self) -> str: return self._separator def getPathSeparator(self) -> str: return self._path_separator def normalize(self, path: str) -> str: return os.path.normpath(path) def resolve(self, parent: str, child: str) -> str: return os.path.join(parent, child) def getLength(self, path: str) -> int: try: return os.path.getsize(path) except Exception: return 0 def getLastModifiedTime(self, path: str) -> int: try: return int(os.path.getmtime(path) * 1000) except Exception: return 0 def getLastAccessTime(self, path: str) -> int: try: st = os.stat(path) return int(st.st_atime * 1000) except Exception: return 0 def getCreationTime(self, path: str) -> int: try: st = os.stat(path) if hasattr(st, 'st_birthtime'): return int(st.st_birthtime * 1000) return int(st.st_ctime * 1000) except Exception: return 0 def isDirectory(self, path: str) -> bool: return os.path.isdir(path) def isFile(self, path: str) -> bool: return os.path.isfile(path) def exists(self, path: str) -> bool: return os.path.exists(path) def canRead(self, path: str) -> bool: return os.access(path, os.R_OK) def canWrite(self, path: str) -> bool: return os.access(path, os.W_OK) def canExecute(self, path: str) -> bool: return os.access(path, os.X_OK) def isAbsolute(self, path: str) -> bool: return os.path.isabs(path) def list(self, path: str) -> List[str]: try: return os.listdir(path) except Exception: return [] def mkdir(self, path: str) -> bool: try: os.mkdir(path) return True except Exception: return False def mkdirs(self, path: str) -> bool: try: os.makedirs(path, exist_ok=True) return True except Exception: return False def rename(self, src: str, dst: str) -> bool: try: os.rename(src, dst) return True except Exception: return False def delete(self, path: str) -> bool: try: if os.path.isdir(path): os.rmdir(path) else: os.unlink(path) return True except Exception: return False def getStatus(self, path: str) -> Optional[FileStatus]: try: st = os.stat(path) return FileStatus( type_=FILE_TYPE_DIRECTORY if stat.S_ISDIR(st.st_mode) else FILE_TYPE_REGULAR if stat.S_ISREG(st.st_mode) else FILE_TYPE_SYMBOLIC_LINK if stat.S_ISLNK(st.st_mode) else FILE_TYPE_UNKNOWN, size=st.st_size, lastModified=int(st.st_mtime * 1000), lastAccess=int(st.st_atime * 1000), creationTime=int(st.st_ctime * 1000), isSymbolicLink=os.path.islink(path), device=st.st_dev, inode=st.st_ino, mode=st.st_mode, uid=st.st_uid, gid=st.st_gid, ) except Exception: return None def getSpace(self, path: str) -> Tuple[int, int, int]: try: st = os.statvfs(path) total = st.f_frsize * st.f_blocks free = st.f_frsize * st.f_bfree available = st.f_frsize * st.f_bavail return (total, free, available) except Exception: return (0, 0, 0) def toString(self) -> str: return f"FileSystem(sep={self._separator})" def equals(self, other: Any) -> bool: return self is other def hashCode(self) -> int: return id(self) & 0x7FFFFFFF class UnixFileSystem(FileSystem): """UnixFileSystem — Unix 文件系统""" def __init__(self): super().__init__() self._separator = "/" self._path_separator = ":" def normalize(self, path: str) -> str: return os.path.normpath(path) def resolve(self, parent: str, child: str) -> str: if not parent: return child if not child: return parent return f"{parent.rstrip('/')}/{child.lstrip('/')}" def toString(self) -> str: return "UnixFileSystem" class WinNTFileSystem(FileSystem): """WinNTFileSystem — Windows NT 文件系统""" def __init__(self): super().__init__() self._separator = "\\" self._path_separator = ";" self._normalized_separator = "\\" def normalize(self, path: str) -> str: return path.replace("/", "\\") def resolve(self, parent: str, child: str) -> str: if not parent: return child if not child: return parent sep = "\\" parent_clean = parent.rstrip(sep) child_clean = child.lstrip(sep) return f"{parent_clean}{sep}{child_clean}" def toString(self) -> str: return "WinNTFileSystem" # ═══════════════════════════════════════════════════════════════ # UserPrincipal / GroupPrincipal / UserPrincipalLookupService # ═══════════════════════════════════════════════════════════════ class UserPrincipal(Stub): """UserPrincipal — 用户主体""" def __init__(self, name: str = ""): super().__init__() self._name = name def getName(self) -> str: return self._name def toString(self) -> str: return self._name def equals(self, other: Any) -> bool: if not isinstance(other, UserPrincipal): return False return self._name == other._name def hashCode(self) -> int: return hash(self._name) & 0x7FFFFFFF class GroupPrincipal(UserPrincipal): """GroupPrincipal — 组主体""" def __init__(self, name: str = ""): super().__init__(name=name) def toString(self) -> str: return f"Group({self._name})" class UserPrincipalLookupService(Stub): """UserPrincipalLookupService — 用户主体查找服务""" def lookupPrincipalByName(self, name: str) -> UserPrincipal: return UserPrincipal(name=name) def lookupPrincipalByGroupName(self, name: str) -> GroupPrincipal: return GroupPrincipal(name=name) def toString(self) -> str: return "UserPrincipalLookupService" # ═══════════════════════════════════════════════════════════════ # FileSystems (工厂) # ═══════════════════════════════════════════════════════════════ class FileSystems(Stub): """FileSystems — 文件系统工厂""" _default: Optional[FileSystem] = None _unix_default: Optional[UnixFileSystem] = None @staticmethod def getDefault() -> FileSystem: if sys.platform == "win32": if FileSystems._default is None: FileSystems._default = WinNTFileSystem() return FileSystems._default else: if FileSystems._unix_default is None: FileSystems._unix_default = UnixFileSystem() return FileSystems._unix_default @staticmethod def newFileSystem(uri: Any, env: Dict[str, Any] = None) -> Any: return None @staticmethod def newFileSystemFromPath(path: Any, loader: Any = None) -> Any: return None def toString(self) -> str: return "FileSystems"