ChioHub/libs/chat_control.py

"""
Дополниельные компоненты для работы плагина chat_control.

- Получени информации о чате.
- Исключение пользователя из чата.
- Управление предупредждениями пользователя
- Отображение динамики изменений показателей чата.

Author: Milinuri Nirvalen
Ver: 2.4 (13)
"""

from chiori import Config
from chiori import Event
from chiori import Context
from libs.vk import get_profiles

import os
from typing import Optional


chats = "data/chats/"

"""
Описание данных чата
====================
	banned: ID участников в чёрном списке чата с указанием времени возврата
	warns: ID участников с предупреждениями, с указанием причины
	last_messages: Время последних сообщений от пользователей
		Количество сообщений в минуту от пользователя
	marriages: Информация о браках: Кто с кем, когда, код редактирования
	m_requests: Отправленные запросы на вступление в брак
	counters: Подсчёт кол-ва слов и сообщений от пользователей за 300 дней
	timers: Время до следующего дневного/недельного обновления счётчиков
	words: Список запрешённых регулярных выражений чата
	edit_code: Код для редактирования данных беседы
	holy_members: ID свяых участников и время снятия их роли
	tokens: Количество жетонов активности каждого участника за 7 дней
	tokens_history: Информация о динамике изменения жетонов активности
		за 14 дней
	silent_mide: Настройка "тихого режима"
	leawes: ID вышедших учстников с указанием даты выхода из беседы
	rules: Сохранённые правила беседыы
"""

chat_data = {
	"banned":{}, "warns":{},
	"last_nessages":{}, "marriages":[], "m_requests":{},
	"counters":{}, "timers":{"day":0, "week":0},
	"words":[], "edit_code":"000000", "holy_members":{}, "tokens":{},
	"tokens_history":[ [0, 0, 0] for x in range(15) ],
	"silent_mode":True, "leaves":{}, "rules":""
}


# Получени данных о чате
# ======================

def get_chat_ids() -> list[str]:
	"""Возвращает ID всех зарегистрированных чатов."""
	return [x.split('.')[0] for x in os.listdir(chats)]

def get_chat(event: Event) -> Config:
	"""Получает данные чата.

	Args:
	    event (Event): Экземпляр события Чио

	Returns:
	    Config: Экземпляр Config с данными чата
	"""

	c = Config(filepath=f"{chats}mirror.toml").file_data
	is_mirror = False

	# Является ли чат зеркалом
	if str(event.get("to.id")) in c:
		cid = c[str(event.get("to.id"))]
		is_mirror = True
	else:
		cid = str(event.get("to.id"))

	# Данные о чате и зеркале
	chat = Config(None, chat_data, filepath=f"{chats}{cid}.toml")
	if chat.file_data:
		chat.file_data["cid"] = cid
		chat.file_data["is_mirror"] = is_mirror

		for k, v in chat_data.items():
			if k not in chat.file_data:
				chat.file_data[k] = v

	return chat


# Работы с временем
# =================

def count_time(seconds: int) -> list[int]:
	"""Переводит секунды в днич, часы, минуты, секунды.

	Args:
	    seconds (int): Колчиество секунд

	Returns:
	    list[int]: Спиок времени [Дней, Часов, Минут, Секунд]
	"""

	if seconds > 0:
		d, h = divmod(seconds, 86400)
		h, m = divmod(h, 3600)
		m, s = divmod(m, 60)
		res = [d, h, m, s]
	else:
		res = [0, 0, 0, 0]

	return res

def get_str_time(time_list: list[int] | int) -> str:
	"""Преобразует список времени в строку прошедшего времени.

	Args:
	    time_list (list[int]): [Дней, Часов, Минут, Секунд]

	Returns:
	    str: Строка с указанием прошедшего времени
	"""

	# Если нам передали число секунд
	if isinstance(time_list, int):
		time_list = count_time(time_list)

	if not sum(time_list[:-1]) and time_list[-1] < 5:
		return "Только что"

	res = ""

	if time_list[0] > 0:
		res += f"{time_list[0]} д."

	res += f" {time_list[1]} ч."
	res += f" {time_list[2]}" if time_list[2] > 0 else "00"

	if time_list[3] > 0:
		res += f":{time_list[3]}"

	return res


# Работа с чатом
# ==============

async def check_chat(e: Event, ctx: Context, prior: Optional[int]=2) -> bool:
	"""проверяющая функции чата.

	Args:
	    e (Event): Экзмепляр события Чио
	    ctx (Context): Экземпляр контекста события
	    prior (int, optional): Требуемый уровень допуска:
	    	0: Все пользователи.
	    	1: Администраторы привязанных чатов.
	    	2: Администраторы зеркал, локальных чатов.

	Returns:
	    bool: True если чат настроен и вы его администратор
	"""

	if e.get("to.is_admin") and e.get("to.is_chat"):
		c = get_chat(e)
		if not str(e.get("to.id")) in get_chat_ids():
			c.save()

		if not prior:
			return True
		elif prior == 1 and e.get("from.is_admin") or e.get("level") == 10:
			return True
		elif not c.file_data["is_mirror"] and prior == 2 and e.get("from.is_admin"):
			return True

