89 lines
3.6 KiB
Python
89 lines
3.6 KiB
Python
"""
|
|
===============================================================================
|
|
FILE: services/snapshots/finder.py
|
|
ROLE: Поиск ближайшего снапшота в SQLite (Smart Snap-to-Grid) и On-Demand экспорт.
|
|
===============================================================================
|
|
"""
|
|
|
|
import logging
|
|
from datetime import datetime
|
|
from typing import Optional, Tuple
|
|
from core.connection import get_connection
|
|
from services.scud_export import run_export
|
|
|
|
logger = logging.getLogger("SNAPSHOT_FINDER")
|
|
|
|
|
|
def find_or_create_snapshot_for_time(
|
|
target_date_str: str,
|
|
target_time_str: str,
|
|
tolerance_minutes: int = 20,
|
|
allow_ondemand_export: bool = True
|
|
) -> Tuple[Optional[str], str]:
|
|
"""
|
|
Ищет ближайший срез за указанную дату и время (±tolerance_minutes).
|
|
Если не найден и allow_ondemand_export=True — запрашивает выгрузку из MS SQL на это время.
|
|
|
|
Возвращает: (snapshot_id, human_message)
|
|
"""
|
|
date_clean = target_date_str.replace('_', '.')
|
|
dt_target = datetime.strptime(f"{date_clean} {target_time_str}", "%d.%m.%Y %H:%M")
|
|
|
|
# 1. Поиск существующих снапшотов за эту дату в SQLite
|
|
with get_connection() as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute("""
|
|
SELECT DISTINCT snapshot_id, snapshot_time
|
|
FROM scud_logs
|
|
WHERE log_date = ? AND snapshot_time IS NOT NULL
|
|
""", (date_clean,))
|
|
rows = cursor.fetchall()
|
|
|
|
best_snapshot = None
|
|
min_diff_seconds = float('inf')
|
|
|
|
for snap_id, snap_time_str in rows:
|
|
try:
|
|
# Формат в базе: YYYY-MM-DD HH:MM:SS или DD.MM.YYYY HH:MM:SS
|
|
raw_time = str(snap_time_str).strip()
|
|
if '.' in raw_time.split()[0]:
|
|
dt_snap = datetime.strptime(raw_time, "%d.%m.%Y %H:%M:%S")
|
|
else:
|
|
dt_snap = datetime.strptime(raw_time, "%Y-%m-%d %H:%M:%S")
|
|
|
|
diff = abs((dt_snap - dt_target).total_seconds())
|
|
if diff < min_diff_seconds:
|
|
min_diff_seconds = diff
|
|
best_snapshot = (snap_id, snap_time_str, dt_snap)
|
|
except Exception:
|
|
continue
|
|
|
|
# Если найден срез в пределах допуска (по умолчанию 20 минут)
|
|
if best_snapshot and min_diff_seconds <= (tolerance_minutes * 60):
|
|
snap_id, snap_time, dt_s = best_snapshot
|
|
diff_mins = round(min_diff_seconds / 60)
|
|
return snap_id, f"Использован готовый срез {snap_id} за {dt_s.strftime('%H:%M')} (разница {diff_mins} мин)."
|
|
|
|
# 2. Если срез не найден и разрешен On-Demand экспорт из MS SQL Орион
|
|
if allow_ondemand_export:
|
|
logger.info(f"Снапшот на {date_clean} {target_time_str} не найден в SQLite. Запуск прямого среза из MS SQL...")
|
|
|
|
run_export(
|
|
input_date=date_clean,
|
|
debug=False,
|
|
save_xlsx=True
|
|
)
|
|
|
|
with get_connection() as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute("""
|
|
SELECT snapshot_id
|
|
FROM scud_logs
|
|
WHERE log_date = ?
|
|
ORDER BY id DESC LIMIT 1
|
|
""", (date_clean,))
|
|
row = cursor.fetchone()
|
|
if row:
|
|
return row[0], f"Создан новый срез {row[0]} из MS SQL на {target_time_str}."
|
|
|
|
return None, f"Срез на {date_clean} {target_time_str} не найден." |