ramjm
/
mrp_bot
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Initial commit

Browse Source
main
ramjm 3 months ago
commit 332cb3733c
25 changed files with 4879 additions and 0 deletions
Show Stats Download Patch File Download Diff File

21
LICENSE
Unescape View File

@ -0,0 +1,21 @@
MIT License
Copyright (c) 2024 Justuser
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.

27
README.md
Unescape View File

@ -0,0 +1,27 @@
# mrp_bot
Бот парсер онлайна проекта Mordor Role Play.
VK, TG
## Установка зависимостей:
```bash
pip install -r requirements.txt
```
## Что требуется для запуска ботов?
Создать файл .env:
```bash
touch .env
```
Заполнить следующие переменные любым текстовым редактором:
* tg_token = "API токен телеграм бота"
* vk_token = "API токен вк бота"
* vk_group_id = 123
ПРИМЕЧАНИЕ: вместо 123 впишите айди группы (тип - int)
---
Запустить ботов командами:
```bash
nohup python3 vk_bot.py &
nohup python3 tg_bot.py &
```
## Основная команда ботов
* /mrp_online - получение онлайна

2
config.py
Unescape View File

@ -0,0 +1,2 @@
# -*- coding: utf-8 -*-
headers = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:108.0) Gecko/20100101 Firefox/108.0"}

14
jconfig/__init__.py
Unescape View File

@ -0,0 +1,14 @@
# -*- coding: utf-8 -*-
"""
:authors: python273
:license: Apache License, Version 2.0, see LICENSE file
:copyright: (c) 2019 python273
"""
__author__ = 'python273'
__version__ = '3.0'
__email__ = 'vk_api@python273.pw'
from .jconfig import Config
from .memory import MemoryConfig

51
jconfig/base.py
Unescape View File

@ -0,0 +1,51 @@
# -*- coding: utf-8 -*-
"""
:authors: python273
:license: Apache License, Version 2.0, see LICENSE file
:copyright: (c) 2019 python273
"""
class BaseConfig(object):
""" Абстрактный базовый класс конфигурации.
У наследуемых классов должен быть определен `__slots__`
:param section: имя подкатегории в конфиге
:param \*\*kwargs: будут переданы в :func:`load`
"""
__slots__ = ('section_name', '_settings', '_section')
def __init__(self, section, **kwargs):
self.section_name = section
self._settings = self.load(**kwargs)
self._section = self._settings.setdefault(section, {})
def __getattr__(self, name):
return self._section.get(name)
__getitem__ = __getattr__
def __setattr__(self, name, value):
try:
super(BaseConfig, self).__setattr__(name, value)
except AttributeError:
self._section[name] = value
__setitem__ = __setattr__
def setdefault(self, k, d=None):
return self._section.setdefault(k, d)
def clear_section(self):
self._section.clear()
def load(self, **kwargs):
"""Абстрактный метод, должен возвращать dict с конфигом"""
raise NotImplementedError
def save(self):
"""Абстрактный метод, должен сохранять конфиг"""
raise NotImplementedError

41
jconfig/jconfig.py
Unescape View File

@ -0,0 +1,41 @@
# -*- coding: utf-8 -*-
"""
:authors: python273
:license: Apache License, Version 2.0, see LICENSE file
:copyright: (c) 2019 python273
"""
import json
from .base import BaseConfig
class Config(BaseConfig):
""" Класс конфигурации в файле
:param filename: имя файла
"""
__slots__ = ('_filename',)
def __init__(self, section, filename='.jconfig'):
self._filename = filename
super(Config, self).__init__(section, filename=filename)
def load(self, filename, **kwargs):
try:
with open(filename, 'r') as f:
settings = json.load(f)
except (IOError, ValueError):
settings = {}
settings.setdefault(self.section_name, {})
return settings
def save(self):
with open(self._filename, 'w') as f:
json.dump(self._settings, f, indent=2, sort_keys=True)

24
jconfig/memory.py
Unescape View File

@ -0,0 +1,24 @@
# -*- coding: utf-8 -*-
"""
:authors: python273
:license: Apache License, Version 2.0, see LICENSE file
:copyright: (c) 2019 python273
"""
from .base import BaseConfig
class MemoryConfig(BaseConfig):
""" Класс конфигурации в памяти
:param settings: существующий dict с конфигом
"""
__slots__ = tuple()
def load(self, settings=None, **kwargs):
return {} if settings is None else settings
def save(self):
pass

5
requirements.txt
Unescape View File

@ -0,0 +1,5 @@
telebot
vk_api
requests
random
python-dotenv

54
tg_bot.py
Unescape View File

@ -0,0 +1,54 @@
# -*- coding: utf-8 -*-
import telebot
from os import getenv
from requests import get
from dotenv import load_dotenv
from pathlib import Path
dotenv_path = f"{Path(__file__).parent.resolve()}/.env"
load_dotenv(dotenv_path=dotenv_path)
bot = telebot.TeleBot(getenv("tg_token"))
@bot.message_handler(commands=["mrp_online"])
def get_online(message):
try:
response = get("https://l.mordor-rp.com/launcher/monitoring/online.php").json()
rponl = 0; obsonl = 0; funonl = 0; text = ""
for element in response:
text += f'[{element["name"]}]: {element["min"]}\n'
obsonl += int(element["min"])
if element["tag"] == "roleplay": rponl += int(element["min"])
if element["tag"] == "fun": funonl += int(element["min"])
text += f"========================\n" \
f"ROLEPLAY ONLINE: {rponl}\n" \
f"FUN ONLINE: {funonl}\n" \
f"========================\n" \
f"FULL ONLINE: {obsonl}"
bot.reply_to(message, text)
except Exception:
bot.reply_to(message, "Ошибка.")
@bot.message_handler(commands=['start', 'help'])
def process_start_command(message):
bot.reply_to(
message,
f"Команды:\n"\
f"- /my_id - получение уникального идентификатора пользователя.\n"\
f"- /mrp_online - получение онлайна на проекте Mordor Role Play"
)
@bot.message_handler(commands=["my_id"])
def user_id(message):
uid = message.from_user.id
bot.reply_to(
message,
f"<a href='tg://user?id={uid}'>Пользователь</a>, "\
f"твой уникальный идентификатор (ID):\n\n"\
f"<code>{uid}</code>",
parse_mode="HTML"
)
if __name__ == '__main__':
while True:
try: bot.polling()
except KeyboardInterrupt: exit()
except: pass

18
vk_api/__init__.py
Unescape View File

@ -0,0 +1,18 @@
# -*- coding: utf-8 -*-
"""
:authors: python273
:license: Apache License, Version 2.0, see LICENSE file
:copyright: (c) 2019 python273
"""
from .enums import *
from .exceptions import *
from .requests_pool import VkRequestsPool, vk_request_one_param_pool
from .tools import VkTools
from .upload import VkUpload
from .vk_api import VkApi
__author__ = 'python273'
__version__ = '11.9.9'
__email__ = 'vk_api@python273.pw'

682
vk_api/audio.py
Unescape View File

@ -0,0 +1,682 @@
# -*- coding: utf-8 -*-
"""
:authors: python273
:license: Apache License, Version 2.0, see LICENSE file
:copyright: (c) 2019 python273
"""
import re
import json
import time
from itertools import islice
from bs4 import BeautifulSoup
from .audio_url_decoder import decode_audio_url
from .exceptions import AccessDenied
from .utils import set_cookies_from_list
RE_ALBUM_ID = re.compile(r'act=audio_playlist(-?\d+)_(\d+)')
RE_ACCESS_HASH = re.compile(r'access_hash=(\w+)')
RE_M3U8_TO_MP3 = re.compile(r'/[0-9a-f]+(/audios)?/([0-9a-f]+)/index.m3u8')
RPS_DELAY_RELOAD_AUDIO = 1.5
RPS_DELAY_LOAD_SECTION = 2.0
TRACKS_PER_USER_PAGE = 2000
TRACKS_PER_ALBUM_PAGE = 2000
ALBUMS_PER_USER_PAGE = 100
class VkAudio(object):
""" Модуль для получения аудиозаписей без использования официального API.
:param vk: Объект :class:`VkApi`
"""
__slots__ = ('_vk', 'user_id', 'convert_m3u8_links')
DEFAULT_COOKIES = [
{ # если не установлено, то первый запрос ломается
'version': 0,
'name': 'remixaudio_show_alert_today',
'value': '0',
'port': None,
'port_specified': False,
'domain': '.vk.com',
'domain_specified': True,
'domain_initial_dot': True,
'path': '/',
'path_specified': True,
'secure': True,
'expires': None,
'discard': False,
'comment': None,
'comment_url': None,
'rfc2109': False,
'rest': {}
}, { # для аудио из постов
'version': 0,
'name': 'remixmdevice',
'value': '1920/1080/2/!!-!!!!',
'port': None,
'port_specified': False,
'domain': '.vk.com',
'domain_specified': True,
'domain_initial_dot': True,
'path': '/',
'path_specified': True,
'secure': True,
'expires': None,
'discard': False,
'comment': None,
'comment_url': None,
'rfc2109': False,
'rest': {}
}
]
def __init__(self, vk, convert_m3u8_links=True):
self.user_id = vk.method('users.get')[0]['id']
self._vk = vk
self.convert_m3u8_links = convert_m3u8_links
set_cookies_from_list(self._vk.http.cookies, self.DEFAULT_COOKIES)
self._vk.http.get('https://m.vk.com/') # load cookies
def get_iter(self, owner_id=None, album_id=None, access_hash=None):
""" Получить список аудиозаписей пользователя (по частям)
:param owner_id: ID владельца (отрицательные значения для групп)
:param album_id: ID альбома
:param access_hash: ACCESS_HASH альбома
"""
if owner_id is None:
owner_id = self.user_id
if album_id is not None:
offset_diff = TRACKS_PER_ALBUM_PAGE
else:
offset_diff = TRACKS_PER_USER_PAGE
offset = 0
while True:
response = self._vk.http.post(
'https://m.vk.com/audio',
data={
'act': 'load_section',
'owner_id': owner_id,
'playlist_id': album_id if album_id else -1,
'offset': offset,
'type': 'playlist',
'access_hash': access_hash,
'is_loading_all': 1
},
allow_redirects=False
).json()
if not response['data'][0]:
raise AccessDenied(
'You don\'t have permissions to browse {}\'s albums'.format(
owner_id
)
)
ids = scrap_ids(
response['data'][0]['list']
)
tracks = scrap_tracks(
ids,
self.user_id,
self._vk.http,
convert_m3u8_links=self.convert_m3u8_links
)
if not tracks:
break
for i in tracks:
yield i
if response['data'][0]['hasMore']:
offset += offset_diff
else:
break
def get(self, owner_id=None, album_id=None, access_hash=None):
""" Получить список аудиозаписей пользователя
:param owner_id: ID владельца (отрицательные значения для групп)
:param album_id: ID альбома
:param access_hash: ACCESS_HASH альбома
"""
return list(self.get_iter(owner_id, album_id, access_hash))
def get_albums_iter(self, owner_id=None):
""" Получить список альбомов пользователя (по частям)
:param owner_id: ID владельца (отрицательные значения для групп)
"""
if owner_id is None:
owner_id = self.user_id
offset = 0
while True:
response = self._vk.http.get(
'https://m.vk.com/audio?act=audio_playlists{}'.format(
owner_id
),
params={
'offset': offset
},
allow_redirects=False
)
if not response.text:
raise AccessDenied(
'You don\'t have permissions to browse {}\'s albums'.format(
owner_id
)
)
albums = scrap_albums(response.text)
if not albums:
break
for i in albums:
yield i
offset += ALBUMS_PER_USER_PAGE
def get_albums(self, owner_id=None):
""" Получить список альбомов пользователя
:param owner_id: ID владельца (отрицательные значения для групп)
"""
return list(self.get_albums_iter(owner_id))
def search_user(self, owner_id=None, q=''):
""" Искать по аудиозаписям пользователя
:param owner_id: ID владельца (отрицательные значения для групп)
:param q: запрос
"""
if owner_id is None:
owner_id = self.user_id
response = self._vk.http.post(
'https://vk.com/al_audio.php',
data={
'al': 1,
'act': 'section',
'claim': 0,
'is_layer': 0,
'owner_id': owner_id,
'section': 'search',
'q': q
}
)
json_response = json.loads(response.text.replace('<!--', ''))
if not json_response['payload'][1]:
raise AccessDenied(
'You don\'t have permissions to browse {}\'s audio'.format(
owner_id
)
)
if json_response['payload'][1][1]['playlists']:
ids = scrap_ids(
json_response['payload'][1][1]['playlists'][0]['list']
)
tracks = scrap_tracks(
ids,
self.user_id,
self._vk.http,
convert_m3u8_links=self.convert_m3u8_links
)
return list(tracks)
else:
return []
def search(self, q, count=100, offset=0):
""" Искать аудиозаписи
:param q: запрос
:param count: количество
:param offset: смещение
"""
return islice(self.search_iter(q, offset=offset), count)
def search_iter(self, q, offset=0):
""" Искать аудиозаписи (генератор)
:param q: запрос
:param offset: смещение
"""
offset_left = 0
response = self._vk.http.post(
'https://vk.com/al_audio.php',
data={
'al': 1,
'act': 'section',
'claim': 0,
'is_layer': 0,
'owner_id': self.user_id,
'section': 'search',
'q': q
}
)
json_response = json.loads(response.text.replace('<!--', ''))
while json_response['payload'][1][1]['playlist']:
ids = scrap_ids(
json_response['payload'][1][1]['playlist']['list']
)
if offset_left + len(ids) >= offset:
if offset_left < offset:
ids = ids[offset - offset_left:]
tracks = scrap_tracks(
ids,
self.user_id,
convert_m3u8_links=self.convert_m3u8_links,
http=self._vk.http
)
if not tracks:
break
for track in tracks:
yield track
offset_left += len(ids)
response = self._vk.http.post(
'https://vk.com/al_audio.php',
data={
'al': 1,
'act': 'load_catalog_section',
'section_id': json_response['payload'][1][1]['sectionId'],
'start_from': json_response['payload'][1][1]['nextFrom']
}
)
json_response = json.loads(response.text.replace('<!--', ''))
def get_updates_iter(self):
""" Искать обновления друзей (генератор) """
response = self._vk.http.post(
'https://vk.com/al_audio.php',
data={
'al': 1,
'act': 'section',
'claim': 0,
'is_layer': 0,
'owner_id': self.user_id,
'section': 'updates'
}
)
json_response = json.loads(response.text.replace('<!--', ''))
while True:
updates = [i['list'] for i in json_response['payload'][1][1]['playlists']]
ids = scrap_ids(
[i[0] for i in updates if i]
)
tracks = scrap_tracks(
ids,
self.user_id,
convert_m3u8_links=self.convert_m3u8_links,
http=self._vk.http
)
if not tracks:
break
for track in tracks:
yield track
if len(updates) < 11:
break
response = self._vk.http.post(
'https://vk.com/al_audio.php',
data={
'al': 1,
'act': 'load_catalog_section',
'section_id': json_response['payload'][1][1]['sectionId'],
'start_from': json_response['payload'][1][1]['nextFrom']
}
)
json_response = json.loads(response.text.replace('<!--', ''))
def get_popular_iter(self, offset=0):
""" Искать популярные аудиозаписи (генератор)
:param offset: смещение
"""
response = self._vk.http.post(
'https://vk.com/audio',
data={
'block': 'chart',
'section': 'explore'
}
)
json_response = json.loads(scrap_json(response.text))
ids = scrap_ids(
json_response['sectionData']['explore']['playlist']['list']
)
if offset:
tracks = scrap_tracks(
ids[offset:],
self.user_id,
convert_m3u8_links=self.convert_m3u8_links,
http=self._vk.http
)
else:
tracks = scrap_tracks(
ids,
self.user_id,
convert_m3u8_links=self.convert_m3u8_links,
http=self._vk.http
)
for track in tracks:
yield track
def get_news_iter(self, offset=0):
""" Искать популярные аудиозаписи (генератор)
:param offset: смещение
"""
offset_left = 0
response = self._vk.http.post(
'https://vk.com/audio',
data={
'block': 'new_songs',
'section': 'explore'
}
)
json_response = json.loads(scrap_json(response.text))
ids = scrap_ids(
json_response['sectionData']['explore']['playlist']['list']
)
if offset_left + len(ids) >= offset:
if offset_left >= offset:
tracks = scrap_tracks(
ids,
self.user_id,
convert_m3u8_links=self.convert_m3u8_links,
http=self._vk.http
)
else:
tracks = scrap_tracks(
ids[offset - offset_left:],
self.user_id,
convert_m3u8_links=self.convert_m3u8_links,
http=self._vk.http
)
for track in tracks:
yield track
offset_left += len(ids)
while True:
response = self._vk.http.post(
'https://vk.com/al_audio.php',
data={
'al': 1,
'act': 'load_catalog_section',
'section_id': json_response['sectionData']['explore']['sectionId'],
'start_from': json_response['sectionData']['explore']['nextFrom']
}
)
json_response = json.loads(response.text.replace('<!--', ''))
ids = scrap_ids(
json_response['payload'][1][1]['playlist']['list']
)
if offset_left + len(ids) >= offset:
if offset_left >= offset:
tracks = scrap_tracks(
ids,
self.user_id,
convert_m3u8_links=self.convert_m3u8_links,
http=self._vk.http
)
else:
tracks = scrap_tracks(
ids[offset - offset_left:],
self.user_id,
convert_m3u8_links=self.convert_m3u8_links,
http=self._vk.http
)
if not tracks:
break
for track in tracks:
yield track
offset_left += len(ids)
def get_audio_by_id(self, owner_id, audio_id):
""" Получить аудиозапись по ID
:param owner_id: ID владельца (отрицательные значения для групп)
:param audio_id: ID аудио
"""
response = self._vk.http.get(
'https://m.vk.com/audio{}_{}'.format(owner_id, audio_id),
allow_redirects=False
)
ids = scrap_ids_from_html(
response.text,
filter_root_el={'class': 'basisDefault'}
)
track = scrap_tracks(
ids,
self.user_id,
http=self._vk.http,
convert_m3u8_links=self.convert_m3u8_links
)
if track:
return next(track)
else:
return []
def get_post_audio(self, owner_id, post_id):
""" Получить список аудиозаписей из поста пользователя или группы
:param owner_id: ID владельца (отрицательные значения для групп)
:param post_id: ID поста
"""
response = self._vk.http.get(
'https://m.vk.com/wall{}_{}'.format(owner_id, post_id)
)
ids = scrap_ids_from_html(
response.text,
filter_root_el={'class': 'audios_list'}
)
tracks = scrap_tracks(
ids,
self.user_id,
http=self._vk.http,
convert_m3u8_links=self.convert_m3u8_links
)
return tracks
def scrap_ids(audio_data):
""" Парсинг списка хэшей аудиозаписей из json объекта """
ids = []
for track in audio_data:
audio_hashes = track[13].split("/")
full_id = (
str(track[1]), str(track[0]), audio_hashes[2], audio_hashes[5]
)
if all(full_id):
ids.append(full_id)
return ids
def scrap_json(html_page):
""" Парсинг списка хэшей ауфдиозаписей новинок или популярных + nextFrom&sessionId """
find_json_pattern = r"new AudioPage\(.*?(\{.*\})"
fr = re.search(find_json_pattern, html_page).group(1)
return fr
def scrap_ids_from_html(html, filter_root_el=None):
""" Парсинг списка хэшей аудиозаписей из html страницы """
if filter_root_el is None:
filter_root_el = {'id': 'au_search_items'}
soup = BeautifulSoup(html, 'html.parser')
ids = []
root_el = soup.find(**filter_root_el)
if root_el is None:
raise ValueError('Could not find root el for audio')
playlist_snippets = soup.find_all('div', {'class': "audioPlaylistSnippet__list"})
for playlist in playlist_snippets:
playlist.decompose()
for audio in root_el.find_all('div', {'class': 'audio_item'}):
if 'audio_item_disabled' in audio['class']:
continue
data_audio = json.loads(audio['data-audio'])
audio_hashes = data_audio[13].split("/")
full_id = (
str(data_audio[1]), str(data_audio[0]), audio_hashes[2], audio_hashes[5]
)
if all(full_id):
ids.append(full_id)
return ids
def scrap_tracks(ids, user_id, http, convert_m3u8_links=True):
last_request = 0.0
for ids_group in [ids[i:i + 10] for i in range(0, len(ids), 10)]:
delay = RPS_DELAY_RELOAD_AUDIO - (time.time() - last_request)
if delay > 0:
time.sleep(delay)
result = http.post(
'https://m.vk.com/audio',
data={'act': 'reload_audio', 'ids': ','.join(['_'.join(i) for i in ids_group])}
).json()
last_request = time.time()
if result['data']:
data_audio = result['data'][0]
for audio in data_audio:
artist = BeautifulSoup(audio[4], 'html.parser').text
title = BeautifulSoup(audio[3].strip(), 'html.parser').text
duration = audio[5]
link = audio[2]
if 'audio_api_unavailable' in link:
link = decode_audio_url(link, user_id)
if convert_m3u8_links and 'm3u8' in link:
link = RE_M3U8_TO_MP3.sub(r'\1/\2.mp3', link)
yield {
'id': audio[0],
'owner_id': audio[1],
'track_covers': audio[14].split(',') if audio[14] else [],
'url': link,
'artist': artist,
'title': title,
'duration': duration,
}
def scrap_albums(html):
""" Парсинг списка альбомов из html страницы """
soup = BeautifulSoup(html, 'html.parser')
albums = []
for album in soup.find_all('div', {'class': 'audioPlaylistsPage__item'}):
link = album.select_one('.audioPlaylistsPage__itemLink')['href']
full_id = tuple(int(i) for i in RE_ALBUM_ID.search(link).groups())
access_hash = RE_ACCESS_HASH.search(link)
stats_text = album.select_one('.audioPlaylistsPage__stats').text
# "1 011 прослушиваний"
try:
plays = int(stats_text.rsplit(' ', 1)[0].replace(' ', ''))
except ValueError:
plays = None
albums.append({
'id': full_id[1],
'owner_id': full_id[0],
'url': 'https://m.vk.com/audio?act=audio_playlist{}_{}'.format(
*full_id
),
'access_hash': access_hash.group(1) if access_hash else None,
'title': album.select_one('.audioPlaylistsPage__title').text,
'artist': album.select_one('.audioPlaylistsPage__author').text,
'plays': plays
})
return albums

