manoraga/scud_ai
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: implement svodka/otchet generators, Y-23:59:59 and snap-to-grid time finder

Browse Source
This commit is contained in:
manoraga
2026-08-28 00:09:57 +03:00
parent fa386b3b79
commit 27e8055a3d
11 changed files with 315 additions and 66 deletions
Download Patch File Download Diff File Expand all files Collapse all files
services/scud_etl/otchet_generator.py
+59
View File
@@ -0,0 +1,59 @@
"""
===============================================================================
FILE: services/scud_etl/otchet_generator.py
ROLE: Генератор Детального Отчета за прошлые смены (строго по итоговому Y-снапшоту).
===============================================================================
"""
import os
import logging
from typing import Dict, Any, Optional
import pandas as pd
from config import DATE_YESTERDAY
from core.database import load_scud_from_db_by_snapshot
from services.scud_etl.pipeline import load_1c_files_for_date
from services.scud_etl.merger import merge_scud_and_1c
from services.excel_exporter import generate_detailed_excel, get_dated_reports_dir, format_date_ru
logger = logging.getLogger("OTCHET_GENERATOR")
def generate_otchet_service(
target_date: Optional[str] = None,
snapshot_id: Optional[str] = None
) -> Dict[str, Any]:
"""
Формирует детальный отчет за прошедшую смену:
- target_date: дата отчета (по умолчанию вчерашний рабочий день).
- snapshot_id: опциональный ID (по умолчанию выбирается итоговый вечерний срез Y).
"""
date_clean = str(target_date or DATE_YESTERDAY).replace('_', '.')
df_scud = load_scud_from_db_by_snapshot(date_clean, snapshot_param=snapshot_id)
if df_scud is None or df_scud.empty:
return {
"status": "error",
"message": f"Итоговый срез СКУД (Y) за {date_clean} не найден в базе данных."
}
df_staff, df_abs = load_1c_files_for_date(date_clean)
df_merged = merge_scud_and_1c(df_scud, df_staff, df_abs)
filename = f"{format_date_ru(date_clean)} отчет.xlsx"
generate_detailed_excel(df_merged, date_str=date_clean, filename=filename)
target_dir = get_dated_reports_dir(date_clean)
full_filepath = os.path.join(target_dir, filename)
return {
"status": "success",
"report_type": "OTCHET",
"date": date_clean,
"snapshot_id": snapshot_id or "AUTO_Y_FINAL",
"filename": filename,
"filepath": full_filepath,
"download_url": f"/api/v1/files/download/reports/{os.path.basename(full_filepath)}",
"total_rows": len(df_merged),
"message": f"Детальный отчет за {date_clean} успешно сформирован."
}
services/scud_etl/pipeline.py
+39 -34
View File
@@ -1,59 +1,64 @@
"""
===============================================================================
FILE: services/scud_etl/pipeline.py
ROLE: Выборка данных СКУД из SQLite (scud_logs) с жестким приоритетом Y-снапшотов.
===============================================================================
"""
import os
import logging
from typing import Optional, Dict, Any, Tuple
import pandas as pd
from core.connection import get_connection
from services.data_loader import load_1c_data_smart
from core.database import load_scud_from_db_by_snapshot
from config import DATA_DIR
from services.data_loader import load_staff_data, load_absent_data
logger = logging.getLogger("SCUD_PIPELINE")
def load_best_snapshot_for_date(date_str: str, prefer_final_y: bool = True) -> pd.DataFrame:
"""Извлекает срез СКУД. Для вчерашнего дня строго берет финальный вечерний срез Y."""
def load_best_snapshot_for_date(date_str: str, prefer_final_y: bool = False) -> Optional[pd.DataFrame]:
"""
Загружает наилучший срез СКУД за дату.
Если prefer_final_y=True — отдает предпочтение финишному Y (23:59:59).
"""
with get_connection() as conn:
df = pd.DataFrame()
cursor = conn.cursor()
target_snap_id = None
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,)
)
cursor.execute("""
SELECT snapshot_id
FROM scud_logs
WHERE log_date = ?
AND (snapshot_id LIKE 'Y%' OR snapshot_time LIKE '%23:59:59' OR snapshot_time LIKE '%22:00:00')
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 row:
target_snap_id = row[0]
if df.empty:
cursor.execute(
"SELECT snapshot_id FROM scud_logs WHERE log_date = ? ORDER BY id DESC LIMIT 1",
(date_str,)
)
if not target_snap_id:
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 row:
target_snap_id = 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)
if not target_snap_id:
return None
return df
return load_scud_from_db_by_snapshot(date_str, snapshot_param=target_snap_id)
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())
def load_1c_files_for_date(date_str: str) -> Tuple[Optional[pd.DataFrame], Optional[pd.DataFrame]]:
"""
Загружает реестры штата и отсутствий 1С на указанную дату через data_loader.
"""
df_staff = load_staff_data(date_str)
df_abs = load_absent_data(date_str)
return df_staff, df_abs
services/scud_etl/svodka_generator.py
+80
View File
@@ -0,0 +1,80 @@
"""
===============================================================================
FILE: services/scud_etl/svodka_generator.py
ROLE: Генератор Ежедневной Сводки (оперативный контроль, текущий срез).
===============================================================================
"""
import os
import logging
from typing import Dict, Any, Optional
import pandas as pd
from config import DATE_TODAY
from core.database import load_scud_from_db_by_snapshot
from services.scud_etl.pipeline import load_1c_files_for_date
from services.scud_etl.merger import merge_scud_and_1c, calculate_summary_metrics
from services.scud_etl.anomaly_detector import detect_registry_anomalies
from services.snapshots.finder import find_or_create_snapshot_for_time
from services.excel_exporter import generate_summary_excel, get_dated_reports_dir, format_date_ru
logger = logging.getLogger("SVODKA_GENERATOR")
def generate_svodka_service(
target_date: Optional[str] = None,
target_time: Optional[str] = None,
snapshot_id: Optional[str] = None
) -> Dict[str, Any]:
"""
Формирует оперативную сводку на указанную дату / время:
- target_date: дата сводки (по умолчанию сегодня).
- target_time: время среза (например '14:30').
- snapshot_id: точный ID среза.
"""
date_clean = str(target_date or DATE_TODAY).replace('_', '.')
applied_note = ""
# Если передано время, но не указан конкретный snapshot_id — ищем ближайший или запрашиваем экспорт
if target_time and not snapshot_id:
found_id, note = find_or_create_snapshot_for_time(date_clean, target_time, allow_ondemand_export=True)
snapshot_id = found_id
applied_note = note
if note:
logger.info(note)
df_scud = load_scud_from_db_by_snapshot(date_clean, snapshot_param=snapshot_id)
if df_scud is None or df_scud.empty:
return {
"status": "error",
"message": f"Срез СКУД за {date_clean} ({applied_note or snapshot_id or 'последний доступный'}) не найден в базе."
}
df_staff, df_abs = load_1c_files_for_date(date_clean)
df_merged = merge_scud_and_1c(df_scud, df_staff, df_abs)
metrics = calculate_summary_metrics(df_merged)
anomalies = detect_registry_anomalies(df_merged, df_raw_scud=df_scud)
# Добавляем суффикс времени в имя файла, если сводка строилась на точный срез
time_suffix = f" на {target_time.replace(':', '-')}" if target_time else ""
filename = f"{format_date_ru(date_clean)} сводка{time_suffix}.xlsx"
generate_summary_excel(df_merged, date_str=date_clean, filename=filename)
target_dir = get_dated_reports_dir(date_clean)
full_filepath = os.path.join(target_dir, filename)
return {
"status": "success",
"report_type": "SVODKA",
"date": date_clean,
"target_time": target_time,
"snapshot_id": snapshot_id or "AUTO_LATEST",
"filename": filename,
"filepath": full_filepath,
"download_url": f"/api/v1/files/download/reports/{os.path.basename(full_filepath)}",
"metrics": metrics,
"anomalies_count": len(anomalies),
"note": applied_note,
"message": f"Ежедневная сводка на {date_clean} {target_time or ''} успешно сформирована."
}
services/scud_export.py
+1 -1
View File
@@ -237,7 +237,7 @@ def run_export(input_date: str | None = None, save_xlsx: bool = True, debug: boo
continue
if is_yesterday:
snapshot_time = f"{processing_date.strftime('%Y-%m-%d')} 22:00:00"
snapshot_time = f"{processing_date.strftime('%Y-%m-%d')} 23:59:59"
else:
snapshot_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
services/snapshots/finder.py
+89
View File
@@ -0,0 +1,89 @@
"""
===============================================================================
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} не найден."