""" =============================================================================== 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} успешно сформирован." }