141
vk_api/audio_url_decoder.py
Unescape View File

@ -0,0 +1,141 @@
# -*- coding: utf-8 -*-
from .exceptions import VkAudioUrlDecodeError
VK_STR = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMN0PQRSTUVWXYZO123456789+/="
def splice(l, a, b, c):
""" JS's Array.prototype.splice
var x = [1, 2, 3],
y = x.splice(0, 2, 1337);
eq
x = [1, 2, 3]
x, y = splice(x, 0, 2, 1337)
"""
return l[:a] + [c] + l[a + b:], l[a:a + b]
def decode_audio_url(string, user_id):
vals = string.split("?extra=", 1)[1].split("#")
tstr = vk_o(vals[0])
ops_list = vk_o(vals[1]).split('\x09')[::-1]
for op_data in ops_list:
split_op_data = op_data.split('\x0b')
cmd = split_op_data[0]
if len(split_op_data) > 1:
arg = split_op_data[1]
else:
arg = None
if cmd == 'v':
tstr = tstr[::-1]
elif cmd == 'r':
tstr = vk_r(tstr, arg)
elif cmd == 'x':
tstr = vk_xor(tstr, arg)
elif cmd == 's':
tstr = vk_s(tstr, arg)
elif cmd == 'i':
tstr = vk_i(tstr, arg, user_id)
else:
raise VkAudioUrlDecodeError(
'Unknown decode cmd: "{}"; Please send bugreport'.format(cmd)
)
return tstr
def vk_o(string):
result = []
index2 = 0
for s in string:
sym_index = VK_STR.find(s)
if sym_index != -1:
if index2 % 4 != 0:
i = (i << 6) + sym_index
else:
i = sym_index
if index2 % 4 != 0:
index2 += 1
shift = -2 * index2 & 6
result += [chr(0xFF & (i >> shift))]
else:
index2 += 1
return ''.join(result)
def vk_r(string, i):
vk_str2 = VK_STR + VK_STR
vk_str2_len = len(vk_str2)
result = []
for s in string:
index = vk_str2.find(s)
if index != -1:
offset = index - int(i)
if offset < 0:
offset += vk_str2_len
result += [vk_str2[offset]]
else:
result += [s]
return ''.join(result)
def vk_xor(string, i):
xor_val = ord(i[0])
return ''.join(chr(ord(s) ^ xor_val) for s in string)
def vk_s_child(t, e):
i = len(t)
if not i:
return []
o = []
e = int(e)
for a in range(i - 1, -1, -1):
e = (i * (a + 1) ^ e + a) % i
o.append(e)
return o[::-1]
def vk_s(t, e):
i = len(t)
if not i:
return t
o = vk_s_child(t, e)
t = list(t)
for a in range(1, i):
t, y = splice(t, o[i - 1 - a], 1, t[a])
t[a] = y[0]
return ''.join(t)
def vk_i(t, e, user_id):
return vk_s(t, int(e) ^ user_id)

287
vk_api/bot_longpoll.py
Unescape View File

@ -0,0 +1,287 @@
# -*- coding: utf-8 -*-
"""
:authors: deker104, python273
:license: Apache License, Version 2.0, see LICENSE file
:copyright: (c) 2019 python273
"""
from enum import Enum
import requests
CHAT_START_ID = int(2E9)
class DotDict(dict):
__getattr__ = dict.get
__setattr__ = dict.__setitem__
__delattr__ = dict.__delitem__
class VkBotEventType(Enum):
MESSAGE_NEW = 'message_new'
MESSAGE_REPLY = 'message_reply'
MESSAGE_EDIT = 'message_edit'
MESSAGE_EVENT = 'message_event'
MESSAGE_TYPING_STATE = 'message_typing_state'
MESSAGE_ALLOW = 'message_allow'
MESSAGE_DENY = 'message_deny'
PHOTO_NEW = 'photo_new'
PHOTO_COMMENT_NEW = 'photo_comment_new'
PHOTO_COMMENT_EDIT = 'photo_comment_edit'
PHOTO_COMMENT_RESTORE = 'photo_comment_restore'
PHOTO_COMMENT_DELETE = 'photo_comment_delete'
AUDIO_NEW = 'audio_new'
VIDEO_NEW = 'video_new'
VIDEO_COMMENT_NEW = 'video_comment_new'
VIDEO_COMMENT_EDIT = 'video_comment_edit'
VIDEO_COMMENT_RESTORE = 'video_comment_restore'
VIDEO_COMMENT_DELETE = 'video_comment_delete'
WALL_POST_NEW = 'wall_post_new'
WALL_REPOST = 'wall_repost'
WALL_REPLY_NEW = 'wall_reply_new'
WALL_REPLY_EDIT = 'wall_reply_edit'
WALL_REPLY_RESTORE = 'wall_reply_restore'
WALL_REPLY_DELETE = 'wall_reply_delete'
BOARD_POST_NEW = 'board_post_new'
BOARD_POST_EDIT = 'board_post_edit'
BOARD_POST_RESTORE = 'board_post_restore'
BOARD_POST_DELETE = 'board_post_delete'
MARKET_COMMENT_NEW = 'market_comment_new'
MARKET_COMMENT_EDIT = 'market_comment_edit'
MARKET_COMMENT_RESTORE = 'market_comment_restore'
MARKET_COMMENT_DELETE = 'market_comment_delete'
GROUP_LEAVE = 'group_leave'
GROUP_JOIN = 'group_join'
USER_BLOCK = 'user_block'
USER_UNBLOCK = 'user_unblock'
POLL_VOTE_NEW = 'poll_vote_new'
GROUP_OFFICERS_EDIT = 'group_officers_edit'
GROUP_CHANGE_SETTINGS = 'group_change_settings'
GROUP_CHANGE_PHOTO = 'group_change_photo'
VKPAY_TRANSACTION = 'vkpay_transaction'
class VkBotEvent(object):
""" Событие Bots Long Poll
:ivar raw: событие, в каком виде было получено от сервера
:ivar type: тип события
:vartype type: VkBotEventType or str
:ivar t: сокращение для type
:vartype t: VkBotEventType or str
:ivar object: объект события, в каком виде был получен от сервера
:ivar obj: сокращение для object
:ivar group_id: ID группы бота
:vartype group_id: int
"""
__slots__ = (
'raw',
't', 'type',
'obj', 'object',
'client_info', 'message',
'group_id'
)
def __init__(self, raw):
self.raw = raw
try:
self.type = VkBotEventType(raw['type'])
except ValueError:
self.type = raw['type']
self.t = self.type # shortcut
self.object = DotDict(raw['object'])
try:
self.message = DotDict(raw['object']['message'])
except KeyError:
self.message = None
self.obj = self.object
try:
self.client_info = DotDict(raw['object']['client_info'])
except KeyError:
self.client_info = None
self.group_id = raw['group_id']
def __repr__(self):
return '<{}({})>'.format(type(self), self.raw)
class VkBotMessageEvent(VkBotEvent):
""" Событие с сообщением Bots Long Poll
:ivar from_user: сообщение от пользователя
:vartype from_user: bool
:ivar from_chat: сообщение из беседы
:vartype from_chat: bool
:ivar from_group: сообщение от группы
:vartype from_group: bool
:ivar chat_id: ID чата
:vartype chat_id: int
"""
__slots__ = ('from_user', 'from_chat', 'from_group', 'chat_id')
def __init__(self, raw):
super(VkBotMessageEvent, self).__init__(raw)
self.from_user = False
self.from_chat = False
self.from_group = False
self.chat_id = None
peer_id = self.obj.peer_id or self.message.peer_id
if peer_id < 0:
self.from_group = True
elif peer_id < CHAT_START_ID:
self.from_user = True
else:
self.from_chat = True
self.chat_id = peer_id - CHAT_START_ID
class VkBotLongPoll(object):
""" Класс для работы с Bots Long Poll сервером
`Подробнее в документации VK API <https://vk.com/dev/bots_longpoll>`__.
:param vk: объект :class:`VkApi`
:param group_id: id группы
:param wait: время ожидания
"""
__slots__ = (
'vk', 'wait', 'group_id',
'url', 'session',
'key', 'server', 'ts'
)
#: Классы для событий по типам
CLASS_BY_EVENT_TYPE = {
VkBotEventType.MESSAGE_NEW.value: VkBotMessageEvent,
VkBotEventType.MESSAGE_REPLY.value: VkBotMessageEvent,
VkBotEventType.MESSAGE_EDIT.value: VkBotMessageEvent,
}
#: Класс для событий
DEFAULT_EVENT_CLASS = VkBotEvent
def __init__(self, vk, group_id, wait=25):
self.vk = vk
self.group_id = group_id
self.wait = wait
self.url = None
self.key = None
self.server = None
self.ts = None
self.session = requests.Session()
self.update_longpoll_server()
def _parse_event(self, raw_event):
event_class = self.CLASS_BY_EVENT_TYPE.get(
raw_event['type'],
self.DEFAULT_EVENT_CLASS
)
return event_class(raw_event)
def update_longpoll_server(self, update_ts=True):
values = {
'group_id': self.group_id
}
response = self.vk.method('groups.getLongPollServer', values)
self.key = response['key']
self.server = response['server']
self.url = self.server
if update_ts:
self.ts = response['ts']
def check(self):
""" Получить события от сервера один раз
:returns: `list` of :class:`Event`
"""
values = {
'act': 'a_check',
'key': self.key,
'ts': self.ts,
'wait': self.wait,
}
response = self.session.get(
self.url,
params=values,
timeout=self.wait + 10
).json()
if 'failed' not in response:
self.ts = response['ts']
return [
self._parse_event(raw_event)
for raw_event in response['updates']
]
elif response['failed'] == 1:
self.ts = response['ts']
elif response['failed'] == 2:
self.update_longpoll_server(update_ts=False)
elif response['failed'] == 3:
self.update_longpoll_server()
return []
def listen(self):
""" Слушать сервер
:yields: :class:`Event`
"""
while True:
for event in self.check():
yield event

