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@tcs.sibcem.ru' ЗамСняСт tcs.sibcem.ru Π½Π° sibcem.ru для поиска Π² AD. """ from ldap3 import Server, Connection, ALL, SUBTREE from config.config import AD_SERVER, AD_USER, AD_PASSWORD, AD_BASE from ldap3.utils.conv import escape_filter_chars try: # ЗамСняСм tcs.sibcem.ru Π½Π° sibcem.ru search_id = user_id.replace("@tcs.sibcem.ru", "@sibcem.ru") short_username = search_id.split("@")[0] if "@" in search_id else user_id safe_id = escape_filter_chars(search_id) safe_short = escape_filter_chars(short_username) logger.info(f"AD lookup: user_id={user_id}, search_id={search_id}, short={short_username}") server = Server(AD_SERVER, get_info=ALL) conn = Connection(server, user=AD_USER, password=AD_PASSWORD, auto_bind=True) # ΠŸΡ€ΠΎΠ±ΡƒΠ΅ΠΌ нСсколько Ρ„ΠΈΠ»ΡŒΡ‚Ρ€ΠΎΠ² + Ρ€Π°Π·Π½Ρ‹Π΅ Π±Π°Π·Ρ‹ поиска filters = [ f"(mail={safe_id})", f"(userPrincipalName={safe_id})", f"(sAMAccountName={safe_short})", ] # AD_BASE + ΠΊΠΎΡ€Π΅Π½ΡŒ Π΄ΠΎΠΌΠ΅Π½Π° для поиска Π²ΠΎ всём Π΄Π΅Ρ€Π΅Π²Π΅ search_bases = [AD_BASE, "DC=sibcem,DC=ru"] for base in search_bases: for filt in filters: conn.search( search_base=base, search_filter=f"(&(objectClass=user)({filt}))", attributes=["cn", "sAMAccountName", "mail", "userPrincipalName"], search_scope=SUBTREE ) if conn.entries: entry = conn.entries[0] cn = str(entry.cn) mail = str(entry.mail) if 'mail' in entry else "N/A" sam = str(entry.sAMAccountName) if 'sAMAccountName' in entry else "N/A" upn = str(entry.userPrincipalName) if 'userPrincipalName' in entry else "N/A" logger.info(f"AD found: cn={cn}, mail={mail}, sam={sam}, upn={upn}, base={base}") return cn logger.warning(f"AD not found for user_id={user_id}, search_id={search_id}, short={short_username}") except Exception as e: logger.error(f"Ошибка получСния CN ΠΈΠ· AD для {user_id}: {e}", exc_info=True) return user_id # fallback β€” ΠΈΡΠΏΠΎΠ»ΡŒΠ·ΡƒΠ΅ΠΌ user_id ΠΊΠ°ΠΊ CN @router.message() async def instruct_router_handler(msg: Message): user_id = msg.from_user.id user_name = getattr(msg.from_user, 'name', 'N/A') user_username = getattr(msg.from_user, 'username', 'N/A') logger.info(f"Instruct: user_id={user_id}, name={user_name}, username={user_username}") 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")