shuiyu1123
506f9ac725
新增模块覆盖: 系统层 (14): system, security, print, opengl, renderscript, transition, test, service_notification, service_controls, service_quicksettings, service_wallpaper, service_autofill, service_dreams, service_vr 应用层 (13): app_admin, app_backup, app_job, app_usage, app_role, app_people, appwidget, companion, credentials, mtp, sip, telecom, speech 硬件媒体层 (15): hardware_display, hardware_fingerprint, hardware_usb, hardware_input, media_audiofx, media_browse, media_session, media_tv, media_midi, media_projection, uwb, se, view_autofill, view_textclassifier, view_translation 总计 94 个模块，75,342 行代码
378 lines
13 KiB
Python
378 lines
13 KiB
Python
"""
|
|
Android Service Notification 组件 — NotificationListenerService, StatusBarNotification, RankingMap 等
|
|
方法签名完全对齐 Android SDK (android.service.notification 包)
|
|
"""
|
|
|
|
import time
|
|
from typing import Any, Callable, Dict, List, Optional, Set, Tuple
|
|
|
|
from .._core import Stub, AndroidOnlyError
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════
|
|
# StatusBarNotification
|
|
# ═══════════════════════════════════════════════════════════════
|
|
|
|
class StatusBarNotification(Stub):
|
|
"""StatusBarNotification — 状态栏通知"""
|
|
|
|
def __init__(self, pkg: str = "", op_pkg: str = "",
|
|
id: int = 0, tag: Optional[str] = None,
|
|
uid: int = 10001, pid: int = 1000,
|
|
user: Any = None, notification: Any = None,
|
|
post_time: int = 0):
|
|
super().__init__()
|
|
self._pkg = pkg
|
|
self._op_pkg = op_pkg or pkg
|
|
self._id = id
|
|
self._tag = tag
|
|
self._uid = uid
|
|
self._pid = pid
|
|
self._user = user
|
|
self._notification = notification
|
|
self._post_time = post_time or int(time.time() * 1000)
|
|
self._is_clearable = True
|
|
self._is_ongoing = False
|
|
self._key = f"{pkg}|{id}|{tag or ''}"
|
|
self._override_group_key: Optional[str] = None
|
|
self._override_public = False
|
|
self._initial_source_id = 0
|
|
self._initial_source_uid = uid
|
|
self._initial_source_pkg = pkg
|
|
|
|
def getPackageName(self) -> str:
|
|
"""获取通知所属包名"""
|
|
return self._pkg
|
|
|
|
def getOpPkg(self) -> str:
|
|
return self._op_pkg
|
|
|
|
def getNotification(self) -> Any:
|
|
"""获取 Notification 对象"""
|
|
return self._notification
|
|
|
|
def getId(self) -> int:
|
|
"""获取通知 ID"""
|
|
return self._id
|
|
|
|
def getTag(self) -> Optional[str]:
|
|
"""获取通知标签"""
|
|
return self._tag
|
|
|
|
def getPostTime(self) -> int:
|
|
"""获取发送时间 (毫秒)"""
|
|
return self._post_time
|
|
|
|
def getUid(self) -> int:
|
|
return self._uid
|
|
|
|
def getPid(self) -> int:
|
|
return self._pid
|
|
|
|
def getUser(self) -> Any:
|
|
return self._user
|
|
|
|
def getKey(self) -> str:
|
|
return self._key
|
|
|
|
def isClearable(self) -> bool:
|
|
"""是否可以清除"""
|
|
return self._is_clearable
|
|
|
|
def isOngoing(self) -> bool:
|
|
"""是否是持续通知"""
|
|
return self._is_ongoing
|
|
|
|
def setClearable(self, clearable: bool):
|
|
self._is_clearable = clearable
|
|
|
|
def setOngoing(self, ongoing: bool):
|
|
self._is_ongoing = ongoing
|
|
|
|
def getOverrideGroupKey(self) -> Optional[str]:
|
|
return self._override_group_key
|
|
|
|
def setOverrideGroupKey(self, override_group_key: Optional[str]):
|
|
self._override_group_key = override_group_key
|
|
|
|
def isGroup(self) -> bool:
|
|
return self._override_group_key is not None
|
|
|
|
def clone(self) -> "StatusBarNotification":
|
|
"""克隆通知"""
|
|
clone = StatusBarNotification(
|
|
pkg=self._pkg, op_pkg=self._op_pkg, id=self._id,
|
|
tag=self._tag, uid=self._uid, pid=self._pid,
|
|
user=self._user, notification=self._notification,
|
|
post_time=self._post_time,
|
|
)
|
|
clone._is_clearable = self._is_clearable
|
|
clone._is_ongoing = self._is_ongoing
|
|
clone._key = self._key
|
|
clone._override_group_key = self._override_group_key
|
|
return clone
|
|
|
|
def toString(self) -> str:
|
|
return (f"StatusBarNotification(pkg={self._pkg}, id={self._id}, "
|
|
f"tag={self._tag}, postTime={self._post_time})")
|
|
|
|
def hashCode(self) -> int:
|
|
return hash(self._key) & 0x7FFFFFFF
|
|
|
|
def equals(self, other) -> bool:
|
|
if not isinstance(other, StatusBarNotification):
|
|
return False
|
|
return self._key == other._key
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════
|
|
# NotificationListenerService
|
|
# ═══════════════════════════════════════════════════════════════
|
|
|
|
class NotificationListenerService(Stub):
|
|
"""NotificationListenerService — 通知监听服务"""
|
|
|
|
# 取消原因
|
|
REASON_LISTENER_CANCEL = 1
|
|
REASON_CLICK = 2
|
|
REASON_CANCEL = 3
|
|
REASON_DELETE = 4
|
|
REASON_PACKAGE_CHANGED = 5
|
|
REASON_USER_STOPPED = 6
|
|
REASON_PACKACE_BANNED = 7
|
|
REASON_APP_CANCEL = 8
|
|
REASON_APP_CANCEL_ALL = 9
|
|
REASON_CHANNEL_BANNED = 10
|
|
REASON_SNOOZED = 11
|
|
REASON_TIMEOUT = 12
|
|
REASON_ERROR = 13
|
|
REASON_GROUP_SUMMARY_CANCELED = 14
|
|
REASON_ASSISTANT_CANCEL = 15
|
|
|
|
# 中断过滤器
|
|
INTERRUPTION_FILTER_NONE = 0
|
|
INTERRUPTION_FILTER_PRIORITY = 1
|
|
INTERRUPTION_FILTER_ALARMS = 2
|
|
INTERRUPTION_FILTER_ALL = 3
|
|
INTERRUPTION_FILTER_UNKNOWN = 4
|
|
|
|
# 提示
|
|
HINT_HOST_DISABLE_CALL_EFFECTS = 1
|
|
HINT_HOST_DISABLE_EFFECTS = 2
|
|
HINT_HOST_DISABLE_NOTIFICATION_EFFECTS = 3
|
|
|
|
# SUW 组件
|
|
TRAMPOLINE_ACTIVITY_CLASS = (
|
|
"com.android.settings.Settings$NotificationAppListActivity"
|
|
)
|
|
|
|
def __init__(self):
|
|
super().__init__()
|
|
self._active_notifications: Dict[str, StatusBarNotification] = {}
|
|
self._keys_list: List[str] = []
|
|
self._interruption_filter: int = (
|
|
NotificationListenerService.INTERRUPTION_FILTER_ALL
|
|
)
|
|
self._ranking_map: Optional["NotificationListenerService.RankingMap"] = None
|
|
self._is_listener_bound = False
|
|
|
|
@staticmethod
|
|
def requestRebind(component_name: Any):
|
|
"""请求重新绑定通知监听器"""
|
|
pass
|
|
|
|
@staticmethod
|
|
def requestUnbind(component_name: Any):
|
|
"""请求解绑通知监听器"""
|
|
pass
|
|
|
|
def onNotificationPosted(self, sbn: StatusBarNotification,
|
|
ranking_map: Optional["RankingMap"] = None):
|
|
"""通知被发送时回调"""
|
|
if sbn:
|
|
self._active_notifications[sbn.getKey()] = sbn
|
|
self._keys_list = list(self._active_notifications.keys())
|
|
if ranking_map:
|
|
self._ranking_map = ranking_map
|
|
|
|
def onNotificationRemoved(self, sbn: StatusBarNotification,
|
|
ranking_map: Optional["RankingMap"] = None,
|
|
reason: int = REASON_LISTENER_CANCEL):
|
|
"""通知被移除时回调"""
|
|
if sbn:
|
|
key = sbn.getKey()
|
|
self._active_notifications.pop(key, None)
|
|
if key in self._keys_list:
|
|
self._keys_list.remove(key)
|
|
|
|
def onListenerConnected(self):
|
|
"""监听器已连接"""
|
|
self._is_listener_bound = True
|
|
|
|
def onListenerDisconnected(self):
|
|
"""监听器已断开"""
|
|
self._is_listener_bound = False
|
|
|
|
def onNotificationRemovedReason(self, sbn: StatusBarNotification,
|
|
ranking_map: "RankingMap",
|
|
reason: int):
|
|
"""通知被移除 (带原因)"""
|
|
self.onNotificationRemoved(sbn, ranking_map, reason)
|
|
|
|
def onNotificationPostedWithRankingMap(self, sbn: StatusBarNotification,
|
|
ranking_map: "RankingMap"):
|
|
"""通知被发送 (带排名映射)"""
|
|
self.onNotificationPosted(sbn, ranking_map)
|
|
|
|
def getActiveNotifications(self) -> List[StatusBarNotification]:
|
|
"""获取活动通知列表"""
|
|
return list(self._active_notifications.values())
|
|
|
|
def getActiveNotificationsWithRankingMap(self) -> Tuple[
|
|
List[StatusBarNotification],
|
|
Optional["RankingMap"]]:
|
|
"""获取活动通知及排名映射"""
|
|
return list(self._active_notifications.values()), self._ranking_map
|
|
|
|
def getCurrentInterruptionFilter(self) -> int:
|
|
"""获取当前中断过滤器"""
|
|
return self._interruption_filter
|
|
|
|
def setInterruptionFilter(self, interruption_filter: int):
|
|
"""设置中断过滤器"""
|
|
self._interruption_filter = interruption_filter
|
|
|
|
def cancelNotification(self, key: str):
|
|
"""取消指定通知"""
|
|
self._active_notifications.pop(key, None)
|
|
if key in self._keys_list:
|
|
self._keys_list.remove(key)
|
|
|
|
def cancelAllNotifications(self):
|
|
"""取消所有通知"""
|
|
self._active_notifications.clear()
|
|
self._keys_list.clear()
|
|
|
|
def requestRebind(self, component_name: Any):
|
|
"""请求重新绑定"""
|
|
pass
|
|
|
|
def requestUnbind(self):
|
|
"""请求解绑"""
|
|
pass
|
|
|
|
def registerAsSystemService(self, context: Any, component_name: Any,
|
|
user: int):
|
|
pass
|
|
|
|
def unregisterAsSystemService(self):
|
|
pass
|
|
|
|
def isBound(self) -> bool:
|
|
return self._is_listener_bound
|
|
|
|
def toString(self) -> str:
|
|
return (f"NotificationListenerService("
|
|
f"notifications={len(self._active_notifications)}, "
|
|
f"filter={self._interruption_filter})")
|
|
|
|
def hashCode(self) -> int:
|
|
return id(self) & 0x7FFFFFFF
|
|
|
|
def equals(self, other) -> bool:
|
|
return self is other
|
|
|
|
# ── RankingMap ──
|
|
|
|
class RankingMap(Stub):
|
|
"""RankingMap — 通知排名映射"""
|
|
|
|
def __init__(self):
|
|
super().__init__()
|
|
self._rankings: Dict[str, "NotificationListenerService.Ranking"] = {}
|
|
|
|
def getOrderedKeys(self) -> List[str]:
|
|
"""获取有序键列表"""
|
|
return list(self._rankings.keys())
|
|
|
|
def getRanking(self, key: str) -> Optional["NotificationListenerService.Ranking"]:
|
|
"""获取指定通知的排名信息"""
|
|
return self._rankings.get(key)
|
|
|
|
def getRawRanking(self, key: str) -> Optional[Any]:
|
|
return self._rankings.get(key)
|
|
|
|
def toString(self) -> str:
|
|
return f"RankingMap(keys={len(self._rankings)})"
|
|
|
|
def hashCode(self) -> int:
|
|
return hash(tuple(self._rankings.keys())) & 0x7FFFFFFF
|
|
|
|
def equals(self, other) -> bool:
|
|
if not isinstance(other, NotificationListenerService.RankingMap):
|
|
return False
|
|
return self._rankings == other._rankings
|
|
|
|
class Ranking(Stub):
|
|
"""Ranking — 通知排名"""
|
|
|
|
def __init__(self):
|
|
super().__init__()
|
|
self._key: str = ""
|
|
self._rank: int = 0
|
|
self._is_ambient: bool = False
|
|
self._matches_interruption_filter: bool = True
|
|
self._suppressed_visual_effects: int = 0
|
|
self._importance: int = 3 # IMPORTANCE_DEFAULT
|
|
self._importance_explanation: Optional[str] = None
|
|
self._group_key: Optional[str] = None
|
|
self._override_group_key: Optional[str] = None
|
|
self._channel: Optional[Any] = None
|
|
self._is_suspended: bool = False
|
|
|
|
def getKey(self) -> str:
|
|
return self._key
|
|
|
|
def getRank(self) -> int:
|
|
return self._rank
|
|
|
|
def isAmbient(self) -> bool:
|
|
return self._is_ambient
|
|
|
|
def matchesInterruptionFilter(self) -> bool:
|
|
return self._matches_interruption_filter
|
|
|
|
def getSuppressedVisualEffects(self) -> int:
|
|
return self._suppressed_visual_effects
|
|
|
|
def getImportance(self) -> int:
|
|
return self._importance
|
|
|
|
def getImportanceExplanation(self) -> Optional[str]:
|
|
return self._importance_explanation
|
|
|
|
def getGroupKey(self) -> Optional[str]:
|
|
return self._group_key
|
|
|
|
def getOverrideGroupKey(self) -> Optional[str]:
|
|
return self._override_group_key
|
|
|
|
def getChannel(self) -> Optional[Any]:
|
|
return self._channel
|
|
|
|
def isSuspended(self) -> bool:
|
|
return self._is_suspended
|
|
|
|
def toString(self) -> str:
|
|
return (f"Ranking(key={self._key}, rank={self._rank}, "
|
|
f"importance={self._importance})")
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════
|
|
# __all__
|
|
# ═══════════════════════════════════════════════════════════════
|
|
|
|
__all__ = [
|
|
"NotificationListenerService",
|
|
"StatusBarNotification",
|
|
] |