85 lines
3.1 KiB
Python
85 lines
3.1 KiB
Python
import os
|
||
import re
|
||
from datetime import datetime, timedelta
|
||
|
||
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
|
||
DATA_DIR = os.path.join(BASE_DIR, "data")
|
||
|
||
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")
|
||
|
||
# Путь к намонтированной сетевой шаре 1С в Linux (вместо \\storage\SCUD\Обмен\Штат)
|
||
SHARE_1C_DIR = "/mnt/scud_share"
|
||
|
||
for folder in [DATA_DIR, SCUD_DIR, ZUP_1C_DIR, OUTPUT_DIR]:
|
||
os.makedirs(folder, exist_ok=True)
|
||
|
||
NOW = datetime.now()
|
||
DATE_TODAY = NOW.strftime("%d.%m.%Y")
|
||
|
||
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")
|
||
|
||
# --- Настройки LLM / Ollama ---
|
||
OLLAMA_URL = "http://192.168.11.3: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")
|
||
|
||
# --- Настройки подключения к MS SQL Server (1С:ЗУП 3.1) ---
|
||
ZUP_SQL_CONFIG = {
|
||
"driver": "{ODBC Driver 18 for SQL Server}", # или "{ODBC Driver 17 for SQL Server}"
|
||
"server": os.getenv("ZUP_SQL_SERVER", "ACCOUNT-01"),
|
||
"database": os.getenv("ZUP_SQL_DB", "ZUP30"), # Замените на точное имя базы ЗУП на ACCOUNT-01
|
||
"user": os.getenv("ZUP_SQL_USER", "scud_reader"), # Логин SQL Server
|
||
"password": os.getenv("ZUP_SQL_PASS", "Rhfcysq90"), # Пароль SQL Server
|
||
"trust_server_certificate": "yes", # Доверять сертификату сервера
|
||
"encrypt": "no" # "yes", если на MS SQL включено обязательное SSL-шифрование
|
||
}
|
||
|
||
|
||
def find_dated_file(prefix, date_str, search_dirs=[ZUP_1C_DIR, SCUD_DIR, DATA_DIR, "."]):
|
||
"""
|
||
Ищет файлы по префиксу и дате (поддерживает и 30.07.2026, и 30_07_2026).
|
||
"""
|
||
date_dots = date_str
|
||
date_underscores = date_str.replace('.', '_')
|
||
|
||
for d in search_dirs:
|
||
if not os.path.exists(d):
|
||
continue
|
||
for f in os.listdir(d):
|
||
if f.endswith('.xlsx') or f.endswith('.csv'):
|
||
if f.lower().startswith(prefix.lower()):
|
||
if date_dots in f or date_underscores in f:
|
||
return os.path.join(d, f)
|
||
return None
|
||
|
||
|
||
def normalize_fio(fio):
|
||
if not fio or not isinstance(fio, str):
|
||
return ""
|
||
fio_clean = re.sub(r'\(.*?\)', '', fio)
|
||
fio_clean = fio_clean.replace('\xa0', ' ')
|
||
parts = fio_clean.strip().split()
|
||
return " ".join(parts).title()
|
||
|
||
|
||
def clean_scud_fio_light(fio_str):
|
||
return normalize_fio(fio_str)
|
||
|
||
|
||
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": []} |