# Copyright (c) 2015 Egor Tensin # 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 import Iterable from collections.abc import Mapping from enum import Enum import json from urllib.error import URLError import urllib.parse from urllib.request import urlopen import vk.error from vk.user import User def _split_url(url): return urllib.parse.urlsplit(url)[:3] def _is_empty_param_value(value): return isinstance(value, str) and not value def _filter_empty_params(params, empty_params=False): if empty_params: return params if isinstance(params, Mapping): return {name: value for name, value in params.items() if not _is_empty_param_value(value)} elif isinstance(params, Iterable): return [(name, value) for name, value in params if not _is_empty_param_value(value)] else: raise TypeError() def _build_url(scheme, host, path, params=None, empty_params=False): if params is None: params = {} if isinstance(params, Mapping) or isinstance(params, Iterable): params = _filter_empty_params(params, empty_params) params = urllib.parse.urlencode(params) elif isinstance(params, str): pass else: raise TypeError() path = urllib.parse.quote(path) return urllib.parse.urlunsplit((scheme, host, path, params, '')) def _join_param_values(values): if isinstance(values, str): return values elif isinstance(values, Iterable): return ','.join(map(str, values)) else: return values def _join_path(base, url): if not base.endswith('/'): base += '/' return urllib.parse.urljoin(base, url) class Version(Enum): V5_73 = '5.73' DEFAULT = V5_73 def __str__(self): return self.value class Language(Enum): EN = 'en' DEFAULT = EN def __str__(self): return self.value class Method(Enum): USERS_GET = 'users.get' FRIENDS_GET = 'friends.get' def __str__(self): return self.value class CommonParameters(Enum): VERSION = 'v' LANGUAGE = 'lang' def __str__(self): return self.value class API: _ROOT_URL = 'https://api.vk.com/method/' _SCHEME, _HOST, _ROOT_PATH = _split_url(_ROOT_URL) def __init__(self, version=Version.DEFAULT, lang=Language.DEFAULT): self._common_params = { CommonParameters.VERSION: version, CommonParameters.LANGUAGE: lang, } def _build_method_url(self, method, **params): path = _join_path(self._ROOT_PATH, str(method)) params.update(self._common_params) return _build_url(self._SCHEME, self._HOST, path, params) def _call_method(self, method, **params): url = self._build_method_url(method, **params) #print(url) try: with urlopen(url) as response: response = json.loads(response.read().decode()) #print(response) if 'response' not in response: raise vk.error.InvalidAPIResponseError(response) return response['response'] except (ConnectionError, URLError) as e: raise vk.error.APIConnectionError(str(e)) from e def _filter_response_with_users(self, user_list, deactivated_users=True): user_list = map(User.from_api_response, user_list) if deactivated_users: return user_list return [user for user in user_list if not user.is_deactivated()] def users_get(self, user_ids, fields=(), deactivated_users=True): return self._filter_response_with_users(self._call_method( Method.USERS_GET, user_ids=_join_param_values(user_ids), fields=_join_param_values(fields)), deactivated_users) def friends_get(self, user_id, fields=(), deactivated_users=True): response = self._call_method( Method.FRIENDS_GET, user_id=user_id, fields=_join_param_values(fields)) if 'items' not in response: raise vk.error.InvalidAPIResponseError(response) return self._filter_response_with_users(response['items'], deactivated_users)