245 lines
8.9 KiB
Python
245 lines
8.9 KiB
Python
#!/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() |