296 lines
10 KiB
Python
296 lines
10 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Проверка подключения к SQL Server и загрузки данных табеля
|
|
за последние 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'),
|
|
}
|
|
|
|
# ============================================================
|
|
# 2. Загрузка паролей из Passwork
|
|
# ============================================================
|
|
DB_USER = ""
|
|
DB_PASSWORD = ""
|
|
|
|
# Пытаемся найти passwork в нескольких местах
|
|
passwork_paths = [
|
|
"/opt/passwork",
|
|
os.path.join(BASE_DIR, "passwork"),
|
|
os.path.join(BASE_DIR, "config", "passwork"),
|
|
]
|
|
|
|
# Добавляем /opt/trueconf_bot в путь, если мы в контейнере
|
|
if BASE_DIR.startswith("/"):
|
|
passwork_paths.append("/passwork")
|
|
|
|
print("🔍 Ищем модуль passwork в путях:")
|
|
for pw_path in passwork_paths:
|
|
exists = "✅" if os.path.isdir(pw_path) else "❌"
|
|
print(f" {exists} {pw_path}")
|
|
|
|
# Добавляем все существующие пути
|
|
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)
|
|
print(f" → Добавлен в sys.path: {pw_path}")
|
|
|
|
# Также добавляем корень проекта
|
|
if BASE_DIR not in sys.path:
|
|
sys.path.insert(0, BASE_DIR)
|
|
|
|
passwork_loaded = False
|
|
try:
|
|
from passwork import get_passwork_secrets
|
|
print("\n✅ Модуль passwork найден!")
|
|
|
|
required_cards = [name for name in CREDENTIALS_MAP.values() if name]
|
|
print(f" Запрашиваем карточки: {required_cards}")
|
|
|
|
passwork_pool = get_passwork_secrets(required_cards=required_cards)
|
|
print(f" Получено карточек: {len(passwork_pool)}")
|
|
|
|
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 == "DB_PASSWORD":
|
|
globals()["DB_USER"] = card_data.get("login", "").strip(' "\'\r\n')
|
|
print(f" ✓ {var_name} загружен из карточки '{clean_card}'")
|
|
|
|
passwork_loaded = True
|
|
|
|
except ImportError as e:
|
|
print(f"\n❌ Модуль passwork не импортируется: {e}")
|
|
print(" Варианты: модуль не установлен, нет прав, неверный путь")
|
|
except Exception as e:
|
|
print(f"\n⚠️ Ошибка Passwork: {e}")
|
|
|
|
# Фоллбэк: проверяем системные переменные окружения
|
|
if not DB_PASSWORD:
|
|
DB_PASSWORD = os.getenv("DB_PASSWORD", "").strip()
|
|
DB_USER = os.getenv("DB_USER", DB_USER).strip()
|
|
if DB_PASSWORD:
|
|
print(" ✓ DB_PASSWORD из переменной окружения")
|
|
|
|
# ============================================================
|
|
# 3. Настройки подключения
|
|
# ============================================================
|
|
DB_SERVER = os.getenv("SQL_SERVER", "SRVKEM-MOBILEIN.sibcem.ru")
|
|
DB_NAME = os.getenv("SQL_DB_NAME", "BossCopy")
|
|
|
|
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 не установлен!")
|
|
print(" pip install 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. Проверка доступности таблиц
|
|
# ============================================================
|
|
print()
|
|
print("=" * 60)
|
|
print(" Проверка доступности таблиц")
|
|
print("=" * 60)
|
|
|
|
tables_to_check = [
|
|
("UOV_SELFSERVICE_PR_EMP", "Таблица сотрудников"),
|
|
("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[:8])}")
|
|
if len(cols) > 8:
|
|
print(f" ... ещё {len(cols) - 8} столбцов")
|
|
else:
|
|
print(f"⚠️ {desc} — таблица пуста")
|
|
except Exception as e:
|
|
print(f"❌ {desc} — ошибка: {e}")
|
|
|
|
# ============================================================
|
|
# 7. Загрузка табеля за последние 5 месяцев
|
|
# ============================================================
|
|
print()
|
|
print("=" * 60)
|
|
print(" Загрузка табеля за последние 5 месяцев")
|
|
print("=" * 60)
|
|
|
|
now = datetime.now()
|
|
months_back = 5
|
|
|
|
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")
|
|
conn.close()
|
|
sys.exit(1)
|
|
test_emp_id = emp_row[0]
|
|
test_card_id = emp_row[1]
|
|
print(f"Тестовый сотрудник: emp_id={test_emp_id}, card_id={test_card_id}")
|
|
except Exception as e:
|
|
print(f"❌ Ошибка: {e}")
|
|
conn.close()
|
|
sys.exit(1)
|
|
|
|
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, (test_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} ч.")
|
|
|
|
# ============================================================
|
|
# 8. Тест с конкретным CARD_ID
|
|
# ============================================================
|
|
print()
|
|
print("=" * 60)
|
|
print(" Тест с конкретным CARD_ID")
|
|
print("=" * 60)
|
|
|
|
test_card_id = input("Введите CARD_ID (или Enter для теста с первым): ").strip()
|
|
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)
|