Получаем удалённые и отредактированные сообщения Telegram без скачивания клиентов

llimonix · 17 май 2024

В данной статье предоставлентестовый код, которая не включает в себя отслеживания сообщений с фотографиями или документами. Этот код сделан лишь для того, чтобы показать как можно реализовать новые фишки Telegram Premium.

1. Покупаем Telegram Premium, если его у вас еще нет
2. Создаем бота в https://t.me/BotFather и включаем в нём Business mode:

3. Подключаем своего бота к профилю Telegram и выбираем чаты, где он будет работать (или же ничего не делаем и он будет отвечать всем кто ему пишет):

Пишем код на Python
1. Скачиваем Python: https://www.python.org/downloads/ (у меня установлена версия Python 3.11.6) (ПРИ УСТАНОВКЕ PYTHON НЕ ЗАБУДЬТЕ ПОСТАВИТЬ ГАЛОЧКУ ADD TO PATH)
2. Открываем консоль и устанавливаем необходимые модули (если чего-то не хватаем докачиваем точно так же):
pip install aiogram
pip install pydantic

Python
pip install aiogram
pip install pydantic
⁡
3. Создаем файл
⁡main.py
⁡ и вставляем мой тестовый код:
import configparser
import json
from typing import Union

import asyncio

from aiogram import (Router, Bot, Dispatcher,
F, types)
import logging

from database import Messagesx

router = Router(name=__name__)

logger = logging.getLogger(__name__)

logging.basicConfig(level=logging.INFO)

config = configparser.ConfigParser()
config.read("config.ini")

TOKEN = config["main"]["bot_token"]
USER_ID = config["main"]["user"]

async def send_msg(message_old: str, message_new: Union[str, None], user_fullname: str, user_id: int, bot: Bot = None):
if message_new is None:
msg = (f' Пользователь {user_fullname} ({user_id})\n'
f' Сообщение удалено:\n'
f' Сообщение:\n<code>{message_old}</code>\n')
else:
msg = (f' Пользователь {user_fullname} ({user_id})\n'
f'✏ Сообщение изменено:\n'
f' Старое сообщение:\n<code>{message_old}</code>\n'
f' Новое сообщение:\n<code>{message_new}</code>')
await bot.send_message(USER_ID, msg, parse_mode='html')

@router.edited_business_message()
async def edited_business_message(message: types.Message):
if message.from_user.id == message.chat.id:
user_msg = Messagesx.get(user_id=message.from_user.id)
data = {message.message_id: message.text}
if user_msg is None:
Messagesx.add(user_id=message.from_user.id, message_history=json.dumps(data))
else:
msg_history = json.loads(user_msg.message_history)
if str(message.message_id) in msg_history:
await send_msg(message_old=msg_history[str(message.message_id)], message_new=message.text,
user_fullname=message.from_user.full_name, user_id=message.chat.id, bot=message.bot)
data = {**msg_history, **data}
Messagesx.update(user_id=message.from_user.id, message_history=json.dumps(data))

@router.deleted_business_messages()
async def deleted_business_messages(message: types.Message):
user_msg = Messagesx.get(user_id=message.chat.id)
if user_msg is not None:
msg_history = json.loads(user_msg.message_history)
for msg_id in message.message_ids:
if str(msg_id) in msg_history:
await send_msg(message_old=msg_history[str(msg_id)], message_new=None,
user_fullname=message.chat.full_name, user_id=message.chat.id, bot=message.bot)
msg_history.pop(str(msg_id))
Messagesx.update(user_id=message.chat.id, message_history=json.dumps(msg_history))

@router.business_message(F.text)
async def business_message(message: types.Message):
if message.from_user.id == message.chat.id:
user_msg = Messagesx.get(user_id=message.from_user.id)
data = {message.message_id: message.text}
if user_msg is None:
Messagesx.add(user_id=message.from_user.id, message_history=json.dumps(data))
else:
msg_history = json.loads(user_msg.message_history)
data = {**msg_history, **data}
Messagesx.update(user_id=message.from_user.id, message_history=json.dumps(data))

async def main() -> None:
Messagesx.create_db()

bot = Bot(token=TOKEN)
dp = Dispatcher()

dp.include_router(router)

await bot.delete_webhook(drop_pending_updates=True)
await dp.start_polling(bot)

asyncio.run(main())

Python
import configparser
import json
from typing import Union

