ramjm
/
voice_logger
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Изменил(а) на 'main.py'

Browse Source
main
ramjm 2 months ago
parent ee7b4d96a5
commit cd1ce7e5cc
1 changed files with 223 additions and 67 deletions
Show Stats Download Patch File Download Diff File

290
main.py
Unescape View File

@ -7,6 +7,7 @@ import pytz # Импортируем pytz
import os
from dotenv import load_dotenv
from pathlib import Path
dotenv_path = f"{Path(__file__).parent.resolve()}/.env"
load_dotenv(dotenv_path=dotenv_path)
@ -17,14 +18,16 @@ intents.message_content = True
intents.members = True
bot = commands.Bot(command_prefix='!', intents=intents)
bot.remove_command('help')
# Путь к файлу для хранения истории
history_file = 'logs/voice_history.json'
# Устанавливаем временную зону для Москвы
moscow_tz = pytz.timezone('Europe/Moscow')
backup_file = f'logs/voice_history_{datetime.datetime.now(moscow_tz).date()}.json'
# Устанавливаем временную зону для Киева
kiev_tz = pytz.timezone('Europe/Kiev')
backup_file = f'logs/voice_history_{datetime.datetime.now(kiev_tz).date()}.json'
# Словарь для хранения истории подключений и отключений пользователей по серверам и каналам
voice_history = {}
cname = os.getenv("cname")
#===================[WORK WITH DB]======================================
def load_voice_history():
@ -59,103 +62,256 @@ async def on_ready():
print(f'Бот {bot.user} запущен!')
# Настраиваем планировщик
scheduler = AsyncIOScheduler()
scheduler.add_job(backup_voice_history, 'cron', hour=17, minute=47) # Запланировать на 00:00
scheduler.add_job(backup_voice_history, 'cron', hour=21, minute=00) # Запланировать на 00:00
scheduler.start()
@bot.event
async def on_voice_state_update(member, before, after):
guild_id = str(member.guild.id) # Получаем ID сервера как строку
channel_id = str(after.channel.id) if after.channel else str(before.channel.id) if before.channel else None
# Проверка на включение логирования
if not voice_history[guild_id]["status"]: return
channel_id = str(after.channel.id) if after.channel else str(before.channel.id) if before.channel else None
# Инициализируем историю для сервера, если её нет
if guild_id not in voice_history:
voice_history[guild_id] = {}
# Инициализируем историю для канала, если её нет
if channel_id not in voice_history[guild_id]:
voice_history[guild_id][channel_id] = []
voice_history[guild_id][channel_id] = {} # Убедитесь, что это словарь
# Если пользователь подключился к голосовому каналу
if before.channel is None and after.channel is not None:
join_time = datetime.datetime.now(moscow_tz) # Получаем текущее время в Москве
voice_history[guild_id][channel_id].append({
'user_id': member.id,
'join_time': join_time.isoformat(), # Сохраняем время в формате ISO
'leave_time': None
})
join_time = datetime.datetime.now(kiev_tz) # Получаем текущее время в Киеве
user_id_str = str(member.id)
# Проверяем, есть ли уже запись для этого пользователя
if user_id_str not in voice_history[guild_id][channel_id]:
voice_history[guild_id][channel_id][user_id_str] = {
'user_id': member.id,
'join_time': join_time.isoformat(), # Сохраняем время в формате ISO
'leave_time': None,
'duration': 0 # Инициализируем продолжительность
}
else:
record = voice_history[guild_id][channel_id][user_id_str]
duration = record["duration"]
# Если leave_time уже установлен, создаем новую запись
voice_history[guild_id][channel_id][user_id_str] = {
'user_id': member.id,
'join_time': join_time.isoformat(),
'duration': duration,
'leave_time': None,
}
save_voice_history() # Сохраняем историю после обновления
# Если пользователь отключился от голосового канала
elif before.channel is not None and after.channel is None:
leave_time = datetime.datetime.now(moscow_tz) # Получаем текущее время в Москве
for record in voice_history[guild_id][channel_id]:
if record['user_id'] == member.id and record['leave_time'] is None:
record['leave_time'] = leave_time.isoformat() # Обновляем последнее подключение
break
save_voice_history() # Сохраняем историю после обновления
leave_time = datetime.datetime.now(kiev_tz) # Получаем текущее время в Киеве
user_id_str = str(member.id)
if user_id_str in voice_history[guild_id][channel_id]:
record = voice_history[guild_id][channel_id][user_id_str]
record['leave_time'] = leave_time.isoformat() # Обновляем время выхода
# Обновляем продолжительность
join_time = datetime.datetime.fromisoformat(record['join_time'])
duration = (leave_time - join_time).total_seconds() # Вычисляем продолжительность
record['duration'] += duration # Суммируем продолжительность
save_voice_history() # Сохраняем историю после обновления
@bot.command(name="clear")
@commands.has_role(os.getenv("role")) # Замените "Admin" на название вашей роли
async def clear_voice_history(ctx):
"""Команда для очистки базы данных голосовой истории."""
if (ctx.channel).name != cname: return
global voice_history
guild_id = str(ctx.guild.id)
voice_history[guild_id] = {"status": voice_history[guild_id]["status"]} # Очищаем историю
save_voice_history() # Сохраняем изменения в файл
await ctx.send("История голосовых подключений очищена.")
# Обработчик ошибок для команды delete
@clear_voice_history.error
async def clear_voice_history_error(ctx, error):
if (ctx.channel).name != cname: return
if isinstance(error, commands.MissingRole):
await ctx.send("У вас нет прав для выполнения этой команды.")
@bot.command(name="start")
async def start_logging(ctx):
"""Команда для начала логирования голосовых каналов."""
if (ctx.channel).name != cname: return
global voice_history
guild_id = str(ctx.guild.id)
voice_history[guild_id]["status"] = True
save_voice_history()
await ctx.send("Логирование голосовых каналов включено.")
@bot.command(name="stop")
async def stop_logging(ctx):
"""Команда для остановки логирования голосовых каналов."""
if (ctx.channel).name != cname: return
global voice_history
guild_id = str(ctx.guild.id)
voice_history[guild_id]["status"] = False
save_voice_history()
await ctx.send("Логирование голосовых каналов отключено.")
@bot.command(name="status")
async def status_logging(ctx):
"""Команда для получения статуса логирования голосовых каналов."""
if (ctx.channel).name != cname: return
global voice_history
guild_id = str(ctx.guild.id)
if guild_id not in list(voice_history.keys()): voice_history[guild_id] = {}
if "status" not in list(voice_history[guild_id].keys()): voice_history[guild_id]["status"] = False
save_voice_history()
await ctx.send("Логирование голосовых каналов включено." if voice_history[guild_id]["status"] else "Логирование голосовых каналов выключено.")
@bot.command(name='help', help="Команда для показа этого сообщения.")
async def custom_help(ctx):
help_message = "Список доступных команд:\n>>> "
for command in bot.commands:
help_message += f"**!{command.name}** - {command.help}\n"
await ctx.send(help_message)
@bot.command(name="log")
async def _log(ctx, channel_name: str):
"""Команда для вывода логов о подключениях и отключениях пользователей в определённом голосовом канале за день."""
async def _log(ctx):
"""Команда для вывода логов о подключениях и отключениях пользователей во всех голосовых каналах за день."""
if (ctx.channel).name != cname: return
load_voice_history()
guild_id = str(ctx.guild.id) # Получаем ID сервера как строку
# Получаем список всех голосовых каналов на сервере
voice_channels = {channel.name: str(channel.id) for channel in ctx.guild.voice_channels}
# Проверяем, существует ли канал с таким именем
channel_id = voice_channels.get(channel_name)
if not channel_id:
await ctx.send(f"Канал с именем '{channel_name}' не найден.")
return
# Проверяем, есть ли записи для этого канала в истории
if guild_id not in voice_history or channel_id not in voice_history[guild_id]:
# Проверяем, есть ли записи для этого сервера в истории
if guild_id not in voice_history:
await ctx.send("Нет записей о подключениях и отключениях.")
return
today = datetime.datetime.now(moscow_tz).date()
today = datetime.datetime.now(kiev_tz).date()
log_messages = []
for record in voice_history[guild_id][channel_id]:
user_id = record['user_id']
join_time = datetime.datetime.fromisoformat(record['join_time']) # Преобразуем обратно в datetime
leave_time = record['leave_time']
try:
member = await ctx.guild.fetch_member(user_id) # Получаем участника по ID
member_mention = member.mention
except discord.NotFound:
await ctx.send('Пользователь не найден.')
continue # Пропускаем, если пользователь не найден
except discord.Forbidden:
await ctx.send('У меня нет прав для получения информации о пользователе.')
continue # Пропускаем, если нет прав
except discord.HTTPException:
await ctx.send('Произошла ошибка при получении информации о пользователе.')
continue # Пропускаем, если произошла ошибка
if leave_time is None: # Если пользователь все еще в голосовом канале
duration = datetime.datetime.now(moscow_tz) - join_time
log_messages.append(
f"{member_mention} подключен к каналу с {join_time.strftime('%Y-%m-%d %H:%M:%S')}\n"\
f"(продолжительность: {duration})\n")
else:
leave_time = datetime.datetime.fromisoformat(leave_time) # Преобразуем обратно в datetime
if leave_time.date() == today: # Проверяем, что отключение произошло сегодня
duration = leave_time - join_time
log_messages.append(
f"{member_mention} подключен к каналу с {join_time.strftime('%Y-%m-%d %H:%M:%S')}\n"\
f"отключен с {leave_time.strftime('%Y-%m-%d %H:%M:%S')}"\
f"\n(продолжительность: {duration})\n"
)
# Проходим по всем голосовым каналам на сервере
for channel_id, records in voice_history[guild_id].items():
if channel_id != "status":
for user_id_str, record in records.items(): # Исправлено: итерируем по элементам словаря
user_id = record['user_id']
join_time = datetime.datetime.fromisoformat(record['join_time']) # Преобразуем обратно в datetime
leave_time = record['leave_time']
try:
member = await ctx.guild.fetch_member(user_id) # Получаем участника по ID
member_mention = member.mention
except discord.NotFound:
continue # Пропускаем, если пользователь не найден
except discord.Forbidden:
continue # Пропускаем, если нет прав
except discord.HTTPException:
continue # Пропускаем, если произошла ошибка
channel = bot.get_channel(int(channel_id))
channel_name = channel.name
if leave_time is None: # Если пользователь все еще в голосовом канале
# Получаем продолжительность из базы данных
db_duration = record['duration'] # Продолжительность в секундах из базы данных (тип float)
current_duration = datetime.datetime.now(kiev_tz) - join_time # Текущая продолжительность
total_duration = db_duration + current_duration.total_seconds() # Суммируем продолжительности
# Округляем общую продолжительность до ближайшей секунды
rounded_duration = round(total_duration)
hours, remainder = divmod(rounded_duration, 3600)
minutes, seconds = divmod(remainder, 60)
log_messages.append(
f"{member_mention} подключен к каналу {channel_name} с {join_time.strftime('%Y-%m-%d %H:%M:%S')}\n"\
f"(продолжительность: {hours}ч {minutes}м {seconds}с)\n"
)
else:
leave_time = datetime.datetime.fromisoformat(leave_time) # Преобразуем обратно в datetime
if leave_time.date() == today: # Проверяем, что отключение произошло сегодня
duration = record["duration"]
# Округляем продолжительность до ближайшей секунды
rounded_duration = round(duration)
hours, remainder = divmod(rounded_duration, 3600)
minutes, seconds = divmod(remainder, 60)
log_messages.append(
f"{member_mention} подключен к каналу {channel_name} с {join_time.strftime('%Y-%m-%d %H:%M:%S')}\n"\
f"(продолжительность: {hours}ч {minutes}м {seconds}с)\n"
)
if log_messages:
await ctx.send("\n".join(log_messages))
else:
await ctx.send(f"Нет записей о подключениях и отключениях в канале '{channel_name}' за сегодня.")
await ctx.send("Нет записей о подключениях и отключениях за сегодня.")
@bot.command(name="send")
async def _send(ctx):
"""Команда для отправки файла логов о подключениях и отключениях пользователей во всех голосовых каналах за день."""
if (ctx.channel).name != cname: return
load_voice_history()
guild_id = str(ctx.guild.id) # Получаем ID сервера как строку
# Проверяем, есть ли записи для этого сервера в истории
if guild_id not in voice_history:
await ctx.send("Нет записей о подключениях и отключениях.")
return
today = datetime.datetime.now(kiev_tz).date()
log_messages = []
# Проходим по всем голосовым каналам на сервере
for channel_id, records in voice_history[guild_id].items():
if channel_id != "status":
for user_id_str, record in records.items(): # Исправлено: итерируем по элементам словаря
user_id = record['user_id']
join_time = datetime.datetime.fromisoformat(record['join_time']) # Преобразуем обратно в datetime
leave_time = record['leave_time']
try:
member = await ctx.guild.fetch_member(user_id) # Получаем участника по ID
m_name = member.name; m_d_name = member.display_name
except discord.NotFound:
continue # Пропускаем, если пользователь не найден
except discord.Forbidden:
continue # Пропускаем, если нет прав
except discord.HTTPException:
continue # Пропускаем, если произошла ошибка
channel = bot.get_channel(int(channel_id))
channel_name = channel.name
if leave_time is None: # Если пользователь все еще в голосовом канале
# Получаем продолжительность из базы данных
db_duration = record['duration'] # Продолжительность в секундах из базы данных (тип float)
current_duration = datetime.datetime.now(kiev_tz) - join_time # Текущая продолжительность
total_duration = db_duration + current_duration.total_seconds() # Суммируем продолжительности
# Округляем общую продолжительность до ближайшей секунды
rounded_duration = round(total_duration)
hours, remainder = divmod(rounded_duration, 3600)
minutes, seconds = divmod(remainder, 60)
log_messages.append(
f"@{m_name} ({m_d_name}) подключен к каналу {channel_name} с {join_time.strftime('%Y-%m-%d %H:%M:%S')}\n"\
f"(продолжительность: {hours}ч {minutes}м {seconds}с)\n"
)
else:
leave_time = datetime.datetime.fromisoformat(leave_time) # Преобразуем обратно в datetime
if leave_time.date() == today: # Проверяем, что отключение произошло сегодня
duration = record["duration"]
# Округляем продолжительность до ближайшей секунды
rounded_duration = round(duration)
hours, remainder = divmod(rounded_duration, 3600)
minutes, seconds = divmod(remainder, 60)
log_messages.append(
f"@{m_name} ({m_d_name}) подключен к каналу {channel_name} с {join_time.strftime('%Y-%m-%d %H:%M:%S')}\n"\
f"(продолжительность: {hours}ч {minutes}м {seconds}с)\n"
)
if log_messages:
with open(f"logs/{guild_id}.txt", "w", encoding="utf-8") as f:
f.write("".join(log_messages))
with open(f"logs/{guild_id}.txt", "r") as f:
await ctx.send("Вот ваш файл:", file=discord.File(f, 'message.txt'))
os.system(f"rm logs/{guild_id}.txt")
else:
await ctx.send("Нет записей о подключениях и отключениях за сегодня.")
bot.run(os.getenv("token"))
bot.run(os.getenv("token"))
Write Preview
Loading…
Cancel
Save