Initial commit. v4.23

master
ssleg 2021-06-14 17:05:26 +03:00
commit 0aeaed5f18
Signed by: anton
GPG Key ID: 50F7E97F96C07ECF
13 changed files with 1416 additions and 0 deletions

51
CONFIG.MD Normal file
View File

@ -0,0 +1,51 @@
### Все настройки бота делаются в файле magic.ini
Секция **bot_config**
+ debug - отладка или продакшен. при отладке пишется расширенный лог и все сообщения присылаемые и отправляемые ботом.
+ auth_from_db - настройка для продвинутых пользователей, см. секцию **bot_db_account**.
+ io_write - записывать или не записывать все сообщения в базу SQLite.
+ logfile_name - имя файла лога.
Секция **bot_admins**
+ admins_ids - перечисление **user_id** администраторов бота. Эти telegram аккаунты смогут смотреть статистику работы
бота и делать массовые рассылки пользователям.
+ Разделитель, знак точка с запятой. При единственном админе можно не ставить.
Команды, доступные для администраторов:
+ `-mw` - присылает статистику работы бота: аптайм, количество пользователей, обработанных сообщений и ошибок.
+ `-send` <сообщение> - посылает сообщение всем пользователям бота, используется для рассылки рекламы, сервисных сообщений и.т.д.
Сообщение отделено пробелом от слова send и дойдет до пользователей вместе с форматированием (жирный/курсив и другие) и ссылками.
Секция **bot_telegram_account**
+ Для получения api_id и api_hash вам надо залогиниться на сайте [my.telegram.org], перейти в раздел **API development
tools** и зарегистрировать свое приложение. Если сервер выдает пустую ошибку (error) заполните поле "описание проекта".
+ session_name - имя файла для хранения сессии бота, может быть любым.
+ bot_token - токен вашего бота, который дает телеграм аккаунт **@BotFather**.
Секция **bot_db_account**
Бот может использовать базу данных PostgreSQL для хранения ключей аутентификации. Для использования такого варианта вам
надо:
+ отредактировать в файле **tg_utils.py** класс **PgServer**, вписав в __init__ имя вашей БД, пользователя и пароль.
+ создать в своей базе таблицу **parameters** из двух текстовых полей **name** и **value**.
+ выбрать имя для своего аккаунта, например Vasya, записать в таблицу два параметра с именами Vasya_api_id и
Vasya_api_hash.
+ записать в ini файл account = Vasya
+ выбрать имя для своего бота, например Bot_Vasi, записать в таблицу его токен с именем Bot_Vasi_key.
+ записать в ini файл bot_account = Bot_Vasi
Все, ваши данные надежно спрятаны от посторонних глаз.
Так же вы найдете несколько полезных классов и функций для **PostgreSQL** и **telegram** в файле **tg_utils.py**.
Поскольку документацию никто и никогда не читает, самая последняя настройка описана здесь:
параметр **bot_stats**, по умолчанию включен, и разрешает отправлять мне анонимную статистику работы.
Через час после запуска и потом каждые сутки отсылается версия приложения, аптайм и количество событий класса ERROR. Мне
интересно, сколько ботов на этом ядре будет сделано. Никакие личные данные не собираются, можете проверить, посмотрев в
файле **magic.py** функцию **stat_upload()**. Можете легко это отключить, назначив параметру значение **False**.
[my.telegram.org]:https://my.telegram.org/apps

32
INSTALL.MD Normal file
View File

@ -0,0 +1,32 @@
### Инсталляция бота на сервер VPS.
На примере стандартного ubuntu сервера на хостингах.
+ обновите репозитории, командой `sudo apt update`
+ установите pip, командой `sudo apt install python3-pip`
+ установите postgres, пригодится :) `sudo apt install postgresql`
+ установите заголовочные файлы (нужны для сборки psycopg2). `sudo apt install libpq-dev`
+ создайте в домашней папке директорию для бота, допустим test_bot
+ скопируйте все файлы бота в папку test_bot
+ перейдите в командной строке в папку test_bot и выполните `sudo pip3 install -r requirements.txt`
+ дайте права на исполнение файлу **magic.py** командой `chmod 755 magic.py`
+ отредактируйте файл **magic.sh** заменив 'пользователь' на имя пользователя сервера
+ дайте права на исполнение файлу **magic.sh** командой `chmod 755 magic.sh`
+ отредактируйте файл **test_bot.service** заменив 'пользователь' на имя пользователя сервера
+ скопируйте его в systemd командой `sudo cp test_bot.service /etc/systemd/system/`
+ обновите список сервисов командой `sudo systemctl daemon-reload`
+ выполните все настройки **magic.ini** как описано в **config.md**
+ включите сервис бота: `sudo systemctl enable test_bot`
+ запустите его: `sudo systemctl start test_bot`
Все, бот стал одним из сервисов linux и работает 24/7/365.
При перезагрузке сервера он запустится автоматически.
Даже если у телеграм будет глобальный сбой, бот оживет, как только он закончится.
При обновлениях вашего кода, вы обновляете файлы в папке бота на сервере и отдаете команду в консоли:
`sudo systemctl restart test_bot `
Перезапуск может занимать до 1 минуты, если очередь сообщений велика, стандартно около 30 секунд.