81
vk_api/enums.py
Unescape View File

@ -0,0 +1,81 @@
# -*- coding: utf-8 -*-
"""
:authors: python273
:license: Apache License, Version 2.0, see LICENSE file
:copyright: (c) 2019 python273
"""
from enum import IntEnum
class VkUserPermissions(IntEnum):
"""
Перечисление прав пользователя.
Список прав получается побитовым сложением (x | y) каждого права.
Подробнее в документации VK API: https://vk.com/dev/permissions
"""
#: Пользователь разрешил отправлять ему уведомления
#: (для flash/iframe-приложений).
#: Не работает с этой библиотекой.
NOTIFY = 1
#: Доступ к друзьям.
FRIEND = 2
#: Доступ к фотографиям.
PHOTOS = 2**2
#: Доступ к аудиозаписям.
#: При отсутствии доступа к закрытому API аудиозаписей это право позволяет
#: только загрузку аудио.
AUDIO = 2**3
#: Доступ к видеозаписям.
VIDEO = 2**4
#: Доступ к историям.
STORIES = 2**6
#: Доступ к wiki-страницам.
PAGES = 2**7
#: Добавление ссылки на приложение в меню слева.
ADD_LINK = 2**8
#: Доступ к статусу пользователя.
STATUS = 2**10
#: Доступ к заметкам пользователя.
NOTES = 2**11
#: Доступ к расширенным методам работы с сообщениями.
MESSAGES = 2**12
#: Доступ к обычным и расширенным методам работы со стеной.
WALL = 2**13
#: Доступ к расширенным методам работы с рекламным API.
ADS = 2**15
#: Доступ к API в любое время. Рекомендуется при работе с этой библиотекой.
OFFLINE = 2**16
#: Доступ к документам.
DOCS = 2**17
#: Доступ к группам пользователя.
GROUPS = 2**18
#: Доступ к оповещениям об ответах пользователю.
NOTIFICATIONS = 2**19
#: Доступ к статистике групп и приложений пользователя, администратором которых он является.
STATS = 2**20
#: Доступ к email пользователя.
EMAIL = 2**22
#: Доступ к товарам.
MARKET = 2**27

180
vk_api/exceptions.py
Unescape View File

@ -0,0 +1,180 @@
# -*- coding: utf-8 -*-
"""
:authors: python273
:license: Apache License, Version 2.0, see LICENSE file
:copyright: (c) 2019 python273
"""
TWOFACTOR_CODE = -2
HTTP_ERROR_CODE = -1
TOO_MANY_RPS_CODE = 6
CAPTCHA_ERROR_CODE = 14
NEED_VALIDATION_CODE = 17
class VkApiError(Exception):
pass
class AccessDenied(VkApiError):
pass
class AuthError(VkApiError):
pass
class LoginRequired(AuthError):
pass
class PasswordRequired(AuthError):
pass
class BadPassword(AuthError):
pass
class AccountBlocked(AuthError):
pass
class TwoFactorError(AuthError):
pass
class SecurityCheck(AuthError):
def __init__(self, phone_prefix=None, phone_postfix=None, response=None):
super(SecurityCheck, self).__init__()
self.phone_prefix = phone_prefix
self.phone_postfix = phone_postfix
self.response = response
def __str__(self):
if self.phone_prefix and self.phone_postfix:
return 'Security check. Enter number: {} ... {}'.format(
self.phone_prefix, self.phone_postfix
)
else:
return ('Security check. Phone prefix and postfix are not detected.'
' Please send bugreport (response in self.response)')
class ApiError(VkApiError):
def __init__(self, vk, method, values, raw, error):
super(ApiError, self).__init__()
self.vk = vk
self.method = method
self.values = values
self.raw = raw
self.code = error['error_code']
self.error = error
def try_method(self):
""" Отправить запрос заново """
return self.vk.method(self.method, self.values, raw=self.raw)
def __str__(self):
return '[{}] {}'.format(self.error['error_code'],
self.error['error_msg'])
class ApiHttpError(VkApiError):
def __init__(self, vk, method, values, raw, response):
super(ApiHttpError, self).__init__()
self.vk = vk
self.method = method
self.values = values
self.raw = raw
self.response = response
def try_method(self):
""" Отправить запрос заново """
return self.vk.method(self.method, self.values, raw=self.raw)
def __str__(self):
return 'Response code {}'.format(self.response.status_code)
class Captcha(VkApiError):
def __init__(self, vk, captcha_sid, func, args=None, kwargs=None, url=None):
super(Captcha, self).__init__()
self.vk = vk
self.sid = captcha_sid
self.func = func
self.args = args or ()
self.kwargs = kwargs or {}
self.code = CAPTCHA_ERROR_CODE
self.key = None
self.url = url
self.image = None
def get_url(self):
""" Получить ссылку на изображение капчи """
if not self.url:
self.url = 'https://api.vk.com/captcha.php?sid={}'.format(self.sid)
return self.url
def get_image(self):
""" Получить изображение капчи (jpg) """
if not self.image:
self.image = self.vk.http.get(self.get_url()).content
return self.image
def try_again(self, key=None):
""" Отправить запрос заново с ответом капчи
:param key: ответ капчи
"""
if key:
self.key = key
self.kwargs.update({
'captcha_sid': self.sid,
'captcha_key': self.key
})
return self.func(*self.args, **self.kwargs)
def __str__(self):
return 'Captcha needed'
class VkAudioException(Exception):
pass
class VkAudioUrlDecodeError(VkAudioException):
pass
class VkToolsException(VkApiError):
def __init__(self, *args, response=None):
super().__init__(*args)
self.response = response
class VkRequestsPoolException(Exception):
def __init__(self, error, *args, **kwargs):
self.error = error
super(VkRequestsPoolException, self).__init__(*args, **kwargs)

102
vk_api/execute.py
Unescape View File

@ -0,0 +1,102 @@
# -*- coding: utf-8 -*-
"""
:authors: python273
:license: Apache License, Version 2.0, see LICENSE file
:copyright: (c) 2019 python273
"""
from .utils import sjson_dumps
from .vk_api import VkApi, VkApiMethod
class VkFunction(object):
""" Обертка над методом execute.
:param code: код функции (VKScript)
:param args: список аргументов (будут конвертированы в JSON)
:param clean_args: список raw аргументов (будут вставлены как строки)
:param return_raw: аргумент raw функции VkApi.method
"""
__slots__ = ('code', '_minified_code', 'args', 'clean_args', 'return_raw')
def __init__(self, code, args=None, clean_args=None, return_raw=False):
self.code = code
self._minified_code = minify(code)
self.args = () if args is None else args
self.clean_args = () if clean_args is None else clean_args
self.return_raw = return_raw
def compile(self, args):
compiled_args = {}
for key, value in args.items():
if key in self.clean_args:
compiled_args[key] = str(value)
else:
compiled_args[key] = sjson_dumps(value)
return self._minified_code % compiled_args
def __call__(self, vk, *args, **kwargs):
"""
:param vk: VkApi или VkApiMethod
:param \*args:
:param \*\*kwargs:
"""
if not isinstance(vk, (VkApi, VkApiMethod)):
raise TypeError(
'The first arg should be VkApi or VkApiMethod instance'
)
if isinstance(vk, VkApiMethod):
vk = vk._vk
args = parse_args(self.args, args, kwargs)
return vk.method(
'execute',
{'code': self.compile(args)},
raw=self.return_raw
)
def minify(code):
return ''.join(i.strip() for i in code.splitlines())
def parse_args(function_args, args, kwargs):
parsed_args = {}
for arg_name in kwargs.keys():
if arg_name in function_args:
parsed_args[arg_name] = kwargs[arg_name]
else:
raise VkFunctionException(
'function got an unexpected keyword argument \'{}\''.format(
arg_name
))
args_count = len(args) + len(kwargs)
func_args_count = len(function_args)
if args_count != func_args_count:
raise VkFunctionException(
'function takes exactly {} argument{} ({} given)'.format(
func_args_count,
's' if func_args_count > 1 else '',
args_count
))
for arg_name, arg_value in zip(function_args, args):
parsed_args[arg_name] = arg_value
return parsed_args
class VkFunctionException(Exception):
pass

303
vk_api/keyboard.py
Unescape View File

@ -0,0 +1,303 @@
# -*- coding: utf-8 -*-
"""
:authors: python273, Helow19274, prostomarkeloff
:license: Apache License, Version 2.0, see LICENSE file
:copyright: (c) 2019 python273
"""
from enum import Enum
from .utils import sjson_dumps
MAX_BUTTONS_ON_LINE = 5
MAX_DEFAULT_LINES = 10
MAX_INLINE_LINES = 6
class VkKeyboardColor(Enum):
""" Возможные цвета кнопок """
#: Синяя
PRIMARY = 'primary'
#: Белая
SECONDARY = 'secondary'
#: Красная
NEGATIVE = 'negative'
#: Зелёная
POSITIVE = 'positive'
class VkKeyboardButton(Enum):
""" Возможные типы кнопки """
#: Кнопка с текстом
TEXT = "text"
#: Кнопка с местоположением
LOCATION = "location"
#: Кнопка с оплатой через VKPay
VKPAY = "vkpay"
#: Кнопка с приложением VK Apps
VKAPPS = "open_app"
#: Кнопка с ссылкой
OPENLINK = "open_link"
#: Callback-кнопка
CALLBACK = "callback"
class VkKeyboard(object):
""" Класс для создания клавиатуры для бота (https://vk.com/dev/bots_docs_3)
:param one_time: Если True, клавиатура исчезнет после нажатия на кнопку
:type one_time: bool
"""
__slots__ = ('one_time', 'lines', 'keyboard', 'inline')
def __init__(self, one_time=False, inline=False):
self.one_time = one_time
self.inline = inline
self.lines = [[]]
self.keyboard = {
'one_time': self.one_time,
'inline': self.inline,
'buttons': self.lines
}
def get_keyboard(self):
""" Получить json клавиатуры """
return sjson_dumps(self.keyboard)
@classmethod
def get_empty_keyboard(cls):
""" Получить json пустой клавиатуры.
Если отправить пустую клавиатуру, текущая у пользователя исчезнет.
"""
keyboard = cls()
keyboard.keyboard['buttons'] = []
return keyboard.get_keyboard()
def add_button(self, label, color=VkKeyboardColor.SECONDARY, payload=None):
""" Добавить кнопку с текстом.
Максимальное количество кнопок на строке - MAX_BUTTONS_ON_LINE
:param label: Надпись на кнопке и текст, отправляющийся при её нажатии.
:type label: str
:param color: цвет кнопки.
:type color: VkKeyboardColor or str
:param payload: Параметр для callback api
:type payload: str or list or dict
"""
current_line = self.lines[-1]
if len(current_line) >= MAX_BUTTONS_ON_LINE:
raise ValueError(f'Max {MAX_BUTTONS_ON_LINE} buttons on a line')
color_value = color
if isinstance(color, VkKeyboardColor):
color_value = color_value.value
if payload is not None and not isinstance(payload, str):
payload = sjson_dumps(payload)
button_type = VkKeyboardButton.TEXT.value
current_line.append({
'color': color_value,
'action': {
'type': button_type,
'payload': payload,
'label': label,
}
})
def add_callback_button(self, label, color=VkKeyboardColor.SECONDARY, payload=None):
""" Добавить callback-кнопку с текстом.
Максимальное количество кнопок на строке - MAX_BUTTONS_ON_LINE
:param label: Надпись на кнопке и текст, отправляющийся при её нажатии.
:type label: str
:param color: цвет кнопки.
:type color: VkKeyboardColor or str
:param payload: Параметр для callback api
:type payload: str or list or dict
"""
current_line = self.lines[-1]
if len(current_line) >= MAX_BUTTONS_ON_LINE:
raise ValueError(f'Max {MAX_BUTTONS_ON_LINE} buttons on a line')
color_value = color
if isinstance(color, VkKeyboardColor):
color_value = color_value.value
if payload is not None and not isinstance(payload, str):
payload = sjson_dumps(payload)
button_type = VkKeyboardButton.CALLBACK.value
current_line.append({
'color': color_value,
'action': {
'type': button_type,
'payload': payload,
'label': label,
}
})
def add_location_button(self, payload=None):
""" Добавить кнопку с местоположением.
Всегда занимает всю ширину линии.
:param payload: Параметр для callback api
:type payload: str or list or dict
"""
current_line = self.lines[-1]
if len(current_line) != 0:
raise ValueError(
'This type of button takes the entire width of the line'
)
if payload is not None and not isinstance(payload, str):
payload = sjson_dumps(payload)
button_type = VkKeyboardButton.LOCATION.value
current_line.append({
'action': {
'type': button_type,
'payload': payload
}
})
def add_vkpay_button(self, hash, payload=None):
""" Добавить кнопку с оплатой с помощью VKPay.
Всегда занимает всю ширину линии.
:param hash: Параметры платежа VKPay и ID приложения
(в поле aid) разделённые &
:type hash: str
:param payload: Параметр для совместимости со старыми клиентами
:type payload: str or list or dict
"""
current_line = self.lines[-1]
if len(current_line) != 0:
raise ValueError(
'This type of button takes the entire width of the line'
)
if payload is not None and not isinstance(payload, str):
payload = sjson_dumps(payload)
button_type = VkKeyboardButton.VKPAY.value
current_line.append({
'action': {
'type': button_type,
'payload': payload,
'hash': hash
}
})
def add_vkapps_button(self, app_id, owner_id, label, hash, payload=None):
""" Добавить кнопку с приложением VK Apps.
Всегда занимает всю ширину линии.
:param app_id: Идентификатор вызываемого приложения с типом VK Apps
:type app_id: int
:param owner_id: Идентификатор сообщества, в котором установлено
приложение, если требуется открыть в контексте сообщества
:type owner_id: int
:param label: Название приложения, указанное на кнопке
:type label: str
:param hash: хэш для навигации в приложении, будет передан в строке
параметров запуска после символа #
:type hash: str
:param payload: Параметр для совместимости со старыми клиентами
:type payload: str or list or dict
"""
current_line = self.lines[-1]
if len(current_line) != 0:
raise ValueError(
'This type of button takes the entire width of the line'
)
if payload is not None and not isinstance(payload, str):
payload = sjson_dumps(payload)
button_type = VkKeyboardButton.VKAPPS.value
current_line.append({
'action': {
'type': button_type,
'app_id': app_id,
'owner_id': owner_id,
'label': label,
'payload': payload,
'hash': hash
}
})
def add_openlink_button(self, label, link, payload=None):
""" Добавить кнопку с ссылкой
Максимальное количество кнопок на строке - MAX_BUTTONS_ON_LINE
:param label: Надпись на кнопке
:type label: str
:param link: ссылка, которую необходимо открыть по нажатию на кнопку
:type link: str
:param payload: Параметр для callback api
:type payload: str or list or dict
"""
current_line = self.lines[-1]
if len(current_line) >= MAX_BUTTONS_ON_LINE:
raise ValueError(f'Max {MAX_BUTTONS_ON_LINE} buttons on a line')
if payload is not None and not isinstance(payload, str):
payload = sjson_dumps(payload)
button_type = VkKeyboardButton.OPENLINK.value
current_line.append({
'action': {
'type': button_type,
'link': link,
'label': label,
'payload': payload
}
})
def add_line(self):
""" Создаёт новую строку, на которой можно размещать кнопки.
Максимальное количество строк:
Стандартное отображение - MAX_DEFAULT_LINES;
Inline-отображение - MAX_INLINE_LINES.
"""
if self.inline:
if len(self.lines) >= MAX_INLINE_LINES:
raise ValueError(f'Max {MAX_INLINE_LINES} lines for inline keyboard')
else:
if len(self.lines) >= MAX_DEFAULT_LINES:
raise ValueError(f'Max {MAX_DEFAULT_LINES} lines for default keyboard')
self.lines.append([])

