Создание git проекта
This commit is contained in:
48
python-bot/utils/admin_utils.py
Normal file
48
python-bot/utils/admin_utils.py
Normal file
@@ -0,0 +1,48 @@
|
||||
import os
|
||||
from utils.logger import logger
|
||||
#from utils.user_data import get_user_info
|
||||
#from utils.user_data import get_user_data
|
||||
from utils.message_sender import send_message
|
||||
|
||||
|
||||
def has_records(data,typer):
|
||||
for user_id, records in data.items():
|
||||
if typer in records:
|
||||
return True
|
||||
return False
|
||||
|
||||
def load_admins(): # utils/admin_utils.py
|
||||
admin_file= os.getenv('TELEGRAM_ADMINS')
|
||||
try:
|
||||
with open(admin_file, 'r') as file:
|
||||
return [line.strip() for line in file]
|
||||
except FileNotFoundError:
|
||||
logger.error("Файл с ID администраторов не найден")
|
||||
return []
|
||||
ADMINS = load_admins()
|
||||
def is_admin(user_id: str) -> bool: # utils/admin_utils.py
|
||||
admins = load_admins()
|
||||
return str(user_id) in admins
|
||||
|
||||
|
||||
def register_handlers(app):
|
||||
app.add_handler(CommandHandler("admin", admin))
|
||||
# другие базовые команды
|
||||
|
||||
def count_memorial_notes(data):
|
||||
total_count = 0
|
||||
|
||||
# Проходим по всем пользователям
|
||||
for user_id, user_data in data.items():
|
||||
# Считаем количество записок "О здравии"
|
||||
if "О здравии" in user_data:
|
||||
total_count += 1
|
||||
# Считаем количество записок "За упокой"
|
||||
if "За упокой" in user_data:
|
||||
total_count += 1
|
||||
|
||||
return total_count
|
||||
|
||||
# Пример использования:
|
||||
# result = count_memorial_notes(USER_DATA)
|
||||
# print(f"Общее количество записок: {result}")
|
||||
21
python-bot/utils/database.py
Normal file
21
python-bot/utils/database.py
Normal file
@@ -0,0 +1,21 @@
|
||||
import json, os
|
||||
from utils.logger import logger
|
||||
database_file = os.getenv('TELEGRAM_DATABASE')
|
||||
USER_DATA ={}
|
||||
def load_database(): # utils/database.py
|
||||
global USER_DATA
|
||||
try:
|
||||
with open(database_file, 'r') as db_file:
|
||||
USER_DATA = json.load(db_file)
|
||||
except (FileNotFoundError, json.JSONDecodeError):
|
||||
USER_DATA = {}
|
||||
|
||||
def save_database(text): #utils/database.py
|
||||
user_data = text
|
||||
try:
|
||||
logger.info(f"Сохраняемые данные: {USER_DATA}")
|
||||
with open(database_file, 'w') as db_file:
|
||||
json.dump(user_data, db_file, ensure_ascii=False, indent=4)
|
||||
except IOError as e:
|
||||
logger.error(f"Ошибка при сохранении базы данных: {e}")
|
||||
load_database()
|
||||
27
python-bot/utils/formating.py
Normal file
27
python-bot/utils/formating.py
Normal file
@@ -0,0 +1,27 @@
|
||||
def format_names(names, numbering_type='default'):
|
||||
"""
|
||||
Функция форматирует список имён с выбранной нумерацией
|
||||
|
||||
Параметры:
|
||||
names (list): список имён для форматирования
|
||||
numbering_type (str): тип нумерации ('default' - обычная нумерация, любой другой символ - использование этого символа)
|
||||
|
||||
Возвращает:
|
||||
str: отформатированную строку с именами
|
||||
"""
|
||||
|
||||
result = '' # создаем пустую строку для результата
|
||||
|
||||
# проверяем тип нумерации
|
||||
if not names:
|
||||
return "Пустая запись"
|
||||
if numbering_type == 'default':
|
||||
# обычная нумерация
|
||||
for i, name in enumerate(names, start=1):
|
||||
result += f'{i}. {name}\n'
|
||||
else:
|
||||
# используем указанный символ
|
||||
for name in names:
|
||||
result += f'{numbering_type} {name}\n'
|
||||
|
||||
return result.strip() # убираем последний лишний перенос строки
|
||||
17
python-bot/utils/logger.py
Normal file
17
python-bot/utils/logger.py
Normal file
@@ -0,0 +1,17 @@
|
||||
import logging
|
||||
|
||||
def setup_logging():
|
||||
logging.basicConfig(
|
||||
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
|
||||
level=logging.INFO,
|
||||
filename='logs/bot.log', # сохранение в файл
|
||||
filemode='a' # добавление в конец файла
|
||||
)
|
||||
console = logging.StreamHandler()
|
||||
console.setLevel(logging.INFO)
|
||||
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
|
||||
console.setFormatter(formatter)
|
||||
logging.getLogger('').addHandler(console)
|
||||
return logging.getLogger(__name__)
|
||||
|
||||
logger = setup_logging()
|
||||
53
python-bot/utils/message_sender.py
Normal file
53
python-bot/utils/message_sender.py
Normal file
@@ -0,0 +1,53 @@
|
||||
# utils/message_sender.py
|
||||
from utils.logger import logger
|
||||
from telegram import Update, Bot
|
||||
from telegram.ext import CallbackContext
|
||||
from telegram.error import TelegramError
|
||||
from telegram.constants import ParseMode
|
||||
|
||||
async def send_message(
|
||||
update: Update,
|
||||
context: CallbackContext,
|
||||
text: str,
|
||||
parse_mode: str = ParseMode.HTML,
|
||||
reply_markup=None,
|
||||
disable_web_page_preview: bool = False,
|
||||
chat_id: int = None,
|
||||
):
|
||||
try:
|
||||
if chat_id:
|
||||
# Отправка по указанному chat_id
|
||||
await context.bot.send_message(
|
||||
chat_id=chat_id,
|
||||
text=text,
|
||||
parse_mode=parse_mode,
|
||||
reply_markup=reply_markup,
|
||||
disable_web_page_preview=disable_web_page_preview
|
||||
)
|
||||
elif update.message:
|
||||
# Отправка в ответ на сообщение
|
||||
await update.message.reply_text(
|
||||
text=text,
|
||||
parse_mode=parse_mode,
|
||||
reply_markup=reply_markup,
|
||||
disable_web_page_preview=disable_web_page_preview
|
||||
)
|
||||
elif update.callback_query:
|
||||
# Отправка в ответ на callback-запрос
|
||||
await update.callback_query.message.reply_text(
|
||||
text=text,
|
||||
parse_mode=parse_mode,
|
||||
reply_markup=reply_markup,
|
||||
disable_web_page_preview=disable_web_page_preview
|
||||
)
|
||||
else:
|
||||
raise ValueError("Некорректный тип обновления")
|
||||
except TelegramError as e:
|
||||
logger.error(f"Ошибка при отправке сообщения: {e}")
|
||||
return False
|
||||
return True
|
||||
|
||||
# Вспомогательная функция для получения всех ID (пример)
|
||||
def get_all_user_ids():
|
||||
# Здесь логика получения всех ID из базы данных
|
||||
return [12345, 67890] # Пример списка ID
|
||||
44
python-bot/utils/parsing.py
Normal file
44
python-bot/utils/parsing.py
Normal file
@@ -0,0 +1,44 @@
|
||||
from telegram import Update
|
||||
from telegram.ext import CallbackContext
|
||||
from utils.prefixes import PREFIXES
|
||||
from utils.logger import logger
|
||||
from utils.database import USER_DATA
|
||||
|
||||
def parse_names_helper(text):
|
||||
words = text.split()
|
||||
names = []
|
||||
current_prefixes = []
|
||||
logger.info(f"Check prefixes: {PREFIXES}")
|
||||
for word in words:
|
||||
# Если слово является префиксом, добавляем его в список текущих >
|
||||
if word in PREFIXES:
|
||||
|
||||
current_prefixes.append(word)
|
||||
# Если слово не префикс - это имя
|
||||
else:
|
||||
# Если есть накопленные префиксы - соединяем их с именем
|
||||
if current_prefixes:
|
||||
names.append(" ".join(current_prefixes + [word.replace(',','')]))
|
||||
current_prefixes = []
|
||||
else:
|
||||
names.append(word.replace(',',''))
|
||||
logger.info(f"Check helper: {names}")
|
||||
return names
|
||||
|
||||
|
||||
async def notify_admins(update: Update, context: CallbackContext):
|
||||
bot = context.bot
|
||||
admin_ids = get_admin_ids() # Функция получения ID админов
|
||||
|
||||
unread_count = get_unread_messages_count() # Функция получения количества непрочитанных
|
||||
|
||||
if unread_count > 0:
|
||||
for admin_id in admin_ids:
|
||||
try:
|
||||
await bot.send_message(
|
||||
chat_id=admin_id,
|
||||
text=f"⚠️ У вас {unread_count} непрочитанных записок\n"
|
||||
f"От {get_users_count()} пользователей"
|
||||
)
|
||||
except TelegramError as e:
|
||||
logger.error(f"Ошибка отправки уведомления админу {admin_id}: {e}")
|
||||
9
python-bot/utils/prefixes.py
Normal file
9
python-bot/utils/prefixes.py
Normal file
@@ -0,0 +1,9 @@
|
||||
import os, json
|
||||
def load_prefixes(): #/utils/prefixes.py
|
||||
try:
|
||||
with open('./prefixes.json', 'r') as file:
|
||||
data = json.load(file)
|
||||
return set(data['prefixes'])
|
||||
except (FileNotFoundError, json.JSONDecodeError):
|
||||
return set()
|
||||
PREFIXES = load_prefixes()
|
||||
33
python-bot/utils/user_data.py
Normal file
33
python-bot/utils/user_data.py
Normal file
@@ -0,0 +1,33 @@
|
||||
from telegram import Update
|
||||
from telegram.ext import CallbackContext
|
||||
from utils.logger import logger
|
||||
# Объединяем функции get_user_name и get_user_id в одну
|
||||
async def get_user_info(update: Update, context: CallbackContext, id = None): #utils/utils.py
|
||||
try:
|
||||
if id:
|
||||
user = context.bot.get_chat(user_id)
|
||||
elif update.callback_query:
|
||||
user = update.callback_query.from_user
|
||||
elif update.message:
|
||||
user = update.message.from_user
|
||||
else:
|
||||
return {"id": "Unknown", "name": "Unknown"}
|
||||
|
||||
user_id = str(user.id)
|
||||
user_name = f"@{user.username}" if user.username else f"{user.first_name} {user.last_name or ''}".strip()
|
||||
|
||||
return {"id": user_id, "name": user_name}
|
||||
except Exception as e:
|
||||
logger.info (f"Ошибка получения данных пользователя: {e}")
|
||||
return {"id": "Unknown", "name": "Unknown"}
|
||||
|
||||
|
||||
def validate_input(text):
|
||||
if len(text) > 6000:
|
||||
raise ValueError("Слишком длинный текст")
|
||||
if not text:
|
||||
raise ValueError("Пустой ввод")
|
||||
return text
|
||||
|
||||
def format_names(names):
|
||||
return "\n".join(f"- {name}" for name in names)
|
||||
Reference in New Issue
Block a user