EDeev/vkrepost_to_tg
1
0
Fork 0
mirror of https://github.com/EDeev/vkrepost_to_tg.git synced 2026-10-07 20:49:50 +03:00
Code Releases 1 Activity

Версия на aiogram 3 — та, что работала на сервере

Browse source 
В репозитории оставался код на aiogram 2. На сервере с апреля–сентября 2024 работала
переписанная версия: aiogram 3, разметка HTML вместо Markdown, код разделён на bot.py /
handlers.py / init.py. Переводы строк приведены к LF.
This commit is contained in:
Деев Егор Викторович 2026-10-05 19:47:59 +00:00
parent c93ffe7c57
commit 490ea8b5d6
8 changed files with 595 additions and 453 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
.gitattributes vendored Normal file
View file

@ -0,0 +1,2 @@
* text=auto eol=lf
*.db binary

375
code/bot.py
View file

@ -1,306 +1,69 @@
import logging, config, asyncio, time, sql, random import asyncio, logging, time, random, scripts, vk_scripts
from aiogram import Bot, Dispatcher, executor, types from init import *
from emoji import emojize from config import *
from handlers import router
from vk_scripts import VkParser
from scripts import login, pars_post
async def timer(wait_for) -> None:
# log level while True:
logging.basicConfig(level=logging.INFO) await asyncio.sleep(wait_for)
# bot init notif_group = list(map(lambda x: x[0], db.all_notifGroup()))
bot = Bot(token=config.botToken) activ_user = list(map(lambda x: [x[0], list(map(int, x[1].split(';')))], db.all_activUser()))
dp = Dispatcher(bot) sub_user = list(map(lambda x: [x[0], list(map(int, x[1].split(';')))], db.all_subUser()))
# connection db for group_id in notif_group:
du = sql.Users('../db/users.db') time.sleep(1)
db = sql.Base('../db/base.db')
tokens = [x[0] for x in activ_user if group_id in x[1]]
users = [x[0] for x in sub_user if group_id in x[1]]
# events
@dp.message_handler(commands=["start", "help"]) if tokens:
async def start(message: types.Message): user_id = tokens[random.randrange(len(tokens))]
login(message.chat.id, du, db) vk = vk_scripts.VkParser(db.get_token(user_id))
else: vk = vk_scripts.VkParser(serviceToken)
buttons = [types.InlineKeyboardButton(text="КОМАНДЫ", callback_data="com"),
types.InlineKeyboardButton(text="АВТОР", callback_data="auth")] try:
output = vk.last_post(owner_id=du.get_vk_id(group_id))
keyboard = types.InlineKeyboardMarkup(row_width=2) last_post = db.get_postGroup(group_id)
keyboard.add(*buttons)
if output[3] > last_post:
await message.answer(text=f'*Portal in VK* - это бот для перепоста постов со страниц в социальной сети ВКонтакте. ' text, audio, media = scripts.pars_post(types, output)
f'Для начала работы вам нужно всего лишь вызвать команду */add* и добавить к ней короткое ' db.update_postGroup(group_id, output[3])
f'имя личной или публичной страницы. После этого, бот начнёт присылать каждый новый пост '
f'с этой страницы форматируя данные с неё под сообщение в Telegram. С полным функционалом ' for sub in users:
f'бота, вы можете ознакомиться по кнопке *КОМАНДЫ*\n\n' tg_id = du.get_tg_id(sub)
f'Если вы хотите ставить лайки или получать уведомления от закрытых страниц на которые ' if audio and media:
f'подписаны в социальной сети, вам необходимо скинуть ссылку из поисковой строки в чат, ' post_message = await bot.send_media_group(chat_id=tg_id, media=media)
f'после подтверждения на [сайте]({config.loginUrl})', await bot.send_media_group(chat_id=tg_id, media=audio,
parse_mode=types.ParseMode.MARKDOWN, reply_markup=keyboard) reply_to_message_id=post_message[0].message_id)
elif audio and media == []:
post_message = await bot.send_message(chat_id=tg_id, text=text)
# ИЛАЙН КЛАВИАТУРА HELP await bot.send_media_group(chat_id=tg_id, media=audio,
@dp.callback_query_handler(text="com") reply_to_message_id=post_message[0].message_id)
async def function(call: types.CallbackQuery): elif audio == [] and media:
await call.message.answer(text='*| КОМАНДЫ |*\n\n' await bot.send_media_group(chat_id=tg_id, media=media)
f'*/like* - лайк на пост, который вы отметили\n' else:
f'*/list* - список страниц от, которых вы получаете уведомления\n' await bot.send_message(chat_id=tg_id, text=text, disable_web_page_preview=True)
f'*/notif* - отписка или подписка от всех уведомлений\n'
f'*/last_post* - получение последнего поста по короткому имени страницы\n' except Exception as err:
f'*/add* и */del* - добавление и удаление странички из списка подписок\n', logging.error("Ошибка обновления поста:", exc_info=True)
parse_mode=types.ParseMode.MARKDOWN)
async def main() -> None:
@dp.callback_query_handler(text="auth") dp.include_router(router)
async def author(call: types.CallbackQuery): await bot.delete_webhook(drop_pending_updates=True)
await call.message.answer(text='*| АВТОР |*\n\n*>>* Этот бот не коммерческий проект, для получения постов из ' await dp.start_polling(bot, allowed_updates=dp.resolve_used_update_types(), skipUpdates=True)
'социальной сети ВКонтакте. Бот работает на сервисном токене VK API, пока вы не '
'предоставите собственный. Ваш токен нужен для получения постов со страниц, '
'доступ к которым есть исключительно у вас. Также это даст вам возможность ставить ' async def process() -> None:
'через Telegram лайки на посты в самой социальной сети!' await asyncio.gather(main(), timer(60))
'\n\nЯ же пишу подобные небольшие проекты, о которых вы можете узнать '
'больше на моём [GitHub](https://github.com/IGlek).', if __name__ == "__main__":
parse_mode=types.ParseMode.MARKDOWN) logging.basicConfig(level=logging.INFO)
try: asyncio.run(process())
# КОМАНДЫ except KeyboardInterrupt: pass
@dp.message_handler(commands=["notif"])
async def notification(message: types.Message):
user_id = login(message.chat.id, du, db)
status = db.get_status(user_id)
groups = list(map(int, db.get_user_groups(user_id).split(";")))
for group in groups: db.update_countGroup(group, -1 if status else 1)
if status: await message.answer("Получение постов из ВК выключено!")
else: await message.answer("Получение постов из ВК включено!")
db.update_status(user_id)
@dp.message_handler(commands=["add", "del"])
async def add_del(message: types.Message):
user_id = login(message.chat.id, du, db)
domain = message.get_args()
if domain:
try:
token = db.get_token(user_id)
if token: vk = VkParser(token)
else: vk = VkParser(config.serviceToken)
group_id, typ, last_post = vk.login(domain)
if not du.group_exists(group_id):
du.add_group(group_id)
group_id = du.get_group_id(group_id)
if not db.infoGroup_exists(group_id):
db.add_infoGroup(group_id, typ, last_post)
groups = db.get_user_groups(user_id)
user_groups = groups.split(";") if groups else []
if str(group_id) not in user_groups:
if message.get_command() == "/del":
await message.answer("Вы и так не подписаны на эту группу!")
else:
if db.get_countUser(user_id) < 10:
user_groups.append(str(group_id))
db.update_user_groups(user_id, ";".join(user_groups))
db.update_countUser(user_id, 1)
if db.get_status(user_id): db.update_countGroup(group_id, 1)
await message.answer("Группа успешно добавлена в список уведомлений!")
else:
await message.answer("У вас добавлено максимально количество групп!")
else:
if message.get_command() == "/add":
await message.answer("Вы уже подписаны на эту страницу!")
else:
user_groups.remove(str(group_id))
db.update_user_groups(user_id, ";".join(user_groups))
db.update_countUser(user_id, -1)
if db.get_status(user_id): db.update_countGroup(group_id, -1)
await message.answer("Группа успешно удалена из списка уведомлений!")
except Exception as e:
await message.answer("Произошла ошибка! Проверьте правильность написания короткого адреса страницы "
"и имеете ли вы доступ к этой странице!")
else:
if message.get_command() == "/add":
await message.answer("Команда введена некорректно, такой страницы не существует или она закрыта для "
"просмотра!\n\nКорректный ввод команд: /add <i>example</i>", types.ParseMode.HTML)
elif message.get_command() == "/del":
await message.answer("Команда введена некорректно, такой страницы не существует или она закрыта для "
"просмотра!\n\nКорректный ввод команд: /del <i>example</i>", types.ParseMode.HTML)
@dp.message_handler(commands=["list"])
async def lst(message: types.Message):
user_id = login(message.chat.id, du, db)
groups = db.get_user_groups(user_id)
user_groups = list(map(int, groups.split(";") if groups else []))
if user_groups:
group_ids = [du.get_vk_id(id) for id in user_groups]
vk = VkParser(config.serviceToken)
peoples, groups = vk.info(group_ids)
text = "<b><i>* Список страниц, от которых вы получаете уведомления!</i></b>\n"
if peoples:
text += "\n<b>Личные страницы пользователей</b>\n"
for i, people in enumerate(peoples):
text += f'{emojize(":green_circle:") if people[2] else emojize(":red_circle:")} ' \
f'{i + 1}. {people[1]} <a href="https://vk.com/{people[0]}">@{people[0]}</a>\n'
if groups:
text += "\n<b>Группы / сообщества / паблики</b>\n"
for i, group in enumerate(groups):
text += f'{i + 1}. {group[1]} <a href="https://vk.com/{group[0]}">@{group[0]}</a>\n'
await message.answer(text, types.ParseMode.HTML)
else:
await message.answer("Вы не добавили ни одной страницы для отслеживания публикаций!")
@dp.message_handler(commands=["like"])
async def like(message: types.Message):
user_id = login(message.chat.id, du, db)
token = db.get_token(user_id)
if token and 'reply_to_message' in message:
url = ''
if 'entities' in message.reply_to_message:
for i in message.reply_to_message.entities:
if i['type'] == 'text_link': url = i['url']; break
elif 'caption_entities' in message.reply_to_message:
for i in message.reply_to_message.caption_entities:
if i['type'] == 'text_link': url = i['url']; break
if url:
try:
[group_id, last_post] = list(map(int, url.split('wall')[-1].split('_')))
vk = VkParser(token)
vk.like(group_id, last_post)
await message.answer(text='Вы поставили лайк 😉')
except Exception as e:
await message.answer(text='По неизвестной причине не удалось поставить лайк!')
else:
await message.answer(text='Вы ответили не на тот пост, для отправки лайка!')
else:
await message.answer(text='Вы не отметили пост который хотите лайкнуть или не прислали свой токен для '
'возможности лайкать посты!')
# ВЗАИМОДЕЙСТВИЕ С ДАННЫМИ ВК ПОСТОВ
@dp.message_handler(commands=["last_post"])
async def last_post(message: types.Message):
user_id = message.chat.id
short_id = login(user_id, du, db)
domain = message.get_args()
if domain:
try:
token = db.get_token(short_id)
if token: vk = VkParser(token)
else: vk = VkParser(config.serviceToken)
output = vk.last_post(domain=domain); vk.login(domain)
text, audio, media = pars_post(types, output)
if audio and media:
post_message = await bot.send_media_group(chat_id=user_id, media=media)
await bot.send_media_group(chat_id=user_id, media=audio, reply_to_message_id=post_message[0].message_id)
elif audio and media == []:
post_message = await bot.send_message(chat_id=user_id, text=text, parse_mode=types.ParseMode.MARKDOWN)
await bot.send_media_group(chat_id=user_id, media=audio, reply_to_message_id=(post_message[0].message_id))
elif audio == [] and media:
await bot.send_media_group(chat_id=user_id, media=media)
else:
await bot.send_message(chat_id=user_id, text=text, parse_mode=types.ParseMode.MARKDOWN)
except Exception as e:
await message.answer("Короткое имя страницы неверное или у вас нет доступа к этой странице!")
else:
await message.answer("Команда введена некорректно! Корректный ввод: /last_post example")
# ПОЛУЧЕНИЕ ТОКЕНА ПОЛЬЗОВАТЕЛЯ
@dp.message_handler(content_types="text")
async def url(message: types.Message):
user_id = login(message.chat.id, du, db)
if config.checkUrl in message.text:
try:
token = message.text.split("&")[0].split("=")[-1]
info = VkParser(token).check()
db.update_token(user_id, token)
await message.answer("Токен успешно сохранён!")
except Exception as e:
await message.answer("Ваша ссылка не корректна, попробуйте повторить копирование!")
else:
await message.answer("Ваше сообщение не корректно, необходима ссылка, что отобразилась в строке поиске после"
" подтверждения доступа!")
# ПРОВЕРКА СТРАНИЦ НА НОВЫЕ ПОСТЫ
async def timer(wait_for):
while True:
await asyncio.sleep(wait_for)
notif_group = list(map(lambda x: x[0], db.all_notifGroup()))
activ_user = list(map(lambda x: [x[0], list(map(int, x[1].split(';')))], db.all_activUser()))
sub_user = list(map(lambda x: [x[0], list(map(int, x[1].split(';')))], db.all_subUser()))
for group_id in notif_group:
time.sleep(1)
tokens = [x[0] for x in activ_user if group_id in x[1]]
users = [x[0] for x in sub_user if group_id in x[1]]
if tokens:
user_id = tokens[random.randrange(len(tokens))]
vk = VkParser(db.get_token(user_id))
else: vk = VkParser(config.serviceToken)
try:
output = vk.last_post(owner_id=du.get_vk_id(group_id))
last_post = db.get_postGroup(group_id)
if output[3] > last_post:
text, audio, media = pars_post(types, output)
db.update_postGroup(group_id, output[3])
for sub in users:
tg_id = du.get_tg_id(sub)
if audio and media:
post_message = await bot.send_media_group(chat_id=tg_id, media=media)
await bot.send_media_group(chat_id=tg_id, media=audio, reply_to_message_id=post_message[0].message_id)
elif audio and media == []:
post_message = await bot.send_message(chat_id=tg_id, text=text, parse_mode=types.ParseMode.MARKDOWN)
await bot.send_media_group(chat_id=tg_id, media=audio, reply_to_message_id=post_message[0].message_id)
elif audio == [] and media:
await bot.send_media_group(chat_id=tg_id, media=media)
else:
await bot.send_message(chat_id=tg_id, text=text, parse_mode=types.ParseMode.MARKDOWN)
except Exception as e:
print(repr(e))
if __name__ == '__main__':
loop = asyncio.get_event_loop()
loop.create_task(timer(60)) # ПРОВЕРКА КАЖДУЮ МИНУТУ
executor.start_polling(dp, skip_updates=True)

