add: check_card_id.py — проверка user_id -> CARD_ID -> телефон

Использует AD_BASES через utils.ad_search.search_by_user_id
2026-07-22 18:32:35 +07:00
parent c76914444c
commit 064451dc7c
1 changed files with 154 additions and 0 deletions
@@ -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()