async def kick(e: Event, ctx: Context, uid: list[str]) -> str:
	"""Исключает участника из чата.

	Args:
	    e (Event): Экземпляр события
	    ctx (Context): Экземпляр контекста события
	    uid (str): Список ID участиков для исключения из чата

	Returns:
	    str: Сообщение со статусов исключения
	"""

	if isinstance(uid, (int, str)):
		uid = [uid]

	profiles = await get_profiles(e, ctx, uid)
	text = ""

	for k, v in profiles.items():
		if v.get("is_admin"):
			text +=  f"\n[{v['name']}] Администратор чата."
			continue

		res = await ctx.request("messages.removeChatUser",
			chat_id=e.get("to.id") - 2000000000, member_id=k
		)

		if res["error"]:
			text += f"\n{v['name']} {res['response']['error_msg']}"
		else:
			text += f"\b{v['name']} исключён."

	return text

async def add_warn(c: Config, profiles: dict, time: int, uid: list[str],
	arg="Причина не указана.") -> str:
	"""Выдаёт предупреждение участнику чата.

	Args:
	    c (Config): Экземпляр настроек чата
	    profiles (dict): Словарь профилей участников чата
	    time (int): Время снятия предупреждения
	    uid (list): Список ID участников для выдачи предупреждения
	    arg (str, optional): Причина выдачи предупреждения

	Returns:
	    str: Сообщение со статусов выдачи предупреждений
	"""

	text = ""
	for x in uid:
		if x not in profiles:
			text += f"\n-- [{x}] Не участник чата."
			continue

		pr = profiles[x]

		if pr.get("is_admin"):
			text += f"\n-- {pr['name']} администратор чата."
			continue

		if x in c.file_data["holy_members"]:
			text += f"\n-- {pr['name']} неприкосновенный(ая)."
			continue

		if len(c.file_data["warns"].get(x, [])) == 5:
			text += f"\n-- {pr['name']} достиг своего предела."
			continue

		if x in c.file_data["warns"]:
			c.file_data["warns"][x][str(time)] = arg
		else:
			c.file_data["warns"][x] = {str(time):arg}

		warns = len(c.file_data["warns"][x])
		text += f"\n-- {pr['name']} выдано предупрждение ({warns}/5)\n> {arg}"

	c.save()
	return text

async def remove_warn(c: Config, profiles: dict, uid: list,
	count: Optional[int] = 1) -> str:
	"""Удаляет предупредждения у участников чата.

	Args:
	    c (Config): Экземпляр данных чата
	    profiles (dict): Словрь профилей участников чата
	    uid (list): Список ID участников для удаления предупреждений
	    count (int, optional): Количество удаляемых предупреждений

	Returns:
	    str: Описание
	"""

	text = ""
	for x in uid:
		if x not in profiles:
			text += f"\n-- [{x}] Нет в списке предупрждений"
			continue

		pr = profiles[x]

		if c not in c.file_data["warns"]:
			text += f"\n-- [{pr['name']}] Нет предупреждений"
			continue

		elif count >= len(c.file_data["warns"].get(x, [])):
			del c.file_data["warns"][x]
			text += f"\n-- [{pr['name']}] Сняты все предупрждения"
			continue

		else:
			for i, k in enumerate(c.file_data["warns"][x]):
				del c.file_data["warns"][x][k]
				if i == count:
					break

			warns = len(c.file_data["warns"][x])
			text += f'\n-- [{pr["name"]}]: -{count} ({warns}/5 пр.)'

	c.save()
	return text


# Динамика изменения жетонов
# ==========================

def get_diff(a: int, b: int) -> str:
	if a == b:
		return "0"
	return f"🔻{a-b}" if a > b else f"🔺️{b-a}"

def tokens_log(c) -> dict:
	text = ""
	data = {"min_tokens": min(map(lambda x: x[0], c)) or 1,
			"max_tokens": max(map(lambda x: x[0], c)) or 1,
			"avarage_tokens": round(sum(map(lambda x: x[0], c))  / 15) or 1,

			"min_lose": min(map(lambda x: x[1], c)) or 1,
			"max_lose": max(map(lambda x: x[1], c)) or 1,
			"avarage_lose": round(sum(map(lambda x: x[1], c))  / 15) or 1,

			"min_added": min(map(lambda x: x[2], c)) or 1,
			"max_added": max(map(lambda x: x[2], c)) or 1,
			"avarage_added": round(sum(map(lambda x: x[2], c))  / 15) or 1}

	for x in c:
		hp = round(x[0] / data["max_tokens"] * 100, 2)
		text += "\n-- ["

		if x[0] == data["max_tokens"]:
			text += "❤"

		if x[1] == data["max_lose"]:
			text += "🔶"

		if x[2] == data["max_added"]:
			text += "🔷️"

		if x[0]:
			text += f"{hp}%; {get_diff(x[1], x[2])}] {x[0]} (-{x[1]}; +{x[2]})"

	data["text"] = text
	return data