промежуточно. Работает все, неверно формируется вчерашний отчет (нет выхода).

This commit is contained in:
2026-07-28 16:41:26 +03:00
parent 522800d317
commit 2603627098
13 changed files with 297 additions and 283 deletions
+42 -56
View File
@@ -1,78 +1,64 @@
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")
# 📂 ПОДПАПКИ DАТА
SCUD_DIR = os.path.join(DATA_DIR, "scud")
ZUP_1C_DIR = os.path.join(DATA_DIR, "1c")
OUTPUT_DIR = os.path.join(BASE_DIR, "output")
os.makedirs(DATA_DIR, exist_ok=True)
os.makedirs(OUTPUT_DIR, exist_ok=True)
# Создаем директории при их отсутствии
for folder in [DATA_DIR, SCUD_DIR, ZUP_1C_DIR, OUTPUT_DIR]:
os.makedirs(folder, 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
# 📅 Даты
NOW = datetime.now()
DATE_TODAY = NOW.strftime("%d.%m.%Y")
DATE_TODAY, DATE_YESTERDAY = get_controlling_dates()
if NOW.weekday() == 0: # Понедельник -> Вчера = Пятница
DATE_YESTERDAY = (NOW - timedelta(days=3)).strftime("%d.%m.%Y")
else:
DATE_YESTERDAY = (NOW - timedelta(days=1)).strftime("%d.%m.%Y")
# 📄 Пути к файлам
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
SCUD_FILE_TODAY = os.path.join(SCUD_DIR, f"Сотрудники_{DATE_TODAY}.xlsx")
SCUD_FILE_YESTERDAY = os.path.join(SCUD_DIR, f"Сотрудники_{DATE_YESTERDAY}.xlsx")
SCUD_FILE = os.path.join(SCUD_DIR, "Сотрудники.xlsx")
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")
STAFF_FILE_1C = os.path.join(ZUP_1C_DIR, "Штатные сотрудники - ОУП.xlsx")
ABSENT_FILE_1C = os.path.join(ZUP_1C_DIR, "Отсутствия сотрудников - выгрузка.xlsx")
# 🤖 Настройки Ollama
OLLAMA_URL = "http://10.121.18.227:11434/api/chat"
# Настройки Ollama / RAG (объявляем оба имени для совместимости)
OLLAMA_URL = "http://10.121.18.227:11434/api/generate"
OLLAMA_MODEL = "qwen2.5:14b"
MODEL_NAME = OLLAMA_MODEL
KNOWLEDGE_BASE_PATH = os.path.join(DATA_DIR, "knowledge_base.json")
EXCEPTIONS_PATH = os.path.join(BASE_DIR, "exceptions.json")
# 🛠 Очистка и нормализация данных
def normalize_fio(val):
"""Строгая нормализация ФИО для точности сопоставления 1С и СКУД"""
if pd.isna(val):
def normalize_fio(fio):
if not fio or not isinstance(fio, str):
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()
parts = fio.strip().split()
return " ".join(parts).title()
def clean_scud_fio_light(val):
"""Легкая очистка ФИО для СКУД"""
if pd.isna(val):
def clean_scud_fio_light(fio_str):
if not isinstance(fio_str, str):
return ""
val = str(val).strip().lower()
val = re.sub(r'\(.*?\)', '', val).replace('ё', 'е')
return re.sub(r'\s+', ' ', val).strip()
fio_clean = fio_str.replace("(сводная)", "").strip()
return normalize_fio(fio_clean)
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", [])]
}
# тестовый комментарий
def load_exceptions():
import json
if os.path.exists(EXCEPTIONS_PATH):
try:
with open(EXCEPTIONS_PATH, "r", encoding="utf-8") as f:
return json.load(f)
except Exception:
pass
return {"fio": [], "departments": [], "positions": [], "position_keywords": []}