import asyncio

from aiogram import (Router, Bot, Dispatcher,
 F, types)
import logging

from database import Messagesx

router = Router(name=__name__)

logger = logging.getLogger(__name__)

logging.basicConfig(level=logging.INFO)

config = configparser.ConfigParser()
config.read("config.ini")

TOKEN = config["main"]["bot_token"]
USER_ID = config["main"]["user"]


async def send_msg(message_old: str, message_new: Union[str, None], user_fullname: str, user_id: int, bot: Bot = None):
 if message_new is None:
 msg = (f' Пользователь {user_fullname} ({user_id})\n'
 f' Сообщение удалено:\n'
 f' Сообщение:\n<code>{message_old}</code>\n')
 else:
 msg = (f' Пользователь {user_fullname} ({user_id})\n'
 f'✏ Сообщение изменено:\n'
 f' Старое сообщение:\n<code>{message_old}</code>\n'
 f' Новое сообщение:\n<code>{message_new}</code>')
 await bot.send_message(USER_ID, msg, parse_mode='html')


@router.edited_business_message()
async def edited_business_message(message: types.Message):
 if message.from_user.id == message.chat.id:
 user_msg = Messagesx.get(user_id=message.from_user.id)
 data = {message.message_id: message.text}
 if user_msg is None:
 Messagesx.add(user_id=message.from_user.id, message_history=json.dumps(data))
 else:
 msg_history = json.loads(user_msg.message_history)
 if str(message.message_id) in msg_history:
 await send_msg(message_old=msg_history[str(message.message_id)], message_new=message.text,
 user_fullname=message.from_user.full_name, user_id=message.chat.id, bot=message.bot)
 data = {**msg_history, **data}
 Messagesx.update(user_id=message.from_user.id, message_history=json.dumps(data))


@router.deleted_business_messages()
async def deleted_business_messages(message: types.Message):
 user_msg = Messagesx.get(user_id=message.chat.id)
 if user_msg is not None:
 msg_history = json.loads(user_msg.message_history)
 for msg_id in message.message_ids:
 if str(msg_id) in msg_history:
 await send_msg(message_old=msg_history[str(msg_id)], message_new=None,
 user_fullname=message.chat.full_name, user_id=message.chat.id, bot=message.bot)
 msg_history.pop(str(msg_id))
 Messagesx.update(user_id=message.chat.id, message_history=json.dumps(msg_history))


@router.business_message(F.text)
async def business_message(message: types.Message):
 if message.from_user.id == message.chat.id:
 user_msg = Messagesx.get(user_id=message.from_user.id)
 data = {message.message_id: message.text}
 if user_msg is None:
 Messagesx.add(user_id=message.from_user.id, message_history=json.dumps(data))
 else:
 msg_history = json.loads(user_msg.message_history)
 data = {**msg_history, **data}
 Messagesx.update(user_id=message.from_user.id, message_history=json.dumps(data))


async def main() -> None:
 Messagesx.create_db()

 bot = Bot(token=TOKEN)
 dp = Dispatcher()

 dp.include_router(router)

 await bot.delete_webhook(drop_pending_updates=True)
 await dp.start_polling(bot)


asyncio.run(main())
4. Создаем файл
⁡database.py
⁡ и вставляем мой код с работой базы данных:
from pydantic import BaseModel

import sqlite3

def dict_factory(cursor, row) -> dict:
save_dict = {}

for idx, col in enumerate(cursor.description):
save_dict[col[0]] = row[idx]

return save_dict

def update_format(sql, parameters: dict) -> tuple[str, list]:
values = ", ".join([
f"{item} = ?" for item in parameters
])
sql += f" {values}"

return sql, list(parameters.values())

def update_format_where(sql, parameters: dict) -> tuple[str, list]:
sql += " WHERE "

sql += " AND ".join([
f"{item} = ?" for item in parameters
])

return sql, list(parameters.values())

class MessageRecord(BaseModel):
user_id: int
message_history: str

# Работа с юзером
class Messagesx:
storage_name = "messages"
PATH_DATABASE = "messages.db"

@staticmethod
def create_db():
with sqlite3.connect('messages.db') as conn:
cursor = conn.cursor()
cursor.execute('''CREATE TABLE IF NOT EXISTS messages
(id INTEGER PRIMARY KEY, user_id INTEGER, message_history TEXT)''')

