Advertisement
Not a member of Pastebin yet?
Sign Up,
it unlocks many cool features!
- import hashlib
- import io
- import json
- import logging
- import os
- import pickle
- import platform
- import re
- import socket
- import time
- import urllib.parse
- import urllib.request
- import uuid
- # Это только для проверки нужно
- if __name__ == '__main__':
- from structures import ObjectDict
- else:
- from .structures import ObjectDict
- __author__ = "Sergei Snegirev (tz4678@gmail.com)"
- __copyright__ = "Copyright (C) 2013-2016 Sergei Snegirev"
- __license__ = "MIT"
- __version__ = "3.0"
- __url__ = "https://github.com/***/vkapi/"
- API_HOST = "api.vk.com"
- API_PATH = "/method/"
- API_VERSION = 5.44
- API_DELAY = 0.34
- TIMEOUT = 15
- TRIES = 3
- USERAGENT_FORMAT = "Mozilla/5.0 (vkapi.client/{ver} Python/{py_ver} +{url})"
- USERAGENT = USERAGENT_FORMAT.format(
- ver=__version__,
- py_ver=platform.python_version(),
- url=__url__
- )
- HTTP = re.compile('https?://', re.I)
- class Client:
- def __init__(self,
- access_token=None,
- access_token_filename=None,
- api_version=API_VERSION,
- api_delay=API_DELAY,
- https=None,
- lang=None,
- opener=None,
- timeout=TIMEOUT,
- tries=TRIES):
- """Клиент для работы с API Вконтакте.
- * Нативная поддержка методов API
- * Работа с подписью
- * Загрузка файлов
- :param access_token: Токен полученный при авторизации
- :type access_token: vkapi.accesstoken.AccessToken instance
- :param access_token_filename: Имя файла в котором хранится токен
- :type access_token_filename: str
- :param api_version: Версия API
- :type api_version: float
- :param api_delay: Задержка между вызовами методов API. Существуют
- ограничения на количество обращений в секунду к методам API
- :type api_delay: int or float
- :param https: Если равен 1, то методы API возвращают https ссылки на
- фотографии и медиа файлы. Работает только если запросы
- осуществляются через https.
- :type https: int
- :param lang: Используемый язык (ru, ua, be, es, fi, de, it)
- :type lang: str
- :param opener: Используется для отправки запросов
- :type opener: urllib.request.OpenerDirector instance
- :param tries: Количество повторных запросов в случае ошибки таймаута.
- Если значение равно -1, то запросы будут посылаться, пока не придет
- ответ.
- :type tries: int
- :param timeout: Таймаут
- :type timeout: int or float
- """
- # if access_token and access_token_filename:
- # raise TypeError("accepts access_token or access_token_filename, "
- # "but NOT both at once")
- self.access_token = access_token
- self.access_token_filename = access_token_filename
- self.api_version = api_version
- self.api_delay = api_delay
- self.lang = lang
- self.https = https
- if not opener:
- # Включаем поддержку cookies и используем системные прокси
- opener = urllib.request.build_opener(
- urllib.request.HTTPCookieProcessor(),
- urllib.request.ProxyHandler()
- )
- opener.addheaders = [('User-Agent', USERAGENT)]
- self.opener = opener
- self.timeout = timeout
- self.tries = tries
- self.logger = logging.getLogger(
- ".".join([self.__class__.__module__, self.__class__.__name__])
- )
- self.last_api_request = 0
- self.api = _Api(self)
- # Пробуем загрузить access_token
- if self.access_token_filename:
- self.load_access_token()
- # Если он установлен, то проверяем является ли он рабочим
- if self.is_authorized:
- if not self.is_valid_access_token:
- self.logger.warning("Validation access_token failed")
- 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:
- return pickle.load(fp)
- 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)
- @property
- def is_authorized(self):
- return bool(self.access_token)
- @property
- def is_valid_access_token(self):
- try:
- return self.api.execute(code="return true;")
- except ApiError:
- return False
- def api_request(self, method, params=None):
- """Делает запрос к API.
- Список методов API: <https://vk.com/dev/methods>
- Подробнее про запросы к API: <https://vk.com/dev/api_requests>
- Пример:
- c = Client()
- r = c.api_request('users.get', {'user_id': 1})
- print("{u.first_name} {u.last_name}".format(u=r[0]))
- Использовать данный метод нет необходимости, так как к API можно
- обращаться "нативно":
- Client.api.<method_name>(**kwargs)
- Client.api.<method_name>(params_dict[, **kwargs])
- Пример выше можно было переписать так:
- r = c.api.users.get(user_id=1)
- # либо так
- r = c.api.users.get({'user_id': 1})
- # или так
- r = c.api.users.get({'user_id': 123, 'fields': 'photo'}, user_id=1)
- В последнем значение user_id в словаре будет заменено на 1. То есть,
- значения словаря идущего первым аргументом заменяется содержимым
- словаря с именованными параметрами (params_dict.update(kwargs)), и
- будут переданы параметры:
- {'user_id': 1, 'fields': 'photo'}
- Если имя именованного параметра совпадает с ключевым словом, то к имени
- нужно добавить подчеркивание (например, _from).
- :param method: Метод API
- :type method: str
- :param params: Передаваемые параметры
- :type params: dict
- :return: Возвращает результат запроса. Это содержимое поля `response`.
- Заметьте, что к элементам словаря можно обращаться как к аттрибутам
- через точечную нотацию.
- """
- params = dict(params or {})
- if self.is_authorized:
- params['access_token'] = self.access_token.token
- # Позволяем в запросе выбрать версию API. Возможно, потребуется для
- # обращения к каким-нибудь depractated методам
- if 'v' not in params and self.api_version:
- params['v'] = self.api_version
- if 'lang' not in params and self.lang:
- params['lang'] = self.lang
- if 'https' not in params and self.https:
- params['https'] = self.https
- # Кстати, через http ответы сервера приходят заметно быстрее
- # <https://vk.com/dev/api_nohttps>
- if hasattr(self.access_token, "secret") and self.access_token.secret:
- scheme = 'http'
- # Исправлен баг с неверной подписью. См. <path/to/lib/notes.txt>.
- # Нам важно чтобы порядок элементов словаря сохранился.
- # Добавим новый элемент
- params['sig'] = ''
- # Сформировали query string из словаря
- qs = urllib.parse.urlencode(params)
- # А теперь вырежем параметр sig
- qs = re.sub('^sig=&|&sig=', '', qs)
- uri = '{}{}?{}{}'.format(
- API_PATH,
- method,
- qs,
- self.access_token.secret
- )
- params["sig"] = hashlib.md5(uri.encode('ascii')).hexdigest()
- else:
- scheme = 'https'
- # !!! params не должен изменяться после добавления sig
- delay = self.api_delay + self.last_api_request - time.time()
- if delay > 0:
- self.logger.debug("Wait %dms", delay * 1000)
- time.sleep(delay)
- self.logger.debug("Call method %r with parameters: %s", method, params)
- api_url = '{}://{}{}{}'.format(scheme, API_HOST, API_PATH, method)
- response = self.request(api_url, params)
- return self.process_api_response(response, method, params)
- def process_api_response(self, response, method, params):
- """Можно переопределить в классах потомках, а можно что-то типа этого
- сделать:
- def process_api_response(self, response, method, params):
- if 'error' in response:
- error = response.error
- if error.error_code is ApiError.TOO_MANY_REQUESTS_PER_SECOND:
- time.sleep(5)
- # Отправляем повторный запрос
- return self.api_request(method, params)
- if error.error_code is ApiError.CAPTCHA_NEEDED:
- # get captcha_key
- ...
- params.update({
- 'captcha_key': captcha_key,
- 'captcha_sid': error.captcha_sid
- })
- return self.api_request(method, params)
- ...
- return super().process_api_response(response, method, params)
- """
- # можно без генерации ошибок
- # return response
- if 'error' in response:
- raise ApiError(response)
- return response.response
- def upload(self, upload_url, files):
- """
- Загружает файлы на сервер. Процесс расписан здесь
- <https://vk.com/dev/upload_files>
- :param upload_url: Адрес сервера для загрузки файлов, который получают,
- например, вызовом Client.api.docs.getWallUploadServer() и т.п.
- :type upload_url: str
- :param files: Словарь где ключами выступают имена полей. Значения могут
- быть строками (путь до файла либо ссылка на него), файловыми
- объектами либо списками/кортежами,где первый элемент - имя файла,
- а второй - его содержимое (бинарное либо текстовое)
- :type files: dict
- :rtype: ObjectDict
- """
- boundary = uuid.uuid4().hex
- buf = io.BytesIO()
- for name, obj in files.items():
- if isinstance(obj, str):
- # Загружаем по URL
- if HTTP.match(obj):
- content = self.get_content(obj)
- # На некоторых ресурсах есть ссылки вида:
- # /image.jpg?w=x&h=y
- filename = urllib.parse.urlparse(obj).path
- filename = os.path.basename(filename)
- # Читаем файл
- else:
- with open(obj, 'rb') as fp:
- content = fp.read()
- filename = os.path.basename(obj)
- else:
- # {'photo': open("path/to/photo.jpg", 'rb')}
- if isinstance(obj, io.IOBase):
- filename = os.path.basename(obj.name)
- content = obj.read()
- # {'photo': ('photo.jpg', b'<binary data>')}
- elif isinstance(obj, (list, tuple)):
- # нам нужны только два первых элемента остальные будем
- # игнорировать
- filename, content, *_ = obj
- else:
- raise ValueError("Unexcepted: {!r}".format(obj))
- if not isinstance(content, (bytearray, bytes)):
- content = str(content).encode('utf-8')
- # Поле Content-Type передавать необязательно, на сервере Вконтакте
- # проверяется расширение файла из поля filename и его содержиме,
- # читаются первые n байтов, и ищутся сигнатуры
- header = ('--{}\r\n'
- 'Content-Disposition: form-data; name="{}"; '
- 'filename="{}"\r\n\r\n')
- # У requests кстати баг, который заставил отказаться от его
- # использования: он неправильно кодирует Non-ASCII поля в
- # заголовках, поэтому файл "Мой файл.txt" в документах будет
- # отображаться как "...???.txt"
- header = header.format(boundary, name, filename)
- buf.write(bytes(header, 'utf-8'))
- buf.write(content)
- buf.write(b'\r\n')
- buf.write(bytes('--{}--\r\n'.format(boundary), 'utf-8'))
- data = buf.getvalue()
- headers = {'Content-Type': 'multipart/form-data; boundary=' + boundary}
- response = self.request(upload_url, data, headers)
- if 'error' in response:
- raise UploadError(response.error)
- return response
- def request(self, url, data=None, headers={}):
- # Для работы с API Вконтакте другие методы кроме POST в принципе не
- # нужны. Ему параметры можно даже через куки передавать (по крайней
- # мере так можно было очень давно).
- if isinstance(data, dict):
- data = urllib.parse.urlencode(data).encode('ascii')
- try:
- content = self.raw_request(url, data, headers).read()
- except urllib.request.HTTPError as e:
- self.logger.debug(e)
- content = e.read()
- content = str(content, "utf-8")
- return json.loads(content, object_hook=ObjectDict)
- def get_content(self, url):
- return self.raw_request(url).read()
- def raw_request(self, url, body=None, headers={}):
- start_time = time.time()
- self.logger.debug("URL: %s", url)
- request = urllib.request.Request(url, body, headers)
- attempts = 0
- while True:
- try:
- response = self.opener.open(request, timeout=self.timeout)
- request_time = time.time() - start_time
- self.logger.debug("Request Time: %dms",
- request_time * 1000)
- return response
- except urllib.request.URLError as e:
- # Если ошибка не является ошибкой таймаута, то бросаем
- # исключение
- if not isinstance(e.reason, socket.timeout):
- raise
- # Если значение равно -1, то запросы будут отправляться пока
- # не будет получен ответ
- if self.tries != - 1:
- # Бросаем ошибку таймаута
- if self.tries <= 1:
- raise
- attempts += 1
- if attempts == self.tries:
- raise ClientError(
- "Maximun tries ({}) exceeded for URL: {}".format(
- self.tries, url))
- self.logger.debug("Retry: %s", url)
- class _Api:
- def __init__(self, client):
- self._client = client
- def __getattr__(self, name):
- if _ApiMethod.METHOD.match(name):
- return _ApiMethod(self._client, name)
- raise AttributeError(name)
- class _ApiMethod:
- METHOD = re.compile("[a-z]+([A-Z][a-z]+)*$")
- def __init__(self, client, method):
- self._client = client
- self._method = method
- def __getattr__(self, name):
- if self.METHOD.match(name):
- return _ApiMethod(self._client, '.'.join([self._method, name]))
- 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_request(self._method, params)
- class ClientError(Exception):
- pass
- class ApiError(ClientError):
- """Подробнее про ошибки API можно прочитать по ссылке
- <https://vk.com/dev/errors>
- """
- UNKNOWN_ERROR_OCCURRED = 1
- UNKNOWN_METHOD_PASSED = 3
- INCORRECT_SIGNATURE = 4
- USER_AUTHORIZATION_FAILED = 5
- TOO_MANY_REQUESTS_PER_SECOND = 6
- INVALID_REQUEST = 8
- FLOOD_CONTROL = 9
- INTERNAL_SERVER_ERROR = 10
- CAPTCHA_NEEDED = 14
- ACCESS_DENIED = 15
- HTTP_AUTHORIZATION_FAILED = 16
- VALIDATION_REQUIRED = 17
- THIS_METHOD_WAS_DISABLED = 23
- CONFIRMATION_REQUIRED = 24
- INVALID_APPLICATION_API_ID = 101
- INVALID_USER_ID = 113
- INVALID_TIMESTAMP = 150
- ACCESS_TO_ALBUM_DENIED = 200
- ACCESS_TO_AUDIO_DENIED = 201
- ACCESS_TO_GROUP_DENIED = 203
- THIS_ALBUM_IS_FULL = 300
- SOME_ADS_ERROR_OCCURED = 603
- def __init__(self, response):
- self.error_code = response.error.error_code
- self.error_msg = response.error.error_msg
- self.captcha_img = response.error.get('captcha_img')
- self.captcha_sid = response.error.get('captcha_sid')
- self.redirect_uri = response.error.get('redirect_uri')
- self.params = params = {
- param.key: param.value
- for param in response.error.request_params
- }
- self.method = params.pop('method')
- # Что это?
- self.oauth = params.pop('oauth')
- def __str__(self):
- return (
- "Error Code: {}, Message: {!r}. An error occured while calling "
- "method {!r} with parameters: {}"
- ).format(self.error_code, self.error_msg, self.method, self.params)
- class UploadError(ClientError):
- pass
- if __name__ == '__main__':
- logging.basicConfig(level=logging.DEBUG)
- c = Client()
- r = c.api.users.get(user_id=1)
- print("{u.first_name} {u.last_name}".format(u=r[0]))
Advertisement
Add Comment
Please, Sign In to add comment
Advertisement