9
LICENSE Normal file
View File

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

39
README.MD Normal file
View File

@ -0,0 +1,39 @@
### Ядро для построения telegram ботов на Python.
Позволяет написать бота любой сложности, не погружаясь в устройство telegram. Доступно даже **начинающим** изучать
python программистам.
Достаточно написать только логику бота и поставить на свой сервер.
Ядро полностью **имплементирует все ограничения** telegram для ботов, вашего бота никогда не забанят.
Основные функции ядра:
+ рассылка сообщений всем подписчикам бота
+ автоматическое уведомление админов бота об ошибках времени исполнения
+ запись всех входящих и исходящих сообщений в базу SQLite
+ высокоскоростное кэширование исходящих сообщений с сохранением их порядка.
+ подробный отладочный лог или лог основных событий в продакшене
+ сбор статистики работы для админа
Структура проекта:
**magic.py** - исполняемый файл бота, загружает все остальное.
**main_module.py** - основной модуль бота (их может быть несколько), место для вашего кода.
**bot_io.py** - модуль ввода-вывода (сообщений).
**bot_io_classes.py** - классы модуля ввода-вывода.
**tg_utils.py** - классы и функции telegram для всех модулей.
**magic.ini** - файл конфигурации бота.
Инструкция по развертыванию на сервере (ubuntu) находится в файле **install.md**
Инструкция по настройкам находится в файле **config.md**
Бот использует библиотеку [Telethon].
Благодарности, вопросы и реквесты фич складывать здесь или в комментариях к [этому посту].
Лицензия на код и документацию MIT. Вы можете свободно использовать, изменять и продавать код при условии сохранения
информации об авторских правах.
[Telethon]:https://docs.telethon.dev/en/latest/
[этому посту]:https://t.me/ssleg/347

388
bot_io.py Normal file
View File

