""" Android os.storage 组件 — StorageManager, StorageVolume, DiskInfo, VolumeInfo 等 方法签名完全对齐 Android SDK """ from __future__ import annotations import os import shutil import threading import uuid from typing import Any, Callable, Dict, List, Optional, Set from .._core import Stub, AndroidOnlyError, resolve_android_path from ..java.lang import RuntimeException IllegalArgumentException = type("IllegalArgumentException", (RuntimeException,), {}) # ═══════════════════════════════════════════════════════════════ # StorageVolume # ═══════════════════════════════════════════════════════════════ class StorageVolume(Stub): """StorageVolume — 存储卷""" STATE_MOUNTED = "mounted" STATE_MOUNTED_READ_ONLY = "mounted_ro" STATE_REMOVED = "removed" STATE_UNMOUNTED = "unmounted" STATE_CHECKING = "checking" STATE_EJECTING = "ejecting" STATE_MOUNTABLE = "mountable" STATE_UNMOUNTABLE = "unmountable" STATE_BAD_REMOVAL = "bad_removal" STATE_DAMAGED = "damaged" STORAGE_TYPE_INTERNAL = 0 STORAGE_TYPE_EXTERNAL = 1 def __init__(self, path: str = "/sdcard", description: str = "Internal shared storage", primary: bool = True, removable: bool = False, state: str = STATE_MOUNTED, max_file_size: int = 4 * 1024 * 1024 * 1024, storage_type: int = STORAGE_TYPE_INTERNAL): super().__init__() self._path = path self._description = description self._primary = primary self._removable = removable self._state = state self._max_file_size = max_file_size self._uuid = str(uuid.uuid4()).replace("-", "")[:16] self._user_label = description self._storage_type = storage_type def getPath(self) -> str: return self._path def getDescription(self, context: Any = None) -> str: return self._description def isPrimary(self) -> bool: return self._primary def isRemovable(self) -> bool: return self._removable def getState(self) -> str: return self._state def getMaxFileSize(self) -> int: return self._max_file_size def getUuid(self) -> str: return self._uuid def getUserLabel(self) -> str: return self._user_label def getDirectory(self) -> Any: from pathlib import Path return Path(self._path) if os.path.isdir(self._path) else None def getStorageType(self) -> int: return self._storage_type def createAccessIntent(self, directoryName: str) -> Any: return None def toString(self) -> str: return (f"StorageVolume{{path={self._path}, primary={self._primary}, " f"state={self._state}, removable={self._removable}}}") # ═══════════════════════════════════════════════════════════════ # DiskInfo # ═══════════════════════════════════════════════════════════════ class DiskInfo(Stub): """DiskInfo — 磁盘信息""" def __init__(self, id_: str = "", label: str = "", size: int = 0, is_sd: bool = False, is_usb: bool = False, is_adoptable: bool = False): super().__init__() self._id = id_ self._label = label self._size = size self._is_sd = is_sd self._is_usb = is_usb self._is_adoptable = is_adoptable def getId(self) -> str: return self._id def getLabel(self) -> str: return self._label def getSize(self) -> int: return self._size def isSd(self) -> bool: return self._is_sd def isUsb(self) -> bool: return self._is_usb def isAdoptable(self) -> bool: return self._is_adoptable def toString(self) -> str: return (f"DiskInfo{{id={self._id}, label={self._label}, " f"size={self._size}, sd={self._is_sd}, usb={self._is_usb}}}") # ═══════════════════════════════════════════════════════════════ # VolumeInfo # ═══════════════════════════════════════════════════════════════ class VolumeInfo(Stub): """VolumeInfo — 卷信息""" TYPE_PRIVATE = 1 TYPE_PUBLIC = 2 TYPE_EMULATED = 3 TYPE_STUB = 4 TYPE_OBB = 5 STATE_UNMOUNTED = 0 STATE_CHECKING = 1 STATE_MOUNTED = 2 STATE_MOUNTED_READ_ONLY = 3 STATE_FORMATTING = 4 STATE_EJECTING = 5 STATE_BAD_REMOVAL = 6 STATE_DAMAGED = 7 def __init__(self, id_: str = "", type_: int = TYPE_EMULATED, disk: Optional[DiskInfo] = None, partition: str = "", path: str = "", internal_path: str = "", fs_type: str = "ext4", fs_uuid: str = "", fs_label: str = "", state: int = STATE_MOUNTED, mount_user_id: int = 0): super().__init__() self._id = id_ self._type = type_ self._disk = disk self._partition = partition self._path = path self._internal_path = internal_path self._fs_type = fs_type self._fs_uuid = fs_uuid self._fs_label = fs_label self._state = state self._mount_user_id = mount_user_id def getId(self) -> str: return self._id def getType(self) -> int: return self._type def getDisk(self) -> Optional[DiskInfo]: return self._disk def getPartition(self) -> str: return self._partition def getPath(self) -> Optional[str]: return self._path def getInternalPath(self) -> Optional[str]: return self._internal_path def getFsType(self) -> str: return self._fs_type def getFsUuid(self) -> str: return self._fs_uuid def getFsLabel(self) -> str: return self._fs_label def getState(self) -> int: return self._state def getMountUserId(self) -> int: return self._mount_user_id def toString(self) -> str: return (f"VolumeInfo{{id={self._id}, type={self._type}, " f"state={self._state}, path={self._path}}}") # ═══════════════════════════════════════════════════════════════ # VolumeRecord # ═══════════════════════════════════════════════════════════════ class VolumeRecord(Stub): """VolumeRecord — 卷记录""" def __init__(self, fs_uuid: str = "", part_guid: str = "", type_: int = VolumeInfo.TYPE_PRIVATE, fs_label: str = "", nickname: str = "", created_millis: int = 0, last_seen_millis: int = 0, approved_for_type: bool = False): super().__init__() self._fs_uuid = fs_uuid self._part_guid = part_guid self._type = type_ self._fs_label = fs_label self._nickname = nickname self._created_millis = created_millis self._last_seen_millis = last_seen_millis self._approved_for_type = approved_for_type self._pseudo_uuid = str(uuid.uuid4()) def getFsUuid(self) -> str: return self._fs_uuid def getPartGuid(self) -> str: return self._part_guid def getType(self) -> int: return self._type def getFsLabel(self) -> str: return self._fs_label def getNickname(self) -> str: return self._nickname def getCreatedMillis(self) -> int: return self._created_millis def getLastSeenMillis(self) -> int: return self._last_seen_millis def isApprovedForType(self) -> bool: return self._approved_for_type def getPseudoUuid(self) -> str: return self._pseudo_uuid def toString(self) -> str: return f"VolumeRecord{{uuid={self._fs_uuid}, type={self._type}, label={self._fs_label}}}" # ═══════════════════════════════════════════════════════════════ # StorageManager # ═══════════════════════════════════════════════════════════════ class StorageManager(Stub): """StorageManager — 存储管理器""" ACTION_MANAGE_STORAGE = "android.os.storage.action.MANAGE_STORAGE" ACTION_CLEAR_APP_CACHE = "android.os.storage.action.CLEAR_APP_CACHE" UUID_DEFAULT = "primary_physical" EXTRA_UUID = "android.os.storage.extra.UUID" EXTRA_REQUESTED_BYTES = "android.os.storage.extra.REQUESTED_BYTES" def __init__(self): super().__init__() self._lock = threading.Lock() self._primary_volume = StorageVolume( path="/sdcard", description="Internal shared storage", primary=True, removable=False, ) self._volumes: List[StorageVolume] = [ self._primary_volume, StorageVolume( path="/sdcard/external", description="External SD Card", primary=False, removable=True, state=StorageVolume.STATE_UNMOUNTED, storage_type=StorageVolume.STORAGE_TYPE_EXTERNAL, ), ] # ── Volumes ── def getStorageVolumes(self) -> List[StorageVolume]: with self._lock: return list(self._volumes) def getVolumeList(self) -> List[StorageVolume]: return self.getStorageVolumes() def getPrimaryStorageVolume(self) -> StorageVolume: return self._primary_volume def getPrimaryStorageSize(self) -> int: stat = shutil.disk_usage(os.path.expanduser("~")) return stat.total def getBestFitPrimaryVolume(self) -> StorageVolume: return self._primary_volume def getMountState(self, mountPoint: str) -> bool: return os.path.ismount(mount_point) if os.path.exists(mount_point) else False # ── Mount / Unmount / Format ── def mount(self, volumeId: str) -> int: return 0 def unmount(self, volumeId: str) -> int: return 0 def format(self, volumeId: str) -> int: return 0 # ── Allocation ── def allocateBytes(self, volumeUuid: str, bytes_: int, flags: int = 0) -> int: stat = shutil.disk_usage(os.path.expanduser("~")) if stat.free >= bytes_: return 0 return -1 def allocateBytesForUuid(self, volumeUuid: str, bytes_: int, flags: int = 0) -> int: return self.allocateBytes(volumeUuid, bytes_, flags) def isAllocationSupported(self, volumeUuid: str) -> bool: return True def cacheBehavior(self, uuid_: str) -> int: return 0 # ── Permission ── def checkPermission(self, volumeUuid: str, pid: int, uid: int) -> int: return 0 # ── OBB ── def isObbMounted(self, filename: str) -> bool: return False def mountObb(self, filename: str, key: str, listener: Any) -> bool: return True def unmountObb(self, filename: str, force: bool, listener: Any) -> bool: return True def getMountedObbPath(self, filename: str) -> Optional[str]: return None def toString(self) -> str: return f"StorageManager@{hex(id(self))}" # ═══════════════════════════════════════════════════════════════ # __all__ # ═══════════════════════════════════════════════════════════════ __all__ = [ "StorageManager", "StorageVolume", "DiskInfo", "VolumeInfo", "VolumeRecord", ]