refactoring

This commit is contained in:
Rykov7
2023-01-28 01:48:05 +03:00
parent 0f44c90c4a
commit b40a22bc3d
10 changed files with 153 additions and 143 deletions
+1
View File
@@ -0,0 +1 @@
from .commands import service
+19 -113
View File
@@ -1,17 +1,15 @@
import shelve import shelve
from datetime import datetime as dt
from flask import request, abort from flask import request, abort
import threading import threading
from time import sleep from time import sleep
from . import stan from . import stan
from . import reminder
from . import reloader
from . import report from . import report
from . import get
from . import rules from . import rules
from .helpers import get_me, represent_as_get, detect_args
from .helpers import represent_as_get, detect_args
from .filters import * from .filters import *
from .config import * from .config import *
@@ -107,30 +105,6 @@ def translate_layout(message):
bot.send_message(message.chat.id, message.reply_to_message.text.translate(ENG_RUS_TABLE)) bot.send_message(message.chat.id, message.reply_to_message.text.translate(ENG_RUS_TABLE))
@bot.message_handler(commands=['me'])
def command_me(message):
""" Send info about user and chat id [Service]. """
bot.send_message(message.chat.id, get_me(message))
@bot.message_handler(commands=['remind'])
def remind_manually(message):
""" Remind holidays manually. """
args = message.text.split()
if len(args) > 1:
try:
today = dt.strptime(args[1], "%m-%d-%Y")
except ValueError as ve:
bot.send_message(message.chat.id, f"Не удалось разобрать дату!\n{ve}")
else:
reminder.remind(message.chat.id, today)
else:
bot.send_message(message.chat.id, f"<b>Формат даты: MM-DD-YYYY</b>\n\n"
f"Примеры:\n"
f"/remind 09-12-2024\n"
f"/remind 09-13-2022")
@bot.message_handler(commands=['quote']) @bot.message_handler(commands=['quote'])
def stan_speak(message): def stan_speak(message):
bot.send_message(message.chat.id, stan.speak(0)) bot.send_message(message.chat.id, stan.speak(0))
@@ -196,48 +170,11 @@ def google_it(message: types.Message):
send_or_reply(message, f'<i>Ищем «{query}» в Гугле...</i>', reply_markup=markup) send_or_reply(message, f'<i>Ищем «{query}» в Гугле...</i>', reply_markup=markup)
@bot.message_handler(func=is_nongrata)
def tease_nongrata(message: types.Message):
""" Reply to non grata mentions. """
bot.reply_to(message, f'у нас тут таких не любят')
"""
[ INLINE ]
"""
@bot.inline_handler(lambda query: True)
def default_query(inline_query):
""" Inline the Zen of Python. """
zen = []
for id_p, phrase in enumerate(ZEN):
q = inline_query.query.casefold()
if phrase.casefold().startswith(q) or ' ' + q in phrase.casefold():
zen.append(types.InlineQueryResultArticle(
f"{id_p}", f'The Zen of Python #{id_p + 1}', types.InputTextMessageContent(
f"<i>{phrase}</i>"), description=phrase))
bot.answer_inline_query(inline_query.id, zen, cache_time=1200)
""" """
[ ADMIN PANEL ] [ ADMIN PANEL ]
""" """
@bot.message_handler(func=is_admin, commands=['ip'])
def get_ip(message):
bot.send_message(message.chat.id, get.my_ip())
@bot.message_handler(func=is_admin, commands=['reload'])
def send_stats(message):
logging.warning('Reloading...')
reloader.reload_modules()
bot.send_message(message.chat.id, 'Reloaded successfully')
@bot.message_handler(func=is_admin, commands=['ddel']) @bot.message_handler(func=is_admin, commands=['ddel'])
def delete_user(message: types.Message): def delete_user(message: types.Message):
if message.reply_to_message: if message.reply_to_message:
@@ -265,58 +202,27 @@ def unban_user(message: types.Message):
logging.warning(f'[UNBAN (M)] {user_id}') logging.warning(f'[UNBAN (M)] {user_id}')
@bot.message_handler(func=is_admin, commands=['jobs']) """
def list_jobs(message): [ INLINE ]
""" List all the jobs in schedule. """ """
bot.send_message(ADMIN_ID, reminder.print_get_jobs())
@bot.message_handler(func=is_white, commands=['add']) @bot.inline_handler(lambda query: True)
def add_stan_quote(message): def default_query(inline_query):
if message.reply_to_message and message.reply_to_message.text: """ Inline the Zen of Python. """
with open('Stan.txt', 'a', encoding='utf8') as stan_quotes: zen = []
if message.reply_to_message.text not in (i.rstrip() for i in open('Stan.txt', 'r', encoding='utf8')): for id_p, phrase in enumerate(ZEN):
stan_quotes.write(message.reply_to_message.text.replace('\n', ' ') + '\n') q = inline_query.query.casefold()
bot.send_message(message.chat.id, if phrase.casefold().startswith(q) or ' ' + q in phrase.casefold():
'✅ <b>Добавил</b>\n └ <i>' + message.reply_to_message.text.replace("\n", zen.append(types.InlineQueryResultArticle(
" ") + '</i>') f"{id_p}", f'The Zen of Python #{id_p + 1}', types.InputTextMessageContent(
else: f"<i>{phrase}</i>"), description=phrase))
bot.send_message(message.chat.id,
f'⛔️ <b>Не добавил</b>, есть токое\n └ <i>{message.reply_to_message.text}</i>')
bot.answer_inline_query(inline_query.id, zen, cache_time=1200)
@bot.message_handler(func=is_white, commands=['remove'])
def remove_stan_quote(message):
if message.reply_to_message and message.reply_to_message.text:
if message.reply_to_message.text in (i.rstrip() for i in open('Stan.txt', 'r', encoding='utf8')):
quotes = list(open('Stan.txt', 'r', encoding='utf8'))
with open('Stan.txt', 'w', encoding='utf8') as stan_quotes:
quotes.remove(message.reply_to_message.text + '\n')
stan_quotes.writelines(quotes)
bot.send_message(message.chat.id, f'✅ <b>Удалил</b>\n └ <i>{message.reply_to_message.text}</i>')
else:
bot.send_message(message.chat.id, f'⛔️ <b>Нет такого</b>\n └ <i>{message.reply_to_message.text}</i>')
@bot.message_handler(func=is_admin, commands=['stats'])
def send_stats(message: types.Message):
if len(message.text.split()) == 1:
rep = report.create_report_text(message.chat.id)
if rep:
bot.send_message(message.chat.id, rep)
else:
bot.send_message(message.chat.id, report.create_report_text(message.text.split()[-1]))
@bot.message_handler(func=is_admin, commands=['reset_stats'], chat_types=['supergroup', 'group'])
def send_stats(message: types.Message):
logging.warning('reset_stats')
report.reset_report_stats(message.chat.id)
bot.send_message(message.chat.id, report.reset_report_stats(message.chat.id))
""" """
[ MAIN MESSAGE HANDLER ] [ COUNTER ]
""" """
@@ -354,7 +260,7 @@ def handle_msg(message: types.Message):
@app.route(f"/bot{TOKEN}/", methods=['POST']) @app.route(f"/bot{TOKEN}/", methods=['POST'])
def bot_webhook(): def webhook():
""" Parse POST requests from Telegram. """ """ Parse POST requests from Telegram. """
if request.headers.get('content-type') == 'application/json': if request.headers.get('content-type') == 'application/json':
json_string = request.get_data().decode('utf-8') json_string = request.get_data().decode('utf-8')
+25
View File
@@ -0,0 +1,25 @@
from urllib.request import urlopen
def my_ip():
ip_service = 'https://icanhazip.com'
ip = urlopen(ip_service, )
return ip.read().decode('utf-8')
def me(message):
msg = f"<b>Telegram ID:</b> <code>{message.from_user.id}</code>\n\
├ <b>Is bot:</b> {message.from_user.is_bot}\n\
├ <b>First name:</b> {message.from_user.first_name}\n"
if message.from_user.last_name:
msg += f" ├ <b>Last name:</b> {message.from_user.last_name}\n"
if message.from_user.username:
msg += f" ├ <b>Username:</b> {message.from_user.username}\n"
msg += f" ├ <b>Language code:</b> {message.from_user.language_code}\n"
if message.from_user.is_premium is True:
msg += f" ├ <b>Is Premium</b>: ⭐️\n"
else:
msg += f" ├ <b>Is Premium</b>: No\n"
msg += f" ├ <b>Chat ID:</b> {message.chat.id}"
return msg
+70
View File
@@ -0,0 +1,70 @@
""" Service commands for Admin purposes. """
import logging
from datetime import datetime as dt
from ..config import bot, ADMIN_ID, types
from .. import reminder
from .. import reloader
from .. import report
from ..filters import is_admin
from .get import me, my_ip
@bot.message_handler(commands=['me'])
def command_me(message):
""" Send info about user and chat id [Service]. """
bot.send_message(message.chat.id, me(message))
@bot.message_handler(func=is_admin, commands=['ip'])
def get_ip(message):
bot.send_message(message.chat.id, my_ip())
@bot.message_handler(commands=['remind'])
def remind_manually(message):
""" Remind holidays manually. """
args = message.text.split()
if len(args) > 1:
try:
today = dt.strptime(args[1], "%m-%d-%Y")
except ValueError as ve:
bot.send_message(message.chat.id, f"Не удалось разобрать дату!\n{ve}")
else:
reminder.remind(message.chat.id, today)
else:
bot.send_message(message.chat.id, f"<b>Формат даты: MM-DD-YYYY</b>\n\n"
f"Примеры:\n"
f"/remind 09-12-2024\n"
f"/remind 09-13-2022")
@bot.message_handler(func=is_admin, commands=['jobs'])
def list_jobs(message):
""" List all the jobs in schedule. """
bot.send_message(ADMIN_ID, reminder.print_get_jobs())
@bot.message_handler(func=is_admin, commands=['stats'])
def send_stats(message: types.Message):
if len(message.text.split()) == 1:
rep = report.create_report_text(message.chat.id)
if rep:
bot.send_message(message.chat.id, rep)
else:
bot.send_message(message.chat.id, report.create_report_text(message.text.split()[-1]))
@bot.message_handler(func=is_admin, commands=['reset_stats'], chat_types=['supergroup', 'group'])
def send_stats(message: types.Message):
logging.warning('reset_stats')
report.reset_report_stats(message.chat.id)
bot.send_message(message.chat.id, report.reset_report_stats(message.chat.id))
@bot.message_handler(func=is_admin, commands=['reload'])
def send_stats(message):
logging.warning('Reloading...')
reloader.reload_modules()
logging.warning('Reloaded!!!')
bot.send_message(message.chat.id, 'Reloaded successfully')
+1 -1
View File
@@ -8,7 +8,7 @@ from dotenv import load_dotenv
logging.basicConfig(level=logging.INFO, format='%(asctime)s %(message)s', datefmt='%Y.%m.%d %H:%M:%S') logging.basicConfig(level=logging.INFO, format='%(asctime)s %(message)s', datefmt='%Y.%m.%d %H:%M:%S')
logging.critical('STAN STARTED!') logging.critical('STAN STARTED!')
logger.setLevel(logging.DEBUG) logger.setLevel(logging.INFO)
load_dotenv() load_dotenv()
TOKEN = os.environ.get('LUTZPYBOT', 'Token not in ENVIRON') TOKEN = os.environ.get('LUTZPYBOT', 'Token not in ENVIRON')
-7
View File
@@ -1,7 +0,0 @@
from urllib.request import urlopen
def my_ip():
ip_service = 'https://icanhazip.com'
ip = urlopen(ip_service, )
return ip.read().decode('utf-8')
-18
View File
@@ -3,24 +3,6 @@ from urllib import parse
from telebot import types from telebot import types
def get_me(message):
msg = f"<b>Telegram ID:</b> <code>{message.from_user.id}</code>\n\
├ <b>Is bot:</b> {message.from_user.is_bot}\n\
├ <b>First name:</b> {message.from_user.first_name}\n"
if message.from_user.last_name:
msg += f" ├ <b>Last name:</b> {message.from_user.last_name}\n"
if message.from_user.username:
msg += f" ├ <b>Username:</b> {message.from_user.username}\n"
msg += f" ├ <b>Language code:</b> {message.from_user.language_code}\n"
if message.from_user.is_premium is True:
msg += f" ├ <b>Is Premium</b>: ⭐️\n"
else:
msg += f" ├ <b>Is Premium</b>: No\n"
msg += f" ├ <b>Chat ID:</b> {message.chat.id}"
return msg
def represent_as_get(message: types.Message): def represent_as_get(message: types.Message):
return parse.quote_plus(detect_args(message)) return parse.quote_plus(detect_args(message))
+2 -1
View File
@@ -3,7 +3,7 @@ from importlib import reload
from . import bot from . import bot
from . import filters from . import filters
from . import config from . import config
from . import get from .commands import get, service
from . import helpers from . import helpers
from . import report from . import report
from . import rules from . import rules
@@ -14,6 +14,7 @@ def reload_modules():
reload(bot) reload(bot)
reload(filters) reload(filters)
reload(get) reload(get)
reload(service)
reload(config) reload(config)
reload(helpers) reload(helpers)
reload(report) reload(report)
-3
View File
@@ -9,9 +9,6 @@ from .config import PYTHONCHATRU, bot
from . import report from . import report
# https://core.telegram.org/bots/api Telegram Bot API
def scheduler(): def scheduler():
while True: while True:
run_pending() run_pending()
+35
View File
@@ -1,8 +1,43 @@
""" Stan quote generator. """ """ Stan quote generator. """
import random import random
from .config import bot, types
from .filters import is_white, is_nongrata
def speak(chance_of): def speak(chance_of):
number = random.randint(0, chance_of) number = random.randint(0, chance_of)
if number == 0: if number == 0:
return random.choice([i.rstrip() for i in open('Stan.txt', 'r', encoding='utf8') if i]) return random.choice([i.rstrip() for i in open('Stan.txt', 'r', encoding='utf8') if i])
@bot.message_handler(func=is_white, commands=['add'])
def add_stan_quote(message):
if message.reply_to_message and message.reply_to_message.text:
with open('Stan.txt', 'a', encoding='utf8') as stan_quotes:
if message.reply_to_message.text not in (i.rstrip() for i in open('Stan.txt', 'r', encoding='utf8')):
stan_quotes.write(message.reply_to_message.text.replace('\n', ' ') + '\n')
bot.send_message(message.chat.id,
'✅ <b>Добавил</b>\n └ <i>' + message.reply_to_message.text.replace("\n",
" ") + '</i>')
else:
bot.send_message(message.chat.id,
f'⛔️ <b>Не добавил</b>, есть токое\n └ <i>{message.reply_to_message.text}</i>')
@bot.message_handler(func=is_white, commands=['remove'])
def remove_stan_quote(message):
if message.reply_to_message and message.reply_to_message.text:
if message.reply_to_message.text in (i.rstrip() for i in open('Stan.txt', 'r', encoding='utf8')):
quotes = list(open('Stan.txt', 'r', encoding='utf8'))
with open('Stan.txt', 'w', encoding='utf8') as stan_quotes:
quotes.remove(message.reply_to_message.text + '\n')
stan_quotes.writelines(quotes)
bot.send_message(message.chat.id, f'✅ <b>Удалил</b>\n └ <i>{message.reply_to_message.text}</i>')
else:
bot.send_message(message.chat.id, f'⛔️ <b>Нет такого</b>\n └ <i>{message.reply_to_message.text}</i>')
@bot.message_handler(func=is_nongrata)
def tease_nongrata(message: types.Message):
""" Reply to non grata mentions. """
bot.reply_to(message, f'у нас тут таких не любят')