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
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
|
# Copyright (c) 2016 Egor Tensin <Egor.Tensin@gmail.com>
# This file is part of the "VK scripts" project.
# For details, see https://github.com/egor-tensin/vk-scripts.
# Distributed under the MIT License.
from collections.abc import Callable
import contextlib
import time
import signal
import vk.error
from vk.user import UserField
class StatusTracker:
DEFAULT_TIMEOUT = 5
def __init__(self, api, timeout=DEFAULT_TIMEOUT):
self._api = api
self._timeout = timeout
self._on_initial_status = []
self._on_status_update = []
self._on_connection_error = []
def _wait_after_connection_error(self):
time.sleep(self._timeout)
def add_database_writer(self, writer):
self.add_initial_status_handler(writer.on_initial_status)
self.add_status_update_handler(writer.on_status_update)
self.add_connection_error_handler(writer.on_connection_error)
def add_initial_status_handler(self, fn):
self._assert_is_callback(fn)
self._on_initial_status.append(fn)
def add_status_update_handler(self, fn):
self._assert_is_callback(fn)
self._on_status_update.append(fn)
def add_connection_error_handler(self, fn):
self._assert_is_callback(fn)
self._on_connection_error.append(fn)
@staticmethod
def _assert_is_callback(fn):
if not isinstance(fn, Callable):
raise TypeError()
_USER_FIELDS = UserField.DOMAIN, UserField.ONLINE, UserField.LAST_SEEN,
def _query_status(self, uids):
user_list = self._api.users_get(uids, self._USER_FIELDS,
deactivated_users=False)
return {user.get_uid(): user for user in user_list}
def _notify_status(self, user):
for fn in self._on_initial_status:
fn(user)
def _notify_status_update(self, user):
for fn in self._on_status_update:
fn(user)
def _notify_connection_error(self, e):
for fn in self._on_connection_error:
fn(e)
def _query_initial_status(self, uids):
while True:
try:
return self._query_status(uids)
except vk.error.APIConnectionError as e:
self._notify_connection_error(e)
self._wait_after_connection_error()
def _query_status_updates(self, uids):
while True:
self._wait_after_connection_error()
try:
return self._query_status(uids)
except vk.error.APIConnectionError as e:
self._notify_connection_error(e)
@staticmethod
def _filter_status_updates(old_users, new_users):
for uid, user in new_users.items():
if old_users[uid].is_online() != user.is_online():
old_users[uid] = user
yield user
def query_status(self, uids):
users = self._query_initial_status(uids)
for user in users.values():
self._notify_status(user)
return users
def _do_loop(self, uids):
users = self.query_status(uids)
while True:
updated_users = self._query_status_updates(uids)
for user in self._filter_status_updates(users, updated_users):
self._notify_status_update(user)
class StopLooping(RuntimeError):
pass
@staticmethod
def _stop_looping(signo, frame):
raise StatusTracker.StopLooping()
@staticmethod
@contextlib.contextmanager
def _handle_signal(signo, handler):
old_handler = signal.getsignal(signo)
signal.signal(signo, handler)
try:
yield
finally:
signal.signal(signal.SIGINT, old_handler)
@staticmethod
def _handle_sigint():
# Python doesn't raise KeyboardInterrupt in case a real SIGINT is sent
# from outside, surprisingly.
return StatusTracker._handle_signal(signal.SIGINT,
StatusTracker._stop_looping)
@staticmethod
def _handle_sigterm():
return StatusTracker._handle_signal(signal.SIGTERM,
StatusTracker._stop_looping)
def loop(self, uids):
with self._handle_sigint(), self._handle_sigterm():
try:
self._do_loop(uids)
except (KeyboardInterrupt, StatusTracker.StopLooping):
pass
|
