#!/usr/bin/env python3 """Диагностика данных сотрудника для модуля ЛК (расчетные листки).""" import calendar import os import sys # ============================================================ # 1. Загрузка .env и путей # ============================================================ BASE_DIR = "/opt/trueconf_bot" if os.path.exists(BASE_DIR): os.chdir(BASE_DIR) else: BASE_DIR = os.path.dirname(os.path.abspath(__file__)) def load_env(filepath): try: with open(filepath, "r", encoding="utf-8") as f: for line in f: line = line.strip() if line and not line.startswith("#") and "=" in line: key, value = line.split("=", 1) os.environ[key.strip()] = value.strip(' "\'\r\n') except FileNotFoundError: print(f"⚠️ Файл {filepath} не найден.") load_env(os.path.join(BASE_DIR, "config", ".env")) # ============================================================ # 2. Загрузка паролей из Passwork (AD и SQL) # ============================================================ 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: 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 required_cards = [name for name in CREDENTIALS_MAP.values() if name] passwork_pool = get_passwork_secrets(required_cards=required_cards) 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", "") SQL_SERVER = os.getenv("SQL_SERVER", "SRVKEM-MOBILEIN.sibcem.ru") 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 # ============================================================ # 3. Поиск сотрудника в Active Directory # ============================================================ email = "la.guseynova@sibcem.ru" search_id = email.replace("@tcs.sibcem.ru", "@sibcem.ru") short_name = search_id.split("@")[0] print("=" * 60) print(f"Диагностика сотрудника: {email}") print("=" * 60) print("\n--- 0. Поиск в AD ---") card_id_ad = None 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( server, user=bind_user, password=AD_PASSWORD, auto_bind=True, auto_referrals=False ) attributes_list = ["cn", "sAMAccountName", "mail", "userPrincipalName", "proxyAddresses", "extensionAttribute2"] search_filter = f"(|(sAMAccountName={short_name})(mail={search_id})(userPrincipalName={search_id})(proxyAddresses=*:{search_id}*)(proxyAddresses=*{short_name}*)(anr={short_name}))" conn.search( search_base=AD_BASE, search_filter=search_filter, attributes=attributes_list ) if conn.entries: entry = conn.entries[0] cn = str(entry.cn) sam = str(entry.sAMAccountName) if 'sAMAccountName' in entry else "" mail_val = str(entry.mail) if 'mail' in entry else "" upn = str(entry.userPrincipalName) if 'userPrincipalName' in entry else "" card_id_ad = str(entry.extensionAttribute2) if 'extensionAttribute2' in entry else None print(" ✅ НАЙДЕН В AD!") print(f" CN: {cn}") print(f" sAMAccountName: {sam}") print(f" mail: {mail_val}") print(f" UPN: {upn}") print(f" extensionAttribute2 (CARD_ID из AD): {card_id_ad}") else: print(" ❌ НЕ НАЙДЕН в AD!") conn.unbind() except Exception as e: print(f" ❌ Ошибка обращения к AD: {e}") # ============================================================ # 4. Подключение к SQL Server # ============================================================ print("\n--- 1. Подключение к SQL Server ---") 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_sql.cursor() print(" ✅ Подключено к SQL Server!") except Exception as e: print(f" ❌ Ошибка подключения к SQL: {e}") sys.exit(1) # ============================================================ # 5. Определение CARD_ID # ============================================================ card_id = card_id_ad if not card_id: print("\n--- 2. Поиск CARD_ID по ФИО в PR_CARD ---") cursor.execute("SELECT c.ID, c.NAME FROM UOV_SELFSERVICE_PR_CARD c WHERE c.NAME LIKE %s", ("%Гусейнова%",)) rows = cursor.fetchall() if rows: card_id = str(rows[0][0]) print(f" ✅ НАЙДЕН по ФИО: CARD_ID={card_id} ({rows[0][1]})") if not card_id: print(" ❌ Не удалось определить CARD_ID!") conn_sql.close() sys.exit(1) print(f"\n Используем CARD_ID: {card_id}") # ============================================================ # 6. Проверка доступных периодов (get_latest_payslip_months) # ============================================================ print("\n--- 3. Расчетные периоды (SL_YTAX) ---") q_months = """ SELECT DISTINCT TOP 6 (t.TYEAR * 12 + t.TMONTH) as CMONTH, t.TYEAR, t.TMONTH FROM UOV_SELFSERVICE_PR_EMP e JOIN UOV_SELFSERVICE_PR_CARD c on c.ID = e.CARD_ID JOIN UOV_SELFSERVICE_SL_YTAX t on t.CARD_ID = e.CARD_ID JOIN UOV_SELFSERVICE_PR_TRANS tr on tr.EMP_ID = e.ID JOIN UOV_SELFSERVICE_ST_APPOINT a on a.ID = tr.APPOINT_ID JOIN UOV_SELFSERVICE_HR_FIRM f on f.ID = e.FIRM_ID WHERE c.ID = %s ORDER BY CMONTH DESC """ cursor.execute(q_months, (card_id,)) month_rows = cursor.fetchall() if not month_rows: print(" ❌ Отсутствуют расчетные периоды для данного CARD_ID!") conn_sql.close() sys.exit(1) print(f" Найдено периодов: {len(month_rows)}") for r in month_rows: print(f" • CMONTH={r[0]} ({r[1]}-{r[2]:02d})") target_cmonth = month_rows[0][0] # ============================================================ # 7. Тест определения организации (q_firm) для периода # ============================================================ print(f"\n--- 4. Определение FIRM_ID для CMONTH={target_cmonth} ---") tyear = (target_cmonth - 1) // 12 tmonth = (target_cmonth - 1) % 12 + 1 _, last_day = calendar.monthrange(tyear, tmonth) date_start = f"{tyear}-{tmonth:02d}-01" date_end = f"{tyear}-{tmonth:02d}-{last_day:02d}" date_params = (date_start, date_start, date_end, date_end, date_end) q_firm = """ select e.FIRM_ID, f.SNAME from (select e.id, 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 ) as e join UOV_SELFSERVICE_PR_TRANS t on t.EMP_ID = e.ID join UOV_SELFSERVICE_HR_FIRM f on f.ID = e.FIRM_ID where ((t.D_FROM <= %s and t.D_TO >= %s) or (t.D_FROM <= %s and t.D_TO >= %s) or (t.D_FROM <= %s and t.D_TO = '1999-12-31')) group by e.FIRM_ID, f.SNAME """ cursor.execute(q_firm, (card_id, *date_params)) firm_row = cursor.fetchone() if firm_row: print(f" ✅ q_firm УСПЕШНО определил FIRM_ID={firm_row[0]} ({firm_row[1]}) за {tyear}-{tmonth:02d}") else: print(f" ❌ q_firm НЕ СМОГ определить FIRM_ID за {tyear}-{tmonth:02d}!") print(f"\n{'=' * 60}") print("Диагностика завершена") print("=" * 60) conn_sql.close()