620
vk_api/longpoll.py
Unescape View File

@ -0,0 +1,620 @@
# -*- coding: utf-8 -*-
"""
:authors: python273
:license: Apache License, Version 2.0, see LICENSE file
:copyright: (c) 2019 python273
"""
from collections import defaultdict
from datetime import datetime
from enum import IntEnum
import requests
CHAT_START_ID = int(2E9) # id с которого начинаются беседы
class VkLongpollMode(IntEnum):
""" Дополнительные опции ответа
`Подробнее в документации VK API
<https://vk.com/dev/using_longpoll?f=1.+Подключение>`_
"""
#: Получать вложения
GET_ATTACHMENTS = 2
#: Возвращать расширенный набор событий
GET_EXTENDED = 2**3
#: возвращать pts для метода `messages.getLongPollHistory`
GET_PTS = 2**5
#: В событии с кодом 8 (друг стал онлайн) возвращать
#: дополнительные данные в поле `extra`
GET_EXTRA_ONLINE = 2**6
#: Возвращать поле `random_id`
GET_RANDOM_ID = 2**7
DEFAULT_MODE = sum(VkLongpollMode)
class VkEventType(IntEnum):
""" Перечисление событий, получаемых от longpoll-сервера.
`Подробнее в документации VK API
<https://vk.com/dev/using_longpoll?f=3.+Структура+событий>`__
"""
#: Замена флагов сообщения (FLAGS:=$flags)
MESSAGE_FLAGS_REPLACE = 1
#: Установка флагов сообщения (FLAGS|=$mask)
MESSAGE_FLAGS_SET = 2
#: Сброс флагов сообщения (FLAGS&=~$mask)
MESSAGE_FLAGS_RESET = 3
#: Добавление нового сообщения.
MESSAGE_NEW = 4
#: Редактирование сообщения.
MESSAGE_EDIT = 5
#: Прочтение всех входящих сообщений в $peer_id,
#: пришедших до сообщения с $local_id.
READ_ALL_INCOMING_MESSAGES = 6
#: Прочтение всех исходящих сообщений в $peer_id,
#: пришедших до сообщения с $local_id.
READ_ALL_OUTGOING_MESSAGES = 7
#: Друг $user_id стал онлайн. $extra не равен 0, если в mode был передан флаг 64.
#: В младшем байте числа extra лежит идентификатор платформы
#: (см. :class:`VkPlatform`).
#: $timestamp — время последнего действия пользователя $user_id на сайте.
USER_ONLINE = 8
#: Друг $user_id стал оффлайн ($flags равен 0, если пользователь покинул сайт и 1,
#: если оффлайн по таймауту). $timestamp — время последнего действия пользователя
#: $user_id на сайте.
USER_OFFLINE = 9
#: Сброс флагов диалога $peer_id.
#: Соответствует операции (PEER_FLAGS &= ~$flags).
#: Только для диалогов сообществ.
PEER_FLAGS_RESET = 10
#: Замена флагов диалога $peer_id.
#: Соответствует операции (PEER_FLAGS:= $flags).
#: Только для диалогов сообществ.
PEER_FLAGS_REPLACE = 11
#: Установка флагов диалога $peer_id.
#: Соответствует операции (PEER_FLAGS|= $flags).
#: Только для диалогов сообществ.
PEER_FLAGS_SET = 12
#: Удаление всех сообщений в диалоге $peer_id с идентификаторами вплоть до $local_id.
PEER_DELETE_ALL = 13
#: Восстановление недавно удаленных сообщений в диалоге $peer_id с
#: идентификаторами вплоть до $local_id.
PEER_RESTORE_ALL = 14
#: Один из параметров (состав, тема) беседы $chat_id были изменены.
#: $self — 1 или 0 (вызваны ли изменения самим пользователем).
CHAT_EDIT = 51
#: Изменение информации чата $peer_id с типом $type_id
#: $info — дополнительная информация об изменениях
CHAT_UPDATE = 52
#: Пользователь $user_id набирает текст в диалоге.
#: Событие приходит раз в ~5 секунд при наборе текста. $flags = 1.
USER_TYPING = 61
#: Пользователь $user_id набирает текст в беседе $chat_id.
USER_TYPING_IN_CHAT = 62
#: Пользователь $user_id записывает голосовое сообщение в диалоге/беседе $peer_id
USER_RECORDING_VOICE = 64
#: Пользователь $user_id совершил звонок с идентификатором $call_id.
USER_CALL = 70
#: Счетчик в левом меню стал равен $count.
MESSAGES_COUNTER_UPDATE = 80
#: Изменились настройки оповещений.
#: $peer_id — идентификатор чата/собеседника,
#: $sound — 1/0, включены/выключены звуковые оповещения,
#: $disabled_until — выключение оповещений на необходимый срок.
NOTIFICATION_SETTINGS_UPDATE = 114
class VkPlatform(IntEnum):
""" Идентификаторы платформ """
#: Мобильная версия сайта или неопознанное мобильное приложение
MOBILE = 1
#: Официальное приложение для iPhone
IPHONE = 2
#: Официальное приложение для iPad
IPAD = 3
#: Официальное приложение для Android
ANDROID = 4
#: Официальное приложение для Windows Phone
WPHONE = 5
#: Официальное приложение для Windows 8
WINDOWS = 6
#: Полная версия сайта или неопознанное приложение
WEB = 7
class VkOfflineType(IntEnum):
""" Выход из сети в событии :attr:`VkEventType.USER_OFFLINE` """
#: Пользователь покинул сайт
EXIT = 0
#: Оффлайн по таймауту
AWAY = 1
class VkMessageFlag(IntEnum):
""" Флаги сообщений """
#: Сообщение не прочитано.
UNREAD = 1
#: Исходящее сообщение.
OUTBOX = 2
#: На сообщение был создан ответ.
REPLIED = 2**2
#: Помеченное сообщение.
IMPORTANT = 2**3
#: Сообщение отправлено через чат.
CHAT = 2**4
#: Сообщение отправлено другом.
#: Не применяется для сообщений из групповых бесед.
FRIENDS = 2**5
#: Сообщение помечено как "Спам".
SPAM = 2**6
#: Сообщение удалено (в корзине).
DELETED = 2**7
#: Сообщение проверено пользователем на спам.
FIXED = 2**8
#: Сообщение содержит медиаконтент
MEDIA = 2**9
#: Приветственное сообщение от сообщества.
HIDDEN = 2**16
#: Сообщение удалено для всех получателей.
DELETED_ALL = 2**17
class VkPeerFlag(IntEnum):
""" Флаги диалогов """
#: Важный диалог
IMPORTANT = 1
#: Неотвеченный диалог
UNANSWERED = 2
class VkChatEventType(IntEnum):
""" Идентификатор типа изменения в чате """
#: Изменилось название беседы
TITLE = 1
#: Сменилась обложка беседы
PHOTO = 2
#: Назначен новый администратор
ADMIN_ADDED = 3
#: Изменены настройки беседы
SETTINGS_CHANGED = 4
#: Закреплено сообщение
MESSAGE_PINNED = 5
#: Пользователь присоединился к беседе
USER_JOINED = 6
#: Пользователь покинул беседу
USER_LEFT = 7
#: Пользователя исключили из беседы
USER_KICKED = 8
#: С пользователя сняты права администратора
ADMIN_REMOVED = 9
#: Бот прислал клавиатуру
KEYBOARD_RECEIVED = 11
MESSAGE_EXTRA_FIELDS = [
'peer_id', 'timestamp', 'text', 'extra_values', 'attachments', 'random_id'
]
MSGID = 'message_id'
EVENT_ATTRS_MAPPING = {
VkEventType.MESSAGE_FLAGS_REPLACE: [MSGID, 'flags'] + MESSAGE_EXTRA_FIELDS,
VkEventType.MESSAGE_FLAGS_SET: [MSGID, 'mask'] + MESSAGE_EXTRA_FIELDS,
VkEventType.MESSAGE_FLAGS_RESET: [MSGID, 'mask'] + MESSAGE_EXTRA_FIELDS,
VkEventType.MESSAGE_NEW: [MSGID, 'flags'] + MESSAGE_EXTRA_FIELDS,
VkEventType.MESSAGE_EDIT: [MSGID, 'mask'] + MESSAGE_EXTRA_FIELDS,
VkEventType.READ_ALL_INCOMING_MESSAGES: ['peer_id', 'local_id'],
VkEventType.READ_ALL_OUTGOING_MESSAGES: ['peer_id', 'local_id'],
VkEventType.USER_ONLINE: ['user_id', 'extra', 'timestamp'],
VkEventType.USER_OFFLINE: ['user_id', 'flags', 'timestamp'],
VkEventType.PEER_FLAGS_RESET: ['peer_id', 'mask'],
VkEventType.PEER_FLAGS_REPLACE: ['peer_id', 'flags'],
VkEventType.PEER_FLAGS_SET: ['peer_id', 'mask'],
VkEventType.PEER_DELETE_ALL: ['peer_id', 'local_id'],
VkEventType.PEER_RESTORE_ALL: ['peer_id', 'local_id'],
VkEventType.CHAT_EDIT: ['chat_id', 'self'],
VkEventType.CHAT_UPDATE: ['type_id', 'peer_id', 'info'],
VkEventType.USER_TYPING: ['user_id', 'flags'],
VkEventType.USER_TYPING_IN_CHAT: ['user_id', 'chat_id'],
VkEventType.USER_RECORDING_VOICE: ['peer_id', 'user_id', 'flags', 'timestamp'],
VkEventType.USER_CALL: ['user_id', 'call_id'],
VkEventType.MESSAGES_COUNTER_UPDATE: ['count'],
VkEventType.NOTIFICATION_SETTINGS_UPDATE: ['values']
}
def get_all_event_attrs():
keys = set()
for l in EVENT_ATTRS_MAPPING.values():
keys.update(l)
return tuple(keys)
ALL_EVENT_ATTRS = get_all_event_attrs()
PARSE_PEER_ID_EVENTS = [
k for k, v in EVENT_ATTRS_MAPPING.items() if 'peer_id' in v
]
PARSE_MESSAGE_FLAGS_EVENTS = [
VkEventType.MESSAGE_FLAGS_REPLACE,
VkEventType.MESSAGE_NEW
]
class Event(object):
""" Событие, полученное от longpoll-сервера.
Имеет поля в соответствии с `документацией
<https://vk.com/dev/using_longpoll_2?f=3.%2BСтруктура%2Bсобытий>`_.
События `MESSAGE_NEW` и `MESSAGE_EDIT` имеют (среди прочих) такие поля:
- `text` - `экранированный <https://ru.wikipedia.org/wiki/Мнемоники_в_HTML>`_ текст
- `message` - оригинальный текст сообщения.
События с полем `timestamp` также дополнительно имеют поле `datetime`.
"""
def __init__(self, raw):
self.raw = raw
self.from_user = False
self.from_chat = False
self.from_group = False
self.from_me = False
self.to_me = False
self.attachments = {}
self.message_data = None
self.message_id = None
self.timestamp = None
self.peer_id = None
self.flags = None
self.extra = None
self.extra_values = None
self.type_id = None
try:
self.type = VkEventType(self.raw[0])
self._list_to_attr(self.raw[1:], EVENT_ATTRS_MAPPING[self.type])
except ValueError:
self.type = self.raw[0]
if self.extra_values:
self._dict_to_attr(self.extra_values)
if self.type in PARSE_PEER_ID_EVENTS:
self._parse_peer_id()
if self.type in PARSE_MESSAGE_FLAGS_EVENTS:
self._parse_message_flags()
if self.type is VkEventType.CHAT_UPDATE:
self._parse_chat_info()
try:
self.update_type = VkChatEventType(self.type_id)
except ValueError:
self.update_type = self.type_id
elif self.type is VkEventType.NOTIFICATION_SETTINGS_UPDATE:
self._dict_to_attr(self.values)
self._parse_peer_id()
elif self.type is VkEventType.PEER_FLAGS_REPLACE:
self._parse_peer_flags()
elif self.type in [VkEventType.MESSAGE_NEW, VkEventType.MESSAGE_EDIT]:
self._parse_message()
elif self.type in [VkEventType.USER_ONLINE, VkEventType.USER_OFFLINE]:
self.user_id = abs(self.user_id)
self._parse_online_status()
elif self.type is VkEventType.USER_RECORDING_VOICE:
if isinstance(self.user_id, list):
self.user_id = self.user_id[0]
if self.timestamp:
self.datetime = datetime.utcfromtimestamp(self.timestamp)
def _list_to_attr(self, raw, attrs):
for i in range(min(len(raw), len(attrs))):
self.__setattr__(attrs[i], raw[i])
def _dict_to_attr(self, values):
for k, v in values.items():
self.__setattr__(k, v)
def _parse_peer_id(self):
if self.peer_id < 0: # Сообщение от/для группы
self.from_group = True
self.group_id = abs(self.peer_id)
elif self.peer_id > CHAT_START_ID: # Сообщение из беседы
self.from_chat = True
self.chat_id = self.peer_id - CHAT_START_ID
if self.extra_values and 'from' in self.extra_values:
self.user_id = int(self.extra_values['from'])
else: # Сообщение от/для пользователя
self.from_user = True
self.user_id = self.peer_id
def _parse_message_flags(self):
self.message_flags = set(
x for x in VkMessageFlag if self.flags & x
)
def _parse_peer_flags(self):
self.peer_flags = set(
x for x in VkPeerFlag if self.flags & x
)
def _parse_message(self):
if self.type is VkEventType.MESSAGE_NEW:
if self.flags & VkMessageFlag.OUTBOX:
self.from_me = True
else:
self.to_me = True
# ВК возвращает сообщения в html-escaped виде,
# при этом переводы строк закодированы как <br> и не экранированы
self.text = self.text.replace('<br>', '\n')
self.message = self.text \
.replace('&lt;', '<') \
.replace('&gt;', '>') \
.replace('&quot;', '"') \
.replace('&amp;', '&')
def _parse_online_status(self):
try:
if self.type is VkEventType.USER_ONLINE:
self.platform = VkPlatform(self.extra & 0xFF)
elif self.type is VkEventType.USER_OFFLINE:
self.offline_type = VkOfflineType(self.flags)
except ValueError:
pass
def _parse_chat_info(self):
if self.type_id == VkChatEventType.ADMIN_ADDED.value:
self.info = {'admin_id': self.info}
elif self.type_id == VkChatEventType.MESSAGE_PINNED.value:
self.info = {'conversation_message_id': self.info}
elif self.type_id in [VkChatEventType.USER_JOINED.value,
VkChatEventType.USER_LEFT.value,
VkChatEventType.USER_KICKED.value,
VkChatEventType.ADMIN_REMOVED.value]:
self.info = {'user_id': self.info}
class VkLongPoll(object):
""" Класс для работы с longpoll-сервером
`Подробнее в документации VK API <https://vk.com/dev/using_longpoll>`__.
:param vk: объект :class:`VkApi`
:param wait: время ожидания
:param mode: дополнительные опции ответа
:param preload_messages: предзагрузка данных сообщений для
получения ссылок на прикрепленные файлы
:param group_id: идентификатор сообщества
(для сообщений сообщества с ключом доступа пользователя)
"""
__slots__ = (
'vk', 'wait', 'mode', 'preload_messages', 'group_id',
'url', 'session',
'key', 'server', 'ts', 'pts'
)
#: Класс для событий
DEFAULT_EVENT_CLASS = Event
#: События, для которых можно загрузить данные сообщений из API
PRELOAD_MESSAGE_EVENTS = [
VkEventType.MESSAGE_NEW,
VkEventType.MESSAGE_EDIT
]
def __init__(self, vk, wait=25, mode=DEFAULT_MODE,
preload_messages=False, group_id=None):
self.vk = vk
self.wait = wait
self.mode = mode.value if isinstance(mode, VkLongpollMode) else mode
self.preload_messages = preload_messages
self.group_id = group_id
self.url = None
self.key = None
self.server = None
self.ts = None
self.pts = mode & VkLongpollMode.GET_PTS
self.session = requests.Session()
self.update_longpoll_server()
def _parse_event(self, raw_event):
return self.DEFAULT_EVENT_CLASS(raw_event)
def update_longpoll_server(self, update_ts=True):
values = {
'lp_version': '3',
'need_pts': self.pts
}
if self.group_id:
values['group_id'] = self.group_id
response = self.vk.method('messages.getLongPollServer', values)
self.key = response['key']
self.server = response['server']
self.url = 'https://' + self.server
if update_ts:
self.ts = response['ts']
if self.pts:
self.pts = response['pts']
def check(self):
""" Получить события от сервера один раз
:returns: `list` of :class:`Event`
"""
values = {
'act': 'a_check',
'key': self.key,
'ts': self.ts,
'wait': self.wait,
'mode': self.mode,
'version': 3
}
response = self.session.get(
self.url,
params=values,
timeout=self.wait + 10
).json()
if 'failed' not in response:
self.ts = response['ts']
if self.pts:
self.pts = response['pts']
events = [
self._parse_event(raw_event)
for raw_event in response['updates']
]
if self.preload_messages:
self.preload_message_events_data(events)
return events
elif response['failed'] == 1:
self.ts = response['ts']
elif response['failed'] == 2:
self.update_longpoll_server(update_ts=False)
elif response['failed'] == 3:
self.update_longpoll_server()
return []
def preload_message_events_data(self, events):
""" Предзагрузка данных сообщений из API
:type events: list of Event
"""
message_ids = set()
event_by_message_id = defaultdict(list)
for event in events:
if event.type in self.PRELOAD_MESSAGE_EVENTS:
message_ids.add(event.message_id)
event_by_message_id[event.message_id].append(event)
if not message_ids:
return
messages_data = self.vk.method(
'messages.getById',
{'message_ids': message_ids}
)
for message in messages_data['items']:
for event in event_by_message_id[message['id']]:
event.message_data = message
def listen(self):
""" Слушать сервер
:yields: :class:`Event`
"""
while True:
for event in self.check():
yield event

