import vk_api
from vk_api.longpoll import VkLongPoll, VkEventType
import json
vk_session = vk_api.VkApi(token='token', api_version=5.131)
api = vk_session.get_api()
longpoll = VkLongPoll(vk_session, preload_messages=True)
print("Бот запущен")
while True:
try:
for event in longpoll.listen():
if event.type != VkEventType.MESSAGE_NEW and event.type != VkEventType.MESSAGE_EDIT: continue
message = event.message_data
# if not message['out']: continue
# Если нужно, чтобы бот реагировал только на сообщения от хозяина страницы, нужно убрать комментирование на строке выше
args = message['text'].split(" ")
cmd = args[0].lower()
if cmd == '!проверка':
# Если нужно, чтобы бот редактировал отправленное сообщение: (я буду использовать именно этот способ)
api.messages.edit(peer_id=message['peer_id'], message_id=message['id'], message='Проверка успешна!', keep_forward_messages=1)
# Если нужно, чтобы бот отвечал новым сообщением:
# api.messages.send(peer_id=message['peer_id'], message='Проверка успешна!', random_id=0, forward=json.dumps({'peer_id': message['peer_id'], 'conversation_message_ids': message['conversation_message_id'], 'is_reply': 1}))
elif cmd == '!напиши':
api.messages.edit(peer_id=message['peer_id'], message_id=message['id'], message=message['text'].replace(cmd+' ', ''), keep_forward_messages=1)
elif cmd == '!параметрысбщ':
api.messages.edit(peer_id=message['peer_id'], message_id=message['id'], message='У сообщения такие параметры:\n'+'\n'.join([str(value)+': '+str(message[value]) for value in message]), keep_forward_messages=1)
except Exception as e:
print(e)
Шаблон страничного бота ВК
Шаблон бота для сообщества ВК
import vk_api
from vk_api.bot_longpoll import VkBotLongPoll, VkBotEventType
import json
vk_session = vk_api.VkApi(token='token', api_version=5.131)
api = vk_session.get_api()
longpoll = VkBotLongPoll(vk_session, group_id=12345) # тут айди группы без минуса
print("Бот запущен")
while True:
try:
for event in longpoll.listen():
if event.type != VkBotEventType.MESSAGE_NEW: continue
message = event.object.message
args = message['text'].split(" ")
cmd = args[0].lower()
if cmd == '!проверка':
api.messages.send(peer_id=message['peer_id'], message='Проверка успешна!', random_id=0, forward=json.dumps({'peer_id': message['peer_id'], 'conversation_message_ids': message['conversation_message_id'], 'is_reply': 1}))
elif cmd == '!напиши':
api.messages.send(peer_id=message['peer_id'], message=message['text'].replace(cmd+' ', ''), random_id=0, forward=json.dumps({'peer_id': message['peer_id'], 'conversation_message_ids': message['conversation_message_id'], 'is_reply': 1}))
elif cmd == '!параметрысбщ':
api.messages.send(peer_id=message['peer_id'], message='У сообщения такие параметры:\n'+'\n'.join([str(value)+': '+str(message[value]) for value in message]), random_id=0, forward=json.dumps({'peer_id': message['peer_id'], 'conversation_message_ids': message['conversation_message_id'], 'is_reply': 1}))
except Exception as e:
print(e)
Шаблон бота Telegram на Telethon
from telethon import TelegramClient, events
telegram = TelegramClient('session', 'api_id', 'api_hash').start()
@telegram.on(events.NewMessage(pattern='(!проверка)'))
async def handler_proverka(message):
await message.reply('Проверка успешна!')
@telegram.on(events.NewMessage(pattern='(!напиши )'))
async def handler_napishi(message):
await message.reply(message.text.replace('!напиши ', ''))
@telegram.on(events.NewMessage(pattern='(!параметрысбщ)'))
async def handler_parametri(message):
await message.reply('У сообщения такие параметры:\n'+'\n'.join([str(value)+': '+str(vars(message)[value]) for value in vars(message) if not value.startswith('_')]))
print('Бот запущен')
telegram.run_until_disconnected()
Шаблон бота Telegram на Aiogram
import logging
from asyncio import run
from aiogram import Bot, Dispatcher
from aiogram.filters import CommandStart, Command
from aiogram.types import Message
from aiogram.client.default import DefaultBotProperties
from aiogram.enums import ParseMode
bot = Bot("TOKEN", default = DefaultBotProperties(parse_mode = ParseMode.HTML))
dp = Dispatcher()
@dp.message(CommandStart())
async def cmd_start(message: Message):
await message.answer("Привет!")
@dp.message(Command("help"))
async def cmd_help(message: Message):
await message.answer("Помощь")
@dp.message()
async def echo(message: Message):
await message.answer(message.text)
async def main() -> None:
await bot.delete_webhook(True)
await dp.start_polling(bot)
logging.basicConfig(level=logging.INFO)
run(main())
Gemini пересказывает все голосовые сообщения, которые вам приходят в Telegram
Возможность подключить этого бота к аккаунту есть только у Premium пользователей! Но использовать его, пересылая ГС боту можно и без Premium.

Это скриншоты в моём ЛС с пользователями:

Бота можно хостить только на нероссийских серверах: API Gemini нельзя использовать с российских IP!
Для установки:
- Заполняем конфиг.
gemini_token — API ключ Gemini. Получать тут: https://aistudio.google.com/app/apikey
- Смените IP-адрес на нероссийский
- Create API Key
- В окне с полем поиска нажмите кнопку выше поля. Скопируйте ключ.
bot_token — токен бота Telegram. https://t.me/botfather

