""" Android Content 组件 — Context, ContentResolver, Uri, SharedPreferences 等 方法签名完全对齐 Android SDK """ import json import mimetypes import os import sys import threading from typing import Any, Dict, List, Optional, Set, Tuple, Union from urllib.parse import urlparse, urlunparse, urlencode, parse_qs, quote, unquote from .._core import ( Stub, AndroidOnlyError, android_context, resolve_android_path, to_android_path, _sdcard_root, _app_data_dir, _package_name, ) try: from .._linux_impl import IS_LINUX, LinuxClipboard, LinuxNotify, LinuxLocation, LinuxProcess _USE_LINUX = IS_LINUX except ImportError: _USE_LINUX = False # ═══════════════════════════════════════════════════════════════ # Uri # ═══════════════════════════════════════════════════════════════ class Uri(Stub): """Android Uri — 统一资源标识符""" def __init__(self, scheme: str = "", authority: str = "", path: str = "", query: str = "", fragment: str = "", host: str = "", port: int = -1, ssp: str = "", user_info: str = ""): super().__init__() self._scheme = scheme self._authority = authority self._path = path self._query = query self._fragment = fragment self._host = host self._port = port self._ssp = ssp self._user_info = user_info self._opaque = bool(ssp) self._cached_string: Optional[str] = None @staticmethod def parse(uri_string: str) -> "Uri": parsed = urlparse(uri_string) host = parsed.hostname or "" port = parsed.port or -1 return Uri( scheme=parsed.scheme, authority=parsed.netloc, path=parsed.path, query=parsed.query, fragment=parsed.fragment, host=host, port=port, user_info=parsed.username or "", ) @staticmethod def fromParts(scheme: str, ssp: str, fragment: str) -> "Uri": uri = Uri(scheme=scheme, ssp=ssp, fragment=fragment) uri._opaque = True return uri @staticmethod def withAppendedPath(base_uri: "Uri", path_segment: str) -> "Uri": new_path = base_uri._path.rstrip("/") + "/" + path_segment return Uri( scheme=base_uri._scheme, authority=base_uri._authority, path=new_path, query=base_uri._query, fragment=base_uri._fragment, host=base_uri._host, port=base_uri._port, ) @staticmethod def encode(s: str) -> str: return quote(s, safe=":/?#[]@!$&'()*+,;=-._~") @staticmethod def decode(s: str) -> str: return unquote(s) def buildUpon(self) -> "Uri.Builder": b = Uri.Builder() b._scheme = self._scheme b._authority = self._authority b._path = self._path b._query = self._query b._fragment = self._fragment return b def getScheme(self) -> Optional[str]: return self._scheme or None def getAuthority(self) -> Optional[str]: return self._authority or None def getPath(self) -> Optional[str]: return self._path or None def getQuery(self) -> Optional[str]: return self._query or None def getFragment(self) -> Optional[str]: return self._fragment or None def getHost(self) -> Optional[str]: return self._host or None def getPort(self) -> int: return self._port def getPathSegments(self) -> List[str]: return [s for s in self._path.split("/") if s] if self._path else [] def getLastPathSegment(self) -> Optional[str]: segments = self.getPathSegments() return segments[-1] if segments else None def getUserInfo(self) -> Optional[str]: return self._user_info or None def isAbsolute(self) -> bool: return bool(self._scheme) def isOpaque(self) -> bool: return self._opaque def isHierarchical(self) -> bool: return not self._opaque def getSchemeSpecificPart(self) -> Optional[str]: return self._ssp or None def normalizeScheme(self) -> "Uri": if self._scheme: self._scheme = self._scheme.lower() return self def toString(self) -> str: if self._cached_string is not None: return self._cached_string if self._opaque and self._ssp: result = f"{self._scheme}:{self._ssp}" if self._fragment: result += f"#{self._fragment}" else: parts = [] if self._scheme: parts.append(f"{self._scheme}:") if self._authority: parts.append(f"//{self._authority}") if self._path: parts.append(self._path) if self._query: parts.append(f"?{self._query}") if self._fragment: parts.append(f"#{self._fragment}") result = "".join(parts) self._cached_string = result return result def __repr__(self): return self.toString() def __str__(self): return self.toString() def __eq__(self, other): if isinstance(other, Uri): return self.toString() == other.toString() return NotImplemented def __hash__(self): return hash(self.toString()) class Builder(Stub): """Uri builder pattern""" def __init__(self): super().__init__() self._scheme = "" self._authority = "" self._path = "" self._query = "" self._fragment = "" self._host = "" self._port = -1 self._segments: List[str] = [] def scheme(self, scheme: str) -> "Uri.Builder": self._scheme = scheme return self def authority(self, authority: str) -> "Uri.Builder": self._authority = authority return self def path(self, path: str) -> "Uri.Builder": self._path = path return self def appendPath(self, new_segment: str) -> "Uri.Builder": self._segments.append(new_segment) return self def query(self, query: str) -> "Uri.Builder": self._query = query return self def appendQueryParameter(self, key: str, value: str) -> "Uri.Builder": params = dict(parse_qs(self._query, keep_blank_values=True)) if self._query else {} if key in params: params[key].append(value) else: params[key] = [value] self._query = urlencode(params, doseq=True) return self def fragment(self, fragment: str) -> "Uri.Builder": self._fragment = fragment return self def opaquePart(self, opaque_part: str) -> "Uri.Builder": return self def build(self) -> "Uri": path = self._path if self._segments: path = (path.rstrip("/") if path else "") + "/" + "/".join(self._segments) return Uri(self._scheme, self._authority, path, self._query, self._fragment) # ═══════════════════════════════════════════════════════════════ # ContentValues # ═══════════════════════════════════════════════════════════════ class ContentValues(Stub): """ContentValues — 类似 dict 的内容值容器""" def __init__(self, from_dict: Optional[Dict[str, Any]] = None): super().__init__() self._values: Dict[str, Any] = dict(from_dict) if from_dict else {} def put(self, key: str, value: Any): self._values[key] = value def putNull(self, key: str): self._values[key] = None def get(self, key: str) -> Optional[Any]: return self._values.get(key) def getAsString(self, key: str) -> Optional[str]: v = self._values.get(key) return str(v) if v is not None else None def getAsInteger(self, key: str) -> Optional[int]: v = self._values.get(key) if v is None: return None try: return int(v) except (ValueError, TypeError): return None def getAsLong(self, key: str) -> Optional[int]: return self.getAsInteger(key) def getAsFloat(self, key: str) -> Optional[float]: v = self._values.get(key) if v is None: return None try: return float(v) except (ValueError, TypeError): return None def getAsDouble(self, key: str) -> Optional[float]: return self.getAsFloat(key) def getAsBoolean(self, key: str) -> Optional[bool]: v = self._values.get(key) if v is None: return None if isinstance(v, bool): return v if isinstance(v, str): return v.lower() in ("true", "1", "yes", "y") return bool(v) def getAsByteArray(self, key: str) -> Optional[bytes]: v = self._values.get(key) if isinstance(v, bytes): return v if isinstance(v, str): return v.encode("utf-8") return None def containsKey(self, key: str) -> bool: return key in self._values def valueSet(self) -> Set[str]: return set(self._values.keys()) def keySet(self) -> Set[str]: return set(self._values.keys()) def size(self) -> int: return len(self._values) def isEmpty(self) -> bool: return len(self._values) == 0 def clear(self): self._values.clear() def remove(self, key: str): self._values.pop(key, None) def putAll(self, other: "ContentValues"): if other: self._values.update(other._values) def equals(self, other) -> bool: if isinstance(other, ContentValues): return self._values == other._values return False def hashCode(self) -> int: return hash(frozenset(self._values.items())) & 0x7FFFFFFF def toString(self) -> str: return f"ContentValues({self._values})" def __getitem__(self, key: str): return self._values[key] def __setitem__(self, key: str, value: Any): self._values[key] = value def __contains__(self, key: str) -> bool: return key in self._values def __len__(self) -> int: return len(self._values) def __repr__(self): return self.toString() # ═══════════════════════════════════════════════════════════════ # ContentResolver # ═══════════════════════════════════════════════════════════════ class ContentResolver(Stub): """ContentResolver — 内容解析器""" NOTIFY_SYNC_TO_NETWORK = 1 NOTIFY_SKIP_NOTIFY_FOR_DESCENDANTS = 2 def __init__(self): super().__init__() self._observers: Dict[str, List[Any]] = {} @staticmethod def _uri_to_file_path(uri) -> Optional[str]: """将 file:// 或纯路径 Uri 转换为本地文件系统路径""" try: uri_str = str(uri) except Exception: return None # 处理字符串输入 if isinstance(uri, str): if uri.startswith("file://"): return uri[7:] if "://" not in uri: return uri if uri else None return None scheme = getattr(uri, "_scheme", "") or "" path = getattr(uri, "_path", "") or "" if scheme == "file": return path if path else None if scheme == "" and path: return path return None def query(self, uri: Uri, projection: Optional[List[str]], selection: Optional[str], selection_args: Optional[List[str]], sort_order: Optional[str], cancellation_signal: Any = None) -> Any: """Query content provider. 对 file:// URI 返回基于目录列表的 Cursor。""" path = self._uri_to_file_path(uri) if path is not None and os.path.isdir(path): try: from .database import Cursor, MatrixCursor cols = projection or ["_id", "name", "size", "last_modified"] cursor = MatrixCursor(cols) for idx, name in enumerate(sorted(os.listdir(path))): full = os.path.join(path, name) try: st = os.stat(full) size = st.st_size mtime = int(st.st_mtime * 1000) except OSError: size = 0 mtime = 0 cursor.addRow([idx, name, size, mtime]) return cursor except Exception: pass return None def insert(self, uri: Uri, values: ContentValues) -> Optional[Uri]: """Insert. 对 file:// URI 将 ContentValues 写为 JSON 文件。""" path = self._uri_to_file_path(uri) if path is not None and isinstance(values, ContentValues): try: os.makedirs(path if os.path.isdir(path) else os.path.dirname(path) or ".", exist_ok=True) target = path if not os.path.isdir(path) else os.path.join(path, "insert.json") data = values._values if hasattr(values, "_values") else {} with open(target, "w", encoding="utf-8") as f: json.dump(data, f, ensure_ascii=False, default=str) return Uri.withAppendedPath(uri, os.path.basename(target)) except OSError: pass return uri def update(self, uri: Uri, values: ContentValues, selection: Optional[str], selection_args: Optional[List[str]]) -> int: """Update. 对 file:// URI 更新对应 JSON 文件。""" path = self._uri_to_file_path(uri) if path is not None and isinstance(values, ContentValues) and os.path.isfile(path): try: data = values._values if hasattr(values, "_values") else {} with open(path, "r", encoding="utf-8") as f: existing = json.load(f) if isinstance(existing, dict): existing.update(data) with open(path, "w", encoding="utf-8") as f: json.dump(existing, f, ensure_ascii=False, default=str) return 1 except (OSError, ValueError): pass return 0 def delete(self, uri: Uri, selection: Optional[str], selection_args: Optional[List[str]]) -> int: """Delete. 对 file:// URI 删除对应文件或目录。""" path = self._uri_to_file_path(uri) if path is not None: try: if os.path.isfile(path): os.remove(path) return 1 if os.path.isdir(path): import shutil shutil.rmtree(path) return 1 except OSError: pass return 0 def applyBatch(self, authority: str, operations: List[Any]) -> List[Any]: """Apply batch operations, returning per-op results.""" results: List[Any] = [] for op in operations: try: op_type = getattr(op, "_type", None) uri = getattr(op, "_uri", None) values = getattr(op, "_values", None) if op_type == 1 and uri is not None: # INSERT results.append(ContentProviderResult(uri=self.insert(uri, values or ContentValues()))) elif op_type == 2 and uri is not None: # UPDATE count = self.update(uri, values or ContentValues(), getattr(op, "_selection", None), getattr(op, "_selection_args", None)) results.append(ContentProviderResult(count=count)) elif op_type == 3 and uri is not None: # DELETE count = self.delete(uri, getattr(op, "_selection", None), getattr(op, "_selection_args", None)) results.append(ContentProviderResult(count=count)) else: results.append(ContentProviderResult(count=0)) except Exception: results.append(ContentProviderResult(count=0)) return results def bulkInsert(self, uri: Uri, values: List[ContentValues]) -> int: count = 0 for v in values or []: if self.insert(uri, v) is not None: count += 1 return count def call(self, uri: Uri, method: str, arg: Optional[str], extras: Optional[ContentValues]) -> Optional[ContentValues]: return None def notifyChange(self, uri: Uri, observer: Any, sync_to_network: bool = False): uri_str = str(uri) if uri_str in self._observers: for obs in list(self._observers[uri_str]): try: obs.onChange(True) except Exception: pass def registerContentObserver(self, uri: Uri, notify_for_descendants: bool, observer: Any): uri_str = str(uri) if uri_str not in self._observers: self._observers[uri_str] = [] self._observers[uri_str].append(observer) def unregisterContentObserver(self, observer: Any): for uri_str in list(self._observers.keys()): if observer in self._observers[uri_str]: self._observers[uri_str].remove(observer) if not self._observers[uri_str]: del self._observers[uri_str] def getType(self, uri) -> Optional[str]: """返回 URI 指向资源的 MIME 类型。""" path = self._uri_to_file_path(uri) if path: mime, _ = mimetypes.guess_type(path) return mime or "application/octet-stream" scheme = getattr(uri, "_scheme", "") or "" if isinstance(uri, str) and "://" in uri: scheme = uri.split("://")[0] if scheme == "content": return "vnd.android.cursor.dir/vnd.android.content" return None def acquireContentProviderClient(self, uri: Uri) -> Optional["ContentProviderClient"]: return ContentProviderClient(uri) def acquireUnstableContentProviderClient(self, uri: Uri) -> Optional["ContentProviderClient"]: return ContentProviderClient(uri) def getStreamTypes(self, uri: Uri, mime_type_filter: str) -> Optional[List[str]]: """返回 URI 支持的流 MIME 类型列表。""" mime = self.getType(uri) if mime is None: return None if mime_type_filter and mime_type_filter.endswith("/*"): prefix = mime_type_filter[:-1] if mime.startswith(prefix): return [mime] return None if mime == mime_type_filter: return [mime] return [mime] if mime else None def openInputStream(self, uri: Uri) -> Any: """打开输入流。对 file:// URI 返回真实文件流。""" path = self._uri_to_file_path(uri) if path and os.path.isfile(path): try: return open(path, "rb") except OSError: return None return None def openOutputStream(self, uri: Uri) -> Any: """打开输出流。对 file:// URI 返回真实文件流。""" path = self._uri_to_file_path(uri) if path: try: d = os.path.dirname(path) if d: os.makedirs(d, exist_ok=True) return open(path, "wb") except OSError: return None return None # ═══════════════════════════════════════════════════════════════ # Context # ═══════════════════════════════════════════════════════════════ class Context(Stub): """Android Context — 应用上下文""" # Service 常量 POWER_SERVICE = "power" WINDOW_SERVICE = "window" LAYOUT_INFLATER_SERVICE = "layout_inflater" ACTIVITY_SERVICE = "activity" ALARM_SERVICE = "alarm" NOTIFICATION_SERVICE = "notification" ACCESSIBILITY_SERVICE = "accessibility" KEYGUARD_SERVICE = "keyguard" LOCATION_SERVICE = "location" SENSOR_SERVICE = "sensor" STORAGE_SERVICE = "storage" VIBRATOR_SERVICE = "vibrator" CONNECTIVITY_SERVICE = "connectivity" WIFI_SERVICE = "wifi" AUDIO_SERVICE = "audio" TELEPHONY_SERVICE = "telephony" INPUT_METHOD_SERVICE = "input_method" CAMERA_SERVICE = "camera" BATTERY_SERVICE = "battery" CLIPBOARD_SERVICE = "clipboard" # Mode 常量 MODE_PRIVATE = 0x0000 MODE_WORLD_READABLE = 0x0001 MODE_WORLD_WRITEABLE = 0x0002 MODE_APPEND = 0x8000 MODE_MULTI_PROCESS = 0x0004 # Flag 常量 BIND_AUTO_CREATE = 0x0001 BIND_DEBUG_UNBIND = 0x0002 BIND_NOT_FOREGROUND = 0x0004 BIND_ABOVE_CLIENT = 0x0008 BIND_ALLOW_OOM_MANAGEMENT = 0x0010 BIND_WAIVE_PRIORITY = 0x0020 BIND_IMPORTANT = 0x0040 BIND_ADJUST_WITH_ACTIVITY = 0x0080 def __init__(self): super().__init__() self._package_name = _package_name self._services: Dict[str, Any] = {} self._shared_prefs: Dict[str, "SharedPreferences"] = {} self._registered_receivers: List[Dict] = [] self._content_resolver = ContentResolver() def getPackageName(self) -> str: return self._package_name def getFilesDir(self): return os.path.join(_app_data_dir, "files") def getCacheDir(self): return os.path.join(_app_data_dir, "cache") def getCodeCacheDir(self): return os.path.join(_app_data_dir, "code_cache") def getNoBackupFilesDir(self): return os.path.join(_app_data_dir, "no_backup") def getExternalFilesDir(self, type_: Optional[str] = None) -> str: base = os.path.join(_sdcard_root, "Android", "data", self._package_name, "files") if type_: base = os.path.join(base, type_) os.makedirs(base, exist_ok=True) return base def getExternalCacheDir(self): base = os.path.join(_sdcard_root, "Android", "data", self._package_name, "cache") os.makedirs(base, exist_ok=True) return base def getObbDir(self): base = os.path.join(_sdcard_root, "Android", "obb", self._package_name) os.makedirs(base, exist_ok=True) return base def getDataDir(self): return _app_data_dir def getPackageCodePath(self) -> str: return os.path.join(_app_data_dir, "base.apk") def getPackageResourcePath(self) -> str: return os.path.join(_app_data_dir, "base.apk") def startActivity(self, intent: Any): sys.stderr.write(f"[Context] startActivity: {intent}\n") def startActivities(self, intents: List[Any], options: Optional[Any] = None): for intent in intents: self.startActivity(intent) def sendBroadcast(self, intent: Any): sys.stderr.write(f"[Context] sendBroadcast: {intent}\n") for receiver in self._registered_receivers: if self._matches_intent_filter(intent, receiver.get("filter")): try: receiver["receiver"].onReceive(self, intent) except Exception as e: sys.stderr.write(f"[Context] broadcast error: {e}\n") def sendOrderedBroadcast(self, intent: Any, receiver_permission: Optional[str] = None): self.sendBroadcast(intent) def sendStickyBroadcast(self, intent: Any): self.sendBroadcast(intent) def registerReceiver(self, receiver: Optional[Any], filter: Any) -> Optional[Any]: if receiver: self._registered_receivers.append({"receiver": receiver, "filter": filter}) return receiver def unregisterReceiver(self, receiver: Any): self._registered_receivers = [ r for r in self._registered_receivers if r["receiver"] is not receiver ] def _matches_intent_filter(self, intent: Any, intent_filter: Any) -> bool: if intent_filter is None: return True action = getattr(intent, "getAction", lambda: None)() if intent_filter._action and action != intent_filter._action: return False return True def getContentResolver(self) -> ContentResolver: return self._content_resolver def getSharedPreferences(self, name: str, mode: int = MODE_PRIVATE) -> "SharedPreferences": if name not in self._shared_prefs: self._shared_prefs[name] = SharedPreferences(name) return self._shared_prefs[name] def getPackageManager(self): from .content_pm import PackageManager return PackageManager() def getResources(self): from .content_res import Resources, AssetManager if not hasattr(self, "_resources") or self._resources is None: self._resources = Resources(self.getAssets()) return self._resources def getAssets(self): from .content_res import AssetManager if not hasattr(self, "_assets") or self._assets is None: self._assets = AssetManager() return self._assets def getMainLooper(self): from .os import Looper return Looper.getMainLooper() def getApplicationContext(self): return self def getSystemService(self, name: str) -> Any: if name == Context.ACTIVITY_SERVICE: from .app import ActivityManager return ActivityManager() if name == Context.ALARM_SERVICE: from .app import AlarmManager return AlarmManager() if name == Context.NOTIFICATION_SERVICE: from .app import NotificationManager return NotificationManager() if name == Context.VIBRATOR_SERVICE: from .os import Vibrator return Vibrator() if name == Context.POWER_SERVICE: from .os import PowerManager return PowerManager() if name == Context.LOCATION_SERVICE: from .location import LocationManager return LocationManager() if name == Context.SENSOR_SERVICE: from .hardware import SensorManager return SensorManager() if name == Context.CONNECTIVITY_SERVICE: from .net import ConnectivityManager return ConnectivityManager() if name == Context.WIFI_SERVICE: from .net_wifi import WifiManager return WifiManager() if name == Context.AUDIO_SERVICE: from .media import AudioManager return AudioManager() if name == Context.TELEPHONY_SERVICE: from .telephony import TelephonyManager return TelephonyManager() if name == Context.INPUT_METHOD_SERVICE: from .view_inputmethod import InputMethodManager return InputMethodManager() if name == Context.CAMERA_SERVICE: from .hardware_camera2 import CameraManager return CameraManager() if name == Context.BATTERY_SERVICE: from .os import BatteryManager return BatteryManager() if name == Context.CLIPBOARD_SERVICE: return ClipboardManager() return self._services.get(name) def startService(self, intent: Any) -> Optional[Any]: sys.stderr.write(f"[Context] startService: {intent}\n") return None def stopService(self, intent: Any) -> bool: return True def bindService(self, intent: Any, conn: Any, flags: int) -> bool: sys.stderr.write(f"[Context] bindService: {intent}\n") return True def unbindService(self, conn: Any): pass def getString(self, res_id: int, *args) -> str: """返回资源字符串。优先委托给 Resources，否则返回合理默认值。""" try: res = self.getResources() val = res.getString(res_id, *args) if val is not None: return val except Exception: pass return str(res_id) def getColor(self, color_id: int) -> int: """返回资源颜色。优先委托给 Resources，否则返回不透明黑色。""" try: res = self.getResources() return res.getColor(color_id) except Exception: return 0xFF000000 def getDrawable(self, drawable_id: int): """返回 Drawable。优先委托给 Resources。""" try: res = self.getResources() return res.getDrawable(drawable_id) except Exception: return None def getWallpaper(self): """返回壁纸 Drawable。桌面端返回 None（无系统壁纸）。""" return None def getWallpaperDesiredMinimumWidth(self) -> int: return 1080 def getWallpaperDesiredMinimumHeight(self) -> int: return 1920 def checkCallingPermission(self, permission: str) -> int: from .content_pm import PackageManager return PackageManager.PERMISSION_GRANTED def checkSelfPermission(self, permission: str) -> int: from .content_pm import PackageManager return PackageManager.PERMISSION_GRANTED def shouldShowRequestPermissionRationale(self, permission: str) -> bool: return False def enforcePermission(self, permission: str, pid: int, uid: int, message: Optional[str]): pass def createPackageContext(self, package_name: str, flags: int) -> "Context": ctx = Context() ctx._package_name = package_name return ctx def isDeviceProtectedStorage(self) -> bool: return False def isCredentialProtectedStorage(self) -> bool: return False def deleteFile(self, name: str) -> bool: path = os.path.join(self.getFilesDir(), name) try: os.remove(path) return True except OSError: return False def fileList(self) -> List[str]: try: return os.listdir(self.getFilesDir()) except OSError: return [] def getDatabasePath(self, name: str): path = os.path.join(_app_data_dir, "databases", name) os.makedirs(os.path.dirname(path), exist_ok=True) return path def databaseList(self) -> List[str]: db_dir = os.path.join(_app_data_dir, "databases") try: return [f for f in os.listdir(db_dir) if f.endswith(".db")] except OSError: return [] def deleteDatabase(self, name: str) -> bool: db_path = os.path.join(_app_data_dir, "databases", name) try: os.remove(db_path) return True except OSError: return False def getDir(self, name: str, mode: int) -> str: path = os.path.join(_app_data_dir, name) os.makedirs(path, exist_ok=True) return path def toString(self) -> str: return f"Context{{{self._package_name}}}" # ─── 文件流 / 权限 (扩展) ───────────────────────────────── def openFileInput(self, name): path = os.path.join(self.getFilesDir(), name) return open(path, 'rb') def openFileOutput(self, name, mode): path = os.path.join(self.getFilesDir(), name) os.makedirs(os.path.dirname(path), exist_ok=True) return open(path, 'wb') def getFileStreamPath(self, name): from ..java.io import File return File(self.getFilesDir(), name) def grantUriPermission(self, toPackage, uri, modeFlags): pass def revokeUriPermission(self, uri, modeFlags): pass def checkUriPermission(self, uri, pid, uid, modeFlags): return 0 # PERMISSION_GRANTED def checkPermission(self, permission, pid, uid): return 0 # PERMISSION_GRANTED def enforceUriPermission(self, uri, pid, uid, modeFlags, message): pass # ─── 资源 / 主题 (扩展) ─────────────────────────────────── def getApplicationInfo(self): from .content_pm import ApplicationInfo return ApplicationInfo() def getTheme(self): return getattr(self, '_theme', 0) def setTheme(self, resid): self._theme = resid def getText(self, resId): """返回资源文本。优先委托给 Resources。""" try: res = self.getResources() val = res.getText(resId) if val is not None: return val except Exception: pass return str(resId) def getQuantityString(self, resId, quantity, *formatArgs): """返回复数字符串。优先委托给 Resources。""" try: res = self.getResources() return res.getQuantityString(resId, quantity, *formatArgs) except Exception: return str(resId) def getStringArray(self, resId): """返回字符串数组。优先委托给 Resources。""" try: res = self.getResources() val = res.getStringArray(resId) if val is not None: return val except Exception: pass return [] def getTextArray(self, resId): """返回文本数组。优先委托给 Resources。""" try: res = self.getResources() val = res.getTextArray(resId) if val is not None: return val except Exception: pass return [] def getIntArray(self, resId): """返回整型数组。优先委托给 Resources。""" try: res = self.getResources() val = res.getIntArray(resId) if val is not None: return val except Exception: pass return [] # ─── Context 派生 / 回调 (扩展) ─────────────────────────── def createConfigurationContext(self, overrideConfiguration): return Context() def createDisplayContext(self, display): return Context() def registerComponentCallbacks(self, callback): pass def unregisterComponentCallbacks(self, callback): pass def getExternalStoragePublicDirectory(self, type_): from ..java.io import File from .os import Environment return Environment.getExternalStoragePublicDirectory(type_) def getDisplayMetrics(self): from .util import DisplayMetrics return DisplayMetrics() class ContextWrapper(Context): """ContextWrapper — Context 的包装器""" def __init__(self, base: Context): super().__init__() self._base = base def getBaseContext(self) -> Context: return self._base def getPackageName(self) -> str: return self._base.getPackageName() def getFilesDir(self): return self._base.getFilesDir() def getCacheDir(self): return self._base.getCacheDir() def startActivity(self, intent: Any): self._base.startActivity(intent) def sendBroadcast(self, intent: Any): self._base.sendBroadcast(intent) def getContentResolver(self) -> ContentResolver: return self._base.getContentResolver() def getSharedPreferences(self, name: str, mode: int) -> "SharedPreferences": return self._base.getSharedPreferences(name, mode) def getSystemService(self, name: str) -> Any: return self._base.getSystemService(name) def startService(self, intent: Any) -> Optional[Any]: return self._base.startService(intent) def bindService(self, intent: Any, conn: Any, flags: int) -> bool: return self._base.bindService(intent, conn, flags) def registerReceiver(self, receiver: Optional[Any], filter: Any) -> Optional[Any]: return self._base.registerReceiver(receiver, filter) def unregisterReceiver(self, receiver: Any): self._base.unregisterReceiver(receiver) # ═══════════════════════════════════════════════════════════════ # IntentFilter # ═══════════════════════════════════════════════════════════════ class IntentFilter(Stub): """IntentFilter — 描述 Intent 匹配规则""" MATCH_CATEGORY_EMPTY = 0x100000 MATCH_CATEGORY_HOST = 0x200000 MATCH_CATEGORY_PORT = 0x400000 MATCH_CATEGORY_PATH = 0x800000 MATCH_CATEGORY_SCHEME = 0x1000000 MATCH_CATEGORY_SCHEME_SPECIFIC_PART = 0x2000000 MATCH_CATEGORY_TYPE = 0x4000000 NO_MATCH_ACTION = -3 NO_MATCH_CATEGORY = -4 NO_MATCH_DATA = -5 NO_MATCH_TYPE = -6 def __init__(self, action: Optional[str] = None): super().__init__() self._action = action self._actions: List[str] = [] self._categories: List[str] = [] self._data_schemes: List[str] = [] self._data_authorities: List[str] = [] self._data_paths: List[str] = [] self._data_types: List[str] = [] self._priority = 0 if action: self.addAction(action) def addAction(self, action: str): if action not in self._actions: self._actions.append(action) def addCategory(self, category: str): if category not in self._categories: self._categories.append(category) def addDataScheme(self, scheme: str): if scheme not in self._data_schemes: self._data_schemes.append(scheme) def addDataAuthority(self, host: str, port: int = -1): auth = f"{host}:{port}" if port >= 0 else host if auth not in self._data_authorities: self._data_authorities.append(auth) def addDataPath(self, path: str, pattern_type: int = 0): if path not in self._data_paths: self._data_paths.append(path) def addDataType(self, type_: str): if type_ not in self._data_types: self._data_types.append(type_) def hasAction(self, action: str) -> bool: return action in self._actions def hasCategory(self, category: str) -> bool: return category in self._categories def hasDataScheme(self, scheme: str) -> bool: return scheme in self._data_schemes def countActions(self) -> int: return len(self._actions) def getAction(self, index: int) -> str: return self._actions[index] def setPriority(self, priority: int): self._priority = priority def getPriority(self) -> int: return self._priority def match(self, action: str, type_: Optional[str], scheme: Optional[str], data: Optional[Any], categories: Optional[Set[str]], tag: str) -> int: if action and action not in self._actions: return self.NO_MATCH_ACTION return self.MATCH_CATEGORY_EMPTY def matchAction(self, action: str) -> bool: return action in self._actions def matchCategories(self, categories: Set[str]) -> Optional[str]: return None def matchData(self, type_: str, scheme: str, data: Any) -> int: return self.MATCH_CATEGORY_EMPTY def dump(self, du: Any, prefix: str): pass def describeContents(self) -> int: return 0 def toString(self) -> str: return f"IntentFilter(actions={self._actions})" # ═══════════════════════════════════════════════════════════════ # BroadcastReceiver # ═══════════════════════════════════════════════════════════════ class BroadcastReceiver(Stub): """BroadcastReceiver — 广播接收器""" RESULT_OK = -1 RESULT_CANCELED = 0 RESULT_REDIRECT = 1 def __init__(self): super().__init__() self._result_code = self.RESULT_OK self._result_data = "" self._result_extras: Optional[ContentValues] = None self._ordered = False self._initial_sticky = False def onReceive(self, context: Context, intent: Any): """Override this method to handle broadcasts""" pass def setResultCode(self, code: int): self._result_code = code def getResultCode(self) -> int: return self._result_code def setResultData(self, data: str): self._result_data = data def getResultData(self) -> str: return self._result_data def setResult(self, code: int, data: str, extras: Optional[ContentValues]): self._result_code = code self._result_data = data self._result_extras = extras def setResultExtras(self, extras: ContentValues): self._result_extras = extras def getResultExtras(self, make_map: bool) -> Optional[ContentValues]: return self._result_extras def abortBroadcast(self): self._result_code = self.RESULT_CANCELED def isOrderedBroadcast(self) -> bool: return self._ordered def isInitialStickyBroadcast(self) -> bool: return self._initial_sticky def getDebugUnregister(self) -> bool: return False def toString(self) -> str: return f"BroadcastReceiver@{hex(id(self))}" # ═══════════════════════════════════════════════════════════════ # SharedPreferences / SharedPreferencesEditor # ═══════════════════════════════════════════════════════════════ _SHARED_PREFS_DIR = os.path.join(_app_data_dir, "shared_prefs") os.makedirs(_SHARED_PREFS_DIR, exist_ok=True) class SharedPreferences(Stub): """SharedPreferences — JSON 文件实现的持久化键值存储""" def __init__(self, name: str): super().__init__() self._name = name self._lock = threading.Lock() self._file_path = os.path.join(_SHARED_PREFS_DIR, f"{name}.json") self._data: Dict[str, Any] = {} self._load() def _load(self): try: if os.path.exists(self._file_path): with open(self._file_path, "r", encoding="utf-8") as f: self._data = json.load(f) else: self._data = {} except (json.JSONDecodeError, OSError): self._data = {} def _save(self): with self._lock: os.makedirs(os.path.dirname(self._file_path), exist_ok=True) with open(self._file_path, "w", encoding="utf-8") as f: json.dump(self._data, f, indent=2, ensure_ascii=False) def getString(self, key: str, def_value: Optional[str] = None) -> Optional[str]: v = self._data.get(key) return str(v) if v is not None else def_value def getInt(self, key: str, def_value: int = 0) -> int: v = self._data.get(key) if isinstance(v, int) or isinstance(v, float): return int(v) return def_value def getLong(self, key: str, def_value: int = 0) -> int: return self.getInt(key, def_value) def getFloat(self, key: str, def_value: float = 0.0) -> float: v = self._data.get(key) if isinstance(v, (int, float)): return float(v) return def_value def getBoolean(self, key: str, def_value: bool = False) -> bool: v = self._data.get(key) if isinstance(v, bool): return v if isinstance(v, str): return v.lower() in ("true", "1", "yes") return def_value def getAll(self) -> Dict[str, Any]: return dict(self._data) def contains(self, key: str) -> bool: return key in self._data def edit(self) -> "SharedPreferencesEditor": return SharedPreferencesEditor(self) def registerOnSharedPreferenceChangeListener(self, listener): if not hasattr(self, '_listeners'): self._listeners = [] self._listeners.append(listener) def unregisterOnSharedPreferenceChangeListener(self, listener): if hasattr(self, '_listeners'): try: self._listeners.remove(listener) except ValueError: pass def getSharedPrefsFile(self): return self._file_path def toString(self) -> str: return f"SharedPreferences({self._name})" class SharedPreferencesEditor(Stub): """SharedPreferences Editor""" def __init__(self, prefs: SharedPreferences): super().__init__() self._prefs = prefs self._modified: Dict[str, Any] = {} self._removed: Set[str] = set() self._clear_all = False def putString(self, key: str, value: str) -> "SharedPreferencesEditor": self._modified[key] = value self._removed.discard(key) return self def putInt(self, key: str, value: int) -> "SharedPreferencesEditor": self._modified[key] = value self._removed.discard(key) return self def putLong(self, key: str, value: int) -> "SharedPreferencesEditor": self._modified[key] = value self._removed.discard(key) return self def putFloat(self, key: str, value: float) -> "SharedPreferencesEditor": self._modified[key] = value self._removed.discard(key) return self def putBoolean(self, key: str, value: bool) -> "SharedPreferencesEditor": self._modified[key] = value self._removed.discard(key) return self def remove(self, key: str) -> "SharedPreferencesEditor": self._removed.add(key) self._modified.pop(key, None) return self def clear(self) -> "SharedPreferencesEditor": self._clear_all = True self._modified.clear() self._removed.clear() return self def apply(self): """异步保存""" self.commit() def commit(self) -> bool: try: if self._clear_all: self._prefs._data.clear() for key in self._removed: self._prefs._data.pop(key, None) self._prefs._data.update(self._modified) self._prefs._save() # 触发监听器 sp = getattr(self, '_prefs', None) if sp: for key in self._modified: for listener in getattr(sp, '_listeners', []): listener.onSharedPreferenceChanged(sp, key) self._modified.clear() self._removed.clear() self._clear_all = False return True except Exception: return False # ═══════════════════════════════════════════════════════════════ # ComponentName # ═══════════════════════════════════════════════════════════════ class ComponentName(Stub): """ComponentName — 标识包中的特定组件""" def __init__(self, pkg: str, cls: str): super().__init__() self._package = pkg self._classname = cls def getPackageName(self) -> str: return self._package def getClassName(self) -> str: return self._classname def getShortClassName(self) -> str: return self._classname.split(".")[-1] if "." in self._classname else self._classname def flattenToString(self) -> str: return f"{self._package}/{self._classname}" def flattenToShortString(self) -> str: return f"{self._package}/{self.getShortClassName()}" @staticmethod def unflattenFromString(str_: str) -> Optional["ComponentName"]: if "/" in str_: parts = str_.split("/", 1) return ComponentName(parts[0], parts[1]) return None def toString(self) -> str: return f"ComponentInfo{{{self._package}/{self._classname}}}" def __repr__(self): return self.toString() # ═══════════════════════════════════════════════════════════════ # ClipData / ClipDescription # ═══════════════════════════════════════════════════════════════ class ClipDescription(Stub): """ClipDescription — 描述剪贴板数据""" MIMETYPE_TEXT_PLAIN = "text/plain" MIMETYPE_TEXT_HTML = "text/html" MIMETYPE_TEXT_URILIST = "text/uri-list" MIMETYPE_TEXT_INTENT = "text/vnd.android.intent" def __init__(self, label: str, mime_types: List[str]): super().__init__() self._label = label self._mime_types = list(mime_types) self._timestamp = 0 def getLabel(self) -> str: return self._label def getMimeType(self, index: int) -> str: if 0 <= index < len(self._mime_types): return self._mime_types[index] return "" def getMimeTypeCount(self) -> int: return len(self._mime_types) def filterMimeTypes(self, mime_type: str) -> bool: return mime_type in self._mime_types def compareMimeTypes(self, mime_type1: str, mime_type2: str) -> bool: return mime_type1 == mime_type2 def getTimestamp(self) -> int: return self._timestamp def toString(self) -> str: return f"ClipDescription({self._label}, {self._mime_types})" class ClipData(Stub): """ClipData — 剪贴板数据""" def __init__(self, description: ClipDescription, item: "ClipData.Item"): super().__init__() self._description = description self._items: List["ClipData.Item"] = [item] if item else [] class Item(Stub): """单个剪贴板条目""" def __init__(self, text: Optional[str] = None, intent: Optional[Any] = None, uri: Optional[Uri] = None): super().__init__() self._text = text self._intent = intent self._uri = uri self._html_text = None def getText(self) -> Optional[str]: return self._text def getIntent(self): return self._intent def getUri(self) -> Optional[Uri]: return self._uri def getHtmlText(self) -> Optional[str]: return self._html_text def setHtmlText(self, html_text: str): self._html_text = html_text def coerceToText(self, context: Context) -> str: if self._text: return self._text if self._uri: return str(self._uri) if self._intent: return str(self._intent) return "" def coerceToHtmlText(self, context: Context) -> Optional[str]: return self._html_text def toString(self) -> str: return f"ClipData.Item({self._text})" @staticmethod def newPlainText(label: str, text: str) -> "ClipData": desc = ClipDescription(label, [ClipDescription.MIMETYPE_TEXT_PLAIN]) return ClipData(desc, ClipData.Item(text=text)) @staticmethod def newUri(cr: ContentResolver, label: str, uri: Uri) -> "ClipData": desc = ClipDescription(label, [ClipDescription.MIMETYPE_TEXT_URILIST]) return ClipData(desc, ClipData.Item(uri=uri)) @staticmethod def newIntent(label: str, intent: Any) -> "ClipData": desc = ClipDescription(label, [ClipDescription.MIMETYPE_TEXT_INTENT]) return ClipData(desc, ClipData.Item(intent=intent)) def getDescription(self) -> ClipDescription: return self._description def addItem(self, item: "ClipData.Item"): self._items.append(item) def getItemAt(self, index: int) -> Optional["ClipData.Item"]: if 0 <= index < len(self._items): return self._items[index] return None def getItemCount(self) -> int: return len(self._items) def toString(self) -> str: return f"ClipData({self._description})" # ═══════════════════════════════════════════════════════════════ # ContentProvider / ContentProviderClient # ═══════════════════════════════════════════════════════════════ class ContentProvider(Stub): """ContentProvider — 内容提供者基类""" def __init__(self): super().__init__() self._context: Optional[Context] = None self._calling_package = "" def onCreate(self) -> bool: return True def query(self, uri: Uri, projection: Optional[List[str]], selection: Optional[str], selection_args: Optional[List[str]], sort_order: Optional[str]) -> Optional[Any]: """默认实现：委托给关联 Context 的 ContentResolver。""" if self._context is not None: return self._context.getContentResolver().query( uri, projection, selection, selection_args, sort_order) return None def insert(self, uri: Uri, values: ContentValues) -> Optional[Uri]: if self._context is not None: return self._context.getContentResolver().insert(uri, values) return None def update(self, uri: Uri, values: ContentValues, selection: Optional[str], selection_args: Optional[List[str]]) -> int: if self._context is not None: return self._context.getContentResolver().update(uri, values, selection, selection_args) return 0 def delete(self, uri: Uri, selection: Optional[str], selection_args: Optional[List[str]]) -> int: if self._context is not None: return self._context.getContentResolver().delete(uri, selection, selection_args) return 0 def getType(self, uri: Uri) -> Optional[str]: """返回 URI 的 MIME 类型。""" path = ContentResolver._uri_to_file_path(uri) if path: mime, _ = mimetypes.guess_type(path) return mime or "application/octet-stream" return None def getStreamTypes(self, uri: Uri, mime_type_filter: str) -> Optional[List[str]]: mime = self.getType(uri) if mime is None: return None return [mime] def openFile(self, uri: Uri, mode: str) -> Any: """打开文件。对 file:// URI 返回真实文件对象。""" path = ContentResolver._uri_to_file_path(uri) if path and os.path.exists(path): try: fmode = "rb" if "r" in mode else "wb" return open(path, fmode) except OSError: return None return None def openAssetFile(self, uri: Uri, mode: str) -> Any: return self.openFile(uri, mode) def attachInfo(self, context: Context, info: Any): self._context = context def getContext(self) -> Optional[Context]: return self._context def getCallingPackage(self) -> str: return self._calling_package def applyBatch(self, operations: List[Any]) -> List[Any]: return [] def bulkInsert(self, uri: Uri, values: List[ContentValues]) -> int: return len(values) if values else 0 def isTemporary(self) -> bool: return False def shutdown(self): pass def toString(self) -> str: return f"ContentProvider@{hex(id(self))}" class ContentProviderClient(Stub): """ContentProviderClient — 内容提供者客户端""" def __init__(self, uri: Optional[Uri] = None): super().__init__() self._uri = uri def query(self, uri: Uri, projection: Optional[List[str]], selection: Optional[str], selection_args: Optional[List[str]], sort_order: Optional[str]) -> Optional[Any]: """委托给 ContentResolver 处理 file:// URI。""" resolver = ContentResolver() return resolver.query(uri, projection, selection, selection_args, sort_order) def insert(self, uri: Uri, values: ContentValues) -> Optional[Uri]: resolver = ContentResolver() return resolver.insert(uri, values) def update(self, uri: Uri, values: ContentValues, selection: Optional[str], selection_args: Optional[List[str]]) -> int: resolver = ContentResolver() return resolver.update(uri, values, selection, selection_args) def delete(self, uri: Uri, selection: Optional[str], selection_args: Optional[List[str]]) -> int: resolver = ContentResolver() return resolver.delete(uri, selection, selection_args) def applyBatch(self, operations: List[Any]) -> List[Any]: resolver = ContentResolver() return resolver.applyBatch("", operations) def bulkInsert(self, uri: Uri, values: List[ContentValues]) -> int: resolver = ContentResolver() return resolver.bulkInsert(uri, values) def getType(self, uri: Uri) -> Optional[str]: """返回 URI 的 MIME 类型。""" path = ContentResolver._uri_to_file_path(uri) if path: mime, _ = mimetypes.guess_type(path) return mime or "application/octet-stream" return None def getStreamTypes(self, uri: Uri, mime_type_filter: str) -> Optional[List[str]]: mime = self.getType(uri) if mime is None: return None return [mime] def call(self, method: str, arg: Optional[str], extras: Optional[ContentValues]) -> Optional[ContentValues]: return None def release(self): pass def close(self): pass def openFile(self, uri: Uri, mode: str) -> Any: """打开文件。对 file:// URI 返回真实文件对象。""" path = ContentResolver._uri_to_file_path(uri) if path and os.path.exists(path): try: fmode = "rb" if "r" in mode else "wb" return open(path, fmode) except OSError: return None return None def openAssetFile(self, uri: Uri, mode: str) -> Any: return self.openFile(uri, mode) def openTypedAssetFile(self, uri: Uri, mime_type: str, opts: Optional[ContentValues]) -> Any: return self.openFile(uri, "r") def getLocalContentProvider(self) -> Optional[ContentProvider]: return None class ContentProviderOperation(Stub): """ContentProviderOperation — 批量操作""" TYPE_INSERT = 1 TYPE_UPDATE = 2 TYPE_DELETE = 3 TYPE_ASSERT = 4 class Builder(Stub): def __init__(self, type_: int, uri: Uri): super().__init__() self._type = type_ self._uri = uri self._values: Optional[ContentValues] = None self._selection: Optional[str] = None self._selection_args: Optional[List[str]] = None def withValue(self, key: str, value: Any) -> "ContentProviderOperation.Builder": if self._values is None: self._values = ContentValues() self._values.put(key, value) return self def withValues(self, values: ContentValues) -> "ContentProviderOperation.Builder": self._values = values return self def withSelection(self, selection: str, selection_args: List[str]) -> "ContentProviderOperation.Builder": self._selection = selection self._selection_args = selection_args return self def build(self) -> "ContentProviderOperation": op = ContentProviderOperation() op._type = self._type op._uri = self._uri op._values = self._values op._selection = self._selection op._selection_args = self._selection_args return op @staticmethod def newInsert(uri: Uri) -> "ContentProviderOperation.Builder": return ContentProviderOperation.Builder(ContentProviderOperation.TYPE_INSERT, uri) @staticmethod def newUpdate(uri: Uri) -> "ContentProviderOperation.Builder": return ContentProviderOperation.Builder(ContentProviderOperation.TYPE_UPDATE, uri) @staticmethod def newDelete(uri: Uri) -> "ContentProviderOperation.Builder": return ContentProviderOperation.Builder(ContentProviderOperation.TYPE_DELETE, uri) @staticmethod def newAssert(uri: Uri) -> "ContentProviderOperation.Builder": return ContentProviderOperation.Builder(ContentProviderOperation.TYPE_ASSERT, uri) class ContentProviderResult(Stub): """ContentProviderResult — 操作结果""" def __init__(self, uri: Optional[Uri] = None, count: Optional[int] = None): super().__init__() self.uri = uri self.count = count # ═══════════════════════════════════════════════════════════════ # Loader / CursorLoader / DialogInterface # ═══════════════════════════════════════════════════════════════ class Loader(Stub): """Loader — 异步数据加载器基类""" def __init__(self, context: Context): super().__init__() self._context = context self._started = False self._abandoned = False self._listeners: List = [] def deliverResult(self, data: Any): for listener in self._listeners: try: listener.onLoadFinished(self, data) except Exception: pass def startLoading(self): self._started = True def forceLoad(self): self.deliverResult(None) def stopLoading(self): self._started = False def abandon(self): self._abandoned = True def isStarted(self) -> bool: return self._started def isAbandoned(self) -> bool: return self._abandoned def reset(self): self._started = False self._abandoned = False def registerListener(self, listener): if listener not in self._listeners: self._listeners.append(listener) def unregisterListener(self, listener): if listener in self._listeners: self._listeners.remove(listener) def dataToString(self, data: Any) -> str: return str(data) def toString(self) -> str: return f"Loader@{hex(id(self))}" class CursorLoader(Loader): """CursorLoader — 游标加载器""" def __init__(self, context: Context, uri: Uri, projection: Optional[List[str]], selection: Optional[str], selection_args: Optional[List[str]], sort_order: Optional[str]): super().__init__(context) self._uri = uri self._projection = projection self._selection = selection self._selection_args = selection_args self._sort_order = sort_order def getUri(self) -> Uri: return self._uri def setUri(self, uri: Uri): self._uri = uri def getProjection(self) -> Optional[List[str]]: return self._projection def setProjection(self, projection: Optional[List[str]]): self._projection = projection def getSelection(self) -> Optional[str]: return self._selection def setSelection(self, selection: Optional[str]): self._selection = selection def getSelectionArgs(self) -> Optional[List[str]]: return self._selection_args def setSelectionArgs(self, args: Optional[List[str]]): self._selection_args = args def getSortOrder(self) -> Optional[str]: return self._sort_order def setSortOrder(self, sort_order: Optional[str]): self._sort_order = sort_order def loadInBackground(self): """在后台加载数据，委托给 ContentResolver.query。""" if self._context is not None: try: resolver = self._context.getContentResolver() return resolver.query(self._uri, self._projection, self._selection, self._selection_args, self._sort_order) except Exception: pass return None class DialogInterface(Stub): """DialogInterface — 对话框接口常量""" BUTTON_POSITIVE = -1 BUTTON_NEGATIVE = -2 BUTTON_NEUTRAL = -3 def cancel(self): pass def dismiss(self): pass # ═══════════════════════════════════════════════════════════════ # SyncAdapter / SyncResult # ═══════════════════════════════════════════════════════════════ class SyncAdapter(Stub): """SyncAdapter — 同步适配器""" def __init__(self, context: Context, auto_initialize: bool, allow_parallel_syncs: bool): super().__init__() self._context = context def onPerformSync(self, account: Any, extras: ContentValues, authority: str, provider: ContentProviderClient, sync_result: "SyncResult"): pass def onSecurityException(self, account: Any, extras: ContentValues, authority: str, sync_result: "SyncResult"): pass class SyncResult(Stub): """SyncResult — 同步结果""" def __init__(self): super().__init__() self.stats = self._Stats() class _Stats: def __init__(self): self.numEntries = 0 self.numUpdates = 0 self.numDeletes = 0 self.numInserts = 0 self.numSkippedEntries = 0 def clear(self): self.stats = self._Stats() def hasError(self) -> bool: return False def hasSoftError(self) -> bool: return False def hasHardError(self) -> bool: return False def madeSomeProgress(self) -> bool: return self.stats.numEntries > 0 def toString(self) -> str: return f"SyncResult({self.stats.numEntries} entries)" # ═══════════════════════════════════════════════════════════════ # ClipboardManager # ═══════════════════════════════════════════════════════════════ class ClipboardManager(Stub): """ClipboardManager — 剪贴板管理器""" def getText(self): if _USE_LINUX: return LinuxClipboard.read() return "" def setText(self, text): if _USE_LINUX: return LinuxClipboard.write(text) return False def hasText(self): if _USE_LINUX: return len(LinuxClipboard.read()) > 0 return False # ═══════════════════════════════════════════════════════════════ # 替换 _core.py 中的 _GlobalContext # ═══════════════════════════════════════════════════════════════ _real_context = Context() # 将 _core 中的 android_context 替换为真实 Context import importlib try: _core_mod = importlib.import_module(".._core", __package__) _core_mod.android_context.__class__ = Context except Exception: pass # ═══════════════════════════════════════════════════════════════ # __all__ # ═══════════════════════════════════════════════════════════════ __all__ = [ "Uri", "ContentValues", "ContentResolver", "Context", "ContextWrapper", "IntentFilter", "BroadcastReceiver", "SharedPreferences", "SharedPreferencesEditor", "ComponentName", "ClipData", "ClipDescription", "ContentProvider", "ContentProviderClient", "ContentProviderOperation", "ContentProviderResult", "CursorLoader", "Loader", "DialogInterface", "SyncAdapter", "SyncResult", "ClipboardManager", ]