261
vk_api/requests_pool.py
Unescape View File

@ -0,0 +1,261 @@
# -*- coding: utf-8 -*-
"""
:authors: python273
:license: Apache License, Version 2.0, see LICENSE file
:copyright: (c) 2019 python273
"""
from collections import namedtuple
from .exceptions import VkRequestsPoolException
from .execute import VkFunction
from .utils import sjson_dumps
PoolRequest = namedtuple('PoolRequest', ['method', 'values', 'result'])
class RequestResult(object):
""" Результат запроса из пула """
__slots__ = ('_result', 'ready', '_error')
def __init__(self):
self._result = None
self.ready = False
self._error = False
@property
def error(self):
"""Ошибка, либо `False`, если запрос прошёл успешно."""
return self._error
@error.setter
def error(self, value):
self._error = value
self.ready = True
@property
def result(self):
"""Результат запроса, если он прошёл успешно."""
if not self.ready:
raise RuntimeError('Result is not available in `with` context')
if self._error:
raise VkRequestsPoolException(
self._error,
'Got error while executing request: [{}] {}'.format(
self.error['error_code'],
self.error['error_msg']
)
)
return self._result
@result.setter
def result(self, result):
self._result = result
self.ready = True
@property
def ok(self):
"""`True`, если результат запроса не содержит ошибок, иначе `False`"""
return self.ready and not self._error
class VkRequestsPool(object):
"""
Позволяет сделать несколько обращений к API за один запрос
за счет метода execute.
Варианты использованя:
- В качестве менеджера контекста: запросы к API добавляются в
открытый пул, и выполняются при его закрытии.
- В качестве объекта пула. запросы к API дабвляются по одному
в пул и выполняются все вместе при выполнении метода execute()
:param vk_session: Объект :class:`VkApi`
"""
__slots__ = ('vk_session', 'pool')
def __init__(self, vk_session):
self.vk_session = vk_session
self.pool = []
def __enter__(self):
return self
def __exit__(self, *args, **kwargs):
self.execute()
def method(self, method, values=None):
""" Добавляет запрос в пул.
Возвращаемое значение будет содержать результат после закрытия пула.
:param method: метод
:type method: str
:param values: параметры
:type values: dict
:rtype: RequestResult
"""
if values is None:
values = {}
result = RequestResult()
self.pool.append(PoolRequest(method, values, result))
return result
def execute(self):
"""
Выполняет все находящиеся в пуле запросы и отчищает пул.
Необходим для использования пула-объекта.
Для пула менеджера контекста вызывается автоматически.
"""
for i in range(0, len(self.pool), 25):
cur_pool = self.pool[i:i + 25]
one_method = check_one_method(cur_pool)
if one_method:
value_list = [i.values for i in cur_pool]
response_raw = vk_one_method(
self.vk_session, one_method, value_list
)
else:
response_raw = vk_many_methods(self.vk_session, cur_pool)
response = response_raw['response']
response_errors = response_raw.get('execute_errors', [])
response_errors_iter = iter(response_errors)
for x, current_response in enumerate(response):
current_result = cur_pool[x].result
if current_response is not False:
current_result.result = current_response
else:
current_result.error = next(response_errors_iter)
self.pool = []
def check_one_method(pool):
""" Возвращает True, если все запросы в пуле к одному методу """
if not pool:
return False
first_method = pool[0].method
if all(req.method == first_method for req in pool[1:]):
return first_method
return False
vk_one_method = VkFunction(
args=('method', 'values'),
clean_args=('method',),
return_raw=True,
code='''
var values = %(values)s,
i = 0,
result = [];
while(i < values.length) {
result.push(API.%(method)s(values[i]));
i = i + 1;
}
return result;
''')
def vk_many_methods(vk_session, pool):
requests = ','.join(
'API.{}({})'.format(i.method, sjson_dumps(i.values))
for i in pool
)
code = 'return [{}];'.format(requests)
return vk_session.method('execute', {'code': code}, raw=True)
def vk_request_one_param_pool(vk_session, method, key, values,
default_values=None):
""" Использовать, если изменяется значение только одного параметра.
Возвращаемое значение содержит tuple из dict с результатами и
dict с ошибками при выполнении
:param vk_session: объект VkApi
:type vk_session: vk_api.VkAPi
:param method: метод
:type method: str
:param default_values: одинаковые значения для запросов
:type default_values: dict
:param key: ключ изменяющегося параметра
:type key: str
:param values: список значений изменяющегося параметра (max: 25)
:type values: list
:rtype: (dict, dict)
"""
result = {}
errors = {}
if default_values is None:
default_values = {}
for i in range(0, len(values), 25):
current_values = values[i:i + 25]
response_raw = vk_one_param(
vk_session, method, current_values, default_values, key
)
response = response_raw['response']
response_errors = response_raw.get('execute_errors', [])
response_errors_iter = iter(response_errors)
for x, r in enumerate(response):
if r is not False:
result[current_values[x]] = r
else:
errors[current_values[x]] = next(response_errors_iter)
return result, errors
vk_one_param = VkFunction(
args=('method', 'values', 'default_values', 'key'),
clean_args=('method', 'key'),
return_raw=True,
code='''
var def_values = %(default_values)s,
values = %(values)s,
result = [],
i = 0;
while(i < values.length) {
def_values.%(key)s = values[i];
result.push(API.%(method)s(def_values));
i = i + 1;
}
return result;
''')

135
vk_api/streaming.py
Unescape View File

@ -0,0 +1,135 @@
# -*- coding: utf-8 -*-
"""
:authors: python273, hdk5
:license: Apache License, Version 2.0, see LICENSE file
:copyright: (c) 2019 python273
"""
from .exceptions import VkApiError
import websocket
import json
class VkStreaming(object):
""" Класс для работы с Streaming API
`Подробнее в документации VK API <https://vk.com/dev/streaming_api_docs>`__.
:param vk: объект :class:`VkApi`
"""
__slots__ = ('vk', 'url', 'key', 'server')
URL_TEMPLATE = '{schema}://{server}/{method}?key={key}'
def __init__(self, vk):
self.vk = vk
self.url = None
self.key = None
self.server = None
self.update_streaming_server()
def update_streaming_server(self):
response = self.vk.method('streaming.getServerUrl')
self.key = response['key']
self.server = response['endpoint']
def get_rules(self):
""" Получить список добавленных правил """
response = self.vk.http.get(self.URL_TEMPLATE.format(
schema='https',
server=self.server,
method='rules',
key=self.key)
).json()
if response['code'] == 200:
return response['rules'] or []
elif response['code'] == 400:
raise VkStreamingError(response['error'])
def add_rule(self, value, tag):
""" Добавить правило
:param value: Строковое представление правила
:type value: str
:param tag: Тег правила
:type tag: str
"""
response = self.vk.http.post(self.URL_TEMPLATE.format(
schema='https',
server=self.server,
method='rules',
key=self.key),
json={'rule': {'value': value, 'tag': tag}}
).json()
if response['code'] == 200:
return True
elif response['code'] == 400:
raise VkStreamingError(response['error'])
def delete_rule(self, tag):
""" Удалить правило
:param tag: Тег правила
:type tag: str
"""
response = self.vk.http.delete(self.URL_TEMPLATE.format(
schema='https',
server=self.server,
method='rules',
key=self.key),
json={'tag': tag}
).json()
if response['code'] == 200:
return True
elif response['code'] == 400:
raise VkStreamingError(response['error'])
def delete_all_rules(self):
for item in self.get_rules():
self.delete_rule(item['tag'])
def listen(self):
""" Слушать сервер """
ws = websocket.create_connection(self.URL_TEMPLATE.format(
schema='wss',
server=self.server,
method='stream',
key=self.key
))
while True:
response = json.loads(ws.recv())
if response['code'] == 100:
yield response['event']
elif response['code'] == 300:
raise VkStreamingServiceMessage(response['service_message'])
class VkStreamingError(VkApiError):
def __init__(self, error):
self.error_code = error['error_code']
self.message = error['message']
def __str__(self):
return '[{}] {}'.format(self.error_code, self.message)
class VkStreamingServiceMessage(VkApiError):
def __init__(self, error):
self.service_code = error['service_code']
self.message = error['message']
def __str__(self):
return '[{}] {}'.format(self.service_code, self.message)

254
vk_api/tools.py
Unescape View File