# Добавление записи
@staticmethod
def add(
user_id: int,
message_history: str,

):

with sqlite3.connect(Messagesx.PATH_DATABASE) as con:
con.row_factory = dict_factory

con.execute(
f"""
INSERT INTO {Messagesx.storage_name} (
user_id,
message_history
) VALUES (?, ?)
""",
[
user_id,
message_history,
],
)

# Получение записи
@staticmethod
def get(**kwargs) -> MessageRecord:
with sqlite3.connect(Messagesx.PATH_DATABASE) as con:
con.row_factory = dict_factory
sql = f"SELECT * FROM {Messagesx.storage_name}"
sql, parameters = update_format_where(sql, kwargs)

response = con.execute(sql, parameters).fetchone()

if response is not None:
response = MessageRecord(**response)

return response

# Получение записей
@staticmethod
def gets(**kwargs) -> list[MessageRecord]:
with sqlite3.connect(Messagesx.PATH_DATABASE) as con:
con.row_factory = dict_factory
sql = f"SELECT * FROM {Messagesx.storage_name}"
sql, parameters = update_format_where(sql, kwargs)

response = con.execute(sql, parameters).fetchall()

if len(response) >= 1:
response = [MessageRecord(**cache_object) for cache_object in response]

return response

# Получение всех записей
@staticmethod
def get_all() -> list[MessageRecord]:
with sqlite3.connect(Messagesx.PATH_DATABASE) as con:
con.row_factory = dict_factory
sql = f"SELECT * FROM {Messagesx.storage_name}"

response = con.execute(sql).fetchall()

if len(response) >= 1:
response = [MessageRecord(**cache_object) for cache_object in response]

return response

# Редактирование записи
@staticmethod
def update(user_id, **kwargs):
with sqlite3.connect(Messagesx.PATH_DATABASE) as con:
con.row_factory = dict_factory
sql = f"UPDATE {Messagesx.storage_name} SET"
sql, parameters = update_format(sql, kwargs)
parameters.append(user_id)

con.execute(sql + "WHERE user_id = ?", parameters)

# Удаление записи
@staticmethod
def delete(**kwargs):
with sqlite3.connect(Messagesx.PATH_DATABASE) as con:
con.row_factory = dict_factory
sql = f"DELETE FROM {Messagesx.storage_name}"
sql, parameters = update_format_where(sql, kwargs)

con.execute(sql, parameters)

# Очистка всех записей
@staticmethod
def clear():
with sqlite3.connect(Messagesx.PATH_DATABASE) as con:
con.row_factory = dict_factory
sql = f"DELETE FROM {Messagesx.storage_name}"

con.execute(sql)

Python
from pydantic import BaseModel

import sqlite3


def dict_factory(cursor, row) -> dict:
 save_dict = {}

 for idx, col in enumerate(cursor.description):
 save_dict[col[0]] = row[idx]

 return save_dict


def update_format(sql, parameters: dict) -> tuple[str, list]:
 values = ", ".join([
 f"{item} = ?" for item in parameters
 ])
 sql += f" {values}"

 return sql, list(parameters.values())


def update_format_where(sql, parameters: dict) -> tuple[str, list]:
 sql += " WHERE "

 sql += " AND ".join([
 f"{item} = ?" for item in parameters
 ])

 return sql, list(parameters.values())


class MessageRecord(BaseModel):
 user_id: int
 message_history: str