28
code/config.py
View file

@ -1,14 +1,14 @@
# TOKENS # TOKENS
botToken = '**********' botToken = '**********'
serviceToken = "**********" serviceToken = "**********"
# URL # URL
loginUrl = "************" loginUrl = "************"
checkUrl = "https://oauth.vk.com/blank.html#access_token=" checkUrl = "https://oauth.vk.com/blank.html#access_token="
# VARIABILITY # VARIABILITY
unvote = {True: 'без возможности переголосовать', False: 'с возможностью переголосовать'} unvote = {True: 'без возможности переголосовать', False: 'с возможностью переголосовать'}
anonymous = {True: 'анонимный опрос', False: 'не анонимный опрос'} anonymous = {True: 'анонимный опрос', False: 'не анонимный опрос'}
# DATA # DATA
data = {0: 'Б', 1: 'КБ', 2: 'МБ', 3: 'ГБ', 4: 'ТБ'} data = {0: 'Б', 1: 'КБ', 2: 'МБ', 3: 'ГБ', 4: 'ТБ'}

304
code/handlers.py Normal file
View file

@ -0,0 +1,304 @@
from aiogram import types, F, Router
from aiogram.types import Message, CallbackQuery, ContentType
from aiogram.filters import Command
from emoji import emojize
from init import *
from config import *
from vk_scripts import VkParser
from scripts import login, pars_post
router = Router()
@router.message(Command("start", "help"))
async def start(msg: Message) -> None:
login(msg.chat.id, du, db)
buttons = [[types.InlineKeyboardButton(text="КОМАНДЫ", callback_data="com"),
types.InlineKeyboardButton(text="АВТОР", callback_data="auth")]]
keyboard = types.InlineKeyboardMarkup(inline_keyboard=buttons, row_width=2)
await msg.answer(text=f'<b>Portal in VK</b> - это бот для перепоста постов со страниц в социальной сети ВКонтакте. '
f'Для начала работы вам нужно всего лишь вызвать команду <b>/add</b> и добавить к ней '
f'короткое имя личной или публичной страницы. После этого, бот начнёт присылать каждый новый '
f'пост с этой страницы форматируя данные с неё под сообщение в Telegram. С полным '
f'функционалом бота, вы можете ознакомиться по кнопке <b>КОМАНДЫ</b>\n\n'
f'Если вы хотите ставить лайки или получать уведомления от закрытых страниц на которые '
f'подписаны в социальной сети, вам необходимо скинуть ссылку из поисковой строки в чат, '
f'после подтверждения на <a href="{loginUrl}">сайте</a>', reply_markup=keyboard)
# ИЛАЙН КЛАВИАТУРА HELP
@router.callback_query(F.data == "com")
async def function(call: CallbackQuery) -> None:
await call.message.answer(text='<b>| КОМАНДЫ |</b>\n\n'
'<b>/like</b> - лайк на пост, который вы отметили\n'
'<b>/list</b> - список страниц от, которых вы получаете уведомления\n'
'<b>/notif</b> - отписка или подписка от всех уведомлений\n'
'<b>/last_post</b> - получение последнего поста по короткому имени страницы\n'
'<b>/add</b> и <b>/del</b> - добавление и удаление странички из списка подписок\n')
@router.callback_query(F.data == "auth")
async def author(call: CallbackQuery) -> None:
await call.message.answer(text='<b>| АВТОР |</b>\n\n<b>>></b> Этот бот не коммерческий проект, для получения '
'постов из социальной сети ВКонтакте. Бот работает на сервисном токене VK API, '
'пока вы не предоставите собственный. Ваш токен нужен для получения постов со '
'страниц, доступ к которым есть исключительно у вас. Также это даст вам '
'возможность ставить через Telegram лайки на посты в самой социальной сети!'
'\n\nЯ же пишу подобные небольшие проекты, о которых вы можете узнать '
'больше на моём <a href="https://github.com/IGlek">GitHub</a>.')
# КОМАНДЫ
@router.message(Command("notif"))
async def notification(msg: Message) -> None:
user_id = login(msg.chat.id, du, db)
status = db.get_status(user_id)
groups = list(map(int, db.get_user_groups(user_id).split(";")))
for group in groups: db.update_countGroup(group, -1 if status else 1)
if status: await msg.answer("Получение постов из ВК выключено!")
else: await msg.answer("Получение постов из ВК включено!")
db.update_status(user_id)
@router.message(Command("add", "del"))
async def add_del(msg: Message) -> None:
user_id = login(msg.chat.id, du, db)
domain = msg.text.split()[-1]
command = msg.text.split()[0]
if "/" not in domain:
try:
token = db.get_token(user_id)
if token: vk = VkParser(token)
else: vk = VkParser(serviceToken)
group_id, typ, last_post = vk.login(domain)
if not du.group_exists(group_id):
du.add_group(group_id)
group_id = du.get_group_id(group_id)
if not db.infoGroup_exists(group_id):
db.add_infoGroup(group_id, typ, last_post)
groups = db.get_user_groups(user_id)
user_groups = groups.split(";") if groups else []
if str(group_id) not in user_groups:
if command == "/del":
await msg.answer("Вы и так не подписаны на эту группу!")
else:
if db.get_countUser(user_id) < 10:
user_groups.append(str(group_id))
db.update_user_groups(user_id, ";".join(user_groups))
db.update_countUser(user_id, 1)
if db.get_status(user_id): db.update_countGroup(group_id, 1)
await msg.answer("Группа успешно добавлена в список уведомлений!")
else:
await msg.answer("У вас добавлено максимально количество групп!")
else:
if command == "/add":
await msg.answer("Вы уже подписаны на эту страницу!")
else:
user_groups.remove(str(group_id))
db.update_user_groups(user_id, ";".join(user_groups))
db.update_countUser(user_id, -1)
if db.get_status(user_id): db.update_countGroup(group_id, -1)
await msg.answer("Группа успешно удалена из списка уведомлений!")
except Exception as err:
logging.info("Нет доступа к страницы:", exc_info=True)
await msg.answer("<b>Произошла ошибка!</b> Проверьте правильность написания <b>короткого адреса</b> "
"страницы, <b>имеете ли вы доступ</b> к этой странице и есть ли на ней <b>хотя бы "
"один пост</b>!")
else:
if command == "/add":
await msg.answer("Команда введена некорректно, такой страницы не существует или она закрыта для "
"просмотра!\n\nКорректный ввод команд: /add <i>example</i>")
elif command == "/del":
await msg.answer("Команда введена некорректно, такой страницы не существует или она закрыта для "
"просмотра!\n\nКорректный ввод команд: /del <i>example</i>")
@router.message(Command("list"))
async def lst(msg: Message) -> None:
user_id = login(msg.chat.id, du, db)
groups = db.get_user_groups(user_id)
user_groups = list(map(int, groups.split(";") if groups else []))
if user_groups:
group_ids = [du.get_vk_id(id) for id in user_groups]
vk = VkParser(serviceToken)
peoples, groups = vk.info(group_ids)
text = "<b><i>* Список страниц, от которых вы получаете уведомления!</i></b>\n"
if peoples:
text += "\n<b>Личные страницы пользователей</b>\n"
for i, people in enumerate(peoples):
text += f'{emojize(":green_circle:") if people[2] else emojize(":red_circle:")} ' \
f'{i + 1}. {people[1]} <a href="https://vk.com/{people[0]}">@{people[0]}</a>\n'
if groups:
text += "\n<b>Группы / сообщества / паблики</b>\n"
for i, group in enumerate(groups):
text += f'{i + 1}. {group[1]} <a href="https://vk.com/{group[0]}">@{group[0]}</a>\n'
await msg.answer(text, disable_web_page_preview=True)
else:
await msg.answer("Вы <b>не добавили ни одной страницы</b> для отслеживания публикаций!")
@router.message(Command("like"))
async def like(msg: Message) -> None:
user_id = login(msg.chat.id, du, db)
token = db.get_token(user_id)
if token and (msg.reply_to_message is not None):
url = ''
if msg.reply_to_message.entities is not None:
for i in list(msg.reply_to_message.entities):
i = dict(i)
if i['type'] == 'text_link':
url = i['url']; break
elif msg.reply_to_message.caption_entities is not None:
for i in list(msg.reply_to_message.caption_entities):
i = dict(i)
if i['type'] == 'text_link':
url = i['url']; break
if url:
try:
[group_id, last_post] = list(map(int, url.split('wall')[-1].split('_')))
vk = VkParser(token)
vk.like(group_id, last_post)
await msg.answer('Вы поставили лайк 😉')
except Exception as err:
logging.error("Не удалось поставить лайк:", exc_info=True)
await msg.answer('По <b>неизвестной причине</b> не удалось поставить лайк!')
else:
await msg.answer('Вы <b>ответили не на тот пост</b>, для отправки лайка!')
else:
await msg.answer('Вы <b>не отметили пост</b> который хотите лайкнуть или <b>не прислали свой токен</b> для '
'возможности лайкать посты!')
@router.message(Command("update"))
async def update(msg: Message) -> None:
user_id = msg.chat.id
short_id = login(msg.chat.id, du, db)
if msg.reply_to_message is not None:
url = ''
if msg.reply_to_message.entities is not None:
for i in list(msg.reply_to_message.entities):
i = dict(i)
if i['type'] == 'text_link':
url = i['url']; break
elif msg.reply_to_message.caption_entities is not None:
for i in list(msg.reply_to_message.caption_entities):
i = dict(i)
if i['type'] == 'text_link':
url = i['url']; break
if url:
try:
[group_id, last_post] = list(map(int, url.split('wall')[-1].split('_')))
token = db.get_token(short_id)
if token: vk = VkParser(token)
else: vk = VkParser(serviceToken)
output = vk.last_post(post=f"{group_id}_{last_post}")
text, audio, media = pars_post(types, output)
delete_r = await bot.delete_message(chat_id=user_id, message_id=msg.reply_to_message.message_id)
delete = await bot.delete_message(chat_id=user_id, message_id=msg.message_id)
if audio and media:
post_message = await bot.send_media_group(chat_id=user_id, media=media)
await bot.send_media_group(chat_id=user_id, media=audio,
reply_to_message_id=post_message[0].message_id)
elif audio and media == []:
post_message = await bot.send_message(chat_id=user_id, text=text)
await bot.send_media_group(chat_id=user_id, media=audio,
reply_to_message_id=post_message[0].message_id)
elif audio == [] and media:
await bot.send_media_group(chat_id=user_id, media=media)
else:
await bot.send_message(chat_id=user_id, text=text, disable_web_page_preview=True)
except Exception as err:
logging.danger("Ошибка обновления поста:", exc_info=True)
await msg.answer(text='По неизвестной причине не удалось обновить пост!')
else: await msg.answer(text='Вы ответили не на тот пост!')
else: await msg.answer(text='Вы не отметили пост который хотите обновить!')
# ВЗАИМОДЕЙСТВИЕ С ДАННЫМИ ВК ПОСТОВ
@router.message(Command("last_post"))
async def last_post(msg: Message) -> None:
user_id = msg.chat.id
short_id = login(user_id, du, db)
domain = msg.text.split()[-1]
if "/" not in domain:
try:
token = db.get_token(short_id)
if token: vk = VkParser(token)
else: vk = VkParser(serviceToken)
output = vk.last_post(domain=domain); vk.login(domain)
text, audio, media = pars_post(types, output)
if audio and media:
post_message = await bot.send_media_group(chat_id=user_id, media=media)
await bot.send_media_group(chat_id=user_id, media=audio, reply_to_message_id=post_message[0].message_id)
elif audio and media == []:
post_message = await bot.send_message(chat_id=user_id, text=text)
await bot.send_media_group(chat_id=user_id, media=audio, reply_to_message_id=post_message[0].message_id)
elif audio == [] and media:
await bot.send_media_group(chat_id=user_id, media=media)
else:
await bot.send_message(chat_id=user_id, text=text, disable_web_page_preview=True)
except Exception as err:
logging.info("Некорректное/недоступное короткое имя:", exc_info=True)
await msg.answer("Короткое <b>имя страницы неверное</b> или у вас <b>нет доступа</b> к этой странице!")
else: await msg.answer("Команда введена некорректно! Корректный ввод: /last_post <b>example</b>")
# ПОЛУЧЕНИЕ ТОКЕНА ПОЛЬЗОВАТЕЛЯ
@router.message(F.content_type == ContentType.TEXT)
async def url(msg: Message) -> None:
user_id = login(msg.chat.id, du, db)
if checkUrl in msg.text:
try:
token = msg.text.split("&")[0].split("=")[-1]
info = VkParser(token).check()
db.update_token(user_id, token)
await msg.answer("Токен успешно сохранён!")
except Exception as err:
logging.info("Некорректная ссылка:", exc_info=True)
await msg.answer("Ваша <b>ссылка не корректна</b>, попробуйте повторить копирование!")
else: await msg.answer("Ваше <b>сообщение не корректно</b>, необходима ссылка, что отобразилась в строке "
"поиске после подтверждения доступа!")