@ -0,0 +1,254 @@
# -*- coding: utf-8 -*-
"""
:authors: python273
:license: Apache License, Version 2.0, see LICENSE file
:copyright: (c) 2019 python273
"""
from .exceptions import ApiError, VkToolsException
from .execute import VkFunction
class VkTools(object):
""" Содержит некоторые вспомогательные функции, которые могут понадобиться
при использовании API
:param vk: Объект :class:`VkApi`
"""
__slots__ = ('vk',)
def __init__(self, vk):
self.vk = vk
def get_all_iter(self, method, max_count, values=None, key='items',
limit=None, stop_fn=None, negative_offset=False):
""" Получить все элементы.
Работает в методах, где в ответе есть count и items или users.
За один запрос получает max_count * 25 элементов
:param method: имя метода
:type method: str
:param max_count: максимальное количество элементов, которое можно
получить за один запрос
:type max_count: int
:param values: параметры
:type values: dict
:param key: ключ элементов, которые нужно получить
:type key: str
:param limit: ограничение на количество получаемых элементов,
но может прийти больше
:type limit: int
:param stop_fn: функция, отвечающая за выход из цикла
:type stop_fn: func
:param negative_offset: True если offset должен быть отрицательный
:type negative_offset: bool
"""
values = values.copy() if values else {}
values['count'] = max_count
offset = max_count if negative_offset else 0
items_count = 0
count = None
while True:
response = vk_get_all_items(
self.vk, method, key, values, count, offset,
offset_mul=-1 if negative_offset else 1
)
if 'execute_errors' in response:
raise VkToolsException(
'Could not load items: {}'.format(
response['execute_errors']
),
response=response
)
response = response['response']
items = response["items"]
items_count += len(items)
for item in items:
yield item
if not response['more']:
break
if limit and items_count >= limit:
break
if stop_fn and stop_fn(items):
break
count = response['count']
offset = response['offset']
def get_all(self, method, max_count, values=None, key='items', limit=None,
stop_fn=None, negative_offset=False):
""" Использовать только если нужно загрузить все объекты в память.
Eсли вы можете обрабатывать объекты по частям, то лучше
использовать get_all_iter
Например если вы записываете объекты в БД, то нет смысла загружать
все данные в память
"""
items = list(
self.get_all_iter(
method, max_count, values, key, limit, stop_fn, negative_offset
)
)
return {'count': len(items), key: items}
def get_all_slow_iter(self, method, max_count, values=None, key='items',
limit=None, stop_fn=None, negative_offset=False):
""" Получить все элементы (без использования execute)
Работает в методах, где в ответе есть count и items или users
:param method: имя метода
:type method: str
:param max_count: максимальное количество элементов, которое можно
получить за один запрос
:type max_count: int
:param values: параметры
:type values: dict
:param key: ключ элементов, которые нужно получить
:type key: str
:param limit: ограничение на количество получаемых элементов,
но может прийти больше
:type limit: int
:param stop_fn: функция, отвечающая за выход из цикла
:type stop_fn: func
:param negative_offset: True если offset должен быть отрицательный
:type negative_offset: bool
"""
values = values.copy() if values else {}
values['count'] = max_count
offset_mul = -1 if negative_offset else 1
offset = max_count if negative_offset else 0
count = None
items_count = 0
while count is None or offset < count:
values['offset'] = offset * offset_mul
response = self.vk.method(method, values)
new_count = response['count']
count_diff = (new_count - count) if count is not None else 0
if count_diff < 0:
offset += count_diff
count = new_count
continue
response_items = response[key]
items = response_items[count_diff:]
items_count += len(items)
for item in items:
yield item
if len(response_items) < max_count - count_diff:
break
if limit and items_count >= limit:
break
if stop_fn and stop_fn(items):
break
offset += max_count
count = new_count
def get_all_slow(self, method, max_count, values=None, key='items',
limit=None, stop_fn=None, negative_offset=False):
""" Использовать только если нужно загрузить все объекты в память.
Eсли вы можете обрабатывать объекты по частям, то лучше
использовать get_all_slow_iter
Например если вы записываете объекты в БД, то нет смысла загружать
все данные в память
"""
items = list(
self.get_all_slow_iter(
method, max_count, values, key, limit, stop_fn, negative_offset
)
)
return {'count': len(items), key: items}
vk_get_all_items = VkFunction(
args=('method', 'key', 'values', 'count', 'offset', 'offset_mul'),
clean_args=('method', 'key', 'offset', 'offset_mul'),
return_raw=True,
code='''
var params = %(values)s,
calls = 0,
items = [],
count = %(count)s,
offset = %(offset)s,
ri;
while(calls < 25) {
calls = calls + 1;
params.offset = offset * %(offset_mul)s;
var response = API.%(method)s(params),
new_count = response.count,
count_diff = (count == null ? 0 : new_count - count);
if (!response) {
return {"_error": 1};
}
if (count_diff < 0) {
offset = offset + count_diff;
} else {
ri = response.%(key)s;
items = items + ri.slice(count_diff);
offset = offset + params.count + count_diff;
if (ri.length < params.count) {
calls = 99;
}
}
count = new_count;
if (count != null && offset >= count) {
calls = 99;
}
};
return {
count: count,
items: items,
offset: offset,
more: calls != 99
};
''')

618
vk_api/upload.py
Unescape View File

