Added some code for starting
This commit is contained in:
10
src/vk/__init__.py
Normal file
10
src/vk/__init__.py
Normal file
@@ -0,0 +1,10 @@
|
||||
|
||||
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
Normal file
177
src/vk/api.py
Normal file
@@ -0,0 +1,177 @@
|
||||
# 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 self.make_request(method_request)
|
||||
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
|
||||
30
src/vk/exceptions.py
Normal file
30
src/vk/exceptions.py
Normal file
@@ -0,0 +1,30 @@
|
||||
|
||||
# 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
|
||||
26
src/vk/logs.py
Normal file
26
src/vk/logs.py
Normal file
@@ -0,0 +1,26 @@
|
||||
|
||||
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
Normal file
215
src/vk/mixins.py
Normal file
@@ -0,0 +1,215 @@
|
||||
# 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()
|
||||
48
src/vk/tests.py
Normal file
48
src/vk/tests.py
Normal file
@@ -0,0 +1,48 @@
|
||||
# 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()
|
||||
157
src/vk/upload.py
Normal file
157
src/vk/upload.py
Normal file
@@ -0,0 +1,157 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
|
||||
"""
|
||||
@author: Kirill Python
|
||||
@contact: https://vk.com/python273
|
||||
@license Apache License, Version 2.0, see LICENSE file
|
||||
|
||||
Copyright (C) 2015
|
||||
"""
|
||||
|
||||
|
||||
class VkUpload(object):
|
||||
def __init__(self, vk):
|
||||
"""
|
||||
|
||||
:param vk: объект VkApi
|
||||
"""
|
||||
|
||||
self.vk = vk
|
||||
# https://vk.com/dev/upload_files
|
||||
|
||||
def photo(self, photos, album_id,
|
||||
latitude=None, longitude=None, caption=None, description=None,
|
||||
group_id=None):
|
||||
""" Загрузка изображений в альбом пользователя
|
||||
|
||||
:param photos: список путей к изображениям, либо путь к изображению
|
||||
:param album_id: идентификатор альбома
|
||||
:param latitude: географическая широта, заданная в градусах
|
||||
(от -90 до 90)
|
||||
:param longitude: географическая долгота, заданная в градусах
|
||||
(от -180 до 180)
|
||||
:param caption: текст описания изображения
|
||||
:param description: текст описания альбома
|
||||
:param group_id: идентификатор сообщества (если загрузка идет в группу)
|
||||
"""
|
||||
|
||||
values = {'album_id': album_id}
|
||||
|
||||
if group_id:
|
||||
values['group_id'] = group_id
|
||||
|
||||
# Получаем ссылку для загрузки
|
||||
url = self.vk.photos.getUploadServer(values)['upload_url']
|
||||
|
||||
# Загружаем
|
||||
photos_files = open_photos(photos)
|
||||
response = self.vk.requests_session.post(url, files=photos_files).json()
|
||||
close_photos(photos_files)
|
||||
|
||||
# Олег Илларионов:
|
||||
# это не могу к сожалению просто пофиксить
|
||||
if 'album_id' not in response:
|
||||
response['album_id'] = response['aid']
|
||||
|
||||
response.update({
|
||||
'latitude': latitude,
|
||||
'longitude': longitude,
|
||||
'caption': caption,
|
||||
'description': description
|
||||
})
|
||||
|
||||
values.update(response)
|
||||
|
||||
# Сохраняем фото в альбоме
|
||||
response = self.vk.photos.save(values)
|
||||
|
||||
return response
|
||||
|
||||
def photo_messages(self, photos):
|
||||
""" Загрузка изображений в сообщения
|
||||
|
||||
:param photos: список путей к изображениям, либо путь к изображению
|
||||
"""
|
||||
|
||||
url = self.vk.method('photos.getMessagesUploadServer')
|
||||
url = url['upload_url']
|
||||
|
||||
photos_files = open_photos(photos)
|
||||
response = self.vk.http.post(url, files=photos_files)
|
||||
close_photos(photos_files)
|
||||
|
||||
response = self.vk.method('photos.saveMessagesPhoto', response.json())
|
||||
|
||||
return response
|
||||
|
||||
def photo_wall(self, photos, user_id=None, group_id=None):
|
||||
""" Загрузка изображений на стену пользователя или в группу
|
||||
|
||||
:param photos: список путей к изображениям, либо путь к изображению
|
||||
:param user_id: идентификатор пользователя
|
||||
:param group_id: идентификатор сообщества (если загрузка идет в группу)
|
||||
"""
|
||||
|
||||
values = {}
|
||||
|
||||
if user_id:
|
||||
values['user_id'] = user_id
|
||||
elif group_id:
|
||||
values['group_id'] = group_id
|
||||
response = self.vk.photos.getWallUploadServer(**values)
|
||||
|
||||
url = response['upload_url']
|
||||
print url
|
||||
photos_files = open_photos(photos)
|
||||
response = self.vk._session.requests_session.post(url, files=photos_files)
|
||||
close_photos(photos_files)
|
||||
|
||||
values.update(response.json())
|
||||
print values
|
||||
response = self.vk.photos.saveWallPhoto(**values)
|
||||
|
||||
return response
|
||||
|
||||
def document(self, file_path, title=None, tags=None, group_id=None):
|
||||
""" Загрузка документа
|
||||
|
||||
:param file_path: путь к документу
|
||||
:param title: название документа
|
||||
:param tags: метки для поиска
|
||||
:param group_id: идентификатор сообщества (если загрузка идет в группу)
|
||||
"""
|
||||
|
||||
values = {'group_id': group_id}
|
||||
url = self.vk.method('docs.getUploadServer', values)['upload_url']
|
||||
|
||||
with open(file_path, 'rb') as file:
|
||||
response = self.vk.http.post(url, files={'file': file}).json()
|
||||
|
||||
response.update({
|
||||
'title': title,
|
||||
'tags': tags
|
||||
})
|
||||
|
||||
response = self.vk.method('docs.save', response)
|
||||
|
||||
return response
|
||||
|
||||
|
||||
def open_photos(photos_paths):
|
||||
if not isinstance(photos_paths, list):
|
||||
photos_paths = [photos_paths]
|
||||
|
||||
photos = []
|
||||
|
||||
for x, filename in enumerate(photos_paths):
|
||||
filetype = filename.split('.')[-1]
|
||||
photos.append(
|
||||
('file%s' % x, ('pic.' + filetype, open(filename, 'rb')))
|
||||
)
|
||||
print photos
|
||||
return photos
|
||||
|
||||
|
||||
def close_photos(photos):
|
||||
for photo in photos:
|
||||
photo[1][1].close()
|
||||
73
src/vk/utils.py
Normal file
73
src/vk/utils.py
Normal file
@@ -0,0 +1,73 @@
|
||||
|
||||
import logging
|
||||
from collections import Iterable
|
||||
import re
|
||||
|
||||
import requests
|
||||
|
||||
|
||||
STRING_TYPES = (str, bytes, bytearray)
|
||||
|
||||
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 = ','.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
|
||||
Reference in New Issue
Block a user