22
code/init.py Normal file
View file

@ -0,0 +1,22 @@
from aiogram import Bot, Dispatcher
from aiogram.enums.parse_mode import ParseMode
from aiogram.fsm.storage.memory import MemoryStorage
from aiogram.client.bot import DefaultBotProperties
from config import botToken
bot = Bot(token=botToken, default=DefaultBotProperties(parse_mode=ParseMode.HTML))
dp = Dispatcher(storage=MemoryStorage())
from sql import Users, Base
du = Users('../db/users.db')
db = Base('../db/base.db')
import logging, functools
logging.basicConfig(level=logging.INFO, filename='../info.log',filemode="w",
format="%(asctime)s %(levelname)s %(message)s")

131
code/scripts.py
View file

@ -1,57 +1,74 @@
import config import config
def login(user_id, du, db): def login(user_id, du, db):
user_id = int(user_id) user_id = int(user_id)
if not du.user_exists(user_id): du.add_user(user_id) if not du.user_exists(user_id): du.add_user(user_id)
user_id = du.get_user_id(user_id) user_id = du.get_user_id(user_id)
if not db.infoUser_exists(user_id): db.add_infoUser(user_id) if not db.infoUser_exists(user_id): db.add_infoUser(user_id)
return user_id return user_id
def pars_post(types, output): def pars_post(types, output):
media, ttl, url, audio, poll, link = [], [], [], [], [], [] media, video, url, audio, poll, link = [], [], [], [], [], []
for group in output[0]: for group in output[0]:
if group[0] == "poll": poll = group[1] if group[0] == "poll": poll = group[1]
elif group[0] == "link": link.append(group[1]) elif group[0] == "link": link.append(group[1])
elif group[0] == "video": ttl.append(group[1]) elif group[0] == "video": video.append(group[1])
elif group[0] == "doc" and group[1][2] != 4: url.append([group[1][0], group[1][1], group[1][3]]) elif group[0] == "doc" and group[1][2] != 4: url.append([group[1][0], group[1][1], group[1][3]])
elif group[0] == "audio": audio.append(types.InputMediaAudio(media=group[1][1], parse_mode=types.ParseMode.MARKDOWN, elif group[0] == "audio": audio.append(types.InputMediaAudio(
caption=f"*Название: {group[1][0]} / Автор: {group[1][2]}*")) media=group[1][1], caption=f"<b>Название: {group[1][0]} / Автор: {group[1][2]}</b>"))
if output[1] != "": text = f"{output[2]}\n\n*{output[1]}*\n" if output[1] != "":
else: text = f"{output[2]}\n" post = [elem_2 for elem_1 in output[1].split("[") for elem_2 in elem_1.split("]")]
for i, elem in enumerate(post):
if poll: if "|" in elem and i != 0:
text += f"\n*> Проводиться {config.anonymous[poll[3]]} {config.unvote[poll[2]]}!" elem = elem.split("|")
text += f"*\n*---------------------*\n*- Вопрос: {poll[0]} [{poll[1]}]*\n" post[i] = f"<a href='https://vk.com/{elem[0]}'>{elem[1]}</a>"
text += ''.join([f'*{i}. Ответ: {ans[0]} [{ans[1]}]*\n' for i, ans in enumerate(poll[4])])
text += "*---------------------*\n" text = f"{output[2]}\n\n{''.join(post)}\n"
else: text = f"{output[2]}\n"
if ttl: text += f"\n*Видео -* находятся в оригинальном посте в количестве {len(ttl)} записей\n"
if poll:
if url: text += f"\n<b>> Проводиться</b> {config.anonymous[poll[3]]} {config.unvote[poll[2]]}! <b>Вопрос:</b> {poll[0]} <b>[{poll[1]}]</b>\n"
for data in url: text += ''.join([f'<b>{i + 1}. Ответ:</b> {ans[0]} <b>[{ans[1]}]</b>\n' for i, ans in enumerate(poll[4])]) + "\n"
k, dt = 0, data[2]
while True: if video:
if dt > 1024: dt = dt // 1024; k += 1 for frame in video:
else: break text += f"\n<b>Видео</b> <i>« {frame[0]} »</i>\n"
text += f"\n*Документ - [* [{data[0]}]({data[1]}) ({dt} {config.data[k]}) *]*"
text += "\n" k, dura = 0, frame[1]
while k < 1:
if link: text += ''.join([f"\n*Ссылка - [* [{s[0]}]({s[1]}) *]*" for s in link]) + "\n" if dura > 60: dura = dura // 60; k += 1
if audio: text += '\n*« Также присутствуют аудио файлы »*' else: break
for i, group in enumerate(output[0]): text += f"<b>Длительность:</b> {dura} {config.dura[k]}; <b>Лайков:</b> {frame[2]};\n"
if group[0] == "photo":
if i: media.append(types.InputMediaPhoto(media=group[1])) if url:
else: media.append(types.InputMediaPhoto(media=group[1], caption=text, parse_mode=types.ParseMode.MARKDOWN)) for data in url:
elif group[0] == "doc": k, dt = 0, data[2]
if group[1][2] == 4: while True:
if i: media.append(types.InputMediaPhoto(media=group[1][1])) if dt > 1024: dt = dt // 1024; k += 1
else: media.append(types.InputMediaPhoto(media=group[1][1], caption=text, parse_mode=types.ParseMode.MARKDOWN)) else: break
text += f"\n<b>Документ - [</b> <a href='{data[1]}'>{data[0]}</a> ({dt} {config.data[k]}) <b>]</b>"
return text, audio, media text += "\n"
if link: text += ''.join([f"\n<b>Ссылка - [</b> [{s[0]}]({s[1]}) <b>]</b>" for s in link]) + "\n"
if audio: text += '\n<b>« Также присутствуют аудио файлы »</b>'
for i, group in enumerate(output[0]):
if group[0] == "photo":
if i: media.append(types.InputMediaPhoto(media=group[1]))
else: media.append(types.InputMediaPhoto(media=group[1], caption=text))
elif group[0] == "video":
if i: media.append(types.InputMediaPhoto(media=group[1][3]))
else: media.append(types.InputMediaPhoto(media=group[1][3], caption=text))
elif group[0] == "doc":
if group[1][2] == 4:
if i: media.append(types.InputMediaPhoto(media=group[1][1]))
else: media.append(types.InputMediaPhoto(media=group[1][1], caption=text))
return text, audio, media

