dddennnisss/trueconf_bot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: lk_diagnostic.py - use AD + correct SQL columns

Browse Source
This commit is contained in:
Денис Кривоченко
2026-08-11 11:30:16 +07:00
parent d2e7dd6234
commit f0d927bcb6
1 changed files with 166 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lk/test/lk_diagnostic.py
+165 -64
View File
@@ -3,11 +3,11 @@
import os
import sys
import pymssql
BASE_DIR = "/opt/trueconf_bot"
os.chdir(BASE_DIR)
# Загрузка .env
def load_env(filepath):
with open(filepath, "r", encoding="utf-8") as f:
for line in f:
@@ -21,6 +21,7 @@ def load_env(filepath):
load_env(os.path.join(BASE_DIR, "config", ".env"))
# Загрузка passwork
for pw_path in ["/opt/passwork", os.path.join(BASE_DIR, "passwork")]:
if os.path.isdir(pw_path) and pw_path not in sys.path:
sys.path.insert(0, pw_path)
@@ -35,66 +36,166 @@ DB_PASSWORD = card_data["password"]
SQL_SERVER = os.getenv("SQL_SERVER", "SRVKEM-MOBILEIN.sibcem.ru")
SQL_DB_NAME = os.getenv("SQL_DB_NAME", "BossCopy")
conn = pymssql.connect(
import pymssql
import ldap3
# --- Подключение к AD ---
print("=" * 60)
print("Диагностика сотрудника: la.guseynova@sibcem.ru")
print("=" * 60)
# Ищем в AD
email = "la.guseynova@sibcem.ru"
search_id = email.replace("@tcs.sibcem.ru", "@sibcem.ru")
short_name = search_id.split("@")[0]
print(f"\n--- 0. Поиск в AD ---")
print(f" Email: {email}")
print(f" Search ID: {search_id}")
print(f" Short: {short_name}")
AD_SERVER = "172.16.20.20"
AD_BASE = "DC=sibcem,DC=ru"
try:
conn = ldap3.Connection(
ldap3.Server(AD_SERVER, connect_timeout=5),
auto_bind=True,
read_only=True
)
# Ищем по mail
conn.search(
search_base=AD_BASE,
search_filter=f"(mail={search_id})",
attributes=["cn", "sAMAccountName", "mail", "userPrincipalName", "extensionAttribute2", "mobile", "personalPager"]
)
if conn.entries:
entry = conn.entries[0]
cn = str(entry.cn)
sam = str(entry.sAMAccountName)
mail = str(entry.mail) if 'mail' in entry else ""
upn = str(entry.userPrincipalName) if 'userPrincipalName' in entry else ""
ext2 = str(entry.extensionAttribute2) if 'extensionAttribute2' in entry else ""
mobile = str(entry.mobile) if 'mobile' in entry else ""
pager = str(entry.personalPager) if 'personalPager' in entry else ""
print(f" CN: {cn}")
print(f" sAMAccountName: {sam}")
print(f" mail: {mail}")
print(f" UPN: {upn}")
print(f" extensionAttribute2 (CARD_ID): {ext2}")
print(f" mobile: {mobile}")
print(f" pager: {pager}")
else:
print(" Не найден по mail. Пробую по sAMAccountName...")
conn.search(
search_base=AD_BASE,
search_filter=f"(sAMAccountName={short_name})",
attributes=["cn", "sAMAccountName", "mail", "userPrincipalName", "extensionAttribute2", "mobile", "personalPager"]
)
if conn.entries:
entry = conn.entries[0]
cn = str(entry.cn)
sam = str(entry.sAMAccountName)
mail = str(entry.mail) if 'mail' in entry else ""
upn = str(entry.userPrincipalName) if 'userPrincipalName' in entry else ""
ext2 = str(entry.extensionAttribute2) if 'extensionAttribute2' in entry else ""
mobile = str(entry.mobile) if 'mobile' in entry else ""
pager = str(entry.personalPager) if 'personalPager' in entry else ""
print(f" CN: {cn}")
print(f" sAMAccountName: {sam}")
print(f" mail: {mail}")
print(f" UPN: {upn}")
print(f" extensionAttribute2 (CARD_ID): {ext2}")
print(f" mobile: {mobile}")
print(f" pager: {pager}")
else:
print(" НЕ НАЙДЕН в AD!")
conn.unbind()
sys.exit(0)
conn.unbind()
except Exception as e:
print(f" Ошибка подключения к AD: {e}")
sys.exit(0)
# --- Подключение к SQL ---
print(f"\n--- 1. Подключение к SQL ---")
try:
conn_sql = pymssql.connect(
server=SQL_SERVER,
user=DB_USER,
password=DB_PASSWORD,
database=SQL_DB_NAME,
charset="cp1251",
login_timeout=10
)
cursor = conn.cursor()
email = "la.guseynova@sibcem.ru"
print("=" * 60)
print(f"Diagnostika sotrudnika: {email}")
print("=" * 60)
# 1. CARD_ID
print("\n--- 1. CARD_ID (iz PR_CARD + PR_EMP) ---")
cursor.execute(
"SELECT c.ID, e.EMAIL_ID FROM UOV_SELFSERVICE_PR_CARD c "
"JOIN UOV_SELFSERVICE_PR_EMP e ON e.CARD_ID = c.ID "
"WHERE LOWER(e.EMAIL_ID) LIKE LOWER(%s)",
(f"%{email.split('@')[0]}@",)
)
card_rows = cursor.fetchall()
if not card_rows:
cursor.execute(
"SELECT c.ID, e.EMAIL_ID FROM UOV_SELFSERVICE_PR_CARD c "
"JOIN UOV_SELFSERVICE_PR_EMP e ON e.CARD_ID = c.ID "
"WHERE LOWER(e.EMAIL_ID) LIKE LOWER(%s)",
(f"%{email.split('@')[0]}%",)
)
card_rows = cursor.fetchall()
if card_rows:
for r in card_rows:
print(f" CARD_ID: {r[0]}, EMAIL: {r[1]}")
card_id = card_rows[0][0]
else:
print(f" NE NAIDEN po email {email}")
conn.close()
cursor = conn_sql.cursor()
print(f" Подключено к {SQL_SERVER}/{SQL_DB_NAME}")
except Exception as e:
print(f" Ошибка подключения к SQL: {e}")
sys.exit(0)
print(f"\n Ispol'zuem CARD_ID: {card_id}")
# Определяем card_id: из AD (extensionAttribute2) или через поиск в БД
card_id = None
# 2. PR_EMP
print("\n--- 2. PR_EMP (profil sotrudnika) ---")
# Пробуем найти CARD_ID через PR_CARD + PR_EMP
print(f"\n--- 2. Поиск CARD_ID в PR_CARD ---")
# Сначала ищем по short_name в PR_EMP
cursor.execute(
"SELECT ID, FIRM_ID, EMAIL_ID FROM UOV_SELFSERVICE_PR_EMP WHERE CARD_ID = %s",
"SELECT c.ID FROM UOV_SELFSERVICE_PR_CARD c "
"JOIN UOV_SELFSERVICE_PR_EMP e ON e.CARD_ID = c.ID "
"WHERE e.EMAIL_ID LIKE %s OR e.USERNAME LIKE %s",
(f"%{short_name}%", f"%{short_name}%")
)
rows = cursor.fetchall()
if rows:
for r in rows:
print(f" CARD_ID (по username): {r[0]}")
card_id = rows[0][0]
else:
print(f" Не найден по username в PR_CARD/PR_EMP")
# Если не нашли, пробуем по email
if not card_id:
cursor.execute(
"SELECT c.ID FROM UOV_SELFSERVICE_PR_CARD c "
"JOIN UOV_SELFSERVICE_PR_EMP e ON e.CARD_ID = c.ID "
"WHERE e.EMAIL_ID LIKE %s",
(f"%{search_id.split('@')[0]}%",)
)
rows = cursor.fetchall()
if rows:
for r in rows:
print(f" CARD_ID (по email): {r[0]}")
card_id = rows[0][0]
else:
print(f" Не найден по email в PR_CARD/PR_EMP")
if not card_id:
print("\n CARD_ID не найден в БД!")
conn_sql.close()
sys.exit(0)
print(f"\n Используем CARD_ID: {card_id}")
# --- 3. PR_EMP ---
print(f"\n--- 3. PR_EMP (профиль сотрудника) ---")
cursor.execute(
"SELECT ID, FIRM_ID, EMAIL_ID, USERNAME FROM UOV_SELFSERVICE_PR_EMP WHERE CARD_ID = %s",
(card_id,)
)
emp_rows = cursor.fetchall()
if emp_rows:
for r in emp_rows:
print(f" EMP_ID: {r[0]}, FIRM_ID: {r[1]}, EMAIL: {r[2]}")
print(f" EMP_ID: {r[0]}, FIRM_ID: {r[1]}, EMAIL: {r[2]}, USERNAME: {r[3]}")
else:
print(" PUSTO - net zapisey v PR_EMP")
print(" ПУСТО - нет записей в PR_EMP")
# 3. PR_TRANS
print("\n--- 3. PR_TRANS (kadrovye naznacheniya) ---")
# --- 4. PR_TRANS ---
print(f"\n--- 4. PR_TRANS (кадровые назначения) ---")
if emp_rows:
emp_ids = [r[0] for r in emp_rows]
placeholders = ",".join(["%s"] * len(emp_ids))
@@ -107,16 +208,16 @@ if emp_rows:
)
trans_rows = cursor.fetchall()
if trans_rows:
print(f" Naideno zapisей: {len(trans_rows)}")
print(f" Найдено записей: {len(trans_rows)}")
for r in trans_rows[:10]:
print(f" EMP_ID: {r[0]}, D_FROM: {r[1]}, D_TO: {r[2]}, TP: {r[3]}, SYSPAY: {r[4]}")
else:
print(" PUSTO - net zapisey v PR_TRANS")
print(" ПУСТО - нет записей в PR_TRANS")
else:
print(" Skip (net PR_EMP)")
print(" Пропуск (нет PR_EMP)")
# 4. SL_YTAX
print("\n--- 4. SL_YTAX (periody s dannymi) ---")
# --- 5. SL_YTAX ---
print(f"\n--- 5. SL_YTAX (периоды с данными) ---")
cursor.execute(
"SELECT DISTINCT TYEAR, TMONTH, CMONTH FROM UOV_SELFSERVICE_SL_YTAX "
"WHERE CARD_ID = %s ORDER BY CMONTH DESC",
@@ -124,14 +225,14 @@ cursor.execute(
)
tax_rows = cursor.fetchall()
if tax_rows:
print(f" Naideno periodov: {len(tax_rows)}")
print(f" Найдено периодов: {len(tax_rows)}")
for r in tax_rows[:10]:
print(f" {r[0]}-{r[1]:02d} (CMONTH={r[2]})")
else:
print(" PUSTO - net periodov v SL_YTAX")
print(" ПУСТО - нет периодов в SL_YTAX")
# 5. Proverka q_firm
print("\n--- 5. Proverka q_firm dlya poslednego perioda ---")
# --- 6. Проверка q_firm ---
print(f"\n--- 6. Проверка q_firm для последнего периода ---")
if tax_rows:
latest_cmonth = tax_rows[0][2]
tyear = (latest_cmonth - 1) // 12
@@ -142,7 +243,7 @@ if tax_rows:
date_end = f"{tyear}-{tmonth:02d}-{last_day:02d}"
date_params = (date_start, date_start, date_end, date_end, date_end)
print(f" Period: {tyear}-{tmonth:02d}")
print(f" Период: {tyear}-{tmonth:02d}")
q_firm = """
select e.FIRM_ID, f.SNAME from (select e.id, e.FIRM_ID from UOV_SELFSERVICE_PR_EMP e
@@ -156,9 +257,9 @@ if tax_rows:
cursor.execute(q_firm, (card_id, *date_params))
firm_row = cursor.fetchone()
if firm_row:
print(f" q_firm NAISHEL: FIRM_ID={firm_row[0]}, SNAME={firm_row[1]}")
print(f" q_firm НАШЁЛ: FIRM_ID={firm_row[0]}, SNAME={firm_row[1]}")
else:
print(f" q_firm NE NAISHEL (NULL) - eto vyzovet oshibku!")
print(f" q_firm НЕ НАШЁЛ (NULL) - это вызовет ошибку!")
q_firm_fb = """
select e.FIRM_ID from UOV_SELFSERVICE_PR_EMP e
join UOV_SELFSERVICE_PR_CARD c on c.ID=e.CARD_ID
@@ -168,23 +269,23 @@ if tax_rows:
cursor.execute(q_firm_fb, (card_id,))
fb_row = cursor.fetchone()
if fb_row:
print(f" Fallbek NAISHEL: FIRM_ID={fb_row[0]}")
print(f" Фоллбэк НАШЁЛ: FIRM_ID={fb_row[0]}")
else:
print(f" Fallbek NE NAISHEL - kriticheskaya oshibka!")
print(f" Фоллбэк НЕ НАШЁЛ - критическая ошибка!")
# 6. Svodka
print("\n--- 6. Svodka ---")
# --- 7. Сводка ---
print(f"\n--- 7. Сводка ---")
cursor.execute(
"SELECT COUNT(DISTINCT CMONTH), MIN(CMONTH), MAX(CMONTH) FROM UOV_SELFSERVICE_SL_YTAX WHERE CARD_ID = %s",
(card_id,)
)
summary = cursor.fetchone()
if summary:
print(f" Periodov v SL_YTAX: {summary[0]}")
print(f" Diapazon: {summary[1]} - {summary[2]}")
print(f" Периодов в SL_YTAX: {summary[0]}")
print(f" Диапазон: {summary[1]} - {summary[2]}")
print("\n" + "=" * 60)
print("Dagnostika zavershena")
print(f"\n{'=' * 60}")
print("Диагностика завершена")
print("=" * 60)
conn.close()
conn_sql.close()