from utils.texts import ( UNKNOWN_MAIN_CMD_TEXT, EMOJI_DIGITS, INSTRUCT_MAIN_MENU_TEXT, INSTRUCT_TRUECONF_TEXT, INSTRUCT_TRUECONF_WITH_AD, INSTRUCT_EMAIL_TEXT, ) import logging import asyncio from trueconf import Router, Message from utils.states import get_state, set_state, clear_state # πŸ”Œ Π˜ΠΌΠΏΠΎΡ€Ρ‚ΠΈΡ€ΡƒΠ΅ΠΌ Ρ†Π΅Π½Ρ‚Ρ€Π°Π»ΠΈΠ·ΠΎΠ²Π°Π½Π½ΡƒΡŽ Ρ„ΡƒΠ½ΠΊΡ†ΠΈΡŽ сбора статистики ΠΈΠ· main from utils.stats_logger import log_menu_stats logger = logging.getLogger(__name__) router = Router() def _get_user_cn(user_id: str) -> str: """ ΠŸΠΎΠ»ΡƒΡ‡ΠΈΡ‚ΡŒ CN ΠΏΠΎΠ»ΡŒΠ·ΠΎΠ²Π°Ρ‚Π΅Π»Ρ ΠΈΠ· AD ΠΏΠΎ Π΅Π³ΠΎ user_id (email ΠΈΠ»ΠΈ Π»ΠΎΠ³ΠΈΠ½). user_id ΠΈΠ· TrueConf β€” это ΠΎΠ±Ρ‹Ρ‡Π½ΠΎ email, Π½Π°ΠΏΡ€ΠΈΠΌΠ΅Ρ€ 'ds.krivochenko@sibcem.ru' """ from ldap3 import Server, Connection, ALL from config.config import AD_SERVER, AD_USER, AD_PASSWORD, AD_BASE from ldap3.utils.conv import escape_filter_chars try: server = Server(AD_SERVER, get_info=ALL) conn = Connection(server, user=AD_USER, password=AD_PASSWORD, auto_bind=True) safe_id = escape_filter_chars(user_id) short_username = user_id.split('@')[0] if '@' in user_id else user_id search_filter = ( f"(&(objectClass=user)(|" f"(mail={safe_id})" f"(userPrincipalName={safe_id})" f"(userPrincipalName={escape_filter_chars(f'{short_username}@sibcem.ru')}))" f"(sAMAccountName={escape_filter_chars(short_username)})))" ) conn.search(AD_BASE, search_filter, attributes=["cn", "sAMAccountName"]) if conn.entries: return str(conn.entries[0].cn) except Exception as e: logger.error(f"Ошибка получСния CN ΠΈΠ· AD для {user_id}: {e}") return user_id # fallback β€” ΠΈΡΠΏΠΎΠ»ΡŒΠ·ΡƒΠ΅ΠΌ user_id ΠΊΠ°ΠΊ CN @router.message() async def instruct_router_handler(msg: Message): user_id = msg.from_user.id current_state = get_state(user_id) if not msg.text: return cmd = msg.text.strip().lower() for raw_num, emoji_num in EMOJI_DIGITS.items(): if cmd == emoji_num: cmd = raw_num break if current_state is None: from utils.menu import MENU_MAP if MENU_MAP.get(cmd) == "INSTRUCT": msg.handled = True set_state(user_id, "INSTRUCT_MODE") await msg.answer(INSTRUCT_MAIN_MENU_TEXT, parse_mode="html") return return if current_state == "INSTRUCT_MODE": msg.handled = True if cmd == "0": log_menu_stats(user_id, "Π˜Π½ΡΡ‚Ρ€ΡƒΠΊΡ†ΠΈΠΈ", "Π’Ρ‹Ρ…ΠΎΠ΄ Π² Π³Π»Π°Π²Π½ΠΎΠ΅ мСню") clear_state(user_id) from utils.menu import MENU_TEXT await msg.answer(MENU_TEXT, parse_mode="html") elif cmd == "1": log_menu_stats(user_id, "Π˜Π½ΡΡ‚Ρ€ΡƒΠΊΡ†ΠΈΠΈ", "Trueconf") set_state(user_id, "INSTRUCT_TRUECONF_VIEW") cn = _get_user_cn(user_id) text = INSTRUCT_TRUECONF_WITH_AD(EMOJI_DIGITS, cn) await msg.answer(text, parse_mode="html") elif cmd == "2": log_menu_stats(user_id, "Π˜Π½ΡΡ‚Ρ€ΡƒΠΊΡ†ΠΈΠΈ", "ΠŸΠΎΡ‡Ρ‚Π°") set_state(user_id, "INSTRUCT_EMAIL_VIEW") await msg.answer(INSTRUCT_EMAIL_TEXT, parse_mode="html") else: await msg.answer(UNKNOWN_MAIN_CMD_TEXT, parse_mode="html") elif current_state in ["INSTRUCT_TRUECONF_VIEW", "INSTRUCT_EMAIL_VIEW"]: msg.handled = True if cmd == "0": clear_state(user_id) from utils.menu import MENU_TEXT await msg.answer(MENU_TEXT, parse_mode="html") elif cmd == "9": set_state(user_id, "INSTRUCT_MODE") await msg.answer(INSTRUCT_MAIN_MENU_TEXT, parse_mode="html") else: await msg.answer(UNKNOWN_MAIN_CMD_TEXT, parse_mode="html")