feat(etl): stable pipeline, exception registry in SQLite, multi-pass aggregation and db_cli

services/scud_etl/pipeline.py

@@ -0,0 +1,59 @@
+"""
+===============================================================================
+FILE: services/scud_etl/pipeline.py
+ROLE: Выборка данных СКУД из SQLite (scud_logs) с жестким приоритетом Y-снапшотов.
+===============================================================================
+"""
+
+import os
+import logging
+import pandas as pd
+from core.connection import get_connection
+from services.data_loader import load_1c_data_smart
+
+logger = logging.getLogger("SCUD_PIPELINE")
+
+
+def load_best_snapshot_for_date(date_str: str, prefer_final_y: bool = True) -> pd.DataFrame:
+    """Извлекает срез СКУД. Для вчерашнего дня строго берет финальный вечерний срез Y."""
+    with get_connection() as conn:
+        df = pd.DataFrame()
+        cursor = conn.cursor()
+
+        if prefer_final_y:
+            cursor.execute(
+                "SELECT snapshot_id FROM scud_logs WHERE log_date = ? AND snapshot_id LIKE 'Y%' ORDER BY id DESC LIMIT 1",
+                (date_str,)
+            )
+            row = cursor.fetchone()
+            if row and row[0]:
+                df = pd.read_sql_query("SELECT * FROM scud_logs WHERE snapshot_id = ?", conn, params=(row[0],))
+
+        if df.empty:
+            cursor.execute(
+                "SELECT snapshot_id FROM scud_logs WHERE log_date = ? ORDER BY id DESC LIMIT 1",
+                (date_str,)
+            )
+            row = cursor.fetchone()
+            if row and row[0]:
+                df = pd.read_sql_query("SELECT * FROM scud_logs WHERE snapshot_id = ?", conn, params=(row[0],))
+
+        if not df.empty:
+            rename_map = {
+                'department': 'Подразделение', 'position': 'Должность', 'fio': 'Сотрудник',
+                'time_in': 'Начало_дня', 'first_activity': 'Первая_активность', 'time_out': 'Конец_дня',
+                'time_in_building': 'Находился_в_здании', 'is_present': 'Пришел'
+            }
+            df = df.rename(columns={k: v for k, v in rename_map.items() if k in df.columns})
+            if 'fio_clean' not in df.columns and 'Сотрудник' in df.columns:
+                df['fio_clean'] = df['Сотрудник'].astype(str).str.strip()
+            if 'Пришел' in df.columns:
+                df['Пришел'] = df['Пришел'].astype(bool)
+
+        return df
+
+
+def load_1c_files_for_date(date_str: str) -> tuple[pd.DataFrame, pd.DataFrame]:
+    df_staff, df_absences = load_1c_data_smart(date_str, use_db=False)
+    return (df_staff if df_staff is not None else pd.DataFrame(),
+            df_absences if df_absences is not None else pd.DataFrame())