diff --git a/check_card_id.py b/check_card_id.py new file mode 100644 index 0000000..59e04fb --- /dev/null +++ b/check_card_id.py @@ -0,0 +1,154 @@ +#!/usr/bin/env python3 +""" +Проверочный скрипт: user_id -> CARD_ID -> телефон. + +Интерактивный режим: после каждого запроса спрашивает, +хочешь ли продолжить. + +config.py сам подтягивает все секреты из Passwork при импорте. +""" + +import sys +import re +import logging + +# ============================================================ +# config.py при импорте заполняет globals(): +# AD_USER, AD_PASSWORD, AD_SERVER, AD_BASES +# DB_USER, DB_PASSWORD, SQL_SERVER, SQL_DB_NAME +# ============================================================ +sys.path.insert(0, "/opt/trueconf_bot") +from config import config + +logging.basicConfig(level=logging.INFO, format="%(message)s") +log = logging.getLogger(__name__) + + +# ============================================================ +# AD: user_id -> CARD_ID +# ============================================================ +def get_card_id_from_ad(user_id: str) -> str: + """Ищем пользователя в AD по всем OU из AD_BASES, возвращаем extensionAttribute2 (CARD_ID).""" + from utils.ad_search import search_by_user_id + + entries = search_by_user_id(user_id, ["extensionAttribute2"]) + + if not entries: + raise Exception("Пользователь не найден в AD") + + entry = entries[0] + + if "extensionAttribute2" in entry and entry.extensionAttribute2.value: + card_id = str(entry.extensionAttribute2.value).strip() + log.info("CARD_ID: {}".format(card_id)) + return card_id + + raise Exception("extensionAttribute2 пуст для этого пользователя в AD") + + +# ============================================================ +# SQL: CARD_ID -> телефон в PR_CARD +# ============================================================ +def get_phone_from_db(card_id: str) -> str: + """Ищем сотрудника в UOV_SELFSERVICE_PR_CARD по CARD_ID.""" + import pymssql + + conn = pymssql.connect( + server=config.SQL_SERVER, + user=config.DB_USER, + password=config.DB_PASSWORD, + database=config.SQL_DB_NAME, + charset="cp1251", + ) + cursor = conn.cursor() + + query = """ + SELECT c.ID, c.PHONE, c.F_NAME, c.L_NAME, c.M_NAME + FROM UOV_SELFSERVICE_PR_CARD c + WHERE c.ID = %s + """ + cursor.execute(query, (card_id,)) + row = cursor.fetchone() + conn.close() + + if not row: + raise Exception("CARD_ID {} не найден в PR_CARD".format(card_id)) + + card_id_db = row[0] + phone = str(row[1]) if row[1] else "Не указан" + f_name = str(row[2]) if row[2] else "" + l_name = str(row[3]) if row[3] else "" + m_name = str(row[4]) if row[4] else "" + + log.info("PR_CARD.ID: {}".format(card_id_db)) + log.info("ФИО: {} {} {}".format(l_name, f_name, m_name)) + log.info("PHONE: {}".format(phone)) + return phone + + +# ============================================================ +# MAIN +# ============================================================ +def run_one(): + """Выполнить один запрос: ввод user_id -> CARD_ID -> телефон.""" + user_id = input("\nВведите user_id (email или sAMAccountName): ").strip() + if not user_id: + print("user_id пустой, выход.") + sys.exit(1) + + print() + print("[1] Поиск CARD_ID в Active Directory") + print(" Сервер: {}".format(config.AD_SERVER)) + print(" Базы поиска: {}".format(config.AD_BASES)) + print(" user_id: {}".format(user_id)) + print() + + try: + card_id = get_card_id_from_ad(user_id) + except Exception as e: + print("Ошибка AD: {}".format(e)) + return None + + print() + print("[2] Поиск телефона в базе BossCopy") + print(" Сервер: {}".format(config.SQL_SERVER)) + print(" CARD_ID: {}".format(card_id)) + print() + + try: + phone = get_phone_from_db(card_id) + except Exception as e: + print("Ошибка SQL: {}".format(e)) + return card_id + + print() + print("=" * 70) + print(" ИТОГ") + print("=" * 70) + print(" user_id: {}".format(user_id)) + print(" CARD_ID: {}".format(card_id)) + print(" Телефон: {}".format(phone)) + print("=" * 70) + print() + + return card_id + + +def main(): + print() + print("=" * 70) + print(" Проверка: user_id -> CARD_ID -> телефон") + print("=" * 70) + print() + + while True: + run_one() + + again = input("Продолжить? (y/n): ").strip().lower() + if again not in ("y", "yes", "д", "да"): + print("Готово.") + break + + +if __name__ == "__main__": + main()