@ -0,0 +1,618 @@
# -*- coding: utf-8 -*-
"""
:authors: python273
:license: Apache License, Version 2.0, see LICENSE file
:copyright: (c) 2019 python273
"""
import requests
from .vk_api import VkApi, VkApiMethod
STORY_ALLOWED_LINK_TEXTS = {
'to_store', 'vote', 'more', 'book', 'order',
'enroll', 'fill', 'signup', 'buy', 'ticket',
'write', 'open', 'learn_more', 'view', 'go_to',
'contact', 'watch', 'play', 'install', 'read'
}
class VkUpload(object):
""" Загрузка файлов через API (https://vk.com/dev/upload_files)
:param vk: объект :class:`VkApi` или :class:`VkApiMethod`
"""
__slots__ = ('vk', 'http')
def __init__(self, vk):
if not isinstance(vk, (VkApi, VkApiMethod)):
raise TypeError(
'The arg should be VkApi or VkApiMethod instance'
)
if isinstance(vk, VkApiMethod):
self.vk = vk
else:
self.vk = vk.get_api()
self.http = requests.Session()
self.http.headers.pop('user-agent')
def photo(self, photos, album_id,
latitude=None, longitude=None, caption=None, description=None,
group_id=None):
""" Загрузка изображений в альбом пользователя
:param photos: путь к изображению(ям) или file-like объект(ы)
:type photos: str or list
: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']
with FilesOpener(photos) as photo_files:
response = self.http.post(url, files=photo_files).json()
if 'album_id' not in response:
response['album_id'] = response['aid']
response.update({
'latitude': latitude,
'longitude': longitude,
'caption': caption,
'description': description
})
values.update(response)
return self.vk.photos.save(**values)
def photo_messages(self, photos, peer_id=None):
""" Загрузка изображений в сообщения
:param photos: путь к изображению(ям) или file-like объект(ы)
:type photos: str or list
:param peer_id: peer_id беседы
:type peer_id: int
"""
url = self.vk.photos.getMessagesUploadServer(peer_id=peer_id)['upload_url']
with FilesOpener(photos) as photo_files:
response = self.http.post(url, files=photo_files)
return self.vk.photos.saveMessagesPhoto(**response.json())
def photo_group_widget(self, photo, image_type):
""" Загрузка изображений в коллекцию сообщества для виджетов приложений сообществ
:param photo: путь к изображению или file-like объект
:type photo: str
:param image_type: тип изображиения в зависимости от выбранного виджета
(https://vk.com/dev/appWidgets.getGroupImageUploadServer)
:type image_type: str
"""
url = self.vk.appWidgets.getGroupImageUploadServer(image_type=image_type)['upload_url']
with FilesOpener(photo, key_format='file') as photo_files:
response = self.http.post(url, files=photo_files)
return self.vk.appWidgets.saveGroupImage(**response.json())
def photo_profile(self, photo, owner_id=None, crop_x=None, crop_y=None,
crop_width=None):
""" Загрузка изображения профиля
:param photo: путь к изображению или file-like объект
:param owner_id: идентификатор сообщества или текущего пользователя.
По умолчанию загрузка идет в профиль текущего пользователя.
При отрицательном значении загрузка идет в группу.
:param crop_x: координата X верхнего правого угла миниатюры.
:param crop_y: координата Y верхнего правого угла миниатюры.
:param crop_width: сторона квадрата миниатюры.
При передаче всех crop_* для фотографии также будет
подготовлена квадратная миниатюра.
"""
values = {}
if owner_id:
values['owner_id'] = owner_id
crop_params = {}
if crop_x is not None and crop_y is not None and crop_width is not None:
crop_params['_square_crop'] = '{},{},{}'.format(
crop_x, crop_y, crop_width
)
response = self.vk.photos.getOwnerPhotoUploadServer(**values)
url = response['upload_url']
with FilesOpener(photo, key_format='file') as photo_files:
response = self.http.post(
url,
data=crop_params,
files=photo_files
)
return self.vk.photos.saveOwnerPhoto(**response.json())
def photo_chat(self, photo, chat_id):
""" Загрузка и смена обложки в беседе
:param photo: путь к изображению или file-like объект
:param chat_id: ID беседы
"""
values = {'chat_id': chat_id}
url = self.vk.photos.getChatUploadServer(**values)['upload_url']
with FilesOpener(photo, key_format='file') as photo_file:
response = self.http.post(url, files=photo_file)
return self.vk.messages.setChatPhoto(
file=response.json()['response']
)
def photo_wall(self, photos, user_id=None, group_id=None, caption=None):
""" Загрузка изображений на стену пользователя или в группу
:param photos: путь к изображению(ям) или file-like объект(ы)
:type photos: str or list
:param user_id: идентификатор пользователя
:param group_id: идентификатор сообщества (если загрузка идет в группу)
:param caption: текст описания фотографии.
"""
values = {}
if user_id:
values['user_id'] = user_id
elif group_id:
values['group_id'] = group_id
if caption:
values['caption'] = caption
response = self.vk.photos.getWallUploadServer(**values)
url = response['upload_url']
with FilesOpener(photos) as photos_files:
response = self.http.post(url, files=photos_files)
values.update(response.json())
return self.vk.photos.saveWallPhoto(**values)
def photo_market(self, photo, group_id, main_photo=False,
crop_x=None, crop_y=None, crop_width=None):
""" Загрузка изображений для товаров в магазине
:param photo: путь к изображению(ям) или file-like объект(ы)
:type photo: str or list
:param group_id: идентификатор сообщества, для которого необходимо загрузить фотографию товара
:type group_id: int
:param main_photo: является ли фотография обложкой товара
:type main_photo: bool
:param crop_x: координата x для обрезки фотографии (верхний правый угол)
:type crop_x: int
:param crop_y: координата y для обрезки фотографии (верхний правый угол)
:type crop_y: int
:param crop_width: ширина фотографии после обрезки в px
:type crop_width: int
"""
if group_id < 0:
group_id = abs(group_id)
values = {
'main_photo': main_photo,
'group_id': group_id,
}
if crop_x is not None:
values['crop_x'] = crop_x
if crop_y is not None:
values['crop_y'] = crop_y
if crop_width is not None:
values['crop_width'] = crop_width
response = self.vk.photos.getMarketUploadServer(**values)
url = response['upload_url']
with FilesOpener(photo) as photos_files:
response = self.http.post(url, files=photos_files)
values.update(response.json())
return self.vk.photos.saveMarketPhoto(**values)
def photo_market_album(self, photo, group_id):
""" Загрузка фотографии для подборки товаров
:param photo: путь к изображению(ям) или file-like объект(ы)
:type photo: str or list
:param group_id: идентификатор сообщества, для которого необходимо загрузить фотографию для подборки товаров
:type group_id: int
"""
if group_id < 0:
group_id = abs(group_id)
values = {
'group_id': group_id,
}
response = self.vk.photos.getMarketAlbumUploadServer(**values)
url = response['upload_url']
with FilesOpener(photo) as photos_files:
response = self.http.post(url, files=photos_files)
values.update(response.json())
return self.vk.photos.saveMarketAlbumPhoto(**values)
def audio(self, audio, artist, title):
""" Загрузка аудио
:param audio: путь к аудиофайлу или file-like объект
:param artist: исполнитель
:param title: название
"""
url = self.vk.audio.getUploadServer()['upload_url']
with FilesOpener(audio, key_format='file') as f:
response = self.http.post(url, files=f).json()
response.update({
'artist': artist,
'title': title
})
return self.vk.audio.save(**response)
def video(self, video_file=None, link=None, name=None, description=None,
is_private=None, wallpost=None, group_id=None,
album_id=None, privacy_view=None, privacy_comment=None,
no_comments=None, repeat=None):
""" Загрузка видео
:param video_file: путь к видеофайлу или file-like объект.
:type video_file: object or str
:param link: url для встраивания видео с внешнего сайта,
например, с Youtube.
:type link: str
:param name: название видеофайла
:type name: str
:param description: описание видеофайла
:type description: str
:param is_private: указывается 1, если видео загружается для отправки
личным сообщением. После загрузки с этим параметром видеозапись
не будет отображаться в списке видеозаписей пользователя и не будет
доступна другим пользователям по ее идентификатору.
:type is_private: bool
:param wallpost: требуется ли после сохранения опубликовать
запись с видео на стене.
:type wallpost: bool
:param group_id: идентификатор сообщества, в которое будет сохранен
видеофайл. По умолчанию файл сохраняется на страницу текущего
пользователя.
:type group_id: int
:param album_id: идентификатор альбома, в который будет загружен
видеофайл.
:type album_id: int
:param privacy_view: настройки приватности просмотра видеозаписи в
специальном формате. (https://vk.com/dev/objects/privacy)
Приватность доступна для видеозаписей, которые пользователь
загрузил в профиль. (список слов, разделенных через запятую)
:param privacy_comment: настройки приватности комментирования
видеозаписи в специальном формате.
(https://vk.com/dev/objects/privacy)
:param no_comments: 1 — закрыть комментарии (для видео из сообществ).
:type no_comments: bool
:param repeat: зацикливание воспроизведения видеозаписи. Флаг.
:type repeat: bool
"""
if not link and not video_file:
raise ValueError('Either link or video_file param is required')
if link and video_file:
raise ValueError('Both params link and video_file aren\'t allowed')
values = {
'name': name,
'description': description,
'is_private': is_private,
'wallpost': wallpost,
'link': link,
'group_id': group_id,
'album_id': album_id,
'privacy_view': privacy_view,
'privacy_comment': privacy_comment,
'no_comments': no_comments,
'repeat': repeat
}
response = self.vk.video.save(**values)
url = response.pop('upload_url')
with FilesOpener(video_file or [], 'video_file') as f:
response.update(self.http.post(
url,
files=f or None
).json())
return response
def document(self, doc, title=None, tags=None, group_id=None,
to_wall=False, message_peer_id=None, doc_type=None):
""" Загрузка документа
:param doc: путь к документу или file-like объект
:param title: название документа
:param tags: метки для поиска
:param group_id: идентификатор сообщества (если загрузка идет в группу)
"""
values = {
'group_id': group_id,
'peer_id': message_peer_id,
'type': doc_type
}
if to_wall:
method = self.vk.docs.getWallUploadServer
elif message_peer_id:
method = self.vk.docs.getMessagesUploadServer
else:
method = self.vk.docs.getUploadServer
url = method(**values)['upload_url']
with FilesOpener(doc, 'file') as files:
response = self.http.post(url, files=files).json()
response.update({
'title': title,
'tags': tags
})
return self.vk.docs.save(**response)
def document_wall(self, doc, title=None, tags=None, group_id=None):
""" Загрузка документа в папку Отправленные,
для последующей отправки документа на стену
или личным сообщением.
:param doc: путь к документу или file-like объект
:param title: название документа
:param tags: метки для поиска
:param group_id: идентификатор сообщества (если загрузка идет в группу)
"""
return self.document(doc, title, tags, group_id, to_wall=True)
def document_message(self, doc, title=None, tags=None, peer_id=None):
""" Загрузка документа для отправки личным сообщением.
:param doc: путь к документу или file-like объект
:param title: название документа
:param tags: метки для поиска
:param peer_id: peer_id беседы
"""
return self.document(doc, title, tags, message_peer_id=peer_id)
def audio_message(self, audio, peer_id=None, group_id=None):
""" Загрузка аудио-сообщения.
:param audio: путь к аудиофайлу или file-like объект
:param peer_id: идентификатор диалога
:param group_id: для токена группы, можно передавать ID группы,
вместо peer_id
"""
return self.document(
audio,
doc_type='audio_message',
message_peer_id=peer_id,
group_id=group_id,
to_wall=group_id is not None
)
def graffiti(self, image, peer_id=None, group_id=None):
""" Загрузка граффити
:param image: путь к png изображению или file-like объект.
:param peer_id: идентификатор диалога (только для авторизации пользователя)
:param group_id: для токена группы, нужно передавать ID группы,
вместо peer_id
"""
return self.document(
image,
doc_type='graffiti',
message_peer_id=peer_id,
group_id=group_id,
to_wall=group_id is not None
)
def photo_cover(self, photo, group_id,
crop_x=None, crop_y=None,
crop_x2=None, crop_y2=None):
""" Загрузка изображения профиля
:param photo: путь к изображению или file-like объект
:param group_id: идентификатор сообщества
:param crop_x: координата X верхнего левого угла для обрезки изображения
:param crop_y: координата Y верхнего левого угла для обрезки изображения
:param crop_x2: коорд. X нижнего правого угла для обрезки изображения
:param crop_y2: коорд. Y нижнего правого угла для обрезки изображения
"""
values = {
'group_id': group_id,
'crop_x': crop_x,
'crop_y': crop_y,
'crop_x2': crop_x2,
'crop_y2': crop_y2
}
url = self.vk.photos.getOwnerCoverPhotoUploadServer(**values)['upload_url']
with FilesOpener(photo, key_format='file') as photo_files:
response = self.http.post(url, files=photo_files)
return self.vk.photos.saveOwnerCoverPhoto(
**response.json()
)
def story(self, file, file_type, add_to_news=True, user_ids=None,
reply_to_story=None, link_text=None,
link_url=None, group_id=None):
""" Загрузка истории
:param file: путь к изображению, гифке или видео или file-like объект
:param file_type: тип истории (photo или video)
:param add_to_news: размещать ли историю в новостях
:param user_ids: идентификаторы пользователей,
которые будут видеть историю
:param reply_to_story: идентификатор истории,
в ответ на которую создается новая
:param link_text: текст ссылки для перехода из истории
:param link_url: адрес ссылки для перехода из истории
:param group_id: идентификатор сообщества,
в которое должна быть загружена история
"""
if user_ids is None:
user_ids = []
if file_type == 'photo':
method = self.vk.stories.getPhotoUploadServer
elif file_type == 'video':
method = self.vk.stories.getVideoUploadServer
else:
raise ValueError('type should be either photo or video')
if not add_to_news and not user_ids:
raise ValueError(
'add_to_news and/or user_ids param is required'
)
if (link_text or link_url) and not group_id:
raise ValueError('Link params available only for communities')
if (not link_text) != (not link_url):
raise ValueError(
'Either both link_text and link_url or neither one are required'
)
if link_text and link_text not in STORY_ALLOWED_LINK_TEXTS:
raise ValueError('Invalid link_text')
if link_url and not link_url.startswith('https://vk.com'):
raise ValueError(
'Only internal https://vk.com links are allowed for link_url'
)
if link_url and len(link_url) > 2048:
raise ValueError('link_url is too long. Max length - 2048')
values = {
'add_to_news': int(add_to_news),
'user_ids': ','.join(map(str, user_ids)),
'reply_to_story': reply_to_story,
'link_text': link_text,
'link_url': link_url,
'group_id': group_id
}
url = method(**values)['upload_url']
with FilesOpener(file, key_format='file') as files:
return self.http.post(url, files=files)
class FilesOpener(object):
def __init__(self, paths, key_format='file{}'):
if not isinstance(paths, list):
paths = [paths]
self.paths = paths
self.key_format = key_format
self.opened_files = []
def __enter__(self):
return self.open_files()
def __exit__(self, type, value, traceback):
self.close_files()
def open_files(self):
self.close_files()
files = []
for x, file in enumerate(self.paths):
if hasattr(file, 'read'):
f = file
if hasattr(file, 'name'):
filename = file.name
else:
filename = '.jpg'
else:
filename = file
f = open(filename, 'rb')
self.opened_files.append(f)
ext = filename.split('.')[-1]
files.append(
(self.key_format.format(x), ('file{}.{}'.format(x, ext), f))
)
return files
def close_files(self):
for f in self.opened_files:
f.close()
self.opened_files = []

166
vk_api/utils.py
Unescape View File

@ -0,0 +1,166 @@
# -*- coding: utf-8 -*-
"""
:authors: python273
:license: Apache License, Version 2.0, see LICENSE file
:copyright: (c) 2019 python273
"""
from __future__ import print_function
import random
try:
import simplejson as json
except ImportError:
import json
from http.cookiejar import Cookie
def search_re(reg, string):
""" Поиск по регулярке """
s = reg.search(string)
if s:
groups = s.groups()
return groups[0]
def clear_string(s):
if s:
return s.strip().replace('&nbsp;', '')
def get_random_id():
""" Get random int32 number (signed) """
return random.getrandbits(31) * random.choice([-1, 1])
def code_from_number(prefix, postfix, number):
prefix_len = len(prefix)
postfix_len = len(postfix)
if number[0] == '+':
number = number[1:]
if (prefix_len + postfix_len) >= len(number):
return
# Сравниваем начало номера
if number[:prefix_len] != prefix:
return
# Сравниваем конец номера
if number[-postfix_len:] != postfix:
return
return number[prefix_len:-postfix_len]
def sjson_dumps(*args, **kwargs):
kwargs['ensure_ascii'] = False
kwargs['separators'] = (',', ':')
return json.dumps(*args, **kwargs)
HTTP_COOKIE_ARGS = [
'version', 'name', 'value',
'port', 'port_specified',
'domain', 'domain_specified',
'domain_initial_dot',
'path', 'path_specified',
'secure', 'expires', 'discard', 'comment', 'comment_url', 'rest', 'rfc2109'
]
def cookie_to_dict(cookie):
cookie_dict = {
k: v for k, v in cookie.__dict__.items() if k in HTTP_COOKIE_ARGS
}
cookie_dict['rest'] = cookie._rest
cookie_dict['expires'] = None
return cookie_dict
def cookie_from_dict(d):
return Cookie(**d)
def cookies_to_list(cookies):
return [cookie_to_dict(cookie) for cookie in cookies]
def set_cookies_from_list(cookie_jar, l):
for cookie in l:
cookie_jar.set_cookie(cookie_from_dict(cookie))
def enable_debug_mode(vk_session, print_content=False):
""" Включает режим отладки:
- Вывод сообщений лога
- Вывод http запросов
:param vk_session: объект VkApi
:param print_content: печатать ответ http запросов
"""
import logging
import sys
import time
import requests
from . import __version__
pypi_version = requests.get(
'https://pypi.org/pypi/vk_api/json'
).json()['info']['version']
if __version__ != pypi_version:
print()
print('######### MODULE IS NOT UPDATED!!1 ##########')
print()
print('Installed vk_api version is:', __version__)
print('PyPI vk_api version is:', pypi_version)
print()
print('######### MODULE IS NOT UPDATED!!1 ##########')
print()
class DebugHTTPAdapter(requests.adapters.HTTPAdapter):
def send(self, request, **kwargs):
start = time.time()
response = super(DebugHTTPAdapter, self).send(request, **kwargs)
end = time.time()
total = end - start
body = request.body
if body and len(body) > 1024:
body = body[:1024] + '[STRIPPED]'
print(
'{:0.2f} {} {} {} {} {} {}'.format(
total,
request.method,
request.url,
request.headers,
repr(body),
response.status_code,
response.history
)
)
if print_content:
print(response.text)
return response
vk_session.http.mount('http://', DebugHTTPAdapter())
vk_session.http.mount('https://', DebugHTTPAdapter())
vk_session.logger.setLevel(logging.INFO)
vk_session.logger.addHandler(logging.StreamHandler(sys.stdout))

737
vk_api/vk_api.py
Unescape View File

@ -0,0 +1,737 @@
# -*- coding: utf-8 -*-
"""
:authors: python273
:license: Apache License, Version 2.0, see LICENSE file
:copyright: (c) 2019 python273
"""
import json
import logging
import random
import re
import threading
import time
import urllib.parse
from hashlib import md5
import requests
import jconfig
from .enums import VkUserPermissions
from .exceptions import *
from .utils import (
code_from_number, search_re, clear_string,
cookies_to_list, set_cookies_from_list
)
RE_LOGIN_TO = re.compile(r'"to":"(.*?)"')
RE_LOGIN_IP_H = re.compile(r'name="ip_h" value="([a-z0-9]+)"')
RE_LOGIN_LG_H = re.compile(r'name="lg_h" value="([a-z0-9]+)"')
RE_LOGIN_LG_DOMAIN_H = re.compile(r'name="lg_domain_h" value="([a-z0-9]+)"')
RE_CAPTCHAID = re.compile(r"onLoginCaptcha\('(\d+)'")
RE_NUMBER_HASH = re.compile(r"al_page: '3', hash: '([a-z0-9]+)'")
RE_AUTH_HASH = re.compile(r"Authcheck\.init\('([a-z_0-9]+)'")
RE_TOKEN_URL = re.compile(r'location\.href = "(.*?)"\+addr;')
RE_PHONE_PREFIX = re.compile(r'label ta_r">\+(.*?)<')
RE_PHONE_POSTFIX = re.compile(r'phone_postfix">.*?(\d+).*?<')
DEFAULT_USERAGENT = 'Mozilla/5.0 (Windows NT 10.0; rv:91.0) Gecko/20100101 Firefox/91.0'
DEFAULT_USER_SCOPE = sum(VkUserPermissions)
def get_unknown_exc_str(s):
return (
f'Unknown error ({s}). Please send a bugreport to GitHub: '
'https://github.com/python273/vk_api/issues'
)
class VkApi(object):
"""
:param login: Логин ВКонтакте (лучше использовать номер телефона для
автоматического обхода проверки безопасности)
:type login: str
:param password: Пароль ВКонтакте (если пароль не передан, то будет
попытка использовать сохраненные данные)
:type password: str
:param token: access_token
:type token: str
:param auth_handler: Функция для обработки двухфакторной аутентификации,
должна возвращать строку с кодом и
булево значение, означающее, стоит ли запомнить
это устройство, для прохождения аутентификации.
:param captcha_handler: Функция для обработки капчи, см. :func:`captcha_handler`
:param config: Класс для сохранения настроек
:type config: :class:`jconfig.base.BaseConfig`
:param config_filename: Расположение config файла для :class:`jconfig.config.Config`
:param api_version: Версия API
:type api_version: str
:param app_id: app_id Standalone-приложения
:type app_id: int
:param scope: Запрашиваемые права, можно передать строкой или числом.
См. :class:`VkUserPermissions`
:type scope: int or str
:param client_secret: Защищенный ключ приложения для Client Credentials Flow
авторизации приложения (https://vk.com/dev/client_cred_flow).
Внимание: Этот способ авторизации устарел, рекомендуется использовать
сервисный ключ из настроек приложения.
`login` и `password` необходимы для автоматического получения токена при помощи
Implicit Flow авторизации пользователя и возможности работы с веб-версией сайта
(включая :class:`vk_api.audio.VkAudio`)
:param session: Кастомная сессия со своими параметрами(из библиотеки requests)
:type session: :class:`requests.Session`
"""
RPS_DELAY = 0.34 # ~3 requests per second
def __init__(self, login=None, password=None, token=None,
auth_handler=None, captcha_handler=None,
config=jconfig.Config, config_filename='vk_config.v2.json',
api_version='5.92', app_id=6222115, scope=DEFAULT_USER_SCOPE,
client_secret=None, session=None):
self.login = login
self.password = password
self.token = {'access_token': token}
self.api_version = api_version
self.app_id = app_id
self.scope = scope
self.client_secret = client_secret
self.storage = config(self.login, filename=config_filename)
self.http = session or requests.Session()
if not session:
self.http.headers['User-agent'] = DEFAULT_USERAGENT
self.last_request = 0.0
self.error_handlers = {
NEED_VALIDATION_CODE: self.need_validation_handler,
CAPTCHA_ERROR_CODE: captcha_handler or self.captcha_handler,
TOO_MANY_RPS_CODE: self.too_many_rps_handler,
TWOFACTOR_CODE: auth_handler or self.auth_handler
}
self.lock = threading.Lock()
self.logger = logging.getLogger('vk_api')
@property
def _sid(self):
return (
self.http.cookies.get('remixsid', domain='.vk.com') or
self.http.cookies.get('remixsid6', domain='.vk.com') or
self.http.cookies.get('remixsid', domain='.vk.ru') or
self.http.cookies.get('remixsid6', domain='.vk.ru')
)
def auth(self, reauth=False, token_only=False):
""" Аутентификация
:param reauth: Позволяет переавторизоваться, игнорируя сохраненные
куки и токен
:param token_only: Включает оптимальную стратегию аутентификации, если
необходим только access_token
Например если сохраненные куки не валидны,
но токен валиден, то аутентификация пройдет успешно
При token_only=False, сначала проверяется
валидность куки. Если кука не будет валидна, то
будет произведена попытка аутетификации с паролем.
Тогда если пароль не верен или пароль не передан,
то аутентификация закончится с ошибкой.
Если вы не делаете запросы к веб версии сайта
используя куки, то лучше использовать
token_only=True
"""
if not self.login:
raise LoginRequired('Login is required to auth')
self.logger.info('Auth with login: {}'.format(self.login))
set_cookies_from_list(
self.http.cookies,
self.storage.setdefault('cookies', [])
)
self.token = self.storage.setdefault(
'token', {}
).setdefault(
'app' + str(self.app_id), {}
).get('scope_' + str(self.scope))
if token_only:
self._auth_token(reauth=reauth)
else:
self._auth_cookies(reauth=reauth)
def _auth_cookies(self, reauth=False):
if reauth:
self.logger.info('Auth forced')
self.storage.clear_section()
self._vk_login()
self._api_login()
return
if not self.check_sid():
self.logger.info(
'remixsid from config is not valid: {}'.format(
self._sid
)
)
self._vk_login()
else:
self._pass_security_check()
if not self._check_token():
self.logger.info(
'access_token from config is not valid: {}'.format(
self.token
)
)
self._api_login()
else:
self.logger.info('access_token from config is valid')
def _auth_token(self, reauth=False):
if not reauth and self._check_token():
self.logger.info('access_token from config is valid')
return
if reauth:
self.logger.info('Auth (API) forced')
if self.check_sid():
self._pass_security_check()
self._api_login()
elif self.password:
self._vk_login()
self._api_login()
def _vk_login(self, captcha_sid=None, captcha_key=None):
""" Авторизация ВКонтакте с получением cookies remixsid
:param captcha_sid: id капчи
:type captcha_key: int or str
:param captcha_key: ответ капчи
:type captcha_key: str
"""
self.logger.info('Logging in...')
if not self.password:
raise PasswordRequired('Password is required to login')
self.http.cookies.clear()
# Get cookies
response = self.http.get('https://vk.com/login')
if response.url.startswith('https://vk.com/429.html?'):
hash429_md5 = md5(self.http.cookies['hash429'].encode('ascii')).hexdigest()
self.http.cookies.pop('hash429')
response = self.http.get(f'{response.url}&key={hash429_md5}')
headers = {
'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8',
'Accept-Language': 'en-US,en;q=0.5',
'Referer': 'https://vk.com/',
'Content-Type': 'application/x-www-form-urlencoded',
'Origin': 'https://vk.com',
}
values = {
'act': 'login',
'role': 'al_frame',
'expire': '',
'to': search_re(RE_LOGIN_TO, response.text),
'recaptcha': '',
'captcha_sid': '',
'captcha_key': '',
'_origin': 'https://vk.com',
'utf8': '1',
'ip_h': search_re(RE_LOGIN_IP_H, response.text),
'lg_h': search_re(RE_LOGIN_LG_H, response.text),
'lg_domain_h': search_re(RE_LOGIN_LG_DOMAIN_H, response.text),
'ul': '',
'email': self.login,
'pass': self.password
}
if captcha_sid and captcha_key:
self.logger.info(
'Using captcha code: {}: {}'.format(
captcha_sid,
captcha_key
)
)
values['captcha_sid'] = captcha_sid
values['captcha_key'] = captcha_key
response = self.http.post(
'https://login.vk.com/?act=login',
data=values,
headers=headers
)
if 'onLoginCaptcha(' in response.text:
self.logger.info('Captcha code is required')
captcha_sid = search_re(RE_CAPTCHAID, response.text)
captcha = Captcha(self, captcha_sid, self._vk_login)
return self.error_handlers[CAPTCHA_ERROR_CODE](captcha)
if 'onLoginReCaptcha(' in response.text:
self.logger.info('Captcha code is required (recaptcha)')
captcha_sid = str(random.random())[2:16]
captcha = Captcha(self, captcha_sid, self._vk_login)
return self.error_handlers[CAPTCHA_ERROR_CODE](captcha)
if 'onLoginFailed(4' in response.text:
raise BadPassword('Bad password')
if 'act=authcheck' in response.text:
self.logger.info('2FA is required')
response = self.http.get('https://vk.com/login?act=authcheck')
self._pass_twofactor(response)
if self._sid:
self.logger.info('Got remixsid')
self.storage.cookies = cookies_to_list(self.http.cookies)
self.storage.save()
else:
raise AuthError(get_unknown_exc_str('AUTH; no sid'))
response = self._pass_security_check(response)
if 'act=blocked' in response.url:
raise AccountBlocked('Account is blocked')
def _pass_twofactor(self, auth_response):
""" Двухфакторная аутентификация
:param auth_response: страница с приглашением к аутентификации
"""
auth_hash = search_re(RE_AUTH_HASH, auth_response.text)
if not auth_hash:
raise TwoFactorError(get_unknown_exc_str('2FA; no hash'))
code, remember_device = self.error_handlers[TWOFACTOR_CODE]()
values = {
'al': '1',
'code': code,
'hash': auth_hash,
'remember': int(remember_device),
}
response = self.http.post(
'https://vk.com/al_login.php?act=a_authcheck_code',
values
)
data = json.loads(response.text.lstrip('<!--'))
status = data['payload'][0]
if status == '4': # OK
path = json.loads(data['payload'][1][0])
return self.http.get(path)
elif status in [0, '8']: # Incorrect code
return self._pass_twofactor(auth_response)
elif status == '2':
raise TwoFactorError('Recaptcha required')
raise TwoFactorError(get_unknown_exc_str('2FA; unknown status'))
def _pass_security_check(self, response=None):
""" Функция для обхода проверки безопасности (запрос номера телефона)
:param response: ответ предыдущего запроса, если есть
"""
self.logger.info('Checking security check request')
if response is None:
response = self.http.get('https://vk.com/settings')
if 'security_check' not in response.url:
self.logger.info('Security check is not required')
return response
phone_prefix = clear_string(search_re(RE_PHONE_PREFIX, response.text))
phone_postfix = clear_string(
search_re(RE_PHONE_POSTFIX, response.text))
code = None
if self.login and phone_prefix and phone_postfix:
code = code_from_number(phone_prefix, phone_postfix, self.login)
if code:
number_hash = search_re(RE_NUMBER_HASH, response.text)
values = {
'act': 'security_check',
'al': '1',
'al_page': '3',
'code': code,
'hash': number_hash,
'to': ''
}
response = self.http.post('https://vk.com/login.php', values)
if response.text.split('<!>')[4] == '4':
return response
if phone_prefix and phone_postfix:
raise SecurityCheck(phone_prefix, phone_postfix)
raise SecurityCheck(response=response)
def check_sid(self):
""" Проверка Cookies remixsid на валидность """
self.logger.info('Checking remixsid...')
if not self._sid:
self.logger.info('No remixsid')
return
response = self.http.get('https://vk.com/feed2.php').json()
if response['user']['id'] != -1:
self.logger.info('remixsid is valid')
return response
self.logger.info('remixsid is not valid')
def _api_login(self):
""" Получение токена через Desktop приложение """
if not self._sid:
raise AuthError('API auth error (no remixsid)')
if not self.http.cookies.get('p', domain='.login.vk.com'):
raise AuthError('API auth error (no login cookies)')
response = self.http.get(
'https://oauth.vk.com/authorize',
params={
'client_id': self.app_id,
'scope': self.scope,
'response_type': 'token'
}
)
if 'act=blocked' in response.url:
raise AccountBlocked('Account is blocked')
if 'access_token' not in response.url:
url = search_re(RE_TOKEN_URL, response.text)
if url:
response = self.http.get(url)
if 'access_token' in response.url:
parsed_url = urllib.parse.urlparse(response.url)
parsed_query = urllib.parse.parse_qs(parsed_url.query)
if 'authorize_url' in parsed_query:
url = parsed_query['authorize_url'][0]
if url.startswith('https%3A'): # double-encoded
url = urllib.parse.unquote(url)
parsed_url = urllib.parse.urlparse(url)
parsed_query = urllib.parse.parse_qs(parsed_url.fragment)
token = {k: v[0] for k, v in parsed_query.items()}
if not isinstance(token.get('access_token'), str):
raise AuthError(get_unknown_exc_str('API AUTH; no access_token'))
self.token = token
self.storage.setdefault(
'token', {}
).setdefault(
'app' + str(self.app_id), {}
)['scope_' + str(self.scope)] = token
self.storage.save()
self.logger.info('Got access_token')
elif 'oauth.vk.com/error' in response.url:
error_data = response.json()
error_text = error_data.get('error_description')
# Deletes confusing error text
if error_text and '@vk.com' in error_text:
error_text = error_data.get('error')
raise AuthError('API auth error: {}'.format(error_text))
else:
raise AuthError('Unknown API auth error')
def server_auth(self):
""" Серверная авторизация """
values = {
'client_id': self.app_id,
'client_secret': self.client_secret,
'v': self.api_version,
'grant_type': 'client_credentials'
}
response = self.http.post(
'https://oauth.vk.com/access_token', values
).json()
if 'error' in response:
raise AuthError(response['error_description'])
else:
self.token = response
def code_auth(self, code, redirect_url):
""" Получение access_token из code """
values = {
'client_id': self.app_id,
'client_secret': self.client_secret,
'v': self.api_version,
'redirect_uri': redirect_url,
'code': code,
}
response = self.http.post(
'https://oauth.vk.com/access_token', values
).json()
if 'error' in response:
raise AuthError(response['error_description'])
else:
self.token = response
return response
def _check_token(self):
""" Проверка access_token юзера на валидность """
if self.token:
try:
self.method('stats.trackVisitor')
except ApiError:
return False
return True
def captcha_handler(self, captcha):
""" Обработчик капчи (http://vk.com/dev/captcha_error)
:param captcha: объект исключения `Captcha`
"""
raise captcha
def need_validation_handler(self, error):
""" Обработчик проверки безопасности при запросе API
(http://vk.com/dev/need_validation)
:param error: исключение
"""
pass # TODO: write me
def http_handler(self, error):
""" Обработчик ошибок соединения
:param error: исключение
"""
pass
def too_many_rps_handler(self, error):
""" Обработчик ошибки "Слишком много запросов в секунду".
Ждет полсекунды и пробует отправить запрос заново
:param error: исключение
"""
self.logger.warning('Too many requests! Sleeping 0.5 sec...')
time.sleep(0.5)
return error.try_method()
def auth_handler(self):
""" Обработчик двухфакторной аутентификации """
raise AuthError('No handler for two-factor authentication')
def get_api(self):
""" Возвращает VkApiMethod(self)
Позволяет обращаться к методам API как к обычным классам.
Например vk.wall.get(...)
"""
return VkApiMethod(self)
def method(self, method, values=None, captcha_sid=None, captcha_key=None,
raw=False):
""" Вызов метода API
:param method: название метода
:type method: str
:param values: параметры
:type values: dict
:param captcha_sid: id капчи
:type captcha_key: int or str
:param captcha_key: ответ капчи
:type captcha_key: str
:param raw: при False возвращает `response['response']`
при True возвращает `response`
(может понадобиться для метода execute для получения
execute_errors)
:type raw: bool
"""
values = values.copy() if values else {}
if 'v' not in values:
values['v'] = self.api_version
if self.token:
values['access_token'] = self.token['access_token']
if captcha_sid and captcha_key:
values['captcha_sid'] = captcha_sid
values['captcha_key'] = captcha_key
with self.lock:
# Ограничение 3 запроса в секунду
delay = self.RPS_DELAY - (time.time() - self.last_request)
if delay > 0:
time.sleep(delay)
response = self.http.post(
'https://api.vk.com/method/' + method,
values,
headers={'Cookie': ''}
)
self.last_request = time.time()
if response.ok:
response = response.json()
else:
error = ApiHttpError(self, method, values, raw, response)
response = self.http_handler(error)
if response is not None:
return response
raise error
if 'error' in response:
error = ApiError(self, method, values, raw, response['error'])
if error.code in self.error_handlers:
if error.code == CAPTCHA_ERROR_CODE:
error = Captcha(
self,
error.error['captcha_sid'],
self.method,
(method,),
{'values': values, 'raw': raw},
error.error['captcha_img']
)
response = self.error_handlers[error.code](error)
if response is not None:
return response
raise error
return response if raw else response['response']
class VkApiGroup(VkApi):
"""Предназначен для авторизации с токеном группы.
Увеличивает частоту обращений к API с 3 до 20 в секунду.
"""
RPS_DELAY = 1 / 20.0
class VkApiMethod(object):
""" Дает возможность обращаться к методам API через:
>>> vk = VkApiMethod(...)
>>> vk.wall.getById(posts='...')
или
>>> vk.wall.get_by_id(posts='...')
"""
__slots__ = ('_vk', '_method')
def __init__(self, vk, method=None):
self._vk = vk
self._method = method
def __getattr__(self, method):
if '_' in method:
m = method.split('_')
method = m[0] + ''.join(i.title() for i in m[1:])
return VkApiMethod(
self._vk,
(self._method + '.' if self._method else '') + method
)
def __call__(self, **kwargs):
for k, v in kwargs.items():
if isinstance(v, (list, tuple)):
kwargs[k] = ','.join(str(x) for x in v)
return self._vk.method(self._method, kwargs)

55
vk_bot.py
Unescape View File

@ -0,0 +1,55 @@
# -*- coding: utf-8 -*-
import vk_api
from requests import get
from os import getenv
from random import randint
from config import headers
from vk_api.bot_longpoll import VkBotLongPoll, VkBotEventType
from dotenv import load_dotenv
from pathlib import Path
dotenv_path = f"{Path(__file__).parent.resolve()}/.env"
load_dotenv(dotenv_path=dotenv_path)
vk_session = vk_api.VkApi(token=getenv("vk_token"))
vk_session._auth_token()
vk_api = vk_session.get_api()
longpoll = VkBotLongPoll(vk_session, getenv("vk_group_id"))
commands = ["/mrp", "/mrp_online", "/pomoidor", "/mordor", "/мордор", "/мрп", "/мрп_онлайн", "/помойдор"]
def get_online():
try:
response = get("https://l.mordor-rp.com/launcher/monitoring/online.php", headers=headers).json()
rponl = 0; obsonl = 0; funonl = 0; text = ""
for element in response:
text += f'[{element["name"]}]: {element["min"]}\n'
obsonl += int(element["min"])
if element["tag"] == "roleplay": rponl += int(element["min"])
if element["tag"] == "fun": funonl += int(element["min"])
text += f"========================\n" \
f"ROLEPLAY ONLINE: {rponl}\n" \
f"FUN ONLINE: {funonl}\n" \
f"========================\n" \
f"FULL ONLINE: {obsonl}"
return(text)
except:
return("Ошибка.")
def send(pid, msg):
rid = randint(1, 2147483647)
vk_api.messages.send(peer_id=pid, message=msg, random_id=rid)
if __name__ == "__main__":
while True:
try:
for event in longpoll.listen():
if event.type == VkBotEventType.MESSAGE_NEW:
peer_id = event.object.message["peer_id"]
text = event.object.message["text"]
if text.lower() in commands:
send(peer_id, f"{get_online()}")
elif text.lower() in ["/help", "/хелп", "/помощь"]:
send(peer_id, "Команды:\nПолучение онлайна Mordor RP:\n- /mrp\n- /mrp_online\n- /pomoidor\n- /mordor\n- /мрп\n- /мрп_онлайн\n- /помойдор\n- /мордор")
except Exception:
pass
Write Preview
Loading…
Cancel
Save