@ -0,0 +1,388 @@
# Bot I/O v1.70
# 14/06/2021
# https://t.me/ssleg © 2020 2021
import logging
import sqlite3
from asyncio import sleep
from datetime import datetime, timedelta
from os import path
from telethon import TelegramClient, errors
from bot_io_classes import TgBotUsers, OutMessagesQueue
from tg_utils import GlobalFlag
debug = GlobalFlag()
io_write = GlobalFlag()
client: TelegramClient
msk_hour = timedelta(hours=3)
sys_wait = 0.045
manager_runned = GlobalFlag()
work_queue = OutMessagesQueue()
tg_flood_wait = GlobalFlag()
last_send_time: datetime
bot_users: TgBotUsers
admins_ids = []
con: sqlite3.Connection
cursor: sqlite3.Cursor
class IncomingMessagesTimeBuffer:
"""класс буфера времени входящих сообщений пользователей"""
__slots__ = ['__buf', '__size', '__banned_to', '__ban_count', '__banned', '__user_id']
@staticmethod
def __sort_key(element):
return element[0]
@staticmethod
def __is_now_banned(timestamp):
now = datetime.now()
delta = now - timestamp
if delta.total_seconds() > 0:
return False
else:
return True
def __init__(self, size, user_id):
self.__buf = []
self.__size = size
self.__user_id = user_id
cursor.execute('select ban_count, ban_to_date from banlist where user_id=?', (user_id,))
row = cursor.fetchone()
if row is not None:
self.__banned_to = datetime.strptime(row[1], '%Y-%m-%d %H:%M:%S')
self.__ban_count = row[0]
self.__banned = self.__is_now_banned(self.__banned_to)
else:
self.__banned_to = datetime(year=2020, month=1, day=1)
self.__ban_count = 0
self.__banned = False
for i in range(0, size):
self.__buf.append((0, 0))
def __is_doubled(self, mess_id):
i = 0
while i < self.__size:
if self.__buf[i][0] == mess_id:
levent = 'обнаружен дубликат сообщения #' + str(mess_id) + ', глубина - ' + str(i)
logging.warning(levent)
return True
i += 1
return False
def __is_valid_timing(self, mess_date):
i = self.__size - 1
sec = 0
while i > self.__size - 4:
prev_date = self.__buf[i][1]
if prev_date != 0:
tmp = mess_date - prev_date
if tmp.seconds < 1:
levent = 'обнаружено слишком частое обращение: ' + str(tmp.seconds) + ' сек.'
logging.warning(levent)
return False
else:
sec += tmp.seconds
mess_date = prev_date
else:
return True
i -= 1
if sec >= 5:
return True
else:
levent = 'обнаружены 4 сообщенния за ' + str(sec) + ' сек.'
logging.warning(levent)
return False
def store_mess(self, mess_id, mess_date):
if not self.__is_doubled(mess_id):
if self.__banned:
status = self.__is_now_banned(self.__banned_to)
if status:
return 1
else:
self.__banned = False
if not self.__is_valid_timing(mess_date):
if self.__user_id not in admins_ids:
now = datetime.now()
if self.__ban_count > 0:
delta = timedelta(days=30)
rez = 4
else:
delta = timedelta(minutes=30)
rez = 3
self.__banned_to = now + delta
self.__banned = True
self.__ban_count += 1
if self.__ban_count == 1:
entry = (self.__user_id, self.__ban_count, str(self.__banned_to)[0:19])
cursor.execute('insert into banlist (user_id, ban_count, ban_to_date) values (?,?,?)', entry)
else:
entry = (self.__ban_count, str(self.__banned_to)[0:19], self.__user_id)
cursor.execute('update banlist set ban_count=?, ban_to_date=? where user_id=?', entry)
con.commit()
return rez
self.__buf.append((mess_id, mess_date))
self.__buf.sort(key=self.__sort_key)
self.__buf.pop(0)
return 0
else:
return 2
class BotIncomingMessagesOrder:
"""класс порядка сообщений в диалогах"""
__slots__ = ['__users', '__orders']
def __init__(self):
self.__users = {}
self.__orders = []
def new_mess(self, message):
mess_id = message.id
mess_date = message.date + msk_hour
user_id = message.peer_id.user_id
txt = message.text
indx = self.__users.get(user_id)
if indx is None:
indx = len(self.__orders)
self.__users[user_id] = indx
self.__orders.append(IncomingMessagesTimeBuffer(20, user_id))
if debug:
levent = 'открыт новый буфер, id - ' + str(user_id)
logging.info(levent)
order_result = self.__orders[indx].store_mess(mess_id, mess_date)
if debug or io_write:
if order_result not in [1, 2]:
entry = (mess_id, str(mess_date)[0:19], user_id, txt)
cursor.execute('insert into messages (mess_id, mess_date, from_id, mess_txt) values (?,?,?,?)', entry)
con.commit()
return order_result
# менеджер отложенной отправки сообщений
async def queue_manager():
if not manager_runned:
manager_runned.set_true()
levent = 'менеджер отложенных сообщений стартовал.'
logging.info(levent)
now = datetime.now()
delta = now - last_send_time
if delta.total_seconds() < 0:
tsec = delta.total_seconds()
seconds = abs(tsec // 1 + 1)
levent = 'ожидание разрешения на отправку - ' + str(seconds) + ' сек.'
logging.info(levent)
await sleep(seconds)
tg_flood_wait.set_false()
mess_count = 0
mess_success = 0
while work_queue.queue_empty() is not True:
if not tg_flood_wait:
entry = work_queue.get_next_message()
user_id = entry[0]
message = entry[1]
file_name = entry[2]
await sleep(sys_wait)
if debug:
levent = 'попытка отправки сообщения для user_id = ' + str(user_id)
logging.info(levent)
result_flag = await send_reply_message(user_id, message, file_name, log_parameter=True, from_queue=True)
work_queue.set_sending_result(result_flag)
if result_flag:
mess_success += 1
mess_count += 1
else:
break
manager_runned.set_false()
levent = 'менеджер отложенных сообщений закончил. попыток - ' + str(mess_count) + ', отправлено - ' + str(
mess_success)
logging.info(levent)
if tg_flood_wait:
client.loop.create_task(queue_manager())
def message_to_queue(user_id, mess, file_name):
work_queue.add_message(user_id, mess, file_name)
if not manager_runned:
client.loop.create_task(queue_manager())
# отправка сообщений пользователю с соблюдением требований тг
async def send_reply_message(user_id, message, file_name=None, log_parameter=False, contact_add=True, from_queue=False):
global last_send_time
now = datetime.now()
delta = now - last_send_time
if not from_queue:
find_flag, indx = work_queue.is_user_in_queue(user_id)
if find_flag:
message_to_queue(user_id, message, file_name)
levent = 'ответ поставлен в очередь. user_id = ' + str(user_id)
logging.warning(levent)
return False
if bot_users.is_bot_user(user_id):
timestamp = bot_users.get_user_mess_timestamp(user_id)
user_delta = now - timestamp
if user_delta.total_seconds() < 1:
if not from_queue:
message_to_queue(user_id, message, file_name)
levent = 'ответ поставлен в очередь, дельта пользователя - ' + str(
user_delta.total_seconds()) + ' сек. user_id = ' + str(user_id)
logging.warning(levent)
return False
if delta.total_seconds() > 0.04:
try:
await client.send_message(user_id, message, file=file_name)
last_send_time = datetime.now()
if log_parameter or debug:
levent = 'ответ отправлен, user_id = ' + str(user_id)
logging.info(levent)
if debug or io_write:
if len(message) > 100:
message = message[0:100]
entry = (str(last_send_time)[0:19], user_id, message)
cursor.execute('insert into messages ( mess_date, to_id, mess_txt) values (?,?,?)', entry)
con.commit()
if contact_add:
if not bot_users.is_bot_user(user_id):
user_entity = await client.get_entity(user_id)
bot_users.new_user_store(user_entity)
else:
bot_users.update_user_mess_timestamp(user_id)
return True
except errors.FloodWaitError as e:
seconds = e.seconds
levent = 'антифлуд телеграм сработал, время ожидания - ' + str(seconds) + ' сек.'
logging.warning(levent)
tg_flood_wait.set_true()
last_send_time = now + timedelta(seconds=seconds + 1)
if not from_queue:
message_to_queue(user_id, message, file_name)
return False
except Exception as e:
levent = 'что-то пошло не так при отправке (user_id = ' + str(user_id) + '): ' + str(e)
if debug:
logging.error(levent)
else:
logging.warning(levent)
return False
else:
if not from_queue:
message_to_queue(user_id, message, file_name)
levent = 'ответ поставлен в очередь, дельта - ' + str(delta.total_seconds()) + ' сек. user_id = ' + str(
user_id)
logging.warning(levent)
return False
# отправка сообщений всем пользователям бота
async def send_message_to_all(mess):
users_list = bot_users.get_users_list()
count = 0
success_count = 0
for user_id in users_list:
result_flag = await send_reply_message(user_id, mess)
await sleep(sys_wait)
if result_flag:
success_count += 1
count += 1
return count, success_count
# инициализация
async def init(cli, debug_mode, adm_ids, io_write_mode):
global client
global last_send_time
global con
global cursor
global bot_users
global admins_ids
admins_ids = adm_ids
client = cli
if debug_mode:
debug.set_true()
if io_write_mode:
io_write.set_true()
if not path.exists('io.sqlite'):
con = sqlite3.connect('io.sqlite')
cursor = con.cursor()
cursor.executescript('''
CREATE TABLE messages
(
mess_id int,
mess_date text,
from_id int,
to_id int,
mess_txt text
);
CREATE TABLE contacts
(
user_id int,
first_name text,
last_name text,
account_name text
);
CREATE TABLE banlist
(
user_id int,
ban_count int,
ban_to_date text
);
''')
con.commit()
else:
con = sqlite3.connect('io.sqlite')
cursor = con.cursor()
bot_users = TgBotUsers(con, cursor)
last_send_time = datetime.now()
levent = 'bot I/O запущен.'
logging.info(levent)
# завершение работы
async def terminate():
if manager_runned:
levent = 'bot I/O, ожидание отправки всех сообщений.'
logging.info(levent)
count = 6
while count > 0:
await sleep(5)
count -= 1
if not manager_runned:
break
con.close()
levent = 'bot I/O остановлен.'
logging.info(levent)

128
bot_io_classes.py Normal file
View File

@ -0,0 +1,128 @@
# Bot I/O classes v1.00
# 14/06/2021
# https://t.me/ssleg © 2020 2021
from datetime import datetime
class TgBotUsers:
"""класс хранения пользователей бота"""
__slots__ = ['__users', '__timestamps', '__con', '__cursor']
def __init__(self, con, cursor):
self.__users = {}
self.__timestamps = []
self.__con = con
self.__cursor = cursor
self.__cursor.execute('select * from contacts')
for row in cursor.fetchall():
user_id = row[0]
f_name = row[1]
l_name = row[2]
user_acc = row[3]
indx = len(self.__timestamps)
self.__users[user_id] = (indx, f_name, l_name, user_acc)
self.__timestamps.append(datetime(year=2020, month=1, day=1))
def is_bot_user(self, user_id):
tmp = self.__users.get(user_id)
if tmp is not None:
return True
else:
return False
def new_user_store(self, userinfo):
user_id = userinfo.id
f_name = userinfo.first_name
l_name = userinfo.last_name
user_acc = userinfo.username
entry = (user_id, f_name, l_name, user_acc)
self.__cursor.execute('''insert into contacts
(user_id, first_name, last_name, account_name)
values (?,?,?,?)''', entry)
self.__con.commit()
timestamp = datetime.now()
indx = len(self.__timestamps)
self.__users[user_id] = (indx, f_name, l_name, user_acc)
self.__timestamps.append(timestamp)
def update_user_mess_timestamp(self, user_id):
timestamp = datetime.now()
user = self.__users.get(user_id)
if user is not None:
indx = user[0]
self.__timestamps[indx] = timestamp
def get_user_mess_timestamp(self, user_id):
user = self.__users.get(user_id)
if user is not None:
indx = user[0]
return self.__timestamps[indx]
else:
return datetime(year=2020, month=1, day=1)
def get_users_list(self):
excluded_users = []
now = datetime.now()
self.__cursor.execute('select user_id, ban_to_date from banlist')
for row in self.__cursor.fetchall():
banned_to = datetime.strptime(row[1], '%Y-%m-%d %H:%M:%S')
delta = now - banned_to
if delta.total_seconds() < 0:
excluded_users.append(row[0])
user_list = []
for key in self.__users:
if key not in excluded_users:
user_list.append(key)
return user_list
class OutMessagesQueue:
"""класс очереди исходящих сообщений"""
__slots__ = ['__queue', '__position_i']
def __init__(self):
self.__queue = []
self.__position_i = 0
def queue_empty(self):
if len(self.__queue) == 0:
return True
return False
def is_user_in_queue(self, user_id):
i = 0
finded_flag = False
while i < len(self.__queue):
if self.__queue[i][0] == user_id:
finded_flag = True
break
i += 1
return finded_flag, i
def add_message(self, user_id, message_text, file_name):
finded_flag, i = self.is_user_in_queue(user_id)
if not finded_flag:
self.__queue.append([user_id])
self.__queue[i].append((user_id, message_text, file_name))
def get_next_message(self):
element = self.__queue[self.__position_i][1]
return element
def set_sending_result(self, flag):
if flag:
user_queue = self.__queue[self.__position_i]
user_queue.pop(1)
if len(user_queue) == 1:
self.__queue.pop(self.__position_i)
else:
self.__position_i += 1
if self.__position_i >= len(self.__queue):
self.__position_i = 0

23
magic.ini Normal file
View File

@ -0,0 +1,23 @@
# настраиваемые параметры работы
[bot_config]
debug = True
auth_from_db = False
bot_stats = True
io_write = True
logfile_name = magic.log
# user_id администраторов бота через ;
[bot_admins]
admins_ids = 222135000; 444049000
# ключи доступа к api телеграм
[bot_telegram_account]
api_id = 123456
api_hash = dfdfdfsdfsfsfssf
session_name = session
bot_token = 1234560000:FFFFQQQQDDD111FDFD-xdfdsfds
# или ключи доступа к api из БД
[bot_db_account]
account =
bot_account =

310
magic.py Executable file
View File

@ -0,0 +1,310 @@
#!/usr/bin/env python3
# Magic wand v4.23
# 14/06/2021
# https://t.me/ssleg © 2020 2021
import logging
import signal
from asyncio import sleep
from configparser import ConfigParser
from datetime import datetime
from hashlib import md5
from logging.handlers import RotatingFileHandler
from os import path
from requests import post
from telethon import TelegramClient, events
import bot_io
import main_module
from tg_utils import get_tg_client, get_bot_key, GlobalFlag, GlobalCounter
# загрузка конфигурации из ini файла
bot_config = ConfigParser()
bot_config.read('magic.ini')
debug = bot_config.getboolean('bot_config', 'debug')
auth_from_db = bot_config.getboolean('bot_config', 'auth_from_db')
bot_stats = bot_config.getboolean('bot_config', 'bot_stats')
io_write = bot_config.getboolean('bot_config', 'io_write')
logfile_name = bot_config.get('bot_config', 'logfile_name')
admins_ids_str = bot_config.get('bot_admins', 'admins_ids')
admins_ids_array = []
if admins_ids_str.find(';') > -1:
admins_ids_str = admins_ids_str.replace(' ', '')
str_split = admins_ids_str.split(';')
for value in str_split:
if value.isdigit():
admins_ids_array.append(int(value))
else:
if admins_ids_str.isdigit():
admins_ids_array.append(int(admins_ids_str))
api_id = bot_config.getint('bot_telegram_account', 'api_id')
api_hash = bot_config.get('bot_telegram_account', 'api_hash')
session_name = bot_config.get('bot_telegram_account', 'session_name')
bot_token = bot_config.get('bot_telegram_account', 'bot_token')
account = bot_config.get('bot_db_account', 'account')
bot_account = bot_config.get('bot_db_account', 'bot_account')
# глобальные переменные
sigterm_flag = GlobalFlag()
logsize = GlobalCounter()
laststring = GlobalCounter()
error_count = GlobalCounter()
started_time: datetime
next_stat = GlobalCounter(3600)
# инициализация лога и параметров подключения
log_file = RotatingFileHandler(logfile_name, 'a', maxBytes=524288, backupCount=10, encoding='utf=8')
log_file.setFormatter(logging.Formatter('%(levelname)s %(module)-13s [%(asctime)s] %(message)s'))
# noinspection PyArgumentList
logging.basicConfig(level=logging.INFO, handlers=[log_file])
sigterm_flag.set_true()
if len(admins_ids_array) == 0:
l_event = 'ни одного админа не указано, бот неуправляем.'
logging.warning(l_event)
admins_ids = ()
else:
admins_ids = tuple(admins_ids_array)
if auth_from_db:
client = get_tg_client(account)
bot_key = get_bot_key(bot_account)
else:
client = TelegramClient(session_name, api_id, api_hash)
bot_key = bot_token
# возвращает строку с версией приложения
def get_my_version():
my_name = path.basename(__file__)
file = open(my_name)
version = ''
for line in file:
line = line[0:len(line) - 1]
if len(line) > 0:
if line[0] == '#':
offset = line.find(' v')
if offset > -1:
version = line[offset + 1:len(line)]
break
file.close()
return version
# записывает начальные параметры лог-файла
def init_log_var():
logsize.set(path.getsize(logfile_name))
file = open(logfile_name, 'r', encoding='utf-8')
allfile = file.readlines()
file.close()
laststring.set(len(allfile))
levent = 'начальные параметры этого логфайла: ' + str(logsize) + ' байт, ' + str(laststring) + ' строк.'
logging.info(levent)
# основная инициализация бота и модулей
async def init():
my_version = get_my_version()
if auth_from_db:
account_id = account
else:
account_id = str(api_id)
levent = 'Magic wand ' + my_version + ' (' + account_id + '), запуск модулей.'
logging.info(levent)
init_log_var()
await bot_io.init(client, debug, admins_ids, io_write)
await main_module.init(client, debug, admins_ids)
levent = 'инициализация завершена.'
logging.info(levent)
# возвращает строку со временем аптайма
def get_uptime():
now = datetime.now()
delta = now - started_time
days = delta.days
secs = delta.seconds
hours = round(secs // 3600)
minutes = round(secs // 60 - hours * 60)
seconds = round(secs - hours * 3600 - minutes * 60)
message = 'Аптайм - ' + str(days) + ' дней, ' + str(hours) + ' час. ' + str(minutes) + ' мин. ' + str(
seconds) + ' сек.\n\n'
return message
# возвращает количество секунд с момента старта
def get_up_seconds():
now = datetime.now()
delta = now - started_time
return delta.days * 86400 + delta.seconds
# эвент хэндлер, отвечающий за команды админа (статистика работы, рассылки)
async def admins_command(event):
from_id = event.message.peer_id.user_id
command = event.message.text
if command == '-mw':
message = get_uptime()
message += main_module.status()
message += 'пользователей - ' + str(len(bot_io.bot_users.get_users_list())) + '\n'
message += 'ошибок - ' + str(error_count)
await bot_io.send_reply_message(from_id, message)
levent = 'запрос статистики выполнен успешно, админ: ' + str(from_id)
logging.info(levent)
if command.find('-send') == 0:
mess = command[6:len(command)]
if mess != '':
levent = 'админ ' + str(from_id) + ' создал рассылку.'
logging.info(levent)
count, success_count = await bot_io.send_message_to_all(mess)
levent = 'массовая рассылка сообщения выполнена, пользователей: ' + str(
count) + ', доставлено сообщений: ' + str(success_count)
logging.info(levent)
await bot_io.send_reply_message(from_id, 'рассылка выполнена для ' + str(
count) + ' пользователей, доставлено сообщений: ' + str(success_count))
else:
await bot_io.send_reply_message(from_id, 'сообщения для рассылки нет.')
# загрузка статистики работы на сервер
def stat_upload():
request_headers = {
'Accept': 'application/json, text/plain, */*',
'Content-Type': 'application/json; charset=utf-8'
}
# noinspection HttpUrlsUsage
stat_upload_url = 'http://77.83.92.107/stat_up'
hash_md5 = md5(bot_key.encode())
request_json = {'protocol_version': '1.1', 'application': 'Magic Wand', 'app_version': get_my_version(),
'uptime': get_up_seconds(), 'errors': error_count.get(), 'fingerprint': hash_md5.hexdigest()}
try:
response = post(stat_upload_url, headers=request_headers, json=request_json, timeout=5)
response_code = response.status_code
if response_code == 200:
if debug:
levent = 'статистика отправлена успешно.'
logging.info(levent)
next_stat.increment(86400)
else:
levent = 'ошибка на сервере статистики, код ' + str(response_code)
logging.warning(levent)
next_stat.increment(300)
except Exception as e:
levent = 'ошибка в http запросе: ' + str(e)
logging.error(levent)
next_stat.increment(300)
# рассылка админам бота уведомления об ошибке
async def send_notices(mess):
error_count.increment()
for admin in admins_ids:
await sleep(0.5)
await bot_io.send_reply_message(admin, mess)
levent = 'уведомление об ошибке отправлено админу: ' + str(admin)
logging.info(levent)
# чтение изменений логфайла и поиск ошибок в нем
def log_reader():
file = open(logfile_name, 'r', encoding='utf-8')
readcount = 0
workcount = laststring.get()
i = 0
flag = False
errmsg = ''
last_read = ''
while i < 1:
string = file.readline()
if string != '':
readcount += 1
if readcount > workcount:
laststring.increment()
if flag:
if len(string) > 1:
if string.find('[20') > -1:
errmsg += '\n' + last_read
client.loop.create_task(send_notices(errmsg))
flag = False
else:
last_read = string
if string.find('ERROR') == 0:
errmsg = string
flag = True
else:
if flag:
errmsg += '\n' + last_read
client.loop.create_task(send_notices(errmsg))
i = 1
file.close()
# наблюдатель за логфайлом, проверяет изменения раз в минуту
def log_watcher():
if sigterm_flag:
filesize = path.getsize(logfile_name)
if filesize > logsize:
logsize.set(filesize)
client.loop.call_soon(log_reader)
if bot_stats:
if get_up_seconds() > next_stat:
client.loop.call_soon(stat_upload)
client.loop.call_later(60, log_watcher)
# завершает работу модулей
async def terminate():
levent = 'остановка началась...'
logging.info(levent)
await main_module.terminate()
await bot_io.terminate()
await client.disconnect()
levent = 'бот остановлен.'
logging.info(levent)
# хэндлер сигнала сигтерм, возникающего при перезапуске системы или бота.
# вызывает процедуру корректного завершения бота и модулей
# noinspection PyUnusedLocal
def sigterm_call(signum, frame):
if sigterm_flag:
sigterm_flag.set_false()
levent = 'получен SIGTERM ' + str(signum)
logging.info(levent)
client.loop.create_task(terminate())
signal.signal(signal.SIGTERM, sigterm_call)
# старт telethon и основной инициализации
client.start(bot_token=bot_key)
client.loop.run_until_complete(init())
started_time = datetime.now()
client.add_event_handler(admins_command, events.NewMessage(chats=admins_ids, incoming=True))
client.loop.call_later(60, log_watcher)
client.run_until_disconnected()

3
magic.sh Executable file
View File

@ -0,0 +1,3 @@
#!/usr/bin/env bash
cd /home/пользователь/test_bot
./magic.py

93
main_module.py Normal file
View File

@ -0,0 +1,93 @@
# Demo bot module v1.00
# 14/06/2021
# https://t.me/ssleg © 2020 2021
import logging
from telethon import TelegramClient, events
import bot_io
from bot_io import BotIncomingMessagesOrder
from tg_utils import GlobalFlag, GlobalCounter, set_int_printable
debug = GlobalFlag()
messages_count = GlobalCounter()
client: TelegramClient
ordnung = BotIncomingMessagesOrder()
admins_ids = []
# хэндлер входящих сообщений
async def mytask(event):
mess = event.message
from_id = mess.peer_id.user_id
txt = mess.text
# антиспам
mess_status = ordnung.new_mess(mess)
if mess_status == 1 or mess_status == 2:
return
elif mess_status == 3:
mess = 'Вы забанены на полчаса за частые обращения к боту (чаще команды в секунду или 4х за 5 сек.).'
await bot_io.send_reply_message(from_id, mess, contact_add=False)
levent = 'пользователь забанен на полчаса - ' + str(from_id)
logging.warning(levent)
return
elif mess_status == 4:
mess = 'Вы забанены на месяц за повторные частые обращения к боту (чаще команды в секунду или 4х за 5 сек.).'
await bot_io.send_reply_message(from_id, mess, contact_add=False)
levent = 'пользователь забанен на месяц - ' + str(from_id)
logging.warning(levent)
return
txt = txt.lower()
txt = txt.replace(' ', '')
# проверка на команду от админа, ee обрабатывает основная программа
# см. magic.py admins_command()
if from_id in admins_ids:
if txt == '-mw' or txt.find('-send') == 0:
return
messages_count.increment()
# стартовое приветствие (когда пользователь нажимает кнопку старт).
if txt == '/start':
mess = 'Привет!'
await bot_io.send_reply_message(from_id, mess)
return
# здесь место для вашего кода. обработайте входящее сообщение и ответьте пользователю.
# всегда используйте функцию bot_io для отправки ответов, это предохранит вас от бана телеграм.
mess = 'ваш user_id: ' + set_int_printable(from_id) + '\n'
mess += 'а я больше ничего не умею.'
await bot_io.send_reply_message(from_id, mess)
# инициализация модуля
async def init(cli, debug_mode, adm_ids):
global client
global admins_ids
client = cli
admins_ids = adm_ids
if debug_mode:
debug.set_true()
client.add_event_handler(mytask, events.NewMessage(incoming=True))
levent = 'основной модуль стартовал.'
logging.info(levent)
# выдача "наверх" статистики работы
def status():
mess = 'выполнено запросов - ' + str(messages_count) + '\n'
return mess
# завершение работы
async def terminate():
levent = 'основной модуль остановлен. выполнил запросов - ' + str(messages_count)
logging.info(levent)

4
requirements.txt Normal file
View File

@ -0,0 +1,4 @@
requests~=2.23.0
Telethon~=1.17.5
psycopg2~=2.8.6
cryptg~=0.2.post1

11
test_bot.service Normal file
View File

@ -0,0 +1,11 @@
[Unit]
Description=test_bot
After=multi-user.target
[Service]
Type=idle
ExecStart=/home/пользователь/test_bot/magic.sh
Restart=always
[Install]
WantedBy=multi-user.target

325
tg_utils.py Normal file
View File

@ -0,0 +1,325 @@
# Tg utils v2.22
# 14/06/2021
# https://t.me/ssleg © 2020 2021
import logging
from datetime import datetime
import psycopg2
from telethon import TelegramClient
class PgServer:
"""класс postgres-сервер"""
__slots__ = ['__con', '__control_flag', '__cursor', '__db_name']
def __init__(self, dbc=None, db_name='mydb'):
if dbc is None:
self.__con = psycopg2.connect(database=db_name,
user='postgres',
password='1234',
host='127.0.0.1',
port='5432')
self.__control_flag = True
self.__db_name = db_name
else:
self.__con = dbc
self.__control_flag = False
self.__db_name = ''
self.__cursor = self.__con.cursor()
def commit(self):
self.__con.commit()
def rollback(self):
self.__con.rollback()
def exec(self, sql, req_data=None, return_type=None, retry_count=0):
try:
if req_data is not None:
self.__cursor.execute(sql, req_data)
else:
self.__cursor.execute(sql)
if return_type is not None:
if return_type == 0:
return self.__cursor.fetchall()
else:
return self.__cursor.fetchone()
else:
return True
except Exception as e:
levent = 'postgres error: ' + str(e)
logging.error(levent)
if self.__control_flag:
if retry_count < 3:
self.__con.close()
self.__con = psycopg2.connect(database=self.__db_name,
user='postgres',
password='1234',
host='127.0.0.1',
port='5432')
self.__cursor = self.__con.cursor()
return self.exec(sql, req_data, return_type, retry_count=retry_count + 1)
return False
def __del__(self):
self.__cursor.close()
if self.__control_flag:
self.__con.close()
else:
self.__con.rollback()
class PgLock:
"""класс блокировки базы данных"""
__slots__ = ['__srv', '__cursor', '__lock_name', '__debug']
def __init__(self, dbc=None, lck='default_lock', debug=False):
self.__srv = PgServer(dbc)
self.__lock_name = lck
self.__debug = debug
def lock(self):
self.__srv.exec('update locks set status=True where lock_name=%s', (self.__lock_name,))
self.__srv.commit()
if self.__debug:
levent = 'locked #' + self.__lock_name
logging.info(levent)
def unlock(self):
self.__srv.exec('update locks set status=False where lock_name=%s', (self.__lock_name,))
self.__srv.commit()
if self.__debug:
levent = 'unlocked #' + self.__lock_name
logging.info(levent)
def lock_status(self):
row = self.__srv.exec('select status from locks where lock_name=%s', (self.__lock_name,), return_type=1)
self.__srv.rollback()
if row is not None:
lock_status = row[0]
return lock_status
else:
return False
class GlobalCounter:
"""глобальный универсальный счетчик"""
__slots__ = ['__count']
@staticmethod
def __is_valid_arg(arg):
if type(arg) is int:
if arg > 0:
return True
return False
def __init__(self, init_count=0):
if GlobalCounter.__is_valid_arg(init_count):
self.__count = init_count
else:
self.__count = 0
def increment(self, step=1):
if GlobalCounter.__is_valid_arg(step):
self.__count += step
return True
return False
def decrement(self, step=1):
if GlobalCounter.__is_valid_arg(step):
if self.__count - step >= 0:
self.__count -= step
return True
return False
def get(self):
return self.__count
def set(self, value):
if GlobalCounter.__is_valid_arg(value):
self.__count = value
return True
return False
def clear(self):
self.__count = 0
def __isub__(self, other):
if GlobalCounter.__is_valid_arg(other):
self.__count -= other
def __iadd__(self, other):
if GlobalCounter.__is_valid_arg(other):
self.__count += other
def __eq__(self, other):
if self.__count == other:
return True
return False
def __ge__(self, other):
if self.__count >= other:
return True
return False
def __gt__(self, other):
if self.__count > other:
return True
return False
def __le__(self, other):
if self.__count <= other:
return True
return False
def __lt__(self, other):
if self.__count <= other:
return True
return False
def __str__(self):
return str(self.__count)
class GlobalFlag:
"""глобальный универсальный флаг"""
__slots__ = ['__flag']
def __init__(self):
self.__flag = False
def set_true(self):
self.__flag = True
def set_false(self):
self.__flag = False
def switch(self):
if self.__flag:
self.__flag = False
else:
self.__flag = True
def get(self):
return self.__flag
def __bool__(self):
return self.__flag
def __str__(self):
return str(self.__flag)
def t_stamp():
now = datetime.now()
stamp = datetime.strftime(now, '%d/%m %Y %H:%M:%S')
return stamp
def t_stamp_sh():
now = datetime.now()
stamp = datetime.strftime(now, '%H:%M')
return stamp
# ловеркейс имени tg канала с проверкой валидности
def set_ch_name_lower(teststr):
i = 0
rezult = ''
while i < len(teststr):
char = teststr[i]
code = ord(char)
if code == 95:
pass
else:
if 47 < code < 58:
pass
else:
if 64 < code < 91:
code += 32
else:
if 96 < code < 123:
pass
else:
levent = 'ошибка конвертации канала: ' + str(i) + ' ' + str(code) + ' ' + char + ' ' + teststr
logging.warning(levent)
return False
rezult += chr(code)
i += 1
return rezult
# сборка имени юзера в читабельное
def set_name_printable(first, last, account=None, phone=None, ad=''):
name = str(first) + ' '
if last is not None:
name = name + str(last) + ' '
if account is not None:
name = name + '(' + ad + str(account) + ')' + ' '
if phone is not None:
name = name + '+' + str(phone) + ' '
ind = len(name)
res = name[0:ind - 1]
return res
# вывод числа с разделителями тысяч
def set_int_printable(integer, razd=' '):
string = '{:,}'.format(integer)
string = string.replace(',', razd)
return string
# версия сервера постгрес
def get_server_version(dbc=None):
srv = PgServer(dbc)
sql = "select setting from pg_config where name='VERSION'"
row = srv.exec(sql, return_type=1)
version = row[0]
return version
# создает обьект клиента с параметрами из бд
def get_tg_client(name, dbc=None, cust_name=None):
srv = PgServer(dbc)
n_api = name + '_api_id'
row = srv.exec('select value from parameters where name=%s', (n_api,), return_type=1)
api_id = int(row[0])
n_hash = name + '_api_hash'
row = srv.exec('select value from parameters where name=%s', (n_hash,), return_type=1)
api_hash = row[0]
if cust_name is None:
client = TelegramClient(name, api_id, api_hash)
else:
client = TelegramClient(cust_name, api_id, api_hash)
return client
# достает ключ бота из бд
def get_bot_key(name, dbc=None):
srv = PgServer(dbc)
bot = name + '_key'
row = srv.exec('select value from parameters where name=%s', (bot,), return_type=1)
bot_key = row[0]
return bot_key