allowed_users — список [ID пользователей, которые смогут привязать бота к бизнесу, либо использовать его в ЛС с ботом] и [ID чатов, в котором бот будет автоматически распознавать все ГС]. Для определения ID используйте https://t.me/myidbot.

- Включите Business Mode в BotFather

- Настройки Telegram arrow_right_alt Telegram для бизнеса arrow_right_alt Чат-боты. Вставляем ссылку на своего бота.

- Всё, ждём голосовых сообщений!
gemini_token = 'https://lolz.live/threads/6818835/'
bot_token = 'https://t.me/botfather'
allowed_users = [657195904, -1001537487920]
# Больше ничего редактировать не нужно
import asyncio
from aiogram import Router, Bot, Dispatcher, F, types
import logging
from io import BytesIO
from pydub import AudioSegment
import google.generativeai as gemini
from google.api_core.exceptions import InternalServerError
from google.api_core.exceptions import ResourceExhausted
from ssl import SSLError
from os import remove
from asyncio import sleep
bot = None
router = Router(name=__name__)
lock = asyncio.Lock()
logger = logging.getLogger(__name__)
logging.basicConfig(level=logging.INFO)
gemini.configure(api_key=gemini_token)
async def telegram_voice_gemini_tts(file_id: str, format: str, message_id: int) -> str:
wav_io = BytesIO()
AudioSegment.from_file((await bot.download_file((await bot.get_file(file_id)).file_path)), format=format).export(wav_io, format="wav")
wav_io.seek(0)
with open(f'vm{message_id}.wav', 'wb') as f: f.write(wav_io.getvalue())
audio_file = gemini.upload_file(f'vm{message_id}.wav', mime_type='audio/wav')
try:
return gemini.GenerativeModel(model_name="gemini-1.5-flash").generate_content([f'Перескажи голосовое сообщение на русском языке. Сообщи об эмоциях автора. В случае, если ты услышал речь, начни свой ответ со слов "В {"голосовом сообщении" if format == "ogg" else "видеосообщении"} говорится". В случае, если ты не услышал в записи речь, сообщи о том, что в {"голосовом сообщении" if format == "ogg" else "видеосообщении"} нет речи.', audio_file], safety_settings=[{"category": "HARM_CATEGORY_HARASSMENT","threshold": "BLOCK_NONE",},{"category": "HARM_CATEGORY_HATE_SPEECH","threshold": "BLOCK_NONE",},{"category": "HARM_CATEGORY_SEXUALLY_EXPLICIT","threshold": "BLOCK_NONE",},{"category": "HARM_CATEGORY_DANGEROUS_CONTENT","threshold": "BLOCK_NONE",},]).text
finally:
remove(f'vm{message_id}.wav')
@router.business_message(F.voice | F.video_note)
async def handle_message(message: types.Message):
async with lock:
if not (await bot.get_business_connection(message.business_connection_id)).user.id in allowed_users: return
logger.info(f"Received voice message from {message.from_user.id}")
recognition_message = await bot.send_message(message.chat.id, '<tg-emoji emoji-id="5296262629558340179">🖥</tg-emoji>Gemini распознаёт голосовое сообщение<tg-emoji emoji-id="5220046725493828505">✍️</tg-emoji>' if message.voice else '<tg-emoji emoji-id="5296262629558340179">🖥</tg-emoji>Gemini распознаёт видеосообщение<tg-emoji emoji-id="5220070652756635426">👀</tg-emoji>', business_connection_id=message.business_connection_id, reply_to_message_id=message.message_id, parse_mode='HTML')
await bot.send_chat_action(message.chat.id, 'upload_voice' if message.voice else 'upload_video_note', business_connection_id=message.business_connection_id)
while True:
try:
tts_result = await telegram_voice_gemini_tts(message.voice.file_id if message.voice else message.video_note.file_id, 'ogg' if message.voice else 'mp4', message.message_id)
return await bot.edit_message_text(chat_id=recognition_message.chat.id, message_id=recognition_message.message_id, text='<tg-emoji emoji-id="5296262629558340179">🖥</tg-emoji><b>Gemini</b>: ' + tts_result, business_connection_id=message.business_connection_id, parse_mode='HTML')
except ValueError:
return await bot.edit_message_text(chat_id=recognition_message.chat.id, message_id=recognition_message.message_id, text='<tg-emoji emoji-id="5807626765874499116">🚫</tg-emoji><tg-emoji emoji-id="5296262629558340179">🖥</tg-emoji>Gemini посчитал этот запрос небезопасным.', business_connection_id=message.business_connection_id, parse_mode='HTML')
except (InternalServerError, SSLError):
sleep(3)
except ResourceExhausted:
return await bot.edit_message_text(chat_id=recognition_message.chat.id, message_id=recognition_message.message_id, text='<tg-emoji emoji-id="5456670136121434320">😔</tg-emoji><tg-emoji emoji-id="5296262629558340179">🖥</tg-emoji>Достигнут лимит запросов к Gemini :(', business_connection_id=message.business_connection_id, parse_mode='HTML')
except Exception as e:
logger.error(e)
await bot.send_message(allowed_users[0], str(e))
# @router.message()
# async def echo(message: types.Message):
# print(message.html_text)
@router.message(F.chat.type == 'private', F.text)
async def handle_message(message: types.Message):
async with lock:
if not message.from_user.id in allowed_users: return
await bot.send_chat_action(message.chat.id, 'typing')
response = gemini.GenerativeModel(model_name="gemini-1.5-pro").generate_content(message.text)
await message.reply(response.text)
@router.message(F.voice | F.video_note)
async def handle_message(message: types.Message):
async with lock:
if not message.chat.id in allowed_users: return
logger.info(f"Received voice message from {message.from_user.id}")
recognition_message = await message.reply('Gemini распознаёт голосовое сообщение...' if message.voice else 'Gemini распознаёт видеосообщение...')
await bot.send_chat_action(message.chat.id, 'upload_voice' if message.voice else 'upload_video_note')
while True:
try:
tts_result = await telegram_voice_gemini_tts(message.voice.file_id if message.voice else message.video_note.file_id, 'ogg' if message.voice else 'mp4', message.message_id)
return await bot.edit_message_text(chat_id=recognition_message.chat.id, message_id=recognition_message.message_id, text='<b>Gemini</b>: ' + tts_result, parse_mode='HTML')
except ValueError:
return await bot.edit_message_text(chat_id=recognition_message.chat.id, message_id=recognition_message.message_id, text='🚫Gemini посчитал этот запрос небезопасным.', parse_mode='HTML')
except (InternalServerError, SSLError):
sleep(3)
except ResourceExhausted:
return await bot.edit_message_text(chat_id=recognition_message.chat.id, message_id=recognition_message.message_id, text='😔Достигнут лимит запросов к Gemini :(')
except Exception as e:
logger.error(e)
await bot.send_message(allowed_users[0], str(e))
sleep(10)
async def main():
global bot
bot = Bot(token=bot_token)
dp = Dispatcher()
dp.include_router(router)
await bot.delete_webhook(drop_pending_updates=True)
await dp.start_polling(bot)
asyncio.run(main())
Простейший вечный онлайн ВК
from urllib.request import urlopen
from time import sleep
tokens = ['token1', 'token2']
while True:
for token in tokens: print(urlopen(f'https://api.vk.com/method/account.setOnline?voip=1&access_token={token}&v=5.131').read().decode())
sleep(120)
Скачивание всей музыки ВК с обложкой и с текстом MusixMatch/Genius
Заполняем конфиг:
- vk_user_agent - User Agent, который будет использоваться при скачивании. Лучше не трогать и оставить как есть.
vk_katemobile_token - токен от ВК от приложения Kate Mobile.
Получить токенopen_in_new(нужно скопировать строку от
access_token=до&expires_in)- genius_token - токен от сайта Genius, получать тут: https://genius.com/api-clients (сначала надо создать аккаунт)
- owner_id - с чьего аккаунта скачивать музыку (по ID профиля)
Если скрипт не сможет найти обложку или текст песни, потребуется вставить их в теги mp3 самому. Это можно сделать через AIMP Tag Editor или Mp3tag.from os import getcwd, name, system, remove
from os.path import isfile
from io import BytesIO
from re import sub
from subprocess import call
if name != 'nt': exit('Скрипт работает неправильно на Linux! Пожалуйста, используйте Windows!')
system('title VKMusic Downloader')
system('cls')
print('Внимание! ВК не позволяет загружать некоторые треки с нероссийским айпи! В то же время, MusixMatch заблокирован РКНом. Примите меры.')
system('pause')
config = {
"vk_user_agent": "KateMobileAndroid/92.2 v1-524 (Android 11; SDK 30; x86; Pixel sdk_gphone_x86_arm; ru)",
"vk_katemobile_token": "",
"genius_token": "",
"owner_id": 1234567
}
try:
from requests import Session, get
from mutagen.mp3 import MP3
from mutagen.id3 import ID3, TT2, TPE1, APIC, USLT
from bs4 import BeautifulSoup
from PIL import Image
except:
system('pip install requests mutagen bs4 pillow')
system('python "'+__file__+'"')
exit()
s, mm = Session(), Session()
s.headers.update({'User-Agent': config['vk_user_agent']})
mm.headers.update({'User-Agent': "Mozilla/5.0 (Windows NT 10.0; Win64; x64)"})
response = s.get(f'https://api.vk.com/method/audio.get?owner_id={config["owner_id"]}&count=6000&access_token={config["vk_katemobile_token"]}&v=5.131').json()['response']
print(f'Будет загружено {response["count"]} треков в папку {getcwd()}')
for song in response['items']:
if isfile(f'{song["artist"]} — {song["title"]}.mp3'):
print(f'Трек {song["artist"]} — {song["title"]}.mp3 уже загружен, пропускаю')
continue
if song['url'] == '':
print(f'Внимание! Нельзя скачать трек {song["artist"]} — {song["title"]}! Скачайте вручную!')
system('pause')
continue
print(f'Скачиваю трек {song["artist"]} — {song["title"]}')
with open(f'{song["artist"]} — {song["title"]}.mp3', 'wb') as file:
file.write(s.get(song["url"]).content)
track = MP3(f'{song["artist"]} — {song["title"]}.mp3', ID3=ID3)
track.delete()
track.tags.add(TT2(encoding=3, text=song['title']))
track.tags.add(TPE1(encoding=3, text=song['artist']))
track.save()
# Получаю текст песни с MusixMatch
lyrics = False
cover_set = False
musixmatch = BeautifulSoup(mm.get(f'https://www.musixmatch.com/search/{song["artist"]} - {song["title"]}/tracks'.replace(' ', '%20')).text, "html.parser")
try:
musixmatch = musixmatch.find_all('ul', {'class': "tracks list thumb-list"})[0]
musixmatch = BeautifulSoup(mm.get("https://www.musixmatch.com"+musixmatch.find_all('a', {'class': 'title'})[0]['href']).text, "html.parser")
if cols := musixmatch.find_all('span', {'class': "lyrics__content__ok"}, string=True):
lyrics = "\n".join(x.text for x in cols)
elif data := musixmatch.find_all('span', {'class': "lyrics__content__warning"}, string=True):
lyrics = "\n".join(x.text for x in data)
elif musixmatch.find_all('h2', {'class': 'mxm-empty__title'})[0].text == 'Instrumental':
print('Инструментальная песня, нет текста')
else:
raise Exception('Трек не найден на MusixMatch')
if lyrics:
track.tags.add(USLT(encoding=3, lang=u'eng', desc='', text=lyrics))
track.save()
else:
lyrics = True
except (IndexError, Exception):
track.save()
print(f'Не удалось найти текст песни {song["artist"]} — {song["title"]}! Ищу на Genius!')
# Получаю обложку трека с MusixMatch
if lyrics:
try:
if musixmatch.find_all('div', {'class': "banner-album-image-desktop"})[0].find_all('img')[0]['src'] == '//s.mxmcdn.net/site/images/albums/nocover_new-350x350.png':
raise Exception('Стандартная обложка не подойдёт')
Image.open(BytesIO(get("https:"+musixmatch.find_all('div', {'class': "banner-album-image-desktop"})[0].find_all('img')[0]['src']).content)).convert('RGB').save('temp_cover.jpg')
track.tags.add(APIC(encoding=0, mime='image/jpg', type=3, desc='', data=open('temp_cover.jpg', 'rb').read()))
track.save()
cover_set = True
continue
except:
print(f'Не удалось найти обложку {song["artist"]} — {song["title"]} на MusixMatch. Ищу на Genius!')
search = get(f'https://api.genius.com/search/?q={song["artist"]} - {song["title"]}&access_token={config["genius_token"]}').json()
try:
if not lyrics:
if not search['response']['hits'][0]['type'] == 'song':
raise Exception('Не удалось найти песню на Genius')
lyrics = sub(r'\[[^\]]*\]', '', "\n".join(x.get_text(separator='\n') for x in BeautifulSoup(get('https://genius.com' + search['response']['hits'][0]['result']['path']).text, "html.parser").find_all('div', {'data-lyrics-container': "true"}))).lstrip('\n')
temp_file = f'Верный ли это текст песни для {song["artist"]} - {song["title"]}?Смотрите в консоль.txt'
with open(temp_file, 'w', encoding='utf-8') as file:
file.write(lyrics)
print(f'Верный ли это текст песни для {song["artist"]} - {song["title"]}? Если да - закройте блокнот. Если нет, измените и сохраните, затем закройте блокнот.')
call(['notepad.exe', temp_file])
system('pause')
with open(temp_file, 'r', encoding='utf-8') as file:
track.tags.add(USLT(encoding=3, lang=u'eng', desc='', text=file.read()))
track.save()
remove(temp_file)
if not cover_set:
if 'default_cover_image.png' in search['response']['hits'][0]['result']['song_art_image_url']:
raise Exception('Стандартная обложка не подойдёт')
Image.open(BytesIO(get(search['response']['hits'][0]['result']['song_art_image_url']).content)).convert('RGB').save('temp_cover.jpg')
track.tags.add(APIC(encoding=0, mime='image/jpg', type=3, desc='', data=open('temp_cover.jpg', 'rb').read()))
track.save()
except Exception as e:
if not lyrics:
print(f'Не удалось найти текст песни {song["artist"]} — {song["title"]}! Установите текст вручную!')
system('pause')
if not cover_set:
print(f'Не удалось найти обложку {song["artist"]} — {song["title"]}! Установите обложку вручную!')
system('pause')
remove('temp_cover.jpg')
Автостатус ВК и автобио Telegram с текущей песней Spotify
Нужно заполнить конфиг, затем, после запуска, войти в Spotify в браузере и войти в Telegram в консоли.
Конфиг:
- vk-default-status - Статус в ВК, когда в Spotify ничего не играет
- telegram-default-status - О себе в Telegram, когда в Spotify ничего не играет
- status-update-timeout - Количество секунд, спустя которое нужно обновить статус
- spotify-client-id - Client ID приложения Spotify
- spotify-client-secret - Client secret приложения Spotify
- spotify-redirect-uri - redirect uri приложения Spotify
- spotify-username - Часть ссылки на ваш профиль Spotify
- vk-token - Токен ВК
- vk-online-when-listen-spotify - Нужно ли делать вас онлайн в ВК, когда включен Spotify (True/False)
- use-telegram - Нужно ли обновлять био в телеграм (поставьте False, если хотите автостатус только в ВК)
- telegram-api-id - API ID приложения Telegram
- telegram-api-hash - API hash приложения Telegram
Где взять параметры, связанные с Spotify:
- Заходим на https://developer.spotify.com/dashboard/, авторизуемся.
- Создаем приложение , вводим любое имя и описание, ставим галочку
- Нажимаем на show client secret
- Два значения уже получили. Теперь нажимаем edit settings
- Вводим
http://localhost:8888/callbackв redirect url. (в конфиге тоже) - Для получения ссылки на свой профиль заходим на https://open.spotify.com/
- Копируем значение после
user/. Это spotify-username.
Где взять параметры, связанные с Telegram:
- Заходим на https://my.telegram.org/, авторизуемся
- Создаем приложение. Вот и данные.
from spotipy.oauth2 import SpotifyOAuth
import spotipy
import vk_api
from time import sleep
config = {
'vk-default-status': "Статус в ВК, когда в Spotify ничего не играет",
'telegram-default-status': 'О себе в Telegram, когда в Spotify ничего не играет',
'status-update-timeout': 60,
'spotify-client-id': "",
'spotify-client-secret': "",
'spotify-redirect-uri': "",
'spotify-username': "as67lwic16uszxqpd6rbfimol",
'vk-token': "token",
'vk-online-when-listen-spotify': True,
'use-telegram': True,
'telegram-api-id': 1,
'telegram-api-hash': ''
}
if config['use-telegram']:
from telethon.tl.functions.account import UpdateProfileRequest
from telethon.sync import TelegramClient
telegram = TelegramClient('telegram', config['telegram-api-id'], config['telegram-api-hash']).start()
spotify = spotipy.Spotify(auth_manager=SpotifyOAuth(scope="user-read-playback-state user-library-read", client_id=config['spotify-client-id'], client_secret=config['spotify-client-secret'], redirect_uri=config['spotify-redirect-uri'], username=config['spotify-username']))
api = vk_api.VkApi(token=config['vk-token']).get_api()
current_status = api.status.get()['text']
def change_status(status):
if not status == current_status:
api.status.set(text=status[:140])
if config['vk-online-when-listen-spotify']: api.account.setOnline(voip=1)
print('Статус в VK обновлен')
if config['use-telegram']:
telegram(UpdateProfileRequest(about=status[:70]))
print('Статус в Telegram обновлен')
print(status)
def default_status():
api.status.set(text=config['vk-default-status'][:140])
print('Статус в VK обновлен')
if config['use-telegram']:
telegram(UpdateProfileRequest(about=config['telegram-default-status'][:70]))
print('Статус в Telegram обновлен')
print('Установлен обычный статус')
def calculate_minutes(ms): return str(int(((ms / (1000*60)) % 60)))
def calculate_seconds(ms): return str(int((ms / 1000) % 60)) if len(str(int((ms / 1000) % 60))) > 1 else '0'+str(int((ms / 1000) % 60))
while True:
current_playing = spotify.current_user_playing_track()
if current_playing is None or current_playing["currently_playing_type"] == "ad":
default_status()
sleep(int(config['status-update-timeout'])*2)
continue
play_emoji = '' if current_playing['is_playing'] == True else ''
change_status(play_emoji+'Слушаю Spotify: '+current_playing["item"]["artists"][0]["name"]+' — '+current_playing["item"]["name"]+' ('+calculate_minutes(current_playing['progress_ms'])+':'+calculate_seconds(current_playing['progress_ms'])+'/'+calculate_minutes(current_playing['item']['duration_ms'])+':'+calculate_seconds(current_playing['item']['duration_ms'])+')')
sleep(int(config['status-update-timeout']))
Папки Личные, Беседы, Боты в ВК Мессенджере
Токен нужно получать именно от приложения VK Мессенджер. Для этого:
- Заходим на web.vk.me
- Нажимаем "Посмотреть код"
- Переходим во вкладку "Сеть" (или "Network")
- Нажимаем "Fetch\XHR"
- Открываем любой диалог слева. Сразу после этого появится несколько запросов, нам нужен любой, который начинается с
messages. - Выбираем раздел "Предварительный просмотр
- Нам нужен параметр "access_token," который, кстати, очень криво отображается. Но это не мешает скопировать его, он начинается с "vk1."
- Подставляем его в переменную "token"
try:
from requests import post
except:
from os import system
system('pip install requests')
system('python '+__file__)
exit()
token = 'token'
personal = []
chats = []
bots = []
offset = 0
while True:
request = post('https://api.vk.me/method/messages.getConversations', headers={'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/108.0.0.0 Safari/537.36'}, params={'v': '5.131'}, data={'count': 200, 'offset': offset, 'access_token': token}).json()
try:
request = request['response']['items']
except:
print(request)
exit()
if request == []: break
for item in request:
if item['conversation']['peer']['type'] == 'user':
personal.append(item['conversation']['peer']['id'])
elif item['conversation']['peer']['type'] == 'chat':
chats.append(item['conversation']['peer']['id'])
elif item['conversation']['peer']['type'] == 'group':
bots.append(item['conversation']['peer']['id'])
offset += 200
print(post('https://api.vk.me/method/messages.createFolder', params={'v': '5.198'}, headers={'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/108.0.0.0 Safari/537.36'},
data={'name': 'Личные', 'included_peer_ids': ','.join(list(map(str, personal[:500]))), 'extended': '1', 'access_token': token}).json())
print(post('https://api.vk.me/method/messages.createFolder', params={'v': '5.198'}, headers={'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/108.0.0.0 Safari/537.36'},
data={'name': 'Беседы', 'included_peer_ids': ','.join(list(map(str, chats[:500]))), 'extended': '1', 'access_token': token}).json())
print(post('https://api.vk.me/method/messages.createFolder', params={'v': '5.198'}, headers={'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/108.0.0.0 Safari/537.36'},
data={'name': 'Боты', 'included_peer_ids': ','.join(list(map(str, bots[:500]))), 'extended': '1', 'access_token': token}).json())
Автоматическое приглашение участника в беседу ВК, как только это возможно
Если в беседе 1000 участников, пригласить нового не получится. Зато этот скрипт пригласит участника сразу же, как только их будет 999.import vk_api
from time import sleep
from random import uniform
chat_id = 123
user_id = 204923959
token = 'token'
api = vk_api.VkApi(token=token).get_api()
while True:
try:
chat = api.messages.getChat(chat_id=chat_id)
print(len(chat['users']))
if len(chat['users']) < 1000:
try:
api.messages.addChatUser(chat_id=chat_id, user_id=chat_id)
exit()
except Exception as e:
print(e)
except Exception as e:
print(e)
sleep(uniform(3.0, 5.0))
Восстановление удаленных сообщений ВК
Запускаем, последние 100 удалённых сообщений восстановятся. Восстанавливаются только сообщения, удалённые не для всех! Восстановить удалённые для всех невозможно! Восстановить возможно только сообщения, удалённые за 24 часа!from requests import get # pip install requests
from time import sleep
token = 'token'
last = get(f'https://api.vk.com/method/messages.send?peer_id={get(f'https://api.vk.com/method/account.getProfileInfo?access_token={token}&v=5.131').json()['response']['id']}&message=Test&random_id=0&access_token={token}&v=5.131').json()['response']
msgs = get(f'https://api.vk.com/method/messages.getById?message_ids={','.join([str(last-i) for i in range(100)])}&access_token={token}&v=5.131').json()['response']['items']
for msg in msgs:
if msg['out'] == 1 and 'deleted' in msg and msg['deleted'] == 1:
print(f'Восстанавливаю сообщение {msg["id"]}:', get(f'https://api.vk.com/method/messages.restore?message_id={msg["id"]}&access_token={token}&v=5.131').json())
sleep(0.5)
Скрипт для чистки от лишних бесед ВК
Почему я не использовал chat_ids, чтобы получить сразу несколько бесед? Потому что существуют фантом-чаты... Допустим, беседы с ID 1, 2, 3 - обычные беседы, с ID 4 - фантом-чат, ID 5 - несуществующая беседа (то есть всего 4 беседы). Если я в chat_ids подставлю 1,4 - будет ошибка. Если 1,2,5 - тоже будет ошибка. Поэтому, после долгих эскперементов, я думаю, что лучше всего получать беседы по одной.import vk_api
token = 'token'
api = vk_api.VkApi(token=token).get_api()
mode = input('Какие беседы показать? 1 - все, 2 - только те, в которых ты состоишь, 3 - только те, где ты не состоишь и не очистил историю, 4 - только те, где ты состоишь и очистил историю: ')
if not int(mode) in [1,2,3,4]: exit('Неверное значение')
print('Получаю беседы...')
for i in range(1, 100000001):
try:
chat = api.messages.getChat(chat_id=i)
except Exception as e:
if str(e) == "[946] Chat not supported": print(str(i)+': Фантом-чат')
elif str(e) == "[100] One of the parameters specified was missing or invalid: chat_id param is incorrect": break
else:
print(e)
break
chat_info = str(chat['id'])+': '+str(chat['title'])+'. '
if 'left' in chat:
chat_info += 'Вышел, '
if mode in ['2', '4']: continue
elif 'kicked' in chat:
chat_info += 'Исключен, '
if mode in ['2', '4']: continue
else:
chat_info += 'В составе, '
if mode == '3': continue
if api.messages.getHistory(peer_id=2000000000+chat['id'])['count'] == 0:
chat_info += 'история очищена.'
if mode == '3': continue
else:
chat_info += 'история не очищена.'
if mode == '4': continue
print(chat_info)
input('Всё!\n')
Чистка друзей ВК: Удаление всех тех, с кем нет общих друзей
import vk_api
token = 'token'
api = vk_api.VkApi(token=token).get_api()
for friend in api.friends.getMutual(target_uids=','.join([str(uid) for uid in api.friends.get()['items']])):
if friend['common_count'] == 0:
friend_name = api.users.get(user_ids=friend['id'])[0]
api.friends.delete(user_id=friend['id'])
print('Удалил друга '+friend_name['first_name']+' '+friend_name['last_name'])
Рандомный статус ВК (автоматически меняется на случайный из списка)
import vk_api
from random import choice
from time import sleep
token = 'token'
statuses = ['Статус', 'Автостатус', 'Рандомный статус', 'Мяу']
api = vk_api.VkApi(token=token).get_api()
while True:
status = choice(statuses)
api.status.set(text=status)
print('Статус изменен на '+status)
sleep(90)
Telegram бот для проведения розыгрышей
- Создаем бота. Пишем BotFather, по стандарту.
- По желанию можно сделать список команд
start - Принять участие в розыгрыше check - Проверить подписку пользователей на каналы (доступно только администраторам) end - Подвести итоги (доступно только администраторам) testend - Тестовые итоги (доступно только администраторам) list - Показать список участников розыгрыша (доступно только администраторам) \ - В token подставляем токен бота
- admins - список айди пользователей, которые могут использовать команды /check, /end, testend, /list. Чтобы узнать свой айди, нужно написать боту IDbot. Чтобы узнать чужой айди, нужно переслать ему сообщение пользователя. Но не у каждого пользователя можно узнать айди, пользователям нужно в настройках приватности сделать вот что:
- channels - список айди каналов, на которые пользователь должен подписаться для участия в розыгрыше. Естественно бота нужно добавить в каждый канал. Узнать айди канала можно через того же IDBota
Всё! Бот запущен, люди могут поучаствовать в розыгрыше, написав боту /start в ЛС. Либо можно добавить бота в группу, и люди смогут участвовать, например, в комментариях к посту в канале
Вы можете время от времени использовать команду /check, чтобы проверить, не отписался ли кто-то из участников от каналов.
Для завершения розыгрыша используйте команду /end
Для просмотра списка участников розыгрыша используйте /list
Для тестового завершения используйте /testend
Автостатус ВК с вашими аудиозаписями
Скрипт получит полный список ваших аудиозаписей в случайном порядке и будет по очереди ставить аудиозаписи в статус. Длительность песни в статусе = длительности песни.import vk_api
from random import shuffle
from time import sleep
token = 'token' # Токен от приложения VK Admin
api = vk_api.VkApi(token=token, api_version=5.131).get_api()
audios = []
while True:
try:
audio = api.audio.get(offset=len(audios))
except: break
if audio['items'] == []: break
else: audios += audio['items']
while True:
shuffle(audios)
for audio in audios:
api.audio.setBroadcast(audio=str(audio['owner_id'])+'_'+str(audio['id']))
sleep(audio['duration'])
Уведомления о том, что пользователь онлайн в ВК
from requests import get
from time import strftime, localtime, sleep
token = 'token'
group_token = 'token'
user_id = 1
sleep_time = 60
# Больше ничего редактировать не нужно
owner_id = get(f'https://api.vk.com/method/account.getProfileInfo?v=5.131&access_token={token}').json()['response']['id']
last_online_time = get(f'https://api.vk.com/method/users.get?fields=online,last_seen&v=5.131&user_ids={user_id}&access_token={token}').json()['response'][0]['last_seen']['time']
last_online = False
while True:
online_info = get(f'https://api.vk.com/method/users.get?fields=online,last_seen&v=5.131&user_ids={user_id}&access_token={token}').json()['response'][0]
if online_info['online'] and not last_online:
get(f'https://api.vk.com/method/messages.send?peer_id={owner_id}&message=[id{user_id}|{online_info["first_name"]} {online_info["last_name"]}] в сети!&random_id=0&v=5.131&access_token={group_token}')
last_online = True
else:
if last_online:
get(f'https://api.vk.com/method/messages.send?peer_id={owner_id}&message=[id{user_id}|{online_info["first_name"]} {online_info["last_name"]}] больше не в сети.&random_id=0&v=5.131&access_token={group_token}')
last_online = False
if online_info['last_seen']['time'] != last_online_time:
get(f'https://api.vk.com/method/messages.send?peer_id={owner_id}&message=[id{user_id}|{online_info["first_name"]} {online_info["last_name"]}] был в сети в {strftime("%H:%M", localtime(online_info["last_seen"]["time"]))}&random_id=0&v=5.131&access_token={group_token}')
last_online_time = online_info['last_seen']['time']
sleep(sleep_time)
Уведомления о событиях в канале Telegram (например, новые подписчики)
Заполняем конфиг:
api_id и api_hash берем здесь: my.telegram.org. Чтобы узнать ID канала, нужно переслать любое сообщение из него боту myidbot
С помощью того же бота можно узнать свой ID, командой /getid, для notifications_chat_id
Токен бота, думаю, знаете, где брать
refresh_rate - раз в сколько секунд бот будет обновлять журнал действий в канале.
Для работы бота требуется войти в аккаунт пользователя-администратора, у которого есть доступ к журналу действий! От имени пользователя-администратора будет выполняться только просмотр журнала активности. Отправка сообщений будет именно от бота.

import json
from os.path import isfile
from time import sleep
from telethon.sync import TelegramClient
config = {
'tg_api_id': '12345',
'tg_api_hash': '3f46s5c7z65875084f39z8d8v89174',
'channel_id': '-1001853994420',
'notifications_chat_id': 657195904,
'bot_token': '12345678:AdfwtKwyBycRyj97JPfd9I',
'refresh_rate': '5'
}
if not isfile('session.session'): print('Войдите от имени администратора канала, у которого есть доступ к журналу событий')
telegram = TelegramClient('session', config['tg_api_id'], config['tg_api_hash']).start()
bot = TelegramClient('bot', config['tg_api_id'], config['tg_api_hash']).start(bot_token=config['bot_token'])
if not isfile('aln_data.malw'):
with open('aln_data.malw', 'w') as file:
file.write('[]')
data = []
else:
with open('aln_data.malw', 'r') as file:
data = json.loads(file.read())
while True:
events = reversed(list(telegram.get_admin_log(entity=int(config['channel_id']), info=False, settings=False, pinned=False, edit=False, delete=False, group_call=False)))
for event in events:
if event.id in data: continue
event2 = event.action.to_dict()
try:
if event2['_'] == 'ChannelAdminLogEventActionParticipantJoinByInvite':
bot.send_message(config['notifications_chat_id'], ''+'['+telegram.get_entity(event.user_id).first_name+'](tg://user?id='+str(event.user_id)+') вступил в канал по ссылке '+event2['invite']['link'], parse_mode='markdown')
elif event2['_'] == 'ChannelAdminLogEventActionParticipantInvite':
bot.send_message(config['notifications_chat_id'], ''+'['+telegram.get_entity(event.user_id).first_name+'](tg://user?id='+str(event.user_id)+') пригласил ['+telegram.get_entity(event2['participant']['user_id']).first_name+'](tg://user?id='+str(event2['participant']['user_id'])+') в канал', parse_mode='markdown')
elif event2['_'] == 'ChannelAdminLogEventActionParticipantJoin':
bot.send_message(config['notifications_chat_id'], ''+'['+telegram.get_entity(event.user_id).first_name+'](tg://user?id='+str(event.user_id)+') подписался на канал!', parse_mode='markdown')
elif event2['_'] == 'ChannelAdminLogEventActionParticipantLeave':
bot.send_message(config['notifications_chat_id'], ''+'['+telegram.get_entity(event.user_id).first_name+'](tg://user?id='+str(event.user_id)+') покинул канал', parse_mode='markdown')
elif event2['_'] == 'ChannelAdminLogEventActionParticipantToggleAdmin':
if event2['prev_participant']['_'] == 'ChannelParticipant' and event2['new_participant']['_'] == 'ChannelParticipantAdmin':
bot.send_message(config['notifications_chat_id'], ''+'['+telegram.get_entity(event2['new_participant']['promoted_by']).first_name+'](tg://user?id='+str(event2['new_participant']['promoted_by'])+') назначил пользователя ['+telegram.get_entity(event2['new_participant']['user_id']).first_name+'](tg://user?id='+str(event2['new_participant']['user_id'])+') администратором')
elif event2['prev_participant']['_'] == 'ChannelParticipantAdmin' and event2['new_participant']['_'] == 'ChannelParticipant':
bot.send_message(config['notifications_chat_id'], ''+'['+telegram.get_entity(event.user_id).first_name+'](tg://user?id='+str(event.user_id)+') разжаловал пользователя ['+telegram.get_entity(event2['new_participant']['user_id']).first_name+'](tg://user?id='+str(event2['new_participant']['user_id'])+'), он больше не администратор')
elif event2['prev_participant']['_'] == 'ChannelParticipantAdmin' and event2['new_participant']['_'] == 'ChannelParticipantAdmin':
bot.send_message(config['notifications_chat_id'], ''+'['+telegram.get_entity(event.user_id).first_name+'](tg://user?id='+str(event.user_id)+') отредактировал права администратора ['+telegram.get_entity(event2['new_participant']['user_id']).first_name+'](tg://user?id='+str(event2['new_participant']['user_id'])+')')
elif event2['_'] == 'ChannelAdminLogEventActionParticipantToggleBan':
if event2['prev_participant']['_'] == 'ChannelParticipantBanned' and event2['new_participant']['_'] == 'ChannelParticipantLeft':
bot.send_message(config['notifications_chat_id'], ''+'['+telegram.get_entity(event.user_id).first_name+'](tg://user?id='+str(event.user_id)+') разблокировал пользователя ['+telegram.get_entity(event2['new_participant']['peer']['user_id']).first_name+'](tg://user?id='+str(event2['new_participant']['peer']['user_id'])+')')
elif event2['prev_participant']['_'] == 'ChannelParticipant' and event2['new_participant']['_'] == 'ChannelParticipantBanned':
bot.send_message(config['notifications_chat_id'], ''+'['+telegram.get_entity(event.user_id).first_name+'](tg://user?id='+str(event.user_id)+') заблокировал пользователя ['+telegram.get_entity(event2['new_participant']['peer']['user_id']).first_name+'](tg://user?id='+str(event2['new_participant']['peer']['user_id'])+')')
data.append(event.id)
except Exception as e:
print(e)
bot.send_message(config['notifications_chat_id'], 'Произошла ошибка :(\nСобытие: '+json.dumps(event2)+'\nОшибка: '+str(e))
with open('aln_data.malw', 'w') as file:
file.write(json.dumps(data))
sleep(int(config['refresh_rate']))
Уведомления о донатах DonationAlerts в Telegram
- Получаем токен оповещений DonationAlerts. Для этого заходим в настройки виджета, возле "Группа 1" нажимаем "Показать ссылку для встраивания" и копируем ее.
В конце этой ссылки будет токен, подставляем его в переменную da_alert_widget_token.
- Создаем бота в Telegram и получаем токен. Для этого используем команду /newbot в боте t.me/BotFather, выбираем имя и адрес боту.
Подставляем токен в переменную tg_bot_token.
- Узнаем ID пользователя или канала, куда вы хотите отправлять сообщения. Для этого ПЕРЕД ДОБАВЛЕНИЕМ БОТА В КАНАЛ берем ссылку
https://api.telegram.org/bot{token}/getUpdates, подставляем туда токен бота вместо{token}, переходим по ней. - Добавляем бота в канал или пишем ему любое сообщение, обновляем страницу и видим там ID пользователя, написавшего сообщение или ID канала, куда бот был приглашен. Тут ID пользователя:
Тут ID канала:
Подставляем ID в переменную
tg_user_id. pip install python-socketio websocket-client requests- Запускаем и ждем донатов :)
import json
import socketio
from requests import get
# pip install python-socketio websocket-client requests
da_alert_widget_token = 'da_token'
tg_bot_token = '12345:tg_token'
tg_user_id = 123457899
sio = socketio.Client(reconnection=True, reconnection_delay=5)
@sio.on('connect')
def on_connect():
sio.emit('add-user', {'token': da_alert_widget_token, "type": "alert_widget"})
print('Бот запущен')
@sio.on('donation')
def on_message(data):
event = json.loads(data)
print(event)
get(f'https://api.telegram.org/bot{tg_bot_token}/sendMessage?chat_id={tg_user_id}&text=Новый донат:\n{event["username"]} - {event["amount"]} {event["currency"]}\n{event["message"]}')
sio.connect('wss://socket.donationalerts.ru:443', transports='websocket')