Advertisement
Not a member of Pastebin yet?
Sign Up,
it unlocks many cool features!
- from . import apierrors, structures
- import hashlib
- import logging
- import os
- import pickle
- import re
- import requests
- import time
- import urllib.parse
- import webbrowser
- __author__ = "Sergei Snegirev (tz4678@gmail.com)"
- __copyright__ = "Copyright (C) 2013-2016 Sergei Snegirev"
- __license__ = "MIT"
- __version__ = "3.0"
- __url__ = "https://github.com/tz4678/vkapi/"
- API_DELAY = 0.34
- API_HOST = "api.vk.com"
- API_PATH = "/method/"
- API_VERSION = 5.44
- RE_API_METHOD = re.compile("[a-z]+([A-Z][a-z]+)*$")
- class Client:
- """Клиент для работы с Api Вконтакте.
- Способы вызова методов Api:
- :instance:`Client`.api.<method_name>(**kwargs)
- :instance:`Client`.api.<method_name>(params_dict[, **kwargs])
- :instance:`Client`.api_call('<method_name>', params_dict)
- Usage::
- >> c = Client()
- >> users = c.api.users.get(user_id=100)
- >> print(users[0].first_name)
- Если имя именованно параметра совпадает с ключевым словом, добавляем
- перед ним подчеркивание (_global, _from).
- """
- def __init__(self,
- access_token=None,
- access_token_filename=None,
- api_delay=API_DELAY,
- api_version=API_VERSION,
- extra_params={},
- session=None):
- """Конструктор.
- :param access_token: Токен доступа
- :type access_token: vkapi.access_token.AccessToken instance
- :param access_token_filename: Имя файла где хранится токен
- :type access_token_filename: str
- :param api_delay: Задержка между вызовами методов Api
- :type api_delay: int
- :param api_version: Версия Api
- :type api_version: float
- :param extra_params: Дополнительные параметры которые будут
- передаваться при каждом запросе к Api (помимо токена и версии Api).
- Например: {'https': 1, 'lang': 'en'}
- :type extra_params: dict
- :param session: Сессия requests
- :type session: requests.Session instance
- """
- self.logger = logging.getLogger(
- ".".join([self.__class__.__module__, self.__class__.__name__]))
- self.access_token = access_token
- self.access_token_filename = access_token_filename
- self.api_delay = api_delay
- self.api_version = api_version
- self.extra_params = extra_params
- self.session = session or requests.session()
- self.last_api_call = 0
- self.captcha_img = None
- self.captcha_sid = None
- self.redirect_uri = None
- # Сахар над вызовом api_call
- self.api = ApiCall(self)
- # Пробуем загрузить access_token
- if self.access_token_filename:
- self.load_access_token()
- def request(self, url, data=None, files=None):
- start_time = time.time()
- if data or files:
- response = self.session.post(url, data, files=files)
- else:
- response = self.session.get(url)
- request_time = (time.time() - start_time) * 1000
- self.logger.debug("Total Request Time: %dms", request_time)
- return response.json(object_hook=structures.JsonObject)
- def api_call(self, method, params={}):
- """Делает запрос к Api.
- Список методов Api: <https://vk.com/dev/methods>
- Подробнее про запросы к Api: <https://vk.com/dev/api_requests>
- :param method: Метод Api
- :type method: str
- :param params: Передаваемые параметры
- :type params: dict
- :return: Возвращает содержимое поля `response`. Заметьте, что к
- элементам словаря можно обращаться как к аттрибутам через точечную
- нотацию. Вместо обычного словаря используется
- :class:`structures.JsonObject`.
- """
- params = dict(params)
- params.update(self.extra_params)
- # Разрешаем использовать произвольную версию Api
- if 'v' not in params:
- params['v'] = self.api_version
- if self.access_token:
- params['access_token'] = self.access_token.key
- if self.captcha_sid:
- params['captcha_sid'] = self.captcha_sid
- path = urllib.parse.urljoin(API_PATH, method)
- scheme = 'https'
- # <https://vk.com/dev/api_nohttps>
- if hasattr(self.access_token, "secret") and self.access_token.secret:
- scheme = 'http'
- # Исправлен баг с неверной подписью. Дело в том, что словари в
- # python неупорядоченные. Новый элемент может быть добавлен как в
- # конец так и любое другое место, да и еще элементы могут
- # поменяться местами. И в итоге имеем, что:
- # md5('foo=bar&baz=qux') != md5('baz=qux&foo=bar')
- # Чтобы закрепить порядок элементов, добавим sig
- params['sig'] = ''
- # Сформировали Query String из словаря
- query = urllib.parse.urlencode(params)
- # А теперь вырежем параметр sig
- query = re.sub('^sig=&|&sig=', '', query)
- uri = '{}?{}{}'.format(path, query, self.access_token.secret)
- sig = hashlib.md5(uri.encode('ascii')).hexdigest()
- params["sig"] = sig
- # !!! params не должен изменяться после добавления sig
- api_endpoint = "{}://{}{}".format(scheme, API_HOST, path)
- delay = self.api_delay + self.last_api_call - time.time()
- if delay > 0:
- self.logger.debug("Wait %dms", delay * 1000)
- time.sleep(delay)
- self.logger.debug(
- "Calling method %r with parameters: %s", method, params)
- response = self.request(api_endpoint, params)
- self.last_api_call = time.time()
- self.captcha_img = response.get('captcha_img')
- self.captcha_sid = response.get('captcha_sid')
- self.redirect_uri = response.get('redirect_uri')
- if 'error' in response:
- self.logger.warning(self.format_api_error(response))
- code = response.error.error_code
- msg = response.error.error_msg
- return self.handle_api_error(code, msg, method, params)
- return response.response
- def handle_api_error(self, code, msg, method, params):
- """Переопределить в случае необходимости"""
- if self.captcha_img:
- params['captcha_key'] = self.handle_captcha()
- # captcha_sid добавлять не нужно, он сам добавится
- return self.api_call(method, params)
- if self.redirect_uri:
- webbrowser.open(self.redirect_uri)
- # Кастомные ошибки (это для наглядности)
- if code is apierrors.TOO_MANY_REQUESTS_PER_SECOND:
- time.sleep(5)
- return self.api_call(method, params)
- raise ApiError(code, msg)
- def handle_captcha(self):
- """Данный метод нуждается в реализации"""
- # Делаем что-то с self.captcha_img и возвращаем текст с картинки
- raise ClientError("Captcha required")
- def format_api_error(self, response):
- details = response.error
- code = details.error_code
- msg = details.error_msg
- params = {param.key: param.value for param in details.request_params}
- method = params.pop('method')
- params.pop('oauth')
- return (
- "Error Code: {}, Message: {!r}. "
- "An error occurred while calling method {!r} with parameters: {}"
- ).format(code, msg, method, params)
- @property
- def is_valid_access_token(self):
- """Проверяет access_token."""
- # Можно короче:
- # return self.api.users.get() == []
- # Но этот способ не будет работать с серверными приложениями
- try:
- return self.api.execute(code="return true;")
- except ApiError:
- return False
- def upload(self, upload_url, files):
- response = self.request(upload_url, None, files)
- if 'error' in response:
- raise ClientError(response.error)
- return response
- def load_access_token(self, filename=None):
- self.access_token_filename = filename or self.access_token_filename
- if (self.access_token_filename and
- os.path.exists(self.access_token_filename)):
- with open(self.access_token_filename, 'rb') as fp:
- self.access_token = pickle.load(fp)
- self.logger.debug("access token loaded from %s",
- self.access_token_filename)
- def save_access_token(self, filename=None):
- self.access_token_filename = filename or self.access_token_filename
- if self.access_token_filename:
- with open(self.access_token_filename, 'wb') as fp:
- data = pickle.dumps(self.access_token)
- fp.write(data)
- self.logger.debug("access token saved %s",
- self.access_token_filename)
- class ClientError(Exception):
- pass
- class ApiError(ClientError):
- def __init__(self, code, msg):
- self.code = code
- self.msg = msg
- def __str__(self):
- return "{}: {}".format(self.code, self.msg)
- class ApiCall:
- def __init__(self, client, method=None):
- self._client = client
- self._method = method
- def __getattr__(self, name):
- if RE_API_METHOD.match(name):
- method = name if not self._method \
- else '.'.join([self._method, name])
- return ApiCall(self._client, method)
- raise AttributeError(name)
- def __call__(self, *args, **kwargs):
- # Если имя именованного параметра совпадает с ключевым словом, то
- # добавляем перед именем подчеркивание:
- params = {k[1:] if len(k) > 1 and k.startswith('_')
- else k: v for k, v in kwargs.items()}
- if len(args):
- # Первым аргументом является словарь
- d = dict(args[0])
- # Если переданы именованные параметры, то обновляем словарь
- d.update(params)
- params = d
- return self._client.api_call(self._method, params)
Advertisement
Add Comment
Please, Sign In to add comment
Advertisement