# Работа с юзером
class Messagesx:
 storage_name = "messages"
 PATH_DATABASE = "messages.db"

 @staticmethod
 def create_db():
 with sqlite3.connect('messages.db') as conn:
 cursor = conn.cursor()
 cursor.execute('''CREATE TABLE IF NOT EXISTS messages
 (id INTEGER PRIMARY KEY, user_id INTEGER, message_history TEXT)''')

 # Добавление записи
 @staticmethod
 def add(
 user_id: int,
 message_history: str,

 ):

 with sqlite3.connect(Messagesx.PATH_DATABASE) as con:
 con.row_factory = dict_factory

 con.execute(
 f"""
 INSERT INTO {Messagesx.storage_name} (
 user_id,
 message_history
 ) VALUES (?, ?)
 """,
 [
 user_id,
 message_history,
 ],
 )

 # Получение записи
 @staticmethod
 def get(**kwargs) -> MessageRecord:
 with sqlite3.connect(Messagesx.PATH_DATABASE) as con:
 con.row_factory = dict_factory
 sql = f"SELECT * FROM {Messagesx.storage_name}"
 sql, parameters = update_format_where(sql, kwargs)

 response = con.execute(sql, parameters).fetchone()

 if response is not None:
 response = MessageRecord(**response)

 return response

 # Получение записей
 @staticmethod
 def gets(**kwargs) -> list[MessageRecord]:
 with sqlite3.connect(Messagesx.PATH_DATABASE) as con:
 con.row_factory = dict_factory
 sql = f"SELECT * FROM {Messagesx.storage_name}"
 sql, parameters = update_format_where(sql, kwargs)

 response = con.execute(sql, parameters).fetchall()

 if len(response) >= 1:
 response = [MessageRecord(**cache_object) for cache_object in response]

 return response

 # Получение всех записей
 @staticmethod
 def get_all() -> list[MessageRecord]:
 with sqlite3.connect(Messagesx.PATH_DATABASE) as con:
 con.row_factory = dict_factory
 sql = f"SELECT * FROM {Messagesx.storage_name}"

 response = con.execute(sql).fetchall()

 if len(response) >= 1:
 response = [MessageRecord(**cache_object) for cache_object in response]

 return response

 # Редактирование записи
 @staticmethod
 def update(user_id, **kwargs):
 with sqlite3.connect(Messagesx.PATH_DATABASE) as con:
 con.row_factory = dict_factory
 sql = f"UPDATE {Messagesx.storage_name} SET"
 sql, parameters = update_format(sql, kwargs)
 parameters.append(user_id)

 con.execute(sql + "WHERE user_id = ?", parameters)

 # Удаление записи
 @staticmethod
 def delete(**kwargs):
 with sqlite3.connect(Messagesx.PATH_DATABASE) as con:
 con.row_factory = dict_factory
 sql = f"DELETE FROM {Messagesx.storage_name}"
 sql, parameters = update_format_where(sql, kwargs)

 con.execute(sql, parameters)

 # Очистка всех записей
 @staticmethod
 def clear():
 with sqlite3.connect(Messagesx.PATH_DATABASE) as con:
 con.row_factory = dict_factory
 sql = f"DELETE FROM {Messagesx.storage_name}"

 con.execute(sql)
5. Создаем файл config.ini и вставляем мой текст, но изменяем токен бота и ставим свой user_id (его можно получить в боте: https://t.me/userinfobot ):
[main]
bot_token=6095.....
user=80....

Код
[main]
bot_token=6095.....
user=80....
6. Вместо
⁡6095.....
⁡ и
⁡80....
⁡ , там где bot_token - вставляем токен вашего бота, который вы получили у BotFather. И в user нужно вставить user_id из пункта 5! (У других бот будет отправлять уведомления вам, если они подключат вашего бота. У telegram не выводит кому было отправлено сообщение, так что узнать ваш ID никак нельзя)
7. Запускаем скрипт и проверяем. Можно написать с другого аккаунта и удалить или изменить сообщение.

Обязательно выберите чата где это будет работать или оставьте все и не забудьте включить бота

Там хранится id пользователя и история сообщений после включения бота ( message_id: text )

Важно! Функционал работает только с теми сообщениями, что были получены во время использования данного бота!

Можете подключить уже готового бота: https://t.me/BusDetectBot

llimonix · 20 май 2024

https://t.me/BusDetectBot

Поднял бота для автоматического отслеживания ивентов редактирования и удаления. В скором времени добавлю так же обработку для файлов, фотографий и тд

Перед началом подключения бота для бизнес чатов, напишите ему /start и все. Потом добавляйте его как бота, для бизнес чатов из гайда в теме! Бот будет присылать вам все замеченные изменения с момента подключения его к чатам

11111111111 · 17 май 2024

видел в ютуб шортсах недавно

MRAC · 20 май 2024

А можно видео гайд? куда блять в конце тыкать?

Signal · 26 май 2024

Traceback (most recent call last):
File "C:\Users\Administrator\Desktop\****\main.py", line 39, in <module>
@router.edited_business_message()
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
AttributeError: 'Router' object has no attribute 'edited_business_message'

Код
Traceback (most recent call last):
 File "C:\Users\Administrator\Desktop\****\main.py", line 39, in <module>
 @router.edited_business_message()
 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
AttributeError: 'Router' object has no attribute 'edited_business_message'
такие дела