1
code/sql.py
View file

@ -166,3 +166,4 @@ class Base:
def close(self): def close(self):
"""Закрываем соединение с БД""" """Закрываем соединение с БД"""
self.connection.close() self.connection.close()

185
code/vk_scripts.py
View file

@ -1,76 +1,109 @@
import vk_api import vk_api
class VkParser: class VkParser:
def __init__(self, token): def __init__(self, token):
"""Подключаемся к API VK""" """Подключаемся к API VK"""
self.session = vk_api.VkApi(token=token) self.session = vk_api.VkApi(token=token)
self.vk = self.session.get_api() self.vk = self.session.get_api()
def check(self): def check(self):
return self.vk.account.getInfo(fields="lang") return self.vk.account.getInfo(fields="lang")
def like(self, owner_id, post_id): def like(self, owner_id, post_id):
return self.vk.likes.add(type="post", owner_id=owner_id, item_id=post_id) return self.vk.likes.add(type="post", owner_id=owner_id, item_id=post_id)
def login(self, domain): def login(self, domain):
posts = self.vk.wall.get(domain=domain, count=3) posts = self.vk.wall.get(domain=domain, count=3)
post = max([[int(pt["id"]), int(pt["owner_id"])] for pt in posts["items"]]) post = max([[int(pt["id"]), int(pt["owner_id"])] for pt in posts["items"]])
[last_post, group_id] = post [last_post, group_id] = post
typ = 0 if group_id > 0 else 1 typ = 0 if group_id > 0 else 1
return group_id, typ, last_post return group_id, typ, last_post
def info(self, group_ids): def info(self, group_ids):
groups, peoples = [], [] groups, peoples = [], []
for id in group_ids: for id in group_ids:
if id > 0: peoples.append(id) if id > 0: peoples.append(id)
else: groups.append(id) else: groups.append(id)
if groups: if groups:
data = self.vk.groups.getById(group_ids=", ".join(list(map(lambda x: str(x)[1:], groups)))) data = self.vk.groups.getById(group_ids=", ".join(list(map(lambda x: str(x)[1:], groups))))
groups = [[elem['screen_name'], elem['name']] for elem in data] groups = [[elem['screen_name'], elem['name']] for elem in data]
if peoples: if peoples:
data = self.vk.users.get(user_ids=", ".join(list(map(str, peoples))), fields="domain, online") data = self.vk.users.get(user_ids=", ".join(list(map(str, peoples))), fields="domain, online")
peoples = [[elem['domain'], elem['first_name'] + ' ' + elem['last_name'], elem['online']] for elem in data] peoples = [[elem['domain'], elem['first_name'] + ' ' + elem['last_name'], elem['online']] for elem in data]
return peoples, groups return peoples, groups
def last_post(self, owner_id=None, domain=None): def last_post(self, owner_id=None, domain=None, post=None):
wall = self.vk.wall.get(owner_id=owner_id, domain=domain, count=2, extended=1) if post: wall = self.vk.wall.getById(posts=post, extended=1)
post = max([[int(pt["id"]), pt] for pt in wall["items"]]) else: wall = self.vk.wall.get(owner_id=owner_id, domain=domain, count=2, extended=1)
[post_id, post] = post post = max([[int(pt["id"]), pt] for pt in wall["items"]])
owner_id, output = post['owner_id'], [] owner_id = post[1]['owner_id']
if owner_id > 0: [name, domain] = [[per['first_name'] + ' ' + per['last_name'], per['screen_name']] if owner_id > 0:
for per in wall["profiles"] if per["id"] == owner_id][0] [name, domain] = [[per['first_name'] + ' ' + per['last_name'], per['screen_name']] for per in wall["profiles"] if per["id"] == owner_id][0]
else: [name, domain] = [[gro['name'], gro['screen_name']] else:
for gro in wall["groups"] if gro["id"] == int(str(owner_id)[1:])][0] [name, domain] = [[gro['name'], gro['screen_name']] for gro in wall["groups"] if gro["id"] == int(str(owner_id)[1:])][0]
attachments = post['attachments'] if 'copy_history' in post[1]:
types = [[typ['type'], typ] for typ in attachments] comment, orig_id = post[1]['text'], post[0]
for typ in types: post = [post[1]['copy_history'][0]['id'], post[1]['copy_history'][0]]
if typ[0] == 'photo': owner_id_r = post[1]['owner_id']
output.append([typ[0], typ[1]['photo']['sizes'][-1]['url']])
elif typ[0] == 'video': if owner_id_r > 0:
output.append([typ[0], typ[1]['video']['title']]) [name_r, domain_r] = [[per['first_name'] + ' ' + per['last_name'], per['screen_name']] for per in wall["profiles"] if per["id"] == owner_id_r][0]
elif typ[0] == "doc": else:
output.append([typ[0], [typ[1]['doc']['title'], typ[1]['doc']['url'], [name_r, domain_r] = [[gro['name'], gro['screen_name']] for gro in wall["groups"] if gro["id"] == int(str(owner_id_r)[1:])][0]
typ[1]['doc']['type'], typ[1]['doc']['size']]])
elif typ[0] == "audio": return get_output(post, name_r, domain_r, [orig_id, owner_id, name, domain, comment])
output.append([typ[0], [typ[1]['audio']['title'], typ[1]['audio']['url'], typ[1]['audio']['artist']]]) else:
elif typ[0] == "poll": return get_output(post, name, domain)
answer = [[que['text'], que['votes']] for que in typ[1]['poll']['answers']]
output.append([typ[0], [typ[1]['poll']['question'], typ[1]['poll']['votes'],
typ[1]['poll']['disable_unvote'], typ[1]['poll']['anonymous'], answer]]) def get_output(post, name, domain, repost=None):
elif typ[0] == "link": [post_id, post] = post
output.append([typ[0], [typ[1]['link']['title'], typ[1]['link']['url']]]) owner_id, output = post['owner_id'], []
return [output, post['text'], attachments = post['attachments']
f"*Автор поста -* [{name}](https://vk.com/{domain}?w=wall{owner_id}_{post_id})", post_id] types = [[typ['type'], typ] for typ in attachments]
for typ in types:
if typ[0] == 'photo':
output.append([typ[0], typ[1]['photo']['sizes'][-1]['url']])
elif typ[0] == 'video':
video = typ[1]['video']
if 'photo_1280' in video:
frame = video['photo_1280']
else:
frame = video['photo_800']
output.append([typ[0], [video['title'], video['duration'], video['views'], frame]])
elif typ[0] == "doc":
output.append([typ[0], [typ[1]['doc']['title'], typ[1]['doc']['url'],
typ[1]['doc']['type'], typ[1]['doc']['size']]])
elif typ[0] == "audio":
output.append([typ[0], [typ[1]['audio']['title'], typ[1]['audio']['url'], typ[1]['audio']['artist']]])
elif typ[0] == "poll":
answer = [[que['text'], que['votes']] for que in typ[1]['poll']['answers']]
output.append([typ[0], [typ[1]['poll']['question'], typ[1]['poll']['votes'],
typ[1]['poll']['disable_unvote'], typ[1]['poll']['anonymous'], answer]])
elif typ[0] == "link":
output.append([typ[0], [typ[1]['link']['title'], typ[1]['link']['url']]])
if repost:
[author_id, author_owner_id, author_name, author_domain, author_comment] = repost
comment = f"<b>Автор репоста -</b> <a href='https://vk.com/{author_domain}?w=wall{author_owner_id}_{author_id}'>{author_name}</a>"
if author_comment != "": comment += f"\n<blockquote>{author_comment}</blockquote>\n"
comment += f"\n<b>Автор поста -</b> <a href='https://vk.com/{domain}?w=wall{owner_id}_{post_id}'>{name}</a>"
return [output, f"{post['text']}", comment, author_id]
else:
return [output, f"{post['text']}",
f"<b>Автор поста -</b> <a href='https://vk.com/{domain}?w=wall{owner_id}_{post_id}'>{name}</a>", post_id]