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

fix: lk_diagnostic.py - check q_firm for ALL periods

Browse Source
This commit is contained in:
Денис Кривоченко
2026-08-11 12:56:52 +07:00
parent 7c2a3f1c4a
commit 941cd569d8
1 changed files with 192 additions and 172 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lk/test/lk_diagnostic.py
+163 -143
View File
@@ -1,222 +1,216 @@
#!/usr/bin/env python3 #!/usr/bin/env python3
"""Диагностика данных сотрудника для модуля ЛК (расчетные листки).""" """Диагностика данных сотрудника для модуля ЛК (расчетnye listki)."""
import calendar
import os import os
import sys import sys
# ============================================================
# 1. Загрузка .env и путей
# ============================================================
BASE_DIR = "/opt/trueconf_bot" BASE_DIR = "/opt/trueconf_bot"
if os.path.exists(BASE_DIR):
os.chdir(BASE_DIR) os.chdir(BASE_DIR)
else:
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
# Загрузка .env
def load_env(filepath): def load_env(filepath):
try:
with open(filepath, "r", encoding="utf-8") as f: with open(filepath, "r", encoding="utf-8") as f:
for line in f: for line in f:
line = line.strip() line = line.strip()
if line and not line.startswith("#") and "=" in line: if line and not line.startswith("#") and "=" in line:
key, value = line.split("=", 1) key, value = line.split("=", 1)
os.environ[key.strip()] = value.strip(' "\'\r\n') value = value.strip(' "')
except FileNotFoundError: value = value.strip("'")
print(f"⚠️ Файл {filepath} не найден.") value = value.strip()
os.environ[key.strip()] = value
load_env(os.path.join(BASE_DIR, "config", ".env")) load_env(os.path.join(BASE_DIR, "config", ".env"))
# ============================================================ # Загрузка passwork
# 2. Загрузка паролей из Passwork (AD и SQL) for pw_path in ["/opt/passwork", os.path.join(BASE_DIR, "passwork")]:
# ============================================================
CREDENTIALS_MAP = {
"AD_PASSWORD": os.getenv("PW_ID_AD", "").strip(' "\'\r\n'),
"DB_PASSWORD": os.getenv("PW_ID_SQL", "").strip(' "\'\r\n'),
}
passwork_paths = [
"/opt/passwork",
os.path.join(BASE_DIR, "passwork"),
os.path.join(BASE_DIR, "config", "passwork"),
]
for pw_path in passwork_paths:
if os.path.isdir(pw_path) and pw_path not in sys.path: if os.path.isdir(pw_path) and pw_path not in sys.path:
sys.path.insert(0, pw_path) sys.path.insert(0, pw_path)
if BASE_DIR not in sys.path:
sys.path.insert(0, BASE_DIR)
AD_USER, AD_PASSWORD, DB_USER, DB_PASSWORD = "", "", "", ""
try:
from passwork import get_passwork_secrets from passwork import get_passwork_secrets
required_cards = [name for name in CREDENTIALS_MAP.values() if name] card_name = os.getenv("PW_ID_SQL", "").strip()
passwork_pool = get_passwork_secrets(required_cards=required_cards) passwork_pool = get_passwork_secrets(required_cards=[card_name])
card_data = passwork_pool[card_name]
ad_card_name = CREDENTIALS_MAP.get("AD_PASSWORD")
if ad_card_name and ad_card_name in passwork_pool:
card_ad = passwork_pool[ad_card_name]
AD_USER = card_ad.get("login", "").strip(' "\'\r\n')
AD_PASSWORD = card_ad.get("password", "").strip(' "\'\r\n')
sql_card_name = CREDENTIALS_MAP.get("DB_PASSWORD")
if sql_card_name and sql_card_name in passwork_pool:
card_sql = passwork_pool[sql_card_name]
DB_USER = card_sql.get("login", "").strip(' "\'\r\n')
DB_PASSWORD = card_sql.get("password", "").strip(' "\'\r\n')
except Exception as e:
print(f"⚠️ Ошибка загрузки из Passwork: {e}")
if not AD_USER: AD_USER = os.getenv("AD_USER", "")
if not AD_PASSWORD: AD_PASSWORD = os.getenv("AD_PASSWORD", "")
if not DB_USER: DB_USER = os.getenv("DB_USER", "")
if not DB_PASSWORD: DB_PASSWORD = os.getenv("DB_PASSWORD", "")
DB_USER = card_data["login"]
DB_PASSWORD = card_data["password"]
SQL_SERVER = os.getenv("SQL_SERVER", "SRVKEM-MOBILEIN.sibcem.ru") SQL_SERVER = os.getenv("SQL_SERVER", "SRVKEM-MOBILEIN.sibcem.ru")
SQL_DB_NAME = os.getenv("SQL_DB_NAME", "BossCopy") SQL_DB_NAME = os.getenv("SQL_DB_NAME", "BossCopy")
AD_SERVER = os.getenv("AD_SERVER", "172.16.20.20")
AD_BASE = os.getenv("AD_BASE", "DC=sibcem,DC=ru")
import ldap3
import pymssql import pymssql
import ldap3
import calendar
# --- Подключение к AD ---
print("=" * 60)
print("Диагностика сотрудника: la.guseynova@sibcem.ru")
print("=" * 60)
# ============================================================
# 3. Поиск сотрудника в Active Directory
# ============================================================
email = "la.guseynova@sibcem.ru" email = "la.guseynova@sibcem.ru"
search_id = email.replace("@tcs.sibcem.ru", "@sibcem.ru") search_id = email.replace("@tcs.sibcem.ru", "@sibcem.ru")
short_name = search_id.split("@")[0] short_name = search_id.split("@")[0]
print("=" * 60) print(f"\n--- 0. Поиск в AD ---")
print(f"Диагностика сотрудника: {email}") print(f" Email: {email}")
print("=" * 60) print(f" Search ID: {search_id}")
print(f" Short (без домена): {short_name}")
print("\n--- 0. Поиск в AD ---") AD_SERVER = "172.16.20.20"
card_id_ad = None AD_BASE = "DC=sibcem,DC=ru"
try: try:
server = ldap3.Server(AD_SERVER, connect_timeout=5)
bind_user = AD_USER
if bind_user and "@" not in bind_user and "\\" not in bind_user:
bind_user = f"{bind_user}@sibcem.ru"
conn = ldap3.Connection( conn = ldap3.Connection(
server, ldap3.Server(AD_SERVER, connect_timeout=5),
user=bind_user,
password=AD_PASSWORD,
auto_bind=True, auto_bind=True,
auto_referrals=False read_only=True
) )
attributes_list = ["cn", "sAMAccountName", "mail", "userPrincipalName", "proxyAddresses", "extensionAttribute2"] search_filter = f"(|(mail={search_id}*)(sAMAccountName={short_name})(userPrincipalName={search_id}*)(anr={short_name}))"
search_filter = f"(|(sAMAccountName={short_name})(mail={search_id})(userPrincipalName={search_id})(proxyAddresses=*:{search_id}*)(proxyAddresses=*{short_name}*)(anr={short_name}))" print(f" LDAP-фильтр: {search_filter}")
conn.search( conn.search(
search_base=AD_BASE, search_base=AD_BASE,
search_filter=search_filter, search_filter=search_filter,
attributes=attributes_list attributes=["cn", "sAMAccountName", "mail", "userPrincipalName", "extensionAttribute2", "mobile", "personalPager"]
) )
if conn.entries: if conn.entries:
entry = conn.entries[0] entry = conn.entries[0]
cn = str(entry.cn) cn = str(entry.cn)
sam = str(entry.sAMAccountName) if 'sAMAccountName' in entry else "" sam = str(entry.sAMAccountName)
mail_val = str(entry.mail) if 'mail' in entry else "" mail = str(entry.mail) if 'mail' in entry else ""
upn = str(entry.userPrincipalName) if 'userPrincipalName' in entry else "" upn = str(entry.userPrincipalName) if 'userPrincipalName' in entry else ""
card_id_ad = str(entry.extensionAttribute2) if 'extensionAttribute2' in entry else None 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(" ✅ НАЙДЕН В AD!") print(f"\n НАЙДЕН:")
print(f" CN: {cn}") print(f" CN: {cn}")
print(f" sAMAccountName: {sam}") print(f" sAMAccountName: {sam}")
print(f" mail: {mail_val}") print(f" mail: {mail}")
print(f" UPN: {upn}") print(f" UPN: {upn}")
print(f" extensionAttribute2 (CARD_ID из AD): {card_id_ad}") print(f" extensionAttribute2 (CARD_ID): {ext2}")
print(f" mobile: {mobile}")
print(f" pager: {pager}")
else: else:
print(" ❌ НЕ НАЙДЕН в AD!") print(" Не найден!")
conn.unbind()
sys.exit(0)
conn.unbind() conn.unbind()
except Exception as e: except Exception as e:
print(f" ❌ Ошибка обращения к AD: {e}") print(f" Ошибка подключения к AD: {e}")
sys.exit(0)
# ============================================================ # --- Подключение к SQL ---
# 4. Подключение к SQL Server print(f"\n--- 1. Подключение к SQL ---")
# ============================================================
print("\n--- 1. Подключение к SQL Server ---")
try: try:
conn_sql = pymssql.connect( conn_sql = pymssql.connect(
server=SQL_SERVER, server=SQL_SERVER,
user=DB_USER, user=DB_USER,
password=DB_PASSWORD, password=DB_PASSWORD,
database=SQL_DB_NAME, database=SQL_DB_NAME,
charset='cp1251', charset="cp1251",
login_timeout=10 login_timeout=10
) )
cursor = conn_sql.cursor() cursor = conn_sql.cursor()
print(" ✅ Подключено к SQL Server!") print(f" Подключено к {SQL_SERVER}/{SQL_DB_NAME}")
except Exception as e: except Exception as e:
print(f" ❌ Ошибка подключения к SQL: {e}") print(f" Ошибка подключения к SQL: {e}")
sys.exit(1) sys.exit(0)
# ============================================================ # --- Получаем CARD_ID ---
# 5. Определение CARD_ID card_id = None
# ============================================================ print(f"\n--- 2. Поиск CARD_ID в PR_CARD ---")
card_id = card_id_ad cursor.execute(
"SELECT c.ID FROM UOV_SELFSERVICE_PR_CARD c "
if not card_id: "JOIN UOV_SELFSERVICE_PR_EMP e ON e.CARD_ID = c.ID "
print("\n--- 2. Поиск CARD_ID по ФИО в PR_CARD ---") "WHERE e.EMAIL_ID LIKE %s OR e.USERNAME LIKE %s",
cursor.execute("SELECT c.ID, c.NAME FROM UOV_SELFSERVICE_PR_CARD c WHERE c.NAME LIKE %s", ("%Гусейнова%",)) (f"%{short_name}%", f"%{short_name}%")
)
rows = cursor.fetchall() rows = cursor.fetchall()
if rows: if rows:
card_id = str(rows[0][0]) for r in rows:
print(f" ✅ НАЙДЕН по ФИО: CARD_ID={card_id} ({rows[0][1]})") print(f" CARD_ID (по username): {r[0]}")
card_id = rows[0][0]
else:
print(f" Не найден по username")
if not card_id: if not card_id:
print(" ❌ Не удалось определить 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" CARD_ID не найден!")
conn_sql.close() conn_sql.close()
sys.exit(1) sys.exit(0)
print(f"\n Используем CARD_ID: {card_id}") print(f"\n Используем CARD_ID: {card_id}")
# ============================================================ # --- PR_EMP ---
# 6. Проверка доступных периодов (get_latest_payslip_months) print(f"\n--- 3. PR_EMP (профиль сотрудника) ---")
# ============================================================ cursor.execute(
print("\n--- 3. Расчетные периоды (SL_YTAX) ---") "SELECT ID, FIRM_ID, EMAIL_ID, USERNAME FROM UOV_SELFSERVICE_PR_EMP WHERE CARD_ID = %s",
q_months = """ (card_id,)
SELECT DISTINCT TOP 6 (t.TYEAR * 12 + t.TMONTH) as CMONTH, t.TYEAR, t.TMONTH )
FROM UOV_SELFSERVICE_PR_EMP e emp_rows = cursor.fetchall()
JOIN UOV_SELFSERVICE_PR_CARD c on c.ID = e.CARD_ID if emp_rows:
JOIN UOV_SELFSERVICE_SL_YTAX t on t.CARD_ID = e.CARD_ID for r in emp_rows:
JOIN UOV_SELFSERVICE_PR_TRANS tr on tr.EMP_ID = e.ID print(f" EMP_ID: {r[0]}, FIRM_ID: {r[1]}, EMAIL: {r[2]}, USERNAME: {r[3]}")
JOIN UOV_SELFSERVICE_ST_APPOINT a on a.ID = tr.APPOINT_ID else:
JOIN UOV_SELFSERVICE_HR_FIRM f on f.ID = e.FIRM_ID print(" ПУСТО - нет записей в PR_EMP")
WHERE c.ID = %s
ORDER BY CMONTH DESC
"""
cursor.execute(q_months, (card_id,))
month_rows = cursor.fetchall()
if not month_rows: # --- PR_TRANS ---
print(" ❌ Отсутствуют расчетные периоды для данного CARD_ID!") print(f"\n--- 4. PR_TRANS (кадровые назначения) ---")
conn_sql.close() if emp_rows:
sys.exit(1) emp_ids = [r[0] for r in emp_rows]
placeholders = ",".join(["%s"] * len(emp_ids))
cursor.execute(
f"SELECT EMP_ID, D_FROM, D_TO, TP_ABBR, SYSPAY_ID "
f"FROM UOV_SELFSERVICE_PR_TRANS "
f"WHERE EMP_ID IN ({placeholders}) "
f"ORDER BY D_FROM DESC",
emp_ids
)
trans_rows = cursor.fetchall()
if 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(" ПУСТО - нет записей в PR_TRANS")
else:
print(" Пропуск (нет PR_EMP)")
print(f" Найдено периодов: {len(month_rows)}") # --- SL_YTAX ---
for r in month_rows: print(f"\n--- 5. SL_YTAX (периоды с данными) ---")
print(f" • CMONTH={r[0]} ({r[1]}-{r[2]:02d})") cursor.execute(
"SELECT DISTINCT TYEAR, TMONTH, CMONTH FROM UOV_SELFSERVICE_SL_YTAX "
"WHERE CARD_ID = %s ORDER BY CMONTH DESC",
(card_id,)
)
tax_rows = cursor.fetchall()
if tax_rows:
print(f" Найдено периодов: {len(tax_rows)}")
for r in tax_rows:
print(f" {r[0]}-{r[1]:02d} (CMONTH={r[2]})")
else:
print(" ПУСТО - нет периодов в SL_YTAX")
target_cmonth = month_rows[0][0] # --- 6. Проверка q_firm для ВСЕХ периодов ---
print(f"\n--- 6. Проверка q_firm для КАЖДОГО периода ---")
# ============================================================ if tax_rows:
# 7. Тест определения организации (q_firm) для периода for tax_row in tax_rows:
# ============================================================ tyear = tax_row[0]
print(f"\n--- 4. Определение FIRM_ID для CMONTH={target_cmonth} ---") tmonth = tax_row[1]
tyear = (target_cmonth - 1) // 12 cmonth = tax_row[2]
tmonth = (target_cmonth - 1) % 12 + 1
_, last_day = calendar.monthrange(tyear, tmonth) _, last_day = calendar.monthrange(tyear, tmonth)
date_start = f"{tyear}-{tmonth:02d}-01" date_start = f"{tyear}-{tmonth:02d}-01"
date_end = f"{tyear}-{tmonth:02d}-{last_day:02d}" date_end = f"{tyear}-{tmonth:02d}-{last_day:02d}"
date_params = (date_start, date_start, date_end, date_end, date_end) date_params = (date_start, date_start, date_end, date_end, date_end)
@@ -232,11 +226,37 @@ q_firm = """
""" """
cursor.execute(q_firm, (card_id, *date_params)) cursor.execute(q_firm, (card_id, *date_params))
firm_row = cursor.fetchone() firm_row = cursor.fetchone()
if firm_row: if firm_row:
print(f" ✅ q_firm УСПЕШНО определил FIRM_ID={firm_row[0]} ({firm_row[1]}) за {tyear}-{tmonth:02d}") print(f" {tyear}-{tmonth:02d} (CMONTH={cmonth}): ✅ FIRM_ID={firm_row[0]}, SNAME={firm_row[1]}")
else: else:
print(f" ❌ q_firm НЕ СМОГ определить FIRM_ID за {tyear}-{tmonth:02d}!") print(f" {tyear}-{tmonth:02d} (CMONTH={cmonth}): ❌ q_firm НЕ НАШЁЛ FIRM_ID!")
# Фоллбэк
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
where c.ID = %s
group by e.ID, e.FIRM_ID
"""
cursor.execute(q_firm_fb, (card_id,))
fb_row = cursor.fetchone()
if fb_row:
print(f" Фоллбэк: FIRM_ID={fb_row[0]}")
else:
print(f" Фоллбэк НЕ НАШЁЛ - критическая ошибка!")
else:
print(" Нет периодов для проверки")
# --- 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" Периодов в SL_YTAX: {summary[0]}")
print(f" Диапазон: {summary[1]} - {summary[2]}")
print(f"\n{'=' * 60}") print(f"\n{'=' * 60}")
print("Диагностика завершена") print("Диагностика завершена")