337 lines
12 KiB
Python
337 lines
12 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
Проверка подключения к SQL Server и загрузки данных табеля
|
||
за последние 5 месяцев для конкретного сотрудника.
|
||
|
||
Поиск сотрудника:
|
||
1. Ищем в AD по email (ds.krivochenko@sibcem.ru) → получаем extensionAttribute2 = CARD_ID
|
||
2. По CARD_ID ищем в SQL (UOV_SELFSERVICE_PR_EMP) → получаем EMP_ID
|
||
3. По EMP_ID загружаем табель (UOV_SELFSERVICE_TB_TABEL) за 5 месяцев
|
||
|
||
Запуск:
|
||
cd /opt/trueconf_bot
|
||
python3 test_tabel_check.py
|
||
"""
|
||
import sys
|
||
import os
|
||
import calendar
|
||
from datetime import datetime
|
||
|
||
# ============================================================
|
||
# 1. Загрузка .env
|
||
# ============================================================
|
||
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
|
||
ENV_PATH = os.path.join(BASE_DIR, "config", ".env")
|
||
|
||
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("#"):
|
||
if "=" in line:
|
||
key, value = line.split("=", 1)
|
||
os.environ[key.strip()] = value.strip(' "\'\r\n')
|
||
except FileNotFoundError:
|
||
print(f"⚠️ Файл {filepath} не найден.")
|
||
|
||
load_env(ENV_PATH)
|
||
|
||
# Карта соответствия: Имя переменной в боте -> Название карточки в сейфе
|
||
CREDENTIALS_MAP = {
|
||
"DB_PASSWORD": os.getenv("PW_ID_SQL", "").strip(' "\'\r\n'),
|
||
"AD_PASSWORD": os.getenv("PW_ID_AD", "").strip(' "\'\r\n'),
|
||
}
|
||
|
||
# ============================================================
|
||
# 2. Загрузка паролей из Passwork
|
||
# ============================================================
|
||
DB_USER = ""
|
||
DB_PASSWORD = ""
|
||
AD_USER = ""
|
||
AD_PASSWORD = ""
|
||
|
||
for pw_path in ["/opt/passwork", os.path.join(BASE_DIR, "passwork"), os.path.join(BASE_DIR, "config", "passwork")]:
|
||
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)
|
||
|
||
try:
|
||
from passwork import get_passwork_secrets
|
||
print("🔐 Загрузка секретов из Passwork...")
|
||
required_cards = [name for name in CREDENTIALS_MAP.values() if name]
|
||
passwork_pool = get_passwork_secrets(required_cards=required_cards)
|
||
for var_name, card_name in CREDENTIALS_MAP.items():
|
||
if not card_name:
|
||
continue
|
||
clean_card = card_name.strip(' "\'\r\n')
|
||
card_data = passwork_pool.get(clean_card)
|
||
if card_data is None:
|
||
print(f" ⚠️ Карточка '{clean_card}' не найдена")
|
||
continue
|
||
globals()[var_name] = card_data["password"].strip(' "\'\r\n')
|
||
if var_name == "AD_PASSWORD":
|
||
globals()["AD_USER"] = card_data.get("login", "").strip(' "\'\r\n')
|
||
elif var_name == "DB_PASSWORD":
|
||
globals()["DB_USER"] = card_data.get("login", "").strip(' "\'\r\n')
|
||
print(f" ✓ {var_name} из карточки '{clean_card}'")
|
||
except ImportError:
|
||
print("⚠️ Модуль passwork не найден")
|
||
except Exception as e:
|
||
print(f"⚠️ Ошибка Passwork: {e}")
|
||
|
||
# Фоллбэк на переменные окружения
|
||
if not DB_PASSWORD:
|
||
DB_PASSWORD = os.getenv("DB_PASSWORD", "").strip()
|
||
DB_USER = os.getenv("DB_USER", DB_USER).strip()
|
||
if not AD_PASSWORD:
|
||
AD_PASSWORD = os.getenv("AD_PASSWORD", "").strip()
|
||
AD_USER = os.getenv("AD_USER", AD_USER).strip()
|
||
|
||
# ============================================================
|
||
# 3. Настройки подключения
|
||
# ============================================================
|
||
DB_SERVER = os.getenv("SQL_SERVER", "SRVKEM-MOBILEIN.sibcem.ru")
|
||
DB_NAME = os.getenv("SQL_DB_NAME", "BossCopy")
|
||
AD_SERVER = os.getenv("AD_SERVER", "ldap://172.16.20.20")
|
||
AD_BASES = (
|
||
"OU=-Пользователи,DC=sibcem,DC=ru",
|
||
"OU=Планшеты,OU=enabled,OU=БезКомпьютеров,DC=sibcem,DC=ru",
|
||
)
|
||
|
||
print()
|
||
print("=" * 60)
|
||
print(" Проверка подключения к SQL Server")
|
||
print("=" * 60)
|
||
print(f" Сервер : {DB_SERVER}")
|
||
print(f" База : {DB_NAME}")
|
||
print(f" Логин : {DB_USER}")
|
||
print(f" Пароль : {'*' * len(DB_PASSWORD) if DB_PASSWORD else '(не задан)'}")
|
||
print()
|
||
|
||
# ============================================================
|
||
# 4. Проверка pymssql
|
||
# ============================================================
|
||
try:
|
||
import pymssql
|
||
print(f"✅ pymssql загружен (версия {pymssql.__version__})")
|
||
except ImportError:
|
||
print("❌ pymssql не установлен!")
|
||
sys.exit(1)
|
||
|
||
# ============================================================
|
||
# 5. Подключение к SQL Server
|
||
# ============================================================
|
||
try:
|
||
conn = pymssql.connect(
|
||
server=DB_SERVER,
|
||
user=DB_USER,
|
||
password=DB_PASSWORD,
|
||
database=DB_NAME,
|
||
charset='cp1251',
|
||
login_timeout=10
|
||
)
|
||
print("✅ Подключение к SQL Server успешно!")
|
||
cursor = conn.cursor()
|
||
except Exception as e:
|
||
print(f"❌ Ошибка подключения: {e}")
|
||
sys.exit(1)
|
||
|
||
# ============================================================
|
||
# 6. Поиск сотрудника в AD по email → CARD_ID
|
||
# ============================================================
|
||
print()
|
||
print("=" * 60)
|
||
print(" Поиск сотрудника в Active Directory")
|
||
print("=" * 60)
|
||
|
||
# Email для поиска
|
||
TARGET_EMAIL = "ds.krivochenko@sibcem.ru"
|
||
print(f" Ищем: {TARGET_EMAIL}")
|
||
|
||
# Импортируем ldap3 и ищем
|
||
try:
|
||
from ldap3 import Server, Connection, ALL
|
||
from ldap3.utils.conv import escape_filter_chars
|
||
|
||
server = Server(AD_SERVER, get_info=ALL)
|
||
ldap_conn = Connection(server, user=AD_USER, password=AD_PASSWORD, auto_bind=True)
|
||
|
||
# Строим фильтр как в ad_search.py
|
||
short_username = TARGET_EMAIL.split("@")[0] if "@" in TARGET_EMAIL else TARGET_EMAIL
|
||
|
||
or_parts = [
|
||
"(mail={})".format(TARGET_EMAIL),
|
||
"(userPrincipalName={})".format(TARGET_EMAIL),
|
||
"(userPrincipalName={})".format(short_username + "@sibcem.ru"),
|
||
"(sAMAccountName={})".format(short_username),
|
||
]
|
||
or_filter = "(|{})".format("".join(or_parts))
|
||
search_filter = "(&{}{})".format("(objectClass=user)", or_filter)
|
||
|
||
print(f" LDAP-фильтр: {search_filter}")
|
||
|
||
card_id_from_ad = None
|
||
ldap_entry = None
|
||
|
||
for base in AD_BASES:
|
||
ldap_conn.search(
|
||
search_base=base,
|
||
search_filter=search_filter,
|
||
attributes=["cn", "sAMAccountName", "mail", "userPrincipalName", "extensionAttribute2"],
|
||
)
|
||
if ldap_conn.entries:
|
||
ldap_entry = ldap_conn.entries[0]
|
||
print(f" ✅ Найден в OU: {base}")
|
||
print(f" CN : {ldap_entry.cn}")
|
||
print(f" sAMAccountName: {ldap_entry.sAMAccountName}")
|
||
print(f" mail : {ldap_entry.mail}")
|
||
ext2 = str(ldap_entry.extensionAttribute2) if hasattr(ldap_entry, 'extensionAttribute2') and ldap_entry.extensionAttribute2 else "(не задан)"
|
||
print(f" extensionAttribute2: {ext2}")
|
||
card_id_from_ad = ext2
|
||
break
|
||
|
||
if not ldap_entry:
|
||
print(f" ❌ Сотрудник не найден в AD по email {TARGET_EMAIL}")
|
||
|
||
except ImportError:
|
||
print("❌ ldap3 не установлен")
|
||
sys.exit(1)
|
||
except Exception as e:
|
||
print(f"❌ Ошибка LDAP: {e}")
|
||
sys.exit(1)
|
||
|
||
# ============================================================
|
||
# 7. Поиск EMP_ID по CARD_ID в SQL
|
||
# ============================================================
|
||
print()
|
||
print("=" * 60)
|
||
print(" Поиск сотрудника в SQL Server")
|
||
print("=" * 60)
|
||
|
||
emp_id = None
|
||
card_id = None
|
||
|
||
if card_id_from_ad:
|
||
card_id = card_id_from_ad
|
||
print(f" CARD_ID из AD: {card_id}")
|
||
cursor.execute("SELECT e.ID, e.CARD_ID FROM UOV_SELFSERVICE_PR_EMP e WHERE e.CARD_ID = %s", (card_id,))
|
||
row = cursor.fetchone()
|
||
if row:
|
||
emp_id = row[0]
|
||
print(f" ✅ Найден: emp_id={emp_id}, CARD_ID={row[1]}")
|
||
else:
|
||
print(f" ⚠️ Сотрудник с CARD_ID={card_id} не найден в UOV_SELFSERVICE_PR_EMP")
|
||
|
||
# Если не нашли по CARD_ID из AD — пробуем по email через SQL
|
||
if not emp_id:
|
||
print(f" ⚠️ Пробуем поиск по email через SQL...")
|
||
cursor.execute(
|
||
"SELECT e.ID, e.CARD_ID, e.EMAIL FROM UOV_SELFSERVICE_PR_EMP e WHERE e.EMAIL LIKE %s OR e.EMAIL LIKE %s ORDER BY e.ID LIMIT 1",
|
||
(f"%{TARGET_EMAIL}%", f"%{short_username}@%")
|
||
)
|
||
row = cursor.fetchone()
|
||
if row:
|
||
emp_id = row[0]
|
||
card_id = row[1]
|
||
print(f" ✅ Найден: emp_id={emp_id}, CARD_ID={card_id}")
|
||
else:
|
||
print(f" ❌ Сотрудник не найден")
|
||
|
||
# Если нашли — выводим инфу
|
||
if emp_id:
|
||
cursor.execute(
|
||
"SELECT f.SNAME, d.NAME, appoint.NAME, e.EMAIL "
|
||
"FROM UOV_SELFSERVICE_PR_EMP 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 "
|
||
"JOIN UOV_SELFSERVICE_HR_DEPT d ON d.ID = t.DEPT_ID "
|
||
"JOIN UOV_SELFSERVICE_ST_APPOINT appoint ON appoint.ID = t.APPOINT_ID "
|
||
"WHERE e.CARD_ID = %s AND t.D_FROM <= CAST(GETDATE() AS DATE) "
|
||
"AND (t.D_TO >= CAST(GETDATE() AS DATE) OR t.D_TO = CAST('1999-12-31' AS DATE))",
|
||
(card_id,)
|
||
)
|
||
row = cursor.fetchone()
|
||
if row:
|
||
print(f"\n 📋 Информация о сотруднике:")
|
||
print(f" Компания : {row[0]}")
|
||
print(f" Отдел : {row[1]}")
|
||
print(f" Должность: {row[2]}")
|
||
print(f" Email : {row[3]}")
|
||
|
||
# ============================================================
|
||
# 8. Загрузка табеля за последние 5 месяцев
|
||
# ============================================================
|
||
print()
|
||
print("=" * 60)
|
||
print(" Загрузка табеля за последние 5 месяцев")
|
||
print("=" * 60)
|
||
|
||
if not emp_id:
|
||
print(" ❌ Не удалось найти сотрудника — завершение")
|
||
conn.close()
|
||
sys.exit(1)
|
||
|
||
now = datetime.now()
|
||
months_back = 5
|
||
|
||
for i in range(months_back):
|
||
month = now.month - i
|
||
year = now.year
|
||
while month <= 0:
|
||
month += 12
|
||
year -= 1
|
||
|
||
month_name = calendar.month_name[month]
|
||
print(f"\n📅 {month_name} {year}")
|
||
print("-" * 40)
|
||
|
||
query = """
|
||
SELECT e.D, e.TDAY_ID, e.H
|
||
FROM UOV_SELFSERVICE_TB_TABEL e
|
||
WHERE e.EMP_ID = %s AND YEAR(e.D) = %s AND MONTH(e.D) = %s
|
||
ORDER BY e.D ASC
|
||
"""
|
||
try:
|
||
cursor.execute(query, (emp_id, year, month))
|
||
records = cursor.fetchall()
|
||
except Exception as e:
|
||
print(f" ❌ Ошибка запроса: {e}")
|
||
continue
|
||
|
||
if not records:
|
||
print(f" ⚠️ Данные табеля отсутствуют")
|
||
continue
|
||
|
||
print(f" Записей: {len(records)}")
|
||
|
||
total_days = 0
|
||
total_hours = 0.0
|
||
codes = {}
|
||
|
||
for r in records:
|
||
tday_id = str(r[1]).strip() if r[1] else "?"
|
||
hours = float(r[2]) if r[2] is not None else 0.0
|
||
codes[tday_id] = codes.get(tday_id, 0) + 1
|
||
if tday_id == 'Я':
|
||
total_days += 1
|
||
total_hours += hours
|
||
|
||
print(f" Явки (Я): {codes.get('Я', 0)} дн., часов: {total_hours:.1f}")
|
||
print(f" Коды: {', '.join(f'{k}={v}' for k, v in codes.items())}")
|
||
|
||
print(f" Первые 5 записей:")
|
||
for r in records[:5]:
|
||
d = r[0].strftime('%d.%m.%Y') if hasattr(r[0], 'strftime') else str(r[0])
|
||
print(f" {d} | {str(r[1]).strip():>3} | {float(r[2]) if r[2] else 0:.1f} ч.")
|
||
|
||
# ============================================================
|
||
# Закрытие
|
||
# ============================================================
|
||
conn.close()
|
||
print()
|
||
print("=" * 60)
|
||
print(" ✅ Проверка завершена")
|
||
print("=" * 60)
|