AdbNotificationForwarder/server.py at main · AlexHalogen/AdbNotificationForwarder · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
import time
import os
import logging
import configparser
from datetime import datetime
from adb_shell.adb_device import AdbDeviceTcp, AdbDeviceUsb
from adb_shell.auth.sign_pythonrsa import PythonRSASigner
from plyer import notification
from adbparser import parse_notifications

'''
Global parameters
'''
packages = set()
interval = 10
address = ""
port = 0
privkey = os.path.join(os.path.expanduser("~"), ".android", "adbkey") # %HOME%/.android/adbkey
pubkey = privkey + ".pub"											  # %HOME%/.android/adbkey.pub


def main():
	parse_config()

	logging.basicConfig(level=logging.INFO)
	logger = logging.getLogger('main')
	logger.setLevel(logging.INFO)

	logger.info("Forwarding notifications for %s" % list(packages))

	device = connect()
	visited = set()
	while True:
		raw = device.shell("dumpsys notification --noredact")
		notifications = parse_notifications(raw)
		if packages:
			notifications = filter_package_names(notifications)

		notifications = filter_duplicate_notifications(notifications, visited)
		for n in notifications:
			logger.info("{pkg} [{time}] \"{title}\" \"{text}\"".format(time=datetime.fromtimestamp(n["time"]/1000), pkg=n["pkg"], title=n["title"], text=n["text"]))
			notification.notify(
				title=n["title"],
				message=n["text"],
				app_icon=None,
				timeout=0,
				toast=True
			)
		time.sleep(interval)
		if len(visited) > 2*1000: # clear
			visited = set()

# Assumption: notifications have unique mUpdateTimeMs values
def filter_duplicate_notifications(notifications, visited):
	ret = []
	for n in notifications:
		if not n["time"] in visited:
			visited.add(n["time"])
			ret.append(n)

	return ret

def filter_package_names(notifications):
	result = filter(lambda n : n["pkg"] in packages, notifications)
	return list(result)

def connect():
	with open(privkey) as f:
		priv = f.read()
	with open(pubkey) as f:
		pub = f.read()

	signer = PythonRSASigner(pub, priv)
	device = AdbDeviceTcp(address, port, default_transport_timeout_s=9.)
	device.connect(rsa_keys=[signer], auth_timeout_s=0.1)
	return device

def parse_config():
	global packages, interval, address, port, privkey, pubkey

	config = configparser.ConfigParser()
	config.read("config.ini")
	section = config['config']
	if "adbprivkey" in section:
		privkey = section["adbprivkey"]
	if "adbpubkey" in section:
		pubkey = section["adbpubkey"]
	if "packages" in section:
		pkgs = section["packages"].split(",")
		packages = set(map(lambda x: x.strip(), pkgs))
	else:
		packages = set()

	address = section["address"]
	port = section.getint("port")
	interval = section.getint("interval")

if __name__ == "__main__":
	main()