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

Первичная зашифрованная версия проекта

Browse Source
This commit is contained in:
manoraga
2026-07-27 21:10:59 +03:00
commit 97b4be7162
14 changed files with 947 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+9
View File
@@ -0,0 +1,9 @@
# Входные и сгенерированные файлы
data/*.xlsx
output/*
# Системные и временные файлы Python
__pycache__/
*.pyc
.venv/
.vscode/
PROJECT_SNAPSHOT.md
+133
View File
@@ -0,0 +1,133 @@
# Архитектура и документация проекта СКУД v2
## 1. Фактическая структура проекта
```text
scud_root_v2/
│
├── config.py # Конфигурация, глобальные пути (DATA_DIR, OUTPUT_DIR), даты (DATE_TODAY, DATE_YESTERDAY)
├── exceptions.json # Фильтры исключений (ОВК, клинеры, конкретные ФИО)
├── PROJECT_SNAPSHOT.md # 📄 Архитектурный снимок состояния проекта
├── main.py # Главный скрипт запуска и оркестрации всего конвейера
│
├── data/ # 📂 Входные файлы (Выгрузки СКУД и 1С)
│ ├── knowledge_base.json # Накопленная база знаний и динамические правила компании
│ ├── Штатные сотрудники - ОУП.xlsx # Выгрузка штата из 1С
│ ├── Отсутствия сотрудников - выгрузка.xlsx # Выгрузка докум. отсутствий из 1С
│ ├── Сотрудники_24.07.2026.xlsx # Выгрузка СКУД за ВЧЕРА (для детального отчета)
│ └── Сотрудники_27.07.2026.xlsx # Выгрузка СКУД за СЕГОДНЯ (для оперативной сводки)
│
├── modules/ # 📂 Исполняемые модули системы
│ ├── __init__.py
│ ├── data_validator.py # 🛡️ Детерминированный контроль актуальности и свежести входных файлов СКУД/1С
│ ├── data_loader.py # Загрузка и объединение Excel-выгрузок СКУД и 1С
│ ├── ai_verifier.py # Интеграция с Ollama (qwen2.5:14b) для исправления опечаток в ФИО из СКУД
│ ├── excel_exporter.py # 📊 Единый модуль генерации Excel-файлов (Сводка + Детальный отчет за вчера)
│ ├── text_reporter.py # 📝 ИИ-аудитор (формирование текстовой сводки с проверкой совмещений)
│ ├── knowledge_base.py # Чтение/запись динамических правил компании
│ └── feedback_loop.py # Модуль обратной связи для обучения базы знаний
│
└── output/ # 📂 Результаты работы (генерируются автоматически):
├── 26 июля 2026 отчет.xlsx
├── 27 июля 2026 сводка.xlsx
└── Сводка_контроллинга_27.07.2026.md
```
---
## 2. Подробное описание модулей и их логики
### ⚙️ `config.py`
**Константы:**
- `OLLAMA_URL = "http://localhost:11434/api/chat"`
- `OLLAMA_MODEL = "qwen2.5:14b"`
- `DATE_TODAY`, `DATE_YESTERDAY` — расчет рабочих дат.
**Основные функции:**
- `normalize_fio()` — приводит ФИО к нижнему регистру, удаляет скобки `(...)`, меняет латинские визуальные дубликаты букв на кириллицу, заменяет `ё` на `е`.
- `clean_scud_fio_light()` — облегченная очистка ФИО из СКУД от системных символов и артефактов.
- `load_exceptions()` — загружает фильтры и исключения из `exceptions.json`.
---
### 🛡️ `modules/data_validator.py`
Модуль детерминированного контроля актуальности данных (**Data Quality & Freshness Control**).
- Проверяет наличие файлов за сегодня и вчера в папке `data/`.
- Проверяет дату и время изменения файлов (`mtime`), предотвращая запуск системы на устаревших выгрузках прошлого дня.
- В случае отсутствия или устаревания данных блокирует запуск основного конвейера и предупреждает оператора.
---
### 🛡️ `exceptions.json`
Содержит конфигурационные списки исключений, отфильтровываемые при сборе аналитики:
- **Департаменты:** отдел ОВК и др.
- **Должности:** Уборщик, Клинер, Дворник и т.д.
- **Персональные исключения:** конкретные ФИО сотрудников или подрядчиков.
---
### 📥 `modules/data_loader.py`
- Выполняет загрузку данных из папки `data/`.
- Определяет факт присутствия на рабочем месте (`Пришел = True`, если зафиксирован проход в СКУД).
- Применяет правила фильтрации `is_excluded()` на основе данных из `exceptions.json`.
---
### 🤖 `modules/ai_verifier.py`
- Отправляет список нераспознанных ФИО из СКУД и эталонный список из 1С в модель Ollama (`qwen2.5:14b`).
- Автоматически выявляет опечатки, несоответствия в написании и возвращает JSON сопоставлений.
---
### 📊 `modules/excel_exporter.py`
Универсальный модуль формирования отчетов в формате Excel. Автоматически формирует имена файлов с названием месяца прописью (например, `27 июля 2026 сводка.xlsx` и `26 июля 2026 отчет.xlsx`).
#### 1. Оперативная сводка за сегодня (`generate_summary_excel`):
- **Древовидная группировка:** Категория *«неизвестно»* раскрыта по умолчанию, остальные категории свернуты под плюсики `[+]` / `[-]`.
- **Настройка структуры:** Принудительно включено отображение боковой панели со значками группировки через `showOutlineSymbols = True`.
- **Палитра:**
- Шапка: `#D9E1F2` *(Светло-голубой)*
- По списку: `#F2F2F2` *(Светло-серый)*
- Неизвестно: `#FCE4D6` *(Персиковый)*
- Категории 1С: Циклическая пастельная палитра (`#FFF2CC`, `#E1D5E7`, `#E1F5FE`, `#FFF0F5`, `#E8F8F5`, `#FCF3CF`)
- Итого на работе: `#E2EFDA` *(Светло-зеленый)*
#### 2. Детальный отчет за вчера (`generate_detailed_excel`):
- **Динамическая ширина колонок:** Ширина столбцов рассчитывается по длине самого широкого значения ячеек.
- **Столбец G (Причина отсутствия):** Ширина фиксирована на `25.0` (180 px), включен автоперенос текста (`wrap_text = True`) с адаптивным расширением высоты строк под длинные тексты.
- **Цветовая маркировка:**
- `#FFF2CC` *(Желтый)* — есть официальная причина в 1С.
- `#FCE4D6` *(Светло-красный)* — приход отсутствует, документ в 1С не найден.
---
### 📝 `modules/text_reporter.py`
- Формирует с помощью Ollama текстовую Markdown-сводку для руководства.
- **Форматирование нумерованного списка неизвестных:** Строгий формат `N. ФИО — [Аббревиатура], Должность`.
- **Контроль математики:** Количество человек в поле *«Неизвестно»* строго совпадает с количеством элементов в сгенерированном нумерованном списке.
- **Подсветка аномалий:** Выявление совмещений и сотрудников с несколькими табельными номерами/картами СКУД (например, Кондрашова Е.В.).
---
### 🚀 `main.py`
Главный исполнительный модуль системы. Координирует конвейер обработки:
1. `check_file_freshness()` — **Этап 0:** жесткая проверка свежести файлов.
2. `load_all_data()` — загрузка реестров 1С и СКУД.
3. `ai_verify()` — сопоставление опечаток в ФИО через ИИ.
4. `merge()` — связывание, фильтрация и категоризация данных.
5. **Генерация отчетов** — создание `[дата] сводка.xlsx` и `[дата] отчет.xlsx`.
6. **Вывод ИИ-сводки** — формирование и сохранение текстового итогового отчета в `output/`.
---
## 3. Текущие и будущие задачи
### 🗄️ Разворачивание локальной базы данных
- Переход от плоских файлов Excel к локальной СУБД (SQLite / PostgreSQL).
- Хранение истории проходов и статусов сотрудников с временными метками для быстрого формирования аналитики за любой выбранный период.
### 🔄 Переход на прямое подключение к СКУД через T-SQL
- Настройка прямого подключения Python к базе данных СКУД через `pyodbc` / `sqlalchemy`.
- Отказ от ручной выгрузки Excel-файлов СКУД оператором.
config.py
+76
View File
@@ -0,0 +1,76 @@
import os
import re
import json
import pandas as pd
from datetime import datetime, timedelta
# 📁 Пути к директориям
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
DATA_DIR = os.path.join(BASE_DIR, "data")
OUTPUT_DIR = os.path.join(BASE_DIR, "output")
os.makedirs(DATA_DIR, exist_ok=True)
os.makedirs(OUTPUT_DIR, exist_ok=True)
# 📅 Динамический расчет дат (СЕГОДНЯ и ВЧЕРА)
def get_controlling_dates():
now = datetime.now()
if now.weekday() == 0: # Если Понедельник -> Вчера = Пятница
yesterday_dt = now - timedelta(days=3)
else:
yesterday_dt = now - timedelta(days=1)
date_today_str = now.strftime("%d.%m.%Y")
date_yesterday_str = yesterday_dt.strftime("%d.%m.%Y")
return date_today_str, date_yesterday_str
DATE_TODAY, DATE_YESTERDAY = get_controlling_dates()
# 📄 Пути к файлам
SCUD_FILE_TODAY = os.path.join(DATA_DIR, f"Сотрудники_{DATE_TODAY}.xlsx")
SCUD_FILE_YESTERDAY = os.path.join(DATA_DIR, f"Сотрудники_{DATE_YESTERDAY}.xlsx")
SCUD_FILE = SCUD_FILE_TODAY if os.path.exists(SCUD_FILE_TODAY) else SCUD_FILE_YESTERDAY
STAFF_FILE = os.path.join(DATA_DIR, "Штатные сотрудники - ОУП.xlsx")
ABSENT_FILE = os.path.join(DATA_DIR, "Отсутствия сотрудников - выгрузка.xlsx")
EXCEPTIONS_FILE = os.path.join(BASE_DIR, "exceptions.json")
# 🤖 Настройки Ollama
OLLAMA_URL = "http://10.121.18.227:11434/api/chat"
OLLAMA_MODEL = "qwen2.5:14b"
# 🛠 Очистка и нормализация данных
def normalize_fio(val):
"""Строгая нормализация ФИО для точности сопоставления 1С и СКУД"""
if pd.isna(val):
return ""
val = str(val).strip().lower()
val = re.sub(r'\(.*?\)', '', val)
trans = str.maketrans('aceopxyABCEHKMOPTX', 'асеорхуАВСЕНКМОРТХ')
val = val.translate(trans).replace('ё', 'е')
return re.sub(r'\s+', ' ', val).strip()
def clean_scud_fio_light(val):
"""Легкая очистка ФИО для СКУД"""
if pd.isna(val):
return ""
val = str(val).strip().lower()
val = re.sub(r'\(.*?\)', '', val).replace('ё', 'е')
return re.sub(r'\s+', ' ', val).strip()
def load_exceptions(file_path=EXCEPTIONS_FILE):
"""Загрузка списка исключений из JSON"""
if not os.path.exists(file_path):
return {"departments": [], "positions": [], "fio": [], "position_keywords": []}
with open(file_path, "r", encoding="utf-8") as f:
data = json.load(f)
return {
"departments": [str(d).strip().upper() for d in data.get("departments", [])],
"positions": [str(p).strip().lower() for p in data.get("positions", [])],
"fio": [normalize_fio(f) for f in data.get("fio", [])],
"position_keywords": [str(k).strip().lower() for k in data.get("position_keywords", [])]
}
data/knowledge_base.json
+9
View File
@@ -0,0 +1,9 @@
{
"rules": [
"Кондрашова Елена Владимировна имеет 2 пропуска в СКУД. Если один 'Отсутствовал', а второй с временем входа — считать ПРИШЕДШЕЙ.",
"Сотрудники подразделения ОА с должностью 'Инженер-программист' часто являются внутренними совместителями."
],
"fio_corrections": {
"кондрашава елена владимировна": "Кондрашова Елена Владимировна"
}
}
exceptions.json
+23
View File
@@ -0,0 +1,23 @@
{
"departments": [
"ОВК"
],
"positions": [
"Уборщик производственных помещений",
"Уборщик служебных помещений"
],
"fio": [
"Шуличенко Иван Иванович",
"Таткало Валерий Валерьевич",
"Пухаренко Юрий Владимирович",
"Чуркина Елена Геннадьевна",
"Михалев Сергей Геннадьевич"
],
"position_keywords": [
"уборщик",
"клинер",
"дворник",
"гардероб",
"рабочий по обслуживанию"
]
}
main.py
+101
View File
@@ -0,0 +1,101 @@
import os
import sys
import pandas as pd
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
from config import DATE_TODAY, DATE_YESTERDAY, OUTPUT_DIR
from modules.data_validator import check_file_freshness
from modules.data_loader import load_all_data
from modules.excel_exporter import generate_summary_excel, generate_detailed_excel
from modules.ai_verifier import ai_verify_scud_against_staff
from modules.text_reporter import generate_markdown_report
from modules.knowledge_base import load_knowledge_base
def main():
print("=" * 60)
print("ЗАПУСК СИСТЕМЫ МОДУЛЬНОГО КОНТРОЛЛИНГА СКУД ⟷ 1С")
print("=" * 60)
# 0. Валидация файлов
print("[0/5] Проверка актуальности и свежести входных файлов...")
is_valid, warnings, errors = check_file_freshness()
if warnings:
print("\n--- ⚠️ ПРЕДУПРЕЖДЕНИЯ ОБ АКТУАЛЬНОСТИ ---")
for w in warnings:
print(f" {w}")
print("-" * 45)
if not is_valid:
print("\n" + "!" * 60)
print("🛑 ОСТАНОВКА ВЫПОЛНЕНИЯ: Обнаружены критические ошибки данных!")
for e in errors:
print(f" {e}")
print("!" * 60)
sys.exit(1)
print("[✓] Проверка актуальности успешно пройдена!\n")
# 1. Загрузка
print("[1/5] Загрузка данных СКУД, 1С и исключений...")
raw_scud_df, staff_fios_clean, df_absent = load_all_data()
# 2. ИИ-сверка
print("[2/5] Проверка опечаток и сопоставление ФИО через ИИ...")
scud_unrecognized = raw_scud_df[~raw_scud_df['fio_clean'].isin(staff_fios_clean)]['fio_clean'].tolist()
fio_mapping = ai_verify_scud_against_staff(scud_unrecognized, staff_fios_clean)
if fio_mapping:
print(f" [ИИ] Найдено и исправлено совпадений/опечаток: {len(fio_mapping)}")
raw_scud_df['fio_clean'] = raw_scud_df['fio_clean'].apply(
lambda x: fio_mapping[x]['staff_fio'] if x in fio_mapping else x
)
# 3. Объединение
print("[3/5] Объединение реестров 1С и СКУД...")
merged = raw_scud_df[raw_scud_df['fio_clean'].isin(staff_fios_clean)].copy()
merged = merged.merge(df_absent, on='fio_clean', how='left')
absent_explained = merged[(merged['Пришел'] == False) & (merged['Вид_отсутствия'].notna())]
absent_unexplained = merged[(merged['Пришел'] == False) & (merged['Вид_отсутствия'].isna())]
scud_present_but_absent_in_1c = merged[(merged['Пришел'] == True) & (merged['Вид_отсутствия'].notna())]
# 4. Формирование Excel-файлов (автоматическое наименование вида "27 июля 2026 сводка.xlsx")
print("[4/5] Формирование Excel-файлов...")
generate_summary_excel(
merged_df=merged,
date_str=DATE_TODAY
)
generate_detailed_excel(
merged_df=merged,
date_str=DATE_YESTERDAY
)
# 5. Текстовый ИИ-отчет
print("[5/5] Запуск ИИ-аудитора...")
report_text = generate_markdown_report(
merged_df=merged,
absent_explained=absent_explained,
absent_unexplained=absent_unexplained,
scud_present_but_absent_in_1c=scud_present_but_absent_in_1c,
raw_scud_df=raw_scud_df,
date_str=DATE_TODAY
)
print("\n" + "=" * 60)
print("ГОТОВАЯ ТЕКСТОВАЯ СВОДКА ИИ:")
print("=" * 60)
print(report_text)
md_report_path = os.path.join(OUTPUT_DIR, f"Сводка_контроллинга_{DATE_TODAY}.md")
with open(md_report_path, "w", encoding="utf-8") as f:
f.write(report_text)
print(f"\n[✓] Текстовый отчет сохранен в: {md_report_path}")
if __name__ == "__main__":
main()
modules/__init__.py
View File
modules/ai_verifier.py
+77
View File
@@ -0,0 +1,77 @@
import json
import re
import requests
from config import OLLAMA_URL, OLLAMA_MODEL
def ask_ollama(prompt, model=OLLAMA_MODEL, system_prompt=None):
"""Базовая функция отправки запросов к Ollama"""
if system_prompt is None:
system_prompt = "Ты — офисный аудитор безопасности и кадрового учета."
payload = {
"model": model,
"messages": [
{"role": "system", "content": system_prompt},
{"role": "user", "content": prompt}
],
"stream": False
}
try:
response = requests.post(OLLAMA_URL, json=payload, timeout=60)
if response.status_code == 200:
return response.json()['message']['content']
else:
return f"Ошибка API Ollama: {response.status_code}"
except Exception as e:
return f"Ошибка подключения к Ollama: {e}"
def ai_verify_scud_against_staff(unrecognized_scud_fios, staff_fios):
"""ИИ-сверка опечаток и несовпадений ФИО между СКУД и 1С:ЗУП"""
if not unrecognized_scud_fios or not staff_fios:
return {}
prompt = f"""
Ты — кадровый аудитор безопасности СКУД.
В СКУД записаны неопознанные ФИО: {json.dumps(unrecognized_scud_fios, ensure_ascii=False)}
В официальном Штатном расписании 1С записаны ЭТАЛОНЫ: {json.dumps(staff_fios, ensure_ascii=False)}
СТРОГИЕ ПРАВИЛА:
1. Запись из Штат 1С — это 100% ПРАВИЛЬНЫЙ эталон.
2. В СКУД допущена опечатка.
3. В поле "warning" описать опечатку.
4. Запрещено выдумывать опечатки!
ОТВЕЧАЙ ТОЛЬКО ИСКЛЮЧИТЕЛЬНО В ФОРМАТЕ JSON!
Формат ответа:
{{
"verified_matches": [
{{
"scud_fio": "ФИО из СКУД",
"staff_fio": "эталон ФИО из Штат 1С",
"warning": "Описание опечатки в СКУД или 'Точное совпадение (без опечаток)'"
}}
]
}}
"""
raw_response = ask_ollama(
prompt,
system_prompt="Ты — JSON API. Выдавай ТОЛЬКО валидный JSON без markdown-разметки."
)
mapping = {}
try:
match = re.search(r'\{.*\}', raw_response, re.DOTALL)
if match:
json_str = match.group(0)
json_str = re.sub(r'\}\s*[^}\]]*$', '}', json_str)
data = json.loads(json_str)
for item in data.get("verified_matches", []):
scud_f = item.get("scud_fio")
staff_f = item.get("staff_fio")
warn = item.get("warning", "Точное совпадение (без опечаток)")
if scud_f and staff_f:
mapping[scud_f] = {"staff_fio": staff_f, "warning": warn}
except Exception as e:
print(f"[!] Ошибка разбора JSON от ИИ при сверке опечаток: {e}")
return mapping
modules/data_loader.py
+79
View File
@@ -0,0 +1,79 @@
import os
import pandas as pd
import warnings
from config import (
normalize_fio, clean_scud_fio_light, load_exceptions,
DATA_DIR, SCUD_FILE, SCUD_FILE_TODAY, SCUD_FILE_YESTERDAY
)
warnings.filterwarnings('ignore', category=UserWarning, module='pandas')
def is_excluded(fio, dept, pos, exceptions):
fio_clean = normalize_fio(fio)
dept_str = str(dept).strip().upper() if pd.notna(dept) else ""
pos_str = str(pos).strip().lower() if pd.notna(pos) else ""
if fio_clean in exceptions["fio"] or dept_str in exceptions["departments"] or pos_str in exceptions["positions"]:
return True
if any(keyword in pos_str for keyword in exceptions["position_keywords"]):
return True
return False
def load_staff_data():
staff_path = os.path.join(DATA_DIR, "Штатные сотрудники - ОУП.xlsx")
if not os.path.exists(staff_path):
raise FileNotFoundError(f"Файл штатного расписания не найден: {staff_path}")
df_staff = pd.read_excel(staff_path, skiprows=7)
df_staff = df_staff.rename(columns={df_staff.columns[1]: 'ФИО'})
df_staff['fio_clean'] = df_staff['ФИО'].apply(normalize_fio)
return df_staff
def load_absent_data():
absent_path = os.path.join(DATA_DIR, "Отсутствия сотрудников - выгрузка.xlsx")
if not os.path.exists(absent_path):
raise FileNotFoundError(f"Файл отсутствий не найден: {absent_path}")
df_absent_raw = pd.read_excel(absent_path, skiprows=3)
df_absent = df_absent_raw[1:].copy().reset_index(drop=True)
df_absent = df_absent.rename(columns={df_absent.columns[0]: 'ФИО', df_absent.columns[2]: 'Вид_отсутствия'})
df_absent['fio_clean'] = df_absent['ФИО'].apply(normalize_fio)
return df_absent[['fio_clean', 'Вид_отсутствия']].dropna(subset=['fio_clean'])
def load_scud_data():
scud_path = SCUD_FILE_TODAY if os.path.exists(SCUD_FILE_TODAY) else (SCUD_FILE_YESTERDAY if os.path.exists(SCUD_FILE_YESTERDAY) else SCUD_FILE)
if not os.path.exists(scud_path):
raise FileNotFoundError(f"Файл СКУД не найден в папке {DATA_DIR}")
df_scud = pd.read_excel(scud_path)
exceptions = load_exceptions()
df_scud['fio_raw'] = df_scud['Сотрудник'].astype(str) if 'Сотрудник' in df_scud.columns else df_scud[df_scud.columns[0]].astype(str)
df_scud['fio_clean'] = df_scud['fio_raw'].apply(clean_scud_fio_light)
start_col = 'Начало_дня' if 'Начало_дня' in df_scud.columns else ('Начало дня' if 'Начало дня' in df_scud.columns else None)
status_col = 'Статус' if 'Статус' in df_scud.columns else None
# 🎯 ФИЗИЧЕСКИЙ ФАКТ ПРИСУТСТВИЯ
def is_person_present(row):
val_status = str(row.get(status_col, '')).strip().lower() if status_col else ''
val_time = str(row.get(start_col, '')).strip() if start_col else ''
# Разделение: Слово "Отсутствовал" в СКУД или пустая метка времени = ФИЗИЧЕСКИ НЕ БЫЛО
if 'отсутствовал' in val_status or val_time in ['нет входа', 'none', 'nan', '', '00:00']:
return False
# Во всех остальных случаях (есть время прохода) считаем, что человек ФИЗИЧЕСКИ переступил порог
return True
df_scud['Пришел'] = df_scud.apply(is_person_present, axis=1)
df_scud['is_excluded'] = df_scud.apply(
lambda r: is_excluded(r.get('Сотрудник', ''), r.get('Подразделение', ''), r.get('Должность', ''), exceptions),
axis=1
)
return df_scud[~df_scud['is_excluded']].copy()
def load_all_data():
df_scud = load_scud_data()
df_staff = load_staff_data()
df_absent = load_absent_data()
staff_fios_clean = df_staff['fio_clean'].dropna().tolist()
return df_scud, staff_fios_clean, df_absent
modules/data_validator.py
+46
View File
@@ -0,0 +1,46 @@
import os
import pandas as pd
from datetime import datetime, timedelta
from config import DATA_DIR, DATE_TODAY, DATE_YESTERDAY
def check_file_freshness():
"""
Проверяет наличие и актуальность файлов СКУД и 1С.
Возвращает (is_valid, warnings_list, errors_list)
"""
errors = []
warnings = []
# Ожидаемые имена файлов
scud_today_file = os.path.join(DATA_DIR, f"Сотрудники_{DATE_TODAY}.xlsx")
scud_yesterday_file = os.path.join(DATA_DIR, f"Сотрудники_{DATE_YESTERDAY}.xlsx")
staff_file = os.path.join(DATA_DIR, "Штатные сотрудники - ОУП.xlsx")
absent_file = os.path.join(DATA_DIR, "Отсутствия сотрудников - выгрузка.xlsx")
now = datetime.now()
# 1. Проверка файла СКУД за СЕГОДНЯ
if not os.path.exists(scud_today_file):
errors.append(f"❌ КРИТИЧЕСКАЯ ОШИБКА: Отсутствует актуальный файл СКУД за сегодня: 'Сотрудники_{DATE_TODAY}.xlsx'!")
else:
# Проверяем дату изменения файла (mtime)
file_mtime = datetime.fromtimestamp(os.path.getmtime(scud_today_file))
if file_mtime.date() < now.date():
errors.append(f"⚠️ ОШИБКА АКТУАЛЬНОСТИ: Файл 'Сотрудники_{DATE_TODAY}.xlsx' существует, но изменялся прошлым днем ({file_mtime.strftime('%d.%m.%Y %H:%M')})!")
# 2. Проверка файла СКУД за ВЧЕРА
if not os.path.exists(scud_yesterday_file):
warnings.append(f"⚠️ ПРЕДУПРЕЖДЕНИЕ: Файл СКУД за вчера ({DATE_YESTERDAY}) не найден. Детальный отчет за вчера может быть неполным.")
# 3. Проверка файлов 1С (Штат и Отсутствия)
for f_path, f_name in [(staff_file, "Штатные сотрудники"), (absent_file, "Выгрузка отсутствий")]:
if not os.path.exists(f_path):
errors.append(f"❌ КРИТИЧЕСКАЯ ОШИБКА: Файл 1С '{f_name}' не найден в папке data/!")
else:
file_mtime = datetime.fromtimestamp(os.path.getmtime(f_path))
days_old = (now - file_mtime).days
if days_old >= 2:
warnings.append(f"⚠️ УСТАРЕВШИЕ ДАННЫЕ 1С: Файл '{os.path.basename(f_path)}' не обновлялся {days_old} дн. (Дата: {file_mtime.strftime('%d.%m.%Y')})")
is_valid = len(errors) == 0
return is_valid, warnings, errors
modules/excel_exporter.py
+250
View File
@@ -0,0 +1,250 @@
import math
import os
import time
import openpyxl
import pandas as pd
from datetime import datetime
from openpyxl import Workbook
from openpyxl.styles import Font, Alignment, PatternFill, Border, Side
from openpyxl.utils import get_column_letter
from config import OUTPUT_DIR
# Словарь месяцев для текстового формата в названиях файлов
MONTHS_RU = {
1: "января", 2: "февраля", 3: "марта", 4: "апреля",
5: "мая", 6: "июня", 7: "июля", 8: "августа",
9: "сентября", 10: "октября", 11: "ноября", 12: "декабря"
}
def format_date_ru(date_str):
"""Преобразует дату формата '27.07.2026' в '27 июля 2026'"""
try:
dt = datetime.strptime(date_str, "%d.%m.%Y")
return f"{dt.day} {MONTHS_RU[dt.month]} {dt.year}"
except Exception:
return date_str
# Границы ячеек
THIN_SIDE = Side(border_style="thin", color="D3D3D3")
THIN_BORDER = Border(left=THIN_SIDE, right=THIN_SIDE, top=THIN_SIDE, bottom=THIN_SIDE)
# Палитра заливки для Сводки
FILL_HEADER = PatternFill(start_color="D9E1F2", end_color="D9E1F2", fill_type="solid")
FILL_TOTAL_LIST = PatternFill(start_color="F2F2F2", end_color="F2F2F2", fill_type="solid")
FILL_UNEXPLAINED = PatternFill(start_color="FCE4D6", end_color="FCE4D6", fill_type="solid")
FILL_PRESENT = PatternFill(start_color="E2EFDA", end_color="E2EFDA", fill_type="solid")
CATEGORY_PASTEL_COLORS = ["FFF2CC", "E1D5E7", "E1F5FE", "FFF0F5", "E8F8F5", "FCF3CF"]
# Палитра для Детального отчета
YELLOW_FILL = PatternFill(start_color="FFF2CC", end_color="FFF2CC", fill_type="solid")
LIGHT_RED_FILL = PatternFill(start_color="FCE4D6", end_color="FCE4D6", fill_type="solid")
def apply_borders_to_cell(cell, border=THIN_BORDER):
cell.border = border
def format_row_cells(ws, r_num, fill_obj, is_bold=False, align_b="right", bold_font=None):
cell_a = ws.cell(row=r_num, column=1)
cell_b = ws.cell(row=r_num, column=2)
if fill_obj:
cell_a.fill = fill_obj
cell_b.fill = fill_obj
apply_borders_to_cell(cell_a)
apply_borders_to_cell(cell_b)
if is_bold and bold_font:
cell_a.font = bold_font
cell_b.font = bold_font
if align_b:
cell_b.alignment = Alignment(horizontal=align_b)
# --- 1. СВОДКА НА СЕГОДНЯ ---
def generate_summary_excel(merged_df, date_str="27.07.2026", filename=None):
if not filename:
filename = f"{format_date_ru(date_str)} сводка.xlsx"
output_path = os.path.join(OUTPUT_DIR, filename)
wb = Workbook()
ws = wb.active
ws.title = "Лист_1"
# 🎯 ДВОЙНОЕ ВКЛЮЧЕНИЕ СИМВОЛОВ СТРУКТУРЫ В EXCEL (Обязательно для вывода плюсиков)
ws.sheet_properties.outlinePr.summaryBelow = False
ws.sheet_properties.outlinePr.summaryRight = False
ws.sheet_properties.outlinePr.showOutlineSymbols = True # Включает боковую панель
ws.sheet_view.showOutlineSymbols = True # Отображает плюсики
bold_font = Font(name="Calibri", size=11, bold=True)
# 1. Шапка
ws.cell(row=1, column=1, value="Сводка на")
ws.cell(row=1, column=2, value=date_str)
format_row_cells(ws, 1, FILL_HEADER, is_bold=True, bold_font=bold_font)
apply_borders_to_cell(ws.cell(row=2, column=1))
apply_borders_to_cell(ws.cell(row=2, column=2))
# 2. По списку
ws.cell(row=3, column=1, value="По списку")
ws.cell(row=3, column=2, value=len(merged_df))
format_row_cells(ws, 3, FILL_TOTAL_LIST, is_bold=True, bold_font=bold_font)
current_row = 4
# 3. НЕИЗВЕСТНО (По умолчанию РАСКРЫТО: hidden=False)
unexplained = merged_df[(merged_df['Пришел'] == False) & (merged_df['Вид_отсутствия'].isna())]
ws.cell(row=current_row, column=1, value="неизвестно")
ws.cell(row=current_row, column=2, value=len(unexplained))
format_row_cells(ws, current_row, FILL_UNEXPLAINED, is_bold=True, bold_font=bold_font)
current_row += 1
for fio in sorted(unexplained['Сотрудник'].dropna().unique()):
ws.cell(row=current_row, column=1, value=fio)
format_row_cells(ws, current_row, FILL_UNEXPLAINED, is_bold=False)
ws.row_dimensions[current_row].outlineLevel = 1
ws.row_dimensions[current_row].hidden = False
current_row += 1
# 4. КАТЕГОРИИ ОТСУТСТВИЙ (Свернуты hidden=True, под плюсики)
absent_groups = merged_df[merged_df['Вид_отсутствия'].notna()].groupby('Вид_отсутствия')
for idx, (cat_name, group) in enumerate(absent_groups):
hex_color = CATEGORY_PASTEL_COLORS[idx % len(CATEGORY_PASTEL_COLORS)]
cat_fill = PatternFill(start_color=hex_color, end_color=hex_color, fill_type="solid")
ws.cell(row=current_row, column=1, value=cat_name)
ws.cell(row=current_row, column=2, value=len(group))
format_row_cells(ws, current_row, cat_fill, is_bold=True, bold_font=bold_font)
current_row += 1
for fio in sorted(group['Сотрудник'].dropna().unique()):
ws.cell(row=current_row, column=1, value=fio)
format_row_cells(ws, current_row, cat_fill, is_bold=False)
ws.row_dimensions[current_row].outlineLevel = 1
ws.row_dimensions[current_row].hidden = True
current_row += 1
# 5. ИТОГО НА РАБОТЕ
present = merged_df[merged_df['Пришел'] == True]
ws.cell(row=current_row, column=1, value="Итого на работе")
ws.cell(row=current_row, column=2, value=len(present))
format_row_cells(ws, current_row, FILL_PRESENT, is_bold=True, bold_font=bold_font)
ws.column_dimensions['A'].width = 35.0
ws.column_dimensions['B'].width = 12.0
try:
wb.save(output_path)
print(f"[✓] Ежедневная сводка сохранена: {output_path}")
except PermissionError:
alt_filename = filename.replace(".xlsx", f"_{int(time.time())}.xlsx")
alt_path = os.path.join(OUTPUT_DIR, alt_filename)
wb.save(alt_path)
print(f"[⚠️] Файл открыт в Excel! Сохранено как: {alt_path}")
# --- 2. ДЕТАЛЬНЫЙ ОТЧЕТ ЗА ВЧЕРА ---
def generate_detailed_excel(merged_df, date_str="26.07.2026", filename=None):
if not filename:
filename = f"{format_date_ru(date_str)} отчет.xlsx"
output_path = os.path.join(OUTPUT_DIR, filename)
wb = Workbook()
ws = wb.active
ws.title = "Детальный_отчет"
ws["B2"] = "Дата:"
ws["D2"] = date_str
ws["B2"].font = Font(name="Arial", size=10, bold=True)
ws["D2"].font = Font(name="Arial", size=10, bold=True)
headers = [
"№", "ФИО", "Подразделение", "время входа", "время выхода",
"находился в здании", "причина отсутствия", "норма", "отклонение от нормы"
]
ws.append([])
ws.append(headers)
header_fill = PatternFill(start_color="D9E1F2", end_color="D9E1F2", fill_type="solid")
for col_idx in range(1, len(headers) + 1):
cell = ws.cell(row=4, column=col_idx)
cell.fill = header_fill
cell.font = Font(name="Arial", size=10, bold=True)
cell.alignment = Alignment(horizontal="center", vertical="center", wrap_text=True)
apply_borders_to_cell(cell)
start_col = 'Начало дня' if 'Начало дня' in merged_df.columns else 'Начало_дня'
end_col = 'Конец дня' if 'Конец дня' in merged_df.columns else 'Конец_дня'
hours_col = 'Часы' if 'Часы' in merged_df.columns else 'Находился_в_здании'
chars_per_line_g = 24
for idx, row in merged_df.reset_index(drop=True).iterrows():
is_present = row.get('Пришел', False)
absence_reason = row.get('Вид_отсутствия', '')
has_reason = pd.notna(absence_reason) and str(absence_reason).strip() != ''
ws.append([
idx + 1,
row.get('Сотрудник', ''),
row.get('Подразделение', ''),
row.get(start_col, 'Нет входа'),
row.get(end_col, 'Нет выхода'),
row.get(hours_col, '0:00'),
absence_reason if has_reason else '',
8,
"-8:00" if not is_present else "0:00"
])
row_num = 5 + idx
row_fill = None
if not is_present:
row_fill = YELLOW_FILL if has_reason else LIGHT_RED_FILL
val_g_str = str(absence_reason) if has_reason else ""
if len(val_g_str) > chars_per_line_g:
needed_lines = math.ceil(len(val_g_str) / chars_per_line_g)
ws.row_dimensions[row_num].height = max(needed_lines * 18, 22)
else:
ws.row_dimensions[row_num].height = 20
for col_idx in range(1, len(headers) + 1):
cell = ws.cell(row=row_num, column=col_idx)
apply_borders_to_cell(cell)
if row_fill:
cell.fill = row_fill
if col_idx == 7:
cell.alignment = Alignment(horizontal="left", vertical="center", wrap_text=True)
elif col_idx in [1, 4, 5, 6, 8, 9]:
cell.alignment = Alignment(horizontal="center", vertical="center")
else:
cell.alignment = Alignment(horizontal="left", vertical="center")
for col in ws.columns:
col_letter = get_column_letter(col[0].column)
if col_letter == 'G':
ws.column_dimensions['G'].width = 25.0 # Ровно 180 пикселей
else:
max_len = 0
for cell in col:
if cell.value is not None:
line_max = max(len(l) for l in str(cell.value).split("\n"))
if line_max > max_len:
max_len = line_max
ws.column_dimensions[col_letter].width = max(max_len + 4, 12)
try:
wb.save(output_path)
print(f"[✓] Детальный отчет сохранен: {output_path}")
except PermissionError:
alt_filename = filename.replace(".xlsx", f"_{int(time.time())}.xlsx")
alt_path = os.path.join(OUTPUT_DIR, alt_filename)
wb.save(alt_path)
print(f"[⚠️] Файл открыт в Excel! Сохранено как: {alt_path}")
def export_raw_scud(df_scud, filename="СКУД_Сырые_данные.xlsx"):
output_path = os.path.join(OUTPUT_DIR, filename)
df_scud.to_excel(output_path, index=False)
modules/feedback_loop.py
+33
View File
@@ -0,0 +1,33 @@
from modules.knowledge_base import add_rule_to_kb
def review_ai_decisions(ai_report, suspicious_cases):
"""
Интерактивный модуль: показывает спорные моменты человеку
и записывает его вердикт в память ИИ.
"""
print("\n" + "="*60)
print("РЕЖИМ ОБУЧЕНИЯ И ПРОВЕРКИ РЕШЕНИЙ ИИ (HUMAN-IN-THE-LOOP)")
print("="*60)
if not suspicious_cases:
print("Спорных случаев для обучения сегодня не обнаружено.")
return
for case in suspicious_cases:
fio = case.get('fio_target')
reason = case.get('reason')
print(f"\n[?] Спорный случай: {fio}")
print(f" Причина аномалии: {reason}")
user_input = input("Согласны с решением ИИ? (д - да / н - нет / введите новое правило): ").strip()
if user_input.lower() == 'н' or len(user_input) > 5:
if user_input.lower() == 'н':
rule_text = input(f"Опишите правило для {fio} (например: 'Считать совместителем'): ")
else:
rule_text = user_input
# Записываем правило в Базу Знаний
full_rule = f"Сотрудник {fio}: {rule_text}"
add_rule_to_kb(full_rule)
print(f"[+] ИИ 'запомнил' это правило и учитывает его со следующего запуска!")
modules/knowledge_base.py
+20
View File
@@ -0,0 +1,20 @@
import json
import os
KB_FILE = os.path.join("data", "knowledge_base.json")
def load_knowledge_base():
"""Загружает накопленный опыт и правила компании"""
if not os.path.exists(KB_FILE):
return {"rules": [], "fio_corrections": {}}
with open(KB_FILE, "r", encoding="utf-8") as f:
return json.load(f)
def add_rule_to_kb(new_rule):
"""Добавляет новое правило, принятое человеком, в базу знаний"""
kb = load_knowledge_base()
if new_rule not in kb["rules"]:
kb["rules"].append(new_rule)
with open(KB_FILE, "w", encoding="utf-8") as f:
json.dump(kb, f, ensure_ascii=False, indent=2)
print(f"[✓] База знаний обновлена! Новое правило: {new_rule}")
modules/text_reporter.py
+91
View File
@@ -0,0 +1,91 @@
import json
from modules.ai_verifier import ask_ollama
from modules.knowledge_base import load_knowledge_base
def generate_markdown_report(merged_df, absent_explained, absent_unexplained, scud_present_but_absent_in_1c, raw_scud_df=None, raw_staff_df=None, date_str="27.07.2026"):
kb = load_knowledge_base()
custom_rules = kb.get("rules", [])
custom_rules_str = "\n".join([f"- {r}" for r in custom_rules]) if custom_rules else "Специфических правил компании пока нет."
detailed_cases_for_ai = []
unexplained_fios = absent_unexplained['fio_clean'].unique() if 'fio_clean' in absent_unexplained.columns else []
for fio_clean in unexplained_fios:
staff_matches = []
if raw_staff_df is not None and not raw_staff_df.empty and 'fio_clean' in raw_staff_df.columns:
st_m = raw_staff_df[raw_staff_df['fio_clean'] == fio_clean]
cols_staff = [c for c in ['ФИО', 'Подразделение', 'Должность', 'Таб_№'] if c in st_m.columns]
staff_matches = st_m[cols_staff].to_dict(orient='records')
scud_matches = []
if raw_scud_df is not None and not raw_scud_df.empty and 'fio_clean' in raw_scud_df.columns:
sc_m = raw_scud_df[raw_scud_df['fio_clean'] == fio_clean]
cols_scud = [c for c in ['Сотрудник', 'Подразделение', 'Должность', 'Начало_дня', 'Статус'] if c in sc_m.columns]
scud_matches = sc_m[cols_scud].to_dict(orient='records')
detailed_cases_for_ai.append({
"fio_clean": fio_clean,
"staff_records_count": len(staff_matches),
"staff_records_1C": staff_matches,
"scud_records_LOGS": scud_matches
})
cols_crit = [c for c in ['Сотрудник', 'Подразделение', 'Должность', 'Вид_отсутствия'] if c in scud_present_but_absent_in_1c.columns]
critical_list = scud_present_but_absent_in_1c[cols_crit].to_dict(orient='records')
# Исходные точные цифры Python
total_staff = len(merged_df)
present_cnt = len(merged_df[merged_df['Пришел'] == True])
explained_cnt = len(absent_explained)
unexplained_cnt = len(absent_unexplained)
prompt = f"""
Ты — старший аудитор кадровой безопасности и систем СКУД. Проанализируй данные за {date_str} и составь итоговый отчет.
📌 ДИНАМИЧЕСКИЕ ПРАВИЛА И НАКОПЛЕННЫЙ ОПЫТ ПРЕДПРИЯТИЯ:
{custom_rules_str}
🚨 СТРОГИЕ ПРАВИЛА МАТЕМАТИКИ И КАТЕГОРИЙ:
1. Цифры сводки ДОЛЖНЫ СТРОГО СОВПАДАТЬ с первично посчитанным массивом:
- Всего офисных сотрудников: {total_staff}
- Присутствуют на рабочем месте: {present_cnt}
- Официально отсутствуют: {explained_cnt}
- Неизвестно (истинно неотмеченные): {unexplained_cnt} чел.
2. Заголовок "#### Неизвестные случаи: {unexplained_cnt}" должен иметь ровно ту же цифру {unexplained_cnt}.
3. Нумерованный список ниже ДОЛЖЕН содержать ровно {unexplained_cnt} сотрудников из массива `detailed_cases_for_ai` без пропусков.
---
📊 ВХОДНЫЕ МИКРО-ДАННЫЕ НЕИЗВЕСТНЫХ ({unexplained_cnt} чел):
{json.dumps(detailed_cases_for_ai, ensure_ascii=False, indent=2)}
🚨 СПИСОК КРИТИЧЕСКИХ АНОМАЛИЙ:
{json.dumps(critical_list, ensure_ascii=False, indent=2)}
СТРОГИЙ ШАБЛОН ОТВЕТА:
**Сводка контроллинга СКУД и 1С:ЗУП на {date_str} в 12:00**
- Всего офисных сотрудников: **{total_staff}**
- Присутствуют на рабочем месте: **{present_cnt}**
- Официально отсутствуют: **{explained_cnt}**
- Неизвестно (истинно неотмеченные после проверки ИИ): **{unexplained_cnt}** чел.
#### ⚠️ Подозрения на ложные срабатывания СКУД / Совмещения:
(Укажи сотрудников с аномалиями/вторыми проходами, например Кондрашову Е.В., если у нее дубликат карты)
#### Критические аномалии (Работа в отпуске / на больничном):
(Детально опиши сотрудников из списка КРИТИЧЕСКИХ АНОМАЛИЙ. Если их нет — напиши "Критических нарушений не выявлено.")
#### Неизвестные случаи: {unexplained_cnt}
(Выведи НУМЕРОВАННЫЙ СПИСОК ВСЕХ {unexplained_cnt} человек строго в формате:
1. ФИО — [Аббревиатура], Должность
2. ФИО — [Аббревиатура], Должность
...
{unexplained_cnt}. ФИО — [Аббревиатура], Должность)
### Рекомендации:
...
"""
sys_prompt = "Ты — кадровый аудитор. Выводи ровно то количество людей в списке, которое указано в переменной unexplained_cnt."
return ask_ollama(prompt, system_prompt=sys_prompt)