# 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.
import argparse
from collections.abc import Callable
import contextlib
import time
import signal
import sys
from vk.api import API
import vk.error
from vk.tracking.db import Format as DatabaseFormat
from vk.user import UserField
import vk.version
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
DEFAULT_TIMEOUT = StatusTracker.DEFAULT_TIMEOUT
DEFAULT_DB_FORMAT = DatabaseFormat.CSV
def _parse_positive_integer(s):
try:
n = int(s)
except ValueError:
raise argparse.ArgumentTypeError('must be a positive integer: ' + s)
if n < 1:
raise argparse.ArgumentTypeError('must be a positive integer: ' + s)
return n
def _parse_database_format(s):
try:
return DatabaseFormat(s)
except ValueError:
raise argparse.ArgumentTypeError('invalid database format: ' + s)
def _parse_args(args=None):
if args is None:
args = sys.argv[1:]
parser = argparse.ArgumentParser(
description='Track when people go online/offline.')
vk.version.add_to_arg_parser(parser)
parser.add_argument('uids', metavar='UID', nargs='+',
help='user IDs or "screen names"')
parser.add_argument('-t', '--timeout', metavar='SECONDS',
type=_parse_positive_integer,
default=DEFAULT_TIMEOUT,
help='set refresh interval')
parser.add_argument('-O', '--only-once', action='store_true',
help='query the status only once and exit')
parser.add_argument('-l', '--log', metavar='PATH', dest='log_path',
help='set log file path (standard output by default)')
parser.add_argument('-f', '--format', dest='db_fmt',
type=_parse_database_format,
choices=DatabaseFormat,
default=DEFAULT_DB_FORMAT,
help='specify database format')
parser.add_argument('-o', '--output', metavar='PATH', dest='db_path',
help='set database file path')
return parser.parse_args(args)
def track_status(
uids, timeout=DEFAULT_TIMEOUT, log_path=None,
db_path=None, db_fmt=DEFAULT_DB_FORMAT, only_once=False):
api = API()
tracker = StatusTracker(api, timeout)
if db_fmt is DatabaseFormat.LOG or db_path is None:
db_fmt = DatabaseFormat.NULL
with DatabaseFormat.LOG.open_output_file(log_path) as log_fd:
log_writer = DatabaseFormat.LOG.create_writer(log_fd)
tracker.add_database_writer(log_writer)
with db_fmt.open_output_file(db_path) as db_fd:
db_writer = db_fmt.create_writer(db_fd)
tracker.add_database_writer(db_writer)
if only_once:
tracker.query_status(uids)
else:
tracker.loop(uids)
def main(args=None):
track_status(**vars(_parse_args(args)))
if __name__ == '__main__':
main()
