From af0caee9638146402e7bfff078e7f0d41f5a5a68 Mon Sep 17 00:00:00 2001 From: dddennnisss Date: Thu, 23 Jul 2026 16:43:47 +0700 Subject: [PATCH] Add test_tabel_check.py - SQL server table check script --- test_tabel_check.py | 253 ++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 253 insertions(+) create mode 100644 test_tabel_check.py diff --git a/test_tabel_check.py b/test_tabel_check.py new file mode 100644 index 0000000..3a472d9 --- /dev/null +++ b/test_tabel_check.py @@ -0,0 +1,253 @@ +#!/usr/bin/env python3 +""" +Проверка подключения к SQL Server и загрузки данных табеля +за последние 5 месяцев. + +Запуск внутри Docker-контейнера проекта: + docker exec -it trueconf-bot python3 test_tabel_check.py + +ИЛИ напрямую на production (если есть доступ к серверу): + cd /opt/trueconf_bot + python3 test_tabel_check.py + +Требования: pymssql, passwork (опционально) +""" +import sys +import os +import calendar +from datetime import datetime, timedelta + +# ============================================================ +# 1. Загрузка .env из /opt/trueconf_bot/config/.env +# ============================================================ +ENV_PATH = "/opt/trueconf_bot/config/.env" + +def load_env(filepath): + """Загружает переменные из .env файла.""" + 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) + +# ============================================================ +# 2. Попытка загрузки из Passwork (как в оригинальном config.py) +# ============================================================ +DB_SERVER = os.getenv("SQL_SERVER", "SRVKEM-MOBILEIN.sibcem.ru") +DB_NAME = os.getenv("SQL_DB_NAME", "BossCopy") +DB_USER = os.getenv("DB_USER", "") +DB_PASSWORD = os.getenv("DB_PASSWORD", "") + +# Если Passwork не загружен, пробуем .db.env +if not DB_PASSWORD: + db_env_path = "/opt/trueconf_bot/config/.db.env" + if os.path.exists(db_env_path): + print(f"📂 Попытка загрузки DB-пароля из {db_env_path}...") + load_env(db_env_path) + DB_USER = os.getenv("DB_USER", DB_USER) + DB_PASSWORD = os.getenv("DB_PASSWORD", DB_PASSWORD) + +# ============================================================ +# 3. Подключение к SQL Server +# ============================================================ +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() + +try: + import pymssql + print(f"✅ pymssql загружен (версия {pymssql.__version__})") +except ImportError: + print("❌ pymssql не установлен!") + print(" pip install pymssql") + sys.exit(1) + +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) + +# ============================================================ +# 4. Проверка доступности таблиц +# ============================================================ +print() +print("=" * 60) +print(" Проверка доступности таблиц") +print("=" * 60) + +tables_to_check = [ + ("UOV_SELFSERVICE_PR_EMP", "Таблица сотрудников (поиск по CARD_ID)"), + ("UOV_SELFSERVICE_TB_TABEL", "Табель учёта рабочего времени"), +] + +for table, desc in tables_to_check: + try: + cursor.execute(f"SELECT TOP 1 * FROM {table}") + row = cursor.fetchone() + if row: + cols = [col[0] for col in cursor.description] + print(f"✅ {desc}") + print(f" Таблица: {table}") + print(f" Столбцы: {', '.join(cols)}") + else: + print(f"⚠️ {desc} — таблица пуста") + except Exception as e: + print(f"❌ {desc} — ошибка: {e}") + +# ============================================================ +# 5. Загрузка табеля за последние 5 месяцев +# ============================================================ +print() +print("=" * 60) +print(" Загрузка табеля за последние 5 месяцев") +print("=" * 60) + +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) + + # Запрос 1: ищем EMP_ID по CARD_ID (тестовый — берём первого попавшегося) + try: + cursor.execute("SELECT TOP 1 e.ID, e.CARD_ID FROM UOV_SELFSERVICE_PR_EMP e ORDER BY e.ID") + emp_row = cursor.fetchone() + if not emp_row: + print(" ⚠️ Нет сотрудников в таблице UOV_SELFSERVICE_PR_EMP") + continue + emp_id = emp_row[0] + card_id = emp_row[1] + print(f" Тестовый сотрудник: emp_id={emp_id}, card_id={card_id}") + except Exception as e: + print(f" ❌ Ошибка поиска сотрудника: {e}") + continue + + # Запрос 2: данные табеля + 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: + day_date = r[0] + 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())}") + + # Показываем первые 5 и последние 5 записей + 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} ч.") + if len(records) > 5: + print(f" ... ещё {len(records) - 5} записей ...") + 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} ч.") + +# ============================================================ +# 6. Тест с конкретным CARD_ID пользователя +# ============================================================ +print() +print("=" * 60) +print(" Тест с конкретным CARD_ID") +print("=" * 60) + +# Спрашиваем CARD_ID у пользователя +test_card_id = input("Введите CARD_ID для проверки (или Enter для теста с первым): ").strip() +if not test_card_id: + # Берём первый + cursor.execute("SELECT TOP 1 CARD_ID FROM UOV_SELFSERVICE_PR_CARD ORDER BY ID") + row = cursor.fetchone() + test_card_id = str(row[0]).strip() if row else "" + print(f" Тестовый CARD_ID (первый): {test_card_id}") + +if test_card_id: + print(f" Поиск сотрудника с CARD_ID={test_card_id}...") + cursor.execute("SELECT e.ID FROM UOV_SELFSERVICE_PR_EMP e WHERE e.CARD_ID = %s", (test_card_id,)) + row = cursor.fetchone() + if not row: + print(f" ⚠️ Сотрудник с CARD_ID={test_card_id} не найден") + else: + emp_id = row[0] + print(f" ✅ Найден: emp_id={emp_id}") + + for i in range(min(3, months_back)): + month = now.month - i + year = now.year + while month <= 0: + month += 12 + year -= 1 + + month_name = calendar.month_name[month] + cursor.execute(query, (emp_id, year, month)) + records = cursor.fetchall() + print(f" {month_name} {year}: {len(records)} записей") + +# ============================================================ +# Закрытие +# ============================================================ +conn.close() +print() +print("=" * 60) +print(" ✅ Проверка завершена") +print("=" * 60)