Removed VK module from repo. Install it via pip. Custom modules has been moved to mysc
This commit is contained in:
parent
c7743fccb9
commit
d1fd6d1861
@ -20,7 +20,6 @@ For other dependencies, do pip install --upgrade -r requirements.txt
|
||||
|
||||
* [Python,](http://python.org) version 2.7.15
|
||||
* [PyEnchant,](http://pythonhosted.org/pyenchant/) version 1.6.6.
|
||||
* [VK API bindings for Python](https://github.com/dimka665/vk) (already included in the SRC directory)
|
||||
* [Pandoc](http://pandoc.org/installing.html) for generating the changelog.
|
||||
|
||||
## Documentation
|
||||
|
||||
@ -1,5 +1,6 @@
|
||||
wxpython
|
||||
pywin32
|
||||
vk
|
||||
configobj
|
||||
pypubsub==3.3.0
|
||||
requests-oauthlib
|
||||
|
||||
@ -18,9 +18,9 @@ from wxUI.tabs import home
|
||||
from pubsub import pub
|
||||
from sessionmanager import session
|
||||
from mysc.thread_utils import call_threaded
|
||||
from mysc import upload
|
||||
from wxUI import commonMessages, menus
|
||||
from vk import upload
|
||||
from vk.exceptions import VkAPIMethodError
|
||||
from vk.exceptions import VkAPIError
|
||||
from utils import add_attachment
|
||||
|
||||
log = logging.getLogger("controller.buffers")
|
||||
@ -79,7 +79,7 @@ class baseBuffer(object):
|
||||
retrieved = True # Control variable for handling unauthorised/connection errors.
|
||||
try:
|
||||
num = getattr(self.session, "get_newsfeed")(show_nextpage=show_nextpage, name=self.name, *self.args, **self.kwargs)
|
||||
except VkAPIMethodError as err:
|
||||
except VkAPIError as err:
|
||||
log.error(u"Error {0}: {1}".format(err.code, err.message))
|
||||
retrieved = err.code
|
||||
return retrieved
|
||||
|
||||
@ -1,6 +1,6 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
import threading
|
||||
from vk import longpool
|
||||
from mysc import longpoll
|
||||
from pubsub import pub
|
||||
from logging import getLogger
|
||||
log = getLogger("controller.longpolThread")
|
||||
@ -10,7 +10,7 @@ class worker(threading.Thread):
|
||||
super(worker, self).__init__()
|
||||
log.debug("Instanciating longPoll server")
|
||||
self.session = session
|
||||
self.l = longpool.LongPoll(self.session.vk.client)
|
||||
self.l = longpoll.LongPoll(self.session.vk.client)
|
||||
|
||||
def run(self):
|
||||
while self.session.is_logged == True:
|
||||
@ -11,9 +11,9 @@ import player
|
||||
import posts
|
||||
import webbrowser
|
||||
import logging
|
||||
import longpoolthread
|
||||
import longpollthread
|
||||
import selector
|
||||
from vk.exceptions import VkAuthError, VkAPIMethodError
|
||||
from vk.exceptions import VkAuthError, VkAPIError
|
||||
from pubsub import pub
|
||||
from mysc.repeating_timer import RepeatingTimer
|
||||
from mysc.thread_utils import call_threaded
|
||||
@ -184,8 +184,8 @@ class Controller(object):
|
||||
self.window.change_status(_(u"Loading items for {0}").format(i.name,))
|
||||
i.get_items()
|
||||
self.window.change_status(_(u"Ready"))
|
||||
self.longpool = longpoolthread.worker(self.session)
|
||||
self.longpool.start()
|
||||
self.longpoll = longpollthread.worker(self.session)
|
||||
self.longpoll.start()
|
||||
self.status_setter = RepeatingTimer(900, self.set_online)
|
||||
self.status_setter.start()
|
||||
self.set_online()
|
||||
@ -369,7 +369,7 @@ class Controller(object):
|
||||
|
||||
def get_chat(self, obj=None):
|
||||
""" Searches or creates a chat buffer with the id of the user that is sending or receiving a message.
|
||||
obj vk.longpool.event: an event wich defines some data from the vk's longpool server."""
|
||||
obj mysc.longpoll.event: an event wich defines some data from the vk's long poll server."""
|
||||
# Set user_id to the id of the friend wich is receiving or sending the message.
|
||||
obj.user_id = obj.from_id
|
||||
buffer = self.search_chat_buffer(obj.user_id)
|
||||
@ -407,7 +407,7 @@ class Controller(object):
|
||||
try:
|
||||
log.debug("Getting possible unread messages.")
|
||||
msgs = self.session.vk.client.messages.getDialogs(count=200, unread=1)
|
||||
except VkAPIMethodError as ex:
|
||||
except VkAPIError as ex:
|
||||
if ex.code == 6:
|
||||
log.exception("Something went wrong when getting messages. Waiting a second to retry")
|
||||
time.sleep(2)
|
||||
@ -429,7 +429,7 @@ class Controller(object):
|
||||
try:
|
||||
log.debug("Create audio albums...")
|
||||
albums = self.session.vk.client.audio.getAlbums(owner_id=user_id)
|
||||
except VkAPIMethodError as ex:
|
||||
except VkAPIError as ex:
|
||||
if ex.code == 6:
|
||||
log.exception("Something went wrong when getting albums. Waiting a second to retry")
|
||||
time.sleep(2)
|
||||
@ -452,7 +452,7 @@ class Controller(object):
|
||||
try:
|
||||
log.debug("Create video albums...")
|
||||
albums = self.session.vk.client.video.getAlbums(owner_id=user_id)
|
||||
except VkAPIMethodError as ex:
|
||||
except VkAPIError as ex:
|
||||
if ex.code == 6:
|
||||
log.exception("Something went wrong when getting albums. Waiting a second to retry")
|
||||
time.sleep(2)
|
||||
|
||||
@ -1,10 +0,0 @@
|
||||
|
||||
from vk.api import logger
|
||||
from vk.api import Session, AuthSession, InteractiveSession, InteractiveAuthSession
|
||||
from vk.api import VERSION
|
||||
from vk.api import API
|
||||
import upload
|
||||
|
||||
__version__ = version = VERSION
|
||||
|
||||
# API = OAuthAPI
|
||||
177
src/vk/api.py
177
src/vk/api.py
@ -1,177 +0,0 @@
|
||||
# coding=utf8
|
||||
|
||||
import logging
|
||||
import logging.config
|
||||
|
||||
from vk.logs import LOGGING_CONFIG
|
||||
from vk.utils import stringify_values, json_iter_parse, LoggingSession
|
||||
from vk.exceptions import VkAuthError, VkAPIMethodError, CAPTCHA_IS_NEEDED, AUTHORIZATION_FAILED
|
||||
from vk.mixins import AuthMixin, InteractiveMixin
|
||||
|
||||
|
||||
VERSION = '2.0a4'
|
||||
|
||||
|
||||
logging.config.dictConfig(LOGGING_CONFIG)
|
||||
logger = logging.getLogger('vk')
|
||||
|
||||
|
||||
class Session(object):
|
||||
API_URL = 'https://api.vk.com/method/'
|
||||
|
||||
def __init__(self, access_token=None):
|
||||
|
||||
logger.debug('API.__init__(access_token=%(access_token)r)', {'access_token': access_token})
|
||||
|
||||
# self.api_version = api_version
|
||||
# self.default_timeout = default_timeout
|
||||
self.access_token = access_token
|
||||
self.access_token_is_needed = False
|
||||
|
||||
# self.requests_session = requests.Session()
|
||||
self.requests_session = LoggingSession()
|
||||
self.requests_session.headers['Accept'] = 'application/json'
|
||||
self.requests_session.headers['Content-Type'] = 'application/x-www-form-urlencoded'
|
||||
|
||||
@property
|
||||
def access_token(self):
|
||||
logger.debug('Check that we need new access token')
|
||||
if self.access_token_is_needed:
|
||||
logger.debug('We need new access token. Try to get it.')
|
||||
self.access_token, self._access_token_expires_in = self.get_access_token()
|
||||
logger.info('Got new access token')
|
||||
logger.debug('access_token = %r, expires in %s', self.censored_access_token, self._access_token_expires_in)
|
||||
return self._access_token
|
||||
|
||||
@access_token.setter
|
||||
def access_token(self, value):
|
||||
self._access_token = value
|
||||
self._access_token_expires_in = None
|
||||
self.access_token_is_needed = not self._access_token
|
||||
|
||||
@property
|
||||
def censored_access_token(self):
|
||||
if self._access_token:
|
||||
return '{}***{}'.format(self._access_token[:4], self._access_token[-4:])
|
||||
|
||||
def get_user_login(self):
|
||||
logger.debug('Do nothing to get user login')
|
||||
|
||||
def get_access_token(self):
|
||||
"""
|
||||
Dummy method
|
||||
"""
|
||||
logger.debug('API.get_access_token()')
|
||||
return self._access_token, self._access_token_expires_in
|
||||
|
||||
def make_request(self, method_request, **method_kwargs):
|
||||
|
||||
logger.debug('Prepare API Method request')
|
||||
|
||||
response = self.send_api_request(method_request)
|
||||
response.raise_for_status()
|
||||
|
||||
# there are may be 2 dicts in one JSON
|
||||
# for example: {'error': ...}{'response': ...}
|
||||
errors = []
|
||||
error_codes = []
|
||||
for data in json_iter_parse(response.text):
|
||||
if 'error' in data:
|
||||
error_data = data['error']
|
||||
if error_data['error_code'] == CAPTCHA_IS_NEEDED:
|
||||
return self.on_captcha_is_needed(error_data, method_request)
|
||||
|
||||
error_codes.append(error_data['error_code'])
|
||||
errors.append(error_data)
|
||||
|
||||
if 'response' in data:
|
||||
for error in errors:
|
||||
logger.warning(str(error))
|
||||
|
||||
return data['response']
|
||||
|
||||
if AUTHORIZATION_FAILED in error_codes: # invalid access token
|
||||
logger.info('Authorization failed. Access token will be dropped')
|
||||
self.access_token = None
|
||||
return False
|
||||
else:
|
||||
raise VkAPIMethodError(errors[0])
|
||||
|
||||
def send_api_request(self, request):
|
||||
url = self.API_URL + request._method_name
|
||||
method_args = request._api._method_default_args.copy()
|
||||
method_args.update(stringify_values(request._method_args))
|
||||
if self.access_token:
|
||||
method_args['access_token'] = self.access_token
|
||||
timeout = request._api._timeout
|
||||
response = self.requests_session.post(url, method_args, timeout=timeout)
|
||||
return response
|
||||
|
||||
def on_captcha_is_needed(self, error_data, method_request):
|
||||
"""
|
||||
Default behavior on CAPTCHA is to raise exception
|
||||
Reload this in child
|
||||
"""
|
||||
raise VkAPIMethodError(error_data)
|
||||
|
||||
def auth_code_is_needed(self, content, session):
|
||||
"""
|
||||
Default behavior on 2-AUTH CODE is to raise exception
|
||||
Reload this in child
|
||||
"""
|
||||
raise VkAuthError('Authorization error (2-factor code is needed)')
|
||||
|
||||
def auth_captcha_is_needed(self, content, session):
|
||||
"""
|
||||
Default behavior on CAPTCHA is to raise exception
|
||||
Reload this in child
|
||||
"""
|
||||
raise VkAuthError('Authorization error (captcha)')
|
||||
|
||||
def phone_number_is_needed(self, content, session):
|
||||
"""
|
||||
Default behavior on PHONE NUMBER is to raise exception
|
||||
Reload this in child
|
||||
"""
|
||||
logger.error('Authorization error (phone number is needed)')
|
||||
raise VkAuthError('Authorization error (phone number is needed)')
|
||||
|
||||
|
||||
class API(object):
|
||||
def __init__(self, session, timeout=10, **method_default_args):
|
||||
self._session = session
|
||||
self._timeout = timeout
|
||||
self._method_default_args = method_default_args
|
||||
|
||||
def __getattr__(self, method_name):
|
||||
return Request(self, method_name)
|
||||
|
||||
def __call__(self, method_name, **method_kwargs):
|
||||
return getattr(self, method_name)(**method_kwargs)
|
||||
|
||||
|
||||
class Request(object):
|
||||
__slots__ = ('_api', '_method_name', '_method_args')
|
||||
|
||||
def __init__(self, api, method_name):
|
||||
self._api = api
|
||||
self._method_name = method_name
|
||||
|
||||
def __getattr__(self, method_name):
|
||||
return Request(self._api, self._method_name + '.' + method_name)
|
||||
|
||||
def __call__(self, **method_args):
|
||||
self._method_args = method_args
|
||||
return self._api._session.make_request(self)
|
||||
|
||||
|
||||
class AuthSession(AuthMixin, Session):
|
||||
pass
|
||||
|
||||
|
||||
class InteractiveSession(InteractiveMixin, Session):
|
||||
pass
|
||||
|
||||
|
||||
class InteractiveAuthSession(InteractiveMixin, AuthSession):
|
||||
pass
|
||||
@ -1,30 +0,0 @@
|
||||
|
||||
# API Error Codes
|
||||
AUTHORIZATION_FAILED = 5 # Invalid access token
|
||||
CAPTCHA_IS_NEEDED = 14
|
||||
ACCESS_DENIED = 15 # No access to call this method
|
||||
|
||||
class VkException(Exception):
|
||||
pass
|
||||
|
||||
|
||||
class VkAuthError(VkException):
|
||||
pass
|
||||
|
||||
|
||||
class VkAPIMethodError(VkException):
|
||||
__slots__ = ['error', 'code', 'message', 'request_params', 'redirect_uri']
|
||||
|
||||
def __init__(self, error):
|
||||
super(VkAPIMethodError, self).__init__()
|
||||
self.error = error
|
||||
self.code = error.get('error_code')
|
||||
self.message = error.get('error_msg')
|
||||
self.request_params = error.get('request_params')
|
||||
self.redirect_uri = error.get('redirect_uri')
|
||||
|
||||
def __str__(self):
|
||||
error_message = '{self.code}. {self.message}. request_params = {self.request_params}'.format(self=self)
|
||||
if self.redirect_uri:
|
||||
error_message += ',\nredirect_uri = "{self.redirect_uri}"'.format(self=self)
|
||||
return error_message
|
||||
@ -1,26 +0,0 @@
|
||||
|
||||
import sys
|
||||
|
||||
|
||||
LOGGING_CONFIG = {
|
||||
'version': 1,
|
||||
'loggers': {
|
||||
'vk': {
|
||||
'level': 'INFO',
|
||||
'handlers': ['vk-stdout'],
|
||||
'propagate': False,
|
||||
},
|
||||
},
|
||||
'handlers': {
|
||||
'vk-stdout': {
|
||||
'class': 'logging.StreamHandler',
|
||||
'stream': sys.stdout,
|
||||
'formatter': 'vk-verbose',
|
||||
},
|
||||
},
|
||||
'formatters': {
|
||||
'vk-verbose': {
|
||||
'format': '%(asctime)s %(name) -5s %(module)s:%(lineno)d %(levelname)s: %(message)s',
|
||||
},
|
||||
},
|
||||
}
|
||||
215
src/vk/mixins.py
215
src/vk/mixins.py
@ -1,215 +0,0 @@
|
||||
# coding=utf8
|
||||
|
||||
import re
|
||||
import logging
|
||||
|
||||
import requests
|
||||
|
||||
from vk.exceptions import VkAuthError
|
||||
from vk.utils import urlparse, parse_qsl, raw_input, get_url_query, LoggingSession, get_form_action
|
||||
|
||||
|
||||
logger = logging.getLogger('vk')
|
||||
|
||||
|
||||
class AuthMixin(object):
|
||||
LOGIN_URL = 'https://m.vk.com'
|
||||
# REDIRECT_URI = 'https://oauth.vk.com/blank.html'
|
||||
AUTHORIZE_URL = 'https://oauth.vk.com/authorize'
|
||||
CAPTCHA_URI = 'https://m.vk.com/captcha.php'
|
||||
|
||||
def __init__(self, app_id=None, user_login='', user_password='', scope='offline', **kwargs):
|
||||
logger.debug('AuthMixin.__init__(app_id=%(app_id)r, user_login=%(user_login)r, user_password=%(user_password)r, **kwargs=%(kwargs)s)',
|
||||
dict(app_id=app_id, user_login=user_login, user_password=user_password, kwargs=kwargs))
|
||||
|
||||
super(AuthMixin, self).__init__(**kwargs)
|
||||
|
||||
self.app_id = app_id
|
||||
self.user_login = user_login
|
||||
self.user_password = user_password
|
||||
self.scope = scope
|
||||
|
||||
@property
|
||||
def user_login(self):
|
||||
if not self._user_login:
|
||||
self._user_login = self.get_user_login()
|
||||
return self._user_login
|
||||
|
||||
@user_login.setter
|
||||
def user_login(self, value):
|
||||
self._user_login = value
|
||||
|
||||
def get_user_login(self):
|
||||
return self._user_login
|
||||
|
||||
@property
|
||||
def user_password(self):
|
||||
if not self._user_password:
|
||||
self._user_password = self.get_user_password()
|
||||
return self._user_password
|
||||
|
||||
@user_password.setter
|
||||
def user_password(self, value):
|
||||
self._user_password = value
|
||||
|
||||
def get_user_password(self):
|
||||
return self._user_password
|
||||
|
||||
def get_access_token(self):
|
||||
"""
|
||||
Get access token using app id and user login and password.
|
||||
"""
|
||||
logger.debug('AuthMixin.get_access_token()')
|
||||
|
||||
auth_session = LoggingSession()
|
||||
with auth_session as self.auth_session:
|
||||
self.auth_session = auth_session
|
||||
self.login()
|
||||
auth_response_url_query = self.oauth2_authorization()
|
||||
|
||||
if 'access_token' in auth_response_url_query:
|
||||
return auth_response_url_query['access_token'], auth_response_url_query['expires_in']
|
||||
else:
|
||||
raise VkAuthError('OAuth2 authorization error')
|
||||
|
||||
def login(self):
|
||||
"""
|
||||
Login
|
||||
"""
|
||||
|
||||
response = self.auth_session.get(self.LOGIN_URL)
|
||||
login_form_action = get_form_action(response.text)
|
||||
if not login_form_action:
|
||||
raise VkAuthError('VK changed login flow')
|
||||
|
||||
login_form_data = {
|
||||
'email': self.user_login,
|
||||
'pass': self.user_password,
|
||||
}
|
||||
response = self.auth_session.post(login_form_action, login_form_data)
|
||||
logger.debug('Cookies: %s', self.auth_session.cookies)
|
||||
|
||||
response_url_query = get_url_query(response.url)
|
||||
|
||||
if 'remixsid' in self.auth_session.cookies or 'remixsid6' in self.auth_session.cookies:
|
||||
return
|
||||
|
||||
if 'sid' in response_url_query:
|
||||
self.auth_captcha_is_needed(response, login_form_data)
|
||||
elif response_url_query.get('act') == 'authcheck':
|
||||
self.auth_check_is_needed(response.text)
|
||||
elif 'security_check' in response_url_query:
|
||||
self.phone_number_is_needed(response.text)
|
||||
else:
|
||||
message = 'Authorization error (incorrect password)'
|
||||
logger.error(message)
|
||||
raise VkAuthError(message)
|
||||
|
||||
def oauth2_authorization(self):
|
||||
"""
|
||||
OAuth2
|
||||
"""
|
||||
auth_data = {
|
||||
'client_id': self.app_id,
|
||||
'display': 'mobile',
|
||||
'response_type': 'token',
|
||||
'scope': self.scope,
|
||||
'v': '5.28',
|
||||
}
|
||||
response = self.auth_session.post(self.AUTHORIZE_URL, auth_data)
|
||||
response_url_query = get_url_query(response.url)
|
||||
if 'access_token' in response_url_query:
|
||||
return response_url_query
|
||||
|
||||
# Permissions is needed
|
||||
logger.info('Getting permissions')
|
||||
# form_action = re.findall(r'<form method="post" action="(.+?)">', auth_response.text)[0]
|
||||
form_action = get_form_action(response.text)
|
||||
logger.debug('Response form action: %s', form_action)
|
||||
if form_action:
|
||||
response = self.auth_session.get(form_action)
|
||||
response_url_query = get_url_query(response.url)
|
||||
return response_url_query
|
||||
|
||||
try:
|
||||
response_json = response.json()
|
||||
except ValueError: # not JSON in response
|
||||
error_message = 'OAuth2 grant access error'
|
||||
else:
|
||||
error_message = 'VK error: [{}] {}'.format(response_json['error'], response_json['error_description'])
|
||||
logger.error('Permissions obtained')
|
||||
raise VkAuthError(error_message)
|
||||
|
||||
def auth_check_is_needed(self, html):
|
||||
logger.info('User enabled 2 factors authorization. Auth check code is needed')
|
||||
auth_check_form_action = get_form_action(html)
|
||||
auth_check_code = self.get_auth_check_code()
|
||||
auth_check_data = {
|
||||
'code': auth_check_code,
|
||||
'_ajax': '1',
|
||||
'remember': '1'
|
||||
}
|
||||
response = self.auth_session.post(auth_check_form_action, data=auth_check_data)
|
||||
|
||||
def auth_captcha_is_needed(self, response, login_form_data):
|
||||
logger.info('Captcha is needed')
|
||||
|
||||
response_url_dict = get_url_query(response.url)
|
||||
|
||||
# form_url = re.findall(r'<form method="post" action="(.+)" novalidate>', response.text)
|
||||
captcha_form_action = get_form_action(response.text)
|
||||
logger.debug('form_url %s', captcha_form_action)
|
||||
if not captcha_form_action:
|
||||
raise VkAuthError('Cannot find form url')
|
||||
|
||||
captcha_url = '%s?s=%s&sid=%s' % (self.CAPTCHA_URI, response_url_dict['s'], response_url_dict['sid'])
|
||||
# logger.debug('Captcha url %s', captcha_url)
|
||||
|
||||
login_form_data['captcha_sid'] = response_url_dict['sid']
|
||||
login_form_data['captcha_key'] = self.on_captcha_is_needed(captcha_url)
|
||||
|
||||
response = self.auth_session.post(captcha_form_action, login_form_data)
|
||||
|
||||
# logger.debug('Cookies %s', self.auth_session.cookies)
|
||||
# if 'remixsid' not in self.auth_session.cookies and 'remixsid6' not in self.auth_session.cookies:
|
||||
# raise VkAuthError('Authorization error (Bad password or captcha key)')
|
||||
|
||||
def phone_number_is_needed(self, text):
|
||||
raise VkAuthError('Phone number is needed')
|
||||
|
||||
def get_auth_check_code(self):
|
||||
raise VkAuthError('Auth check code is needed')
|
||||
|
||||
|
||||
class InteractiveMixin(object):
|
||||
def get_user_login(self):
|
||||
user_login = raw_input('VK user login: ')
|
||||
return user_login.strip()
|
||||
|
||||
def get_user_password(self):
|
||||
import getpass
|
||||
user_password = getpass.getpass('VK user password: ')
|
||||
return user_password
|
||||
|
||||
def get_access_token(self):
|
||||
logger.debug('InteractiveMixin.get_access_token()')
|
||||
access_token, access_token_expires_in = super(InteractiveMixin, self).get_access_token()
|
||||
if not access_token:
|
||||
access_token = raw_input('VK API access token: ')
|
||||
access_token_expires_in = None
|
||||
return access_token, access_token_expires_in
|
||||
|
||||
def on_captcha_is_needed(self, url):
|
||||
"""
|
||||
Read CAPTCHA key from shell
|
||||
"""
|
||||
print('Open captcha url:', url)
|
||||
captcha_key = raw_input('Enter captcha key: ')
|
||||
return captcha_key
|
||||
|
||||
def get_auth_check_code(self):
|
||||
"""
|
||||
Read Auth code from shell
|
||||
"""
|
||||
auth_check_code = raw_input('Auth check code: ')
|
||||
return auth_check_code.strip()
|
||||
@ -1,48 +0,0 @@
|
||||
# coding=utf8
|
||||
|
||||
import os
|
||||
import sys
|
||||
import time
|
||||
|
||||
import unittest
|
||||
|
||||
import vk
|
||||
|
||||
sys.path.append(os.path.join(os.path.dirname(__file__), '..'))
|
||||
|
||||
# copy to test_props.py and fill it
|
||||
USER_LOGIN = '' # user email or phone number
|
||||
USER_PASSWORD = '' # user password
|
||||
APP_ID = '' # aka API/Client ID
|
||||
|
||||
from test_props import USER_LOGIN, USER_PASSWORD, APP_ID
|
||||
|
||||
|
||||
class VkTestCase(unittest.TestCase):
|
||||
|
||||
def setUp(self):
|
||||
auth_session = vk.AuthSession(app_id=APP_ID, user_login=USER_LOGIN, user_password=USER_PASSWORD)
|
||||
access_token, _ = auth_session.get_access_token()
|
||||
|
||||
session = vk.Session(access_token=access_token)
|
||||
self.vk_api = vk.API(session, lang='ru')
|
||||
|
||||
def test_get_server_time(self):
|
||||
time_1 = time.time() - 1
|
||||
time_2 = time_1 + 10
|
||||
server_time = self.vk_api.getServerTime()
|
||||
self.assertTrue(time_1 <= server_time <= time_2)
|
||||
|
||||
def test_get_server_time_via_token_api(self):
|
||||
time_1 = time.time() - 1
|
||||
time_2 = time_1 + 10
|
||||
server_time = self.vk_api.getServerTime()
|
||||
self.assertTrue(time_1 <= server_time <= time_2)
|
||||
|
||||
def test_get_profiles_via_token(self):
|
||||
profiles = self.vk_api.users.get(user_id=1)
|
||||
self.assertEqual(profiles[0]['last_name'], u'Дуров')
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
@ -1,73 +0,0 @@
|
||||
|
||||
import logging
|
||||
from collections import Iterable
|
||||
import re
|
||||
|
||||
import requests
|
||||
|
||||
|
||||
STRING_TYPES = (str, bytes, bytearray, unicode)
|
||||
|
||||
logger = logging.getLogger('vk')
|
||||
|
||||
|
||||
try:
|
||||
# Python 2
|
||||
from urllib import urlencode
|
||||
from urlparse import urlparse, parse_qsl
|
||||
except ImportError:
|
||||
# Python 3
|
||||
from urllib.parse import urlparse, parse_qsl, urlencode
|
||||
|
||||
|
||||
try:
|
||||
import simplejson as json
|
||||
except ImportError:
|
||||
import json
|
||||
|
||||
|
||||
try:
|
||||
# Python 2
|
||||
raw_input = raw_input
|
||||
except NameError:
|
||||
# Python 3
|
||||
raw_input = input
|
||||
|
||||
|
||||
def json_iter_parse(response_text):
|
||||
decoder = json.JSONDecoder(strict=False)
|
||||
idx = 0
|
||||
while idx < len(response_text):
|
||||
obj, idx = decoder.raw_decode(response_text, idx)
|
||||
yield obj
|
||||
|
||||
|
||||
def stringify_values(dictionary):
|
||||
stringified_values_dict = {}
|
||||
for key, value in dictionary.items():
|
||||
if isinstance(value, Iterable) and not isinstance(value, STRING_TYPES):
|
||||
value = u','.join(map(str, value))
|
||||
stringified_values_dict[key] = value
|
||||
return stringified_values_dict
|
||||
|
||||
|
||||
def get_url_query(url):
|
||||
parsed_url = urlparse(url)
|
||||
url_query = parse_qsl(parsed_url.fragment)
|
||||
# login_response_url_query can have multiple key
|
||||
url_query = dict(url_query)
|
||||
return url_query
|
||||
|
||||
|
||||
def get_form_action(html):
|
||||
form_action = re.findall(r'<form(?= ).* action="(.+)"', html)
|
||||
if form_action:
|
||||
return form_action[0]
|
||||
|
||||
|
||||
class LoggingSession(requests.Session):
|
||||
def request(self, method, url, **kwargs):
|
||||
logger.debug('Request: %s %s, params=%r, data=%r', method, url, kwargs.get('params'), kwargs.get('data'))
|
||||
response = super(LoggingSession, self).request(method, url, **kwargs)
|
||||
logger.debug('Response: %s %s', response.status_code, response.url)
|
||||
return response
|
||||
Loading…
Reference in New Issue
Block a user