Add test_tabel_check.py - SQL server table check script

2026-07-23 16:43:47 +07:00
parent 0dca28ef20
commit af0caee963
1 changed files with 253 additions and 0 deletions
@@ -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)