""" =============================================================================== FILE: services/scud_export.py ROLE: Прямой экспорт данных СКУД Орион (MS SQL) в SQLite и чистый Excel (XlsxWriter). Корректная фильтрация транзитных проходов турникетов парковки и двора. Учет только левого PERCo (DoorIndex = 1), правого PERCo (DoorIndex = 2) и входа через Флигель (DoorIndex = 23). =============================================================================== """ import argparse import logging import os import sys import warnings from datetime import datetime, timedelta CURRENT_DIR = os.path.dirname(os.path.abspath(__file__)) ROOT_DIR = os.path.abspath(os.path.join(CURRENT_DIR, "..")) if ROOT_DIR not in sys.path: sys.path.insert(0, ROOT_DIR) import pandas as pd import pyodbc import xlsxwriter from services.scud_etl.sql_queries import SQL_QUERY_TEMPLATE, SQL_RAW_EVENTS_QUERY from config import SCUD_DIR, clean_scud_fio_light, load_exceptions from core.database import save_scud_to_db, has_scud_logs_for_date, has_yesterday_final_snapshot from core.repositories.scud_repo import save_raw_events_to_db warnings.filterwarnings("ignore", message="pandas only supports SQLAlchemy connectable") SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__)) LOG_DIR = os.path.join(SCRIPT_DIR, "..", "logs") os.makedirs(LOG_DIR, exist_ok=True) TODAY_DATE_STR = datetime.now().strftime("%d.%m.%Y") LOG_FILE = os.path.join(LOG_DIR, f"export_{TODAY_DATE_STR}.log") logger = logging.getLogger("scud_export") logger.setLevel(logging.INFO) logger.propagate = False if not logger.handlers: try: _file_handler = logging.FileHandler(LOG_FILE, encoding="utf-8") _formatter = logging.Formatter("[%(asctime)s] [%(levelname)s] %(message)s", datefmt="%Y-%m-%d %H:%M:%S") _file_handler.setFormatter(_formatter) logger.addHandler(_file_handler) except (PermissionError, OSError) as e: sys.stderr.write(f"Предупреждение: невозможно создать лог-файл {LOG_FILE}: {e}\n") _console_handler = logging.StreamHandler(sys.stdout) _formatter = logging.Formatter("[%(asctime)s] [%(levelname)s] %(message)s", datefmt="%Y-%m-%d %H:%M:%S") _console_handler.setFormatter(_formatter) logger.addHandler(_console_handler) def log(message: str, level: str = "INFO"): level_map = { "INFO": logging.INFO, "ERROR": logging.ERROR, "WARNING": logging.WARNING, "SUCCESS": logging.INFO, } if level == "SUCCESS": logger.info(f"[SUCCESS] {message}") else: logger.log(level_map.get(level, logging.INFO), message) SERVER_NAME = r"172.16.200.147\SQL" DATABASE_NAME = "Orion-14.01.21-1" SQL_USER = "sa" SQL_PASSWORD = "123456" ODBC_DRIVER = "ODBC Driver 18 for SQL Server" SQL_QUERY_TEMPLATE = r""" DECLARE @InputDate DATE = '{target_date}'; DECLARE @TargetDate DATE = @InputDate; DECLARE @StartDate DATETIME = CAST(@TargetDate AS DATETIME); DECLARE @EndDate DATETIME = {end_datetime_sql}; WITH PercoPassages AS ( -- Физические факты прохода (Event = 32) SELECT log.HozOrgan AS EmployeeID, log.TimeVal, CASE WHEN log.Mode = 1 THEN 'IN' WHEN log.Mode = 2 THEN 'OUT' ELSE 'OTHER' END AS Direction FROM pLogData log WITH (NOLOCK) INNER JOIN pList p WITH (NOLOCK) ON log.HozOrgan = p.ID LEFT JOIN PDivision div WITH (NOLOCK) ON p.Section = div.ID WHERE log.TimeVal BETWEEN @StartDate AND @EndDate AND log.HozOrgan IS NOT NULL AND log.HozOrgan > 0 AND log.Event = 32 AND log.Mode IN (1, 2) AND ( -- Контур 1: Левый турникет открыт для всех log.DoorIndex = 1 OR -- Контур 2: Правый турникет разрешен только для реестра двора ( log.DoorIndex = 2 AND ({turnstile_filter_sql}) ) OR -- Контур 3: Флигель 1 эт. разрешен только для реестра флигеля ( log.DoorIndex = 23 AND ({fligel_filter_sql}) ) ) ), Passages AS ( SELECT EmployeeID, MIN(TimeVal) AS FirstRawEvent, MAX(TimeVal) AS LastRawEvent, MIN(CASE WHEN Direction = 'IN' THEN TimeVal END) AS FirstIn, MAX(CASE WHEN Direction = 'OUT' THEN TimeVal END) AS FinalOut FROM PercoPassages GROUP BY EmployeeID ), EvaluatedPassages AS ( SELECT p.*, CASE WHEN p.FinalOut IS NOT NULL AND p.FirstIn IS NOT NULL AND p.FinalOut > DATEADD(MINUTE, 5, p.FirstIn) THEN p.FinalOut ELSE NULL END AS FilteredLastOut FROM Passages p ) SELECT N'ЛЕНМОРНИИПРОЕКТ' AS [Фирма], ISNULL(CAST(div.Name AS NVARCHAR(255)), N'Без подразделения') AS [Подразделение], LTRIM(RTRIM( ISNULL(CAST(p.Name AS NVARCHAR(255)), N'') + CASE WHEN p.FirstName IS NOT NULL AND CAST(p.FirstName AS NVARCHAR(255)) <> '' THEN N' ' + CAST(p.FirstName AS NVARCHAR(255)) ELSE N'' END + CASE WHEN p.MidName IS NOT NULL AND CAST(p.MidName AS NVARCHAR(255)) <> '' THEN N' ' + CAST(p.MidName AS NVARCHAR(255)) ELSE N'' END )) AS [Сотрудник], ISNULL(CAST(post.Name AS NVARCHAR(255)), N'—') AS [Должность], ISNULL(CAST(p.TabNumber AS NVARCHAR(50)), N'—') AS [Таб_№], CONVERT(VARCHAR(10), @TargetDate, 104) AS [Дата], ISNULL(CAST(CONVERT(VARCHAR(8), pass.FirstIn, 108) AS NVARCHAR(20)), N'Нет входа') AS [Начало_дня], CASE WHEN pass.FirstIn IS NULL AND pass.FirstRawEvent IS NOT NULL THEN CAST(CONVERT(VARCHAR(8), pass.FirstRawEvent, 108) AS NVARCHAR(20)) ELSE N'—' END AS [Первая_активность], CASE WHEN pass.FilteredLastOut IS NOT NULL THEN CAST(CONVERT(VARCHAR(8), pass.FilteredLastOut, 108) AS NVARCHAR(20)) ELSE N'Нет выхода' END AS [Конец_дня], CASE WHEN pass.EmployeeID IS NOT NULL AND (pass.FirstIn IS NOT NULL OR pass.FirstRawEvent IS NOT NULL) THEN RIGHT('0' + CAST(DATEDIFF(MINUTE, ISNULL(pass.FirstIn, pass.FirstRawEvent), CASE WHEN pass.FilteredLastOut IS NOT NULL THEN pass.FilteredLastOut ELSE @EndDate END) / 60 AS VARCHAR), 2) + ':' + RIGHT('0' + CAST(DATEDIFF(MINUTE, ISNULL(pass.FirstIn, pass.FirstRawEvent), CASE WHEN pass.FilteredLastOut IS NOT NULL THEN pass.FilteredLastOut ELSE @EndDate END) % 60 AS VARCHAR), 2) ELSE N'00:00' END AS [Находился_в_здании], CASE WHEN pass.EmployeeID IS NOT NULL THEN N'Присутствовал' ELSE N'Отсутствовал (Нет событий)' END AS [Статус] FROM pList p WITH (NOLOCK) LEFT JOIN PDivision div WITH (NOLOCK) ON p.Section = div.ID LEFT JOIN PPost post WITH (NOLOCK) ON p.Post = post.ID LEFT JOIN EvaluatedPassages pass ON p.ID = pass.EmployeeID WHERE ISNULL(p.StatusRecord, 0) = 0 AND p.DateTimeInArchive IS NULL AND ISNULL(CAST(div.Name AS NVARCHAR(255)), N'') NOT LIKE N'Аренд%' AND ISNULL(CAST(div.Name AS NVARCHAR(255)), N'') NOT IN (N'Без подразделения', N'') AND p.Name NOT LIKE N'бр.%' AND p.Name NOT LIKE N'Гость%' AND ISNULL(CAST(div.Name AS NVARCHAR(255)), N'') NOT IN (N'БГИ', N'КНР') AND ISNULL(CAST(div.Name AS NVARCHAR(255)), N'') NOT LIKE N'Рабоч%' AND ISNULL(CAST(div.Name AS NVARCHAR(255)), N'') NOT LIKE N'Врем%' AND ISNULL(CAST(div.Name AS NVARCHAR(255)), N'') NOT LIKE N'Практика%' AND ISNULL(CAST(div.Name AS NVARCHAR(255)), N'') NOT LIKE N'тест%' AND ISNULL(CAST(post.Name AS NVARCHAR(255)), N'') NOT LIKE N'Практикант%' ORDER BY p.Name ASC; """ SQL_RAW_EVENTS_QUERY = r""" DECLARE @InputDate DATE = '{target_date}'; DECLARE @StartDate DATETIME = CAST(@InputDate AS DATETIME); DECLARE @EndDate DATETIME = {end_datetime_sql}; SELECT log.TimeVal, log.HozOrgan, LTRIM(RTRIM( ISNULL(CAST(p.Name AS NVARCHAR(255)), N'') + CASE WHEN p.FirstName IS NOT NULL AND CAST(p.FirstName AS NVARCHAR(255)) <> '' THEN N' ' + CAST(p.FirstName AS NVARCHAR(255)) ELSE N'' END + CASE WHEN p.MidName IS NOT NULL AND CAST(p.MidName AS NVARCHAR(255)) <> '' THEN N' ' + CAST(p.MidName AS NVARCHAR(255)) ELSE N'' END )) AS [Сотрудник], ISNULL(CAST(div.Name AS NVARCHAR(255)), N'Без подразделения') AS [Подразделение], log.Event, log.Mode, log.DoorIndex, CASE WHEN log.Mode = 2 THEN 'OUT' WHEN log.Mode = 1 THEN 'IN' WHEN log.Event IN (2, 27, 29, 33, 55, 65) THEN 'OUT' WHEN log.Event IN (1, 21, 26, 54, 64) THEN 'IN' ELSE 'OTHER' END AS Direction FROM pLogData log WITH (NOLOCK) INNER JOIN pList p WITH (NOLOCK) ON log.HozOrgan = p.ID LEFT JOIN PDivision div WITH (NOLOCK) ON p.Section = div.ID LEFT JOIN PPost post WITH (NOLOCK) ON p.Post = post.ID WHERE log.TimeVal BETWEEN @StartDate AND @EndDate AND log.HozOrgan IS NOT NULL AND log.HozOrgan > 0 AND log.Event IN (28, 32) -- ⭐️ Фильтрация сотрудников Ленморниипроект (исключение арендаторов и гостей) AND ISNULL(p.StatusRecord, 0) = 0 AND p.DateTimeInArchive IS NULL AND ISNULL(CAST(div.Name AS NVARCHAR(255)), N'') NOT LIKE N'Аренд%' AND ISNULL(CAST(div.Name AS NVARCHAR(255)), N'') NOT IN (N'Без подразделения', N'') AND p.Name NOT LIKE N'бр.%' AND p.Name NOT LIKE N'Гость%' AND ISNULL(CAST(div.Name AS NVARCHAR(255)), N'') NOT IN (N'БГИ', N'КНР') AND ISNULL(CAST(div.Name AS NVARCHAR(255)), N'') NOT LIKE N'Рабоч%' AND ISNULL(CAST(div.Name AS NVARCHAR(255)), N'') NOT LIKE N'Врем%' AND ISNULL(CAST(div.Name AS NVARCHAR(255)), N'') NOT LIKE N'Практика%' AND ISNULL(CAST(div.Name AS NVARCHAR(255)), N'') NOT LIKE N'тест%' AND ISNULL(CAST(post.Name AS NVARCHAR(255)), N'') NOT LIKE N'Практикант%' ORDER BY log.TimeVal ASC; """ def save_df_to_clean_excel(df: pd.DataFrame, file_path: str, sheet_name: str = "Отчет"): workbook = xlsxwriter.Workbook(file_path, {'constant_memory': False}) worksheet = workbook.add_worksheet(sheet_name) fmt_header = workbook.add_format({ 'bold': True, 'bg_color': '#D9E1F2', 'border': 1, 'border_color': '#D3D3D3', 'align': 'center', 'valign': 'vcenter', 'font_name': 'Calibri', 'font_size': 11 }) fmt_cell = workbook.add_format({ 'border': 1, 'border_color': '#D3D3D3', 'valign': 'vcenter', 'align': 'left', 'font_name': 'Calibri', 'font_size': 11 }) headers = list(df.columns) col_widths = [len(str(h)) for h in headers] for col_idx, header in enumerate(headers): worksheet.write(0, col_idx, str(header), fmt_header) for row_idx, row_values in enumerate(df.values, start=1): for col_idx, val in enumerate(row_values): if pd.isna(val) or val is None: val_str = "" elif isinstance(val, bool): val_str = "Да" if val else "Нет" else: val_str = str(val) worksheet.write(row_idx, col_idx, val_str, fmt_cell) if len(val_str) > col_widths[col_idx]: col_widths[col_idx] = len(val_str) for col_idx, width in enumerate(col_widths): worksheet.set_column(col_idx, col_idx, min(max(width + 3, 10), 45)) workbook.close() def get_targets(input_date: str | None, input_time: str | None = None): targets = [] if input_date: try: parsed = datetime.strptime(input_date.replace('_', '.'), "%d.%m.%Y").date() targets.append({"name": "Указанная дата", "date": parsed, "target_time": input_time}) except ValueError: log(f"ОШИБКА: Неверный формат даты '{input_date}'. Используйте ДД.ММ.ГГГГ", "ERROR") sys.exit(1) else: now = datetime.now() yesterday = (now - timedelta(days=3 if now.weekday() == 0 else 1)).date() today = now.date() targets.append({"name": "Вчера", "date": yesterday, "target_time": None}) targets.append({"name": "Сегодня", "date": today, "target_time": None}) return targets def run_export(input_date: str | None = None, input_time: str | None = None, save_xlsx: bool = True, debug: bool = False): if debug: logger.setLevel(logging.DEBUG) log("=== ВКЛЮЧЕН РЕЖИМ ОТЛАДКИ (DEBUG MODE) ===", "WARNING") log("=== [ЭТАП 0] Выгрузка свежих данных СКУД напрямую из БД Орион ===") os.makedirs(SCUD_DIR, exist_ok=True) # ⭐️ 1. ПОЛУЧАСОВАЯ СИНХРОНИЗАЦИЯ 1С:ЗУП (Файлы с шары + База данных) try: log("[1C] Проверка сетевой шары и синхронизация файлов 1С...") copy_1c_files_from_share() target_sync_date = input_date or datetime.now().strftime("%d.%m.%Y") # Прямой опрос свежих отсутствий и сохранение в SQLite df_fresh_abs = load_absent_data(target_sync_date) if df_fresh_abs is not None and not df_fresh_abs.empty: save_absences_to_db(df_fresh_abs, target_sync_date) log(f"[1C] [✓] Актуализировано {len(df_fresh_abs)} отсутствий в SQLite за {target_sync_date}", "SUCCESS") # Если на шаре появился свежий штат — также обновляем его в базе df_fresh_staff = load_staff_data(target_sync_date) if df_fresh_staff is not None and not df_fresh_staff.empty: save_staff_to_db(df_fresh_staff, target_sync_date) log(f"[1C] [✓] Актуализирован штат ({len(df_fresh_staff)} чел.) в SQLite за {target_sync_date}", "SUCCESS") except Exception as e: log(f"[1C] ⚠️ Ошибка фоновой синхронизации 1С: {e}", "WARNING") targets = get_targets(input_date, input_time) conn_str = ( f"DRIVER={{{ODBC_DRIVER}}};" f"SERVER={SERVER_NAME};" f"DATABASE={DATABASE_NAME};" f"UID={SQL_USER};" f"PWD={SQL_PASSWORD};" f"TrustServerCertificate=yes;" f"Encrypt=no;" ) success = True for target in targets: processing_date = target["date"] processing_date_str = processing_date.strftime("%d.%m.%Y") period_label = target["name"] target_time = target["target_time"] today_date = datetime.now().date() is_past_day = (processing_date < today_date) or (period_label == "Вчера") if is_past_day and not target_time and has_yesterday_final_snapshot(processing_date_str): log(f"[ℹ️] День ({processing_date_str}) уже зафиксирован финишным снапшотом _FINAL. Пропускаем.") continue exc_data = load_exceptions() # 1. Формируем фильтр правого турникета (DoorIndex = 2) t_fios = [f.replace("'", "''") for f in exc_data.get('turnstile_fio', []) if f] t_depts = [d.replace("'", "''") for d in exc_data.get('turnstile_departments', []) if d] full_fio_sql = ( "LTRIM(RTRIM(" "ISNULL(CAST(p.Name AS NVARCHAR(255)), N'') + " "CASE WHEN p.FirstName IS NOT NULL AND CAST(p.FirstName AS NVARCHAR(255)) <> '' THEN N' ' + CAST(p.FirstName AS NVARCHAR(255)) ELSE N'' END + " "CASE WHEN p.MidName IS NOT NULL AND CAST(p.MidName AS NVARCHAR(255)) <> '' THEN N' ' + CAST(p.MidName AS NVARCHAR(255)) ELSE N'' END" "))" ) conditions = [] if t_fios: fio_in = ", ".join([f"N'{f}'" for f in t_fios]) conditions.append(f"{full_fio_sql} IN ({fio_in})") if t_depts: dept_in = ", ".join([f"N'{d}'" for d in t_depts]) conditions.append(f"ISNULL(CAST(div.Name AS NVARCHAR(255)), N'') IN ({dept_in})") turnstile_filter_sql = " OR ".join(conditions) if conditions else "1 = 0" # 1.1 Формируем фильтр Флигеля (DoorIndex = 23) fl_fios = [f.replace("'", "''") for f in exc_data.get('fligel_fio', []) if f] fl_depts = [d.replace("'", "''") for d in exc_data.get('fligel_departments', []) if d] fl_conditions = [] if fl_fios: fl_fio_in = ", ".join([f"N'{f}'" for f in fl_fios]) fl_conditions.append(f"{full_fio_sql} IN ({fl_fio_in})") if fl_depts: fl_dept_in = ", ".join([f"N'{d}'" for d in fl_depts]) fl_conditions.append(f"ISNULL(CAST(div.Name AS NVARCHAR(255)), N'') IN ({fl_dept_in})") fligel_filter_sql = " OR ".join(fl_conditions) if fl_conditions else "1 = 0" # 2. Безопасное математическое определение @EndDate через DATEADD if target_time: t_clean = target_time.strip() t_parts = t_clean.split(":") h = int(t_parts[0]) m = int(t_parts[1]) if len(t_parts) > 1 else 0 s = int(t_parts[2]) if len(t_parts) > 2 else 0 snapshot_time = f"{processing_date.strftime('%Y-%m-%d')} {h:02d}:{m:02d}:{s:02d}" end_datetime_sql = f"DATEADD(SECOND, {s}, DATEADD(MINUTE, {m}, DATEADD(HOUR, {h}, @StartDate)))" is_final = False elif is_past_day: snapshot_time = f"{processing_date.strftime('%Y-%m-%d')} 23:59:59" end_datetime_sql = "DATEADD(SECOND, -1, DATEADD(DAY, 1, @StartDate))" is_final = True else: snapshot_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S") end_datetime_sql = "GETDATE()" is_final = False log(f"--- Обработка периода: {period_label} ({processing_date_str}) --- [Срез: {snapshot_time}]") sql_query = SQL_QUERY_TEMPLATE.format( target_date=processing_date.strftime("%Y-%m-%d"), end_datetime_sql=end_datetime_sql, turnstile_filter_sql=turnstile_filter_sql, fligel_filter_sql=fligel_filter_sql ) connection = None try: connection = pyodbc.connect(conn_str, timeout=120) df = pd.read_sql(sql_query, connection) log(f"База вернула {len(df)} строк за {processing_date_str}") if len(df) > 0: df['fio_clean'] = df['Сотрудник'].apply(clean_scud_fio_light) df['anomaly_flag'] = 'NONE' mask_anomaly = (df['Начало_дня'] == 'Нет входа') & (df['Первая_активность'] != '—') df.loc[mask_anomaly, 'anomaly_flag'] = 'ANOMALY_NO_IN_HAS_ACTIVITY' df['Пришел'] = df['Статус'].str.contains('Присутствовал', case=False, na=False) & (~mask_anomaly) save_scud_to_db(df, processing_date_str, snapshot_time=snapshot_time, is_yesterday=is_final) raw_sql = SQL_RAW_EVENTS_QUERY.format( target_date=processing_date.strftime("%Y-%m-%d"), end_datetime_sql=end_datetime_sql ) df_raw = pd.read_sql(raw_sql, connection) if len(df_raw) > 0: df_raw['fio_clean'] = df_raw['Сотрудник'].apply(clean_scud_fio_light) inserted_count = save_raw_events_to_db(df_raw, processing_date_str) log(f"[✓] В scud_events_raw сохранено {inserted_count} сырых событий за {processing_date_str}!", "SUCCESS") log(f"[✓] Записи за {processing_date_str} успешно сохранены в SQLite!", "SUCCESS") if save_xlsx: file_name = f"Сотрудники_{processing_date_str}.xlsx" file_path = os.path.join(SCUD_DIR, file_name) if os.path.exists(file_path): try: os.remove(file_path) except OSError as e: log(f"ОШИБКА при удалении старого файла {file_name}: {e}", "ERROR") save_df_to_clean_excel(df, file_path, sheet_name="Отчет") log(f"[✓] Успешно экспортирован файл: data/scud/{file_name}", "SUCCESS") else: log(f"Запрос за {processing_date_str} вернул 0 строк.", "WARNING") except Exception as e: log(f"🛑 ОШИБКА выгрузки СКУД за {processing_date_str}: {e}", "ERROR") success = False finally: if connection is not None: connection.close() log("=== Выгрузка СКУД завершена ===\n") return success if __name__ == "__main__": parser = argparse.ArgumentParser() parser.add_argument("--date", dest="input_date", default=None, help="Дата среза (ДД.ММ.ГГГГ)") parser.add_argument("--time", dest="input_time", default=None, help="Время среза (ЧЧ:ММ)") parser.add_argument("-d", "--debug", action="store_true") parser.add_argument("--no-xlsx", dest="save_xlsx", action="store_false", default=True) args = parser.parse_args() run_export(args.input_date, input_time=args.input_time, save_xlsx=args.save_xlsx, debug=args.debug)