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

перед внедрением API и веб интерфейса.

Browse Source
This commit is contained in:
manoraga
2026-08-04 17:43:00 +03:00
parent d667b2d846
commit 7eea236417
11 changed files with 357 additions and 87 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Project_Analysis_and_Recommendations_04.08.2026.docx
BIN
View File
Binary file not shown.
config.py
+12
View File
@@ -23,6 +23,7 @@ if NOW.weekday() == 0:
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
@@ -30,6 +31,17 @@ 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, "."]):
"""
data/scud_orion_ai.db
BIN
View File
Binary file not shown.
modules/data_loader.py
+37 -26
View File
@@ -5,6 +5,8 @@ from config import (
normalize_fio, clean_scud_fio_light, load_exceptions,
DATE_TODAY, DATE_YESTERDAY, find_dated_file, ZUP_1C_DIR, SCUD_DIR
)
# Импортируем прямую SQL-выгрузку отсутствий из 1С
from modules.zup_extractor import fetch_zup_absences_from_sql
warnings.filterwarnings('ignore', category=UserWarning, module='pandas')
@@ -24,42 +26,51 @@ def is_excluded(fio, dept, pos, exceptions):
def load_staff_data(date_str):
"""
Загружает файл Штатного расписания 1С за конкретную дату.
Если файла за эту дату нет — возвращает None без подмены на старые даты.
Загружает файл Штатного расписания 1С за конкретную дату из папки 1c.
"""
filepath = find_dated_file("Штат", date_str)
if not filepath:
return None
fallback_path = os.path.join(ZUP_1C_DIR, "штат.xlsx")
if os.path.exists(fallback_path):
filepath = fallback_path
else:
return None
df_raw = pd.read_excel(filepath, skiprows=8)
df_staff = df_raw.iloc[:, [1, 5, 12]].copy()
df_staff.columns = ['ФИО', 'Подразделение', 'Должность']
df_staff = df_staff.dropna(subset=['ФИО']).reset_index(drop=True)
df_staff = df_staff[~df_staff['ФИО'].astype(str).str.contains('Всего|Организация|Сотрудник|ФИО', case=False, na=False)]
df_staff['fio_clean'] = df_staff['ФИО'].apply(normalize_fio)
return df_staff
try:
df_raw = pd.read_excel(filepath, skiprows=8)
# Индексы столбцов в файле 1С: 1 — ФИО, 5 — Подразделение, 12 — Должность
df_staff = df_raw.iloc[:, [1, 5, 12]].copy()
df_staff.columns = ['ФИО', 'Подразделение', 'Должность']
df_staff = df_staff.dropna(subset=['ФИО']).reset_index(drop=True)
df_staff = df_staff[~df_staff['ФИО'].astype(str).str.contains('Всего|Организация|Сотрудник|ФИО', case=False, na=False)]
df_staff['fio_clean'] = df_staff['ФИО'].apply(normalize_fio)
return df_staff
except Exception as e:
print(f"[❌] Ошибка загрузки штата из {filepath}: {e}")
return None
def load_absent_data(date_str):
"""
Загружает файл Отсутствий сотрудников 1С за конкретную дату.
Если файла за эту дату нет — возвращает None без подмены на старые даты.
Загружает документально подтвержденные отсутствия сотрудников
НАПРЯМУЮ из базы MS SQL 1С:ЗУП за указанную дату.
"""
filepath = find_dated_file("Отсутствия", date_str)
if not filepath:
try:
df_absent = fetch_zup_absences_from_sql(date_str)
if df_absent is not None and not df_absent.empty:
df_absent = df_absent.dropna(subset=['ФИО', 'Вид_отсутствия'])
df_absent = df_absent[~df_absent['ФИО'].astype(str).str.contains('АО "ЛЕНМОРНИИПРОЕКТ"|Сотрудник', case=False, na=False)]
df_absent['fio_clean'] = df_absent['ФИО'].apply(normalize_fio)
return df_absent[['fio_clean', 'Вид_отсутствия']].dropna(subset=['fio_clean'])
else:
return None
except Exception as e:
print(f"[⚠️] Ошибка SQL-выгрузки отсутствий за {date_str}: {e}")
return None
df_raw = pd.read_excel(filepath, skiprows=3)
df_absent = df_raw[1:].copy().reset_index(drop=True)
df_absent = df_absent.rename(columns={
df_absent.columns[0]: 'ФИО',
df_absent.columns[2]: 'Вид_отсутствия'
})
df_absent = df_absent[['ФИО', 'Вид_отсутствия']].dropna(subset=['ФИО', 'Вид_отсутствия'])
df_absent = df_absent[~df_absent['ФИО'].astype(str).str.contains('АО "ЛЕНМОРНИИПРОЕКТ"|Сотрудник', case=False, na=False)]
df_absent['fio_clean'] = df_absent['ФИО'].apply(normalize_fio)
return df_absent[['fio_clean', 'Вид_отсутствия']].dropna(subset=['fio_clean'])
def load_scud_data(target_date_type="today"):
"""Загружает файл СКУД за 'today' или 'yesterday'."""
@@ -100,7 +111,7 @@ def load_scud_data(target_date_type="today"):
def load_all_data():
"""
Загружает слитные датасеты за Сегодня и за Вчера.
Если файла 1С за Сегодня нет, соответствующие датасеты вернет как None.
Если данных за Сегодня нет, возвращает для них None.
"""
df_scud_today = load_scud_data(target_date_type="today")
try:
modules/database.py
+25 -6
View File
@@ -1,5 +1,6 @@
import os
import sqlite3
import json
import pandas as pd
from datetime import datetime
from config import DATA_DIR
@@ -94,21 +95,38 @@ def init_db():
conn.commit()
# Автоматически загружаем базовые правила компании
# Автоматически загружаем базовые правила и системный промпт
sync_knowledge_base_to_db()
def sync_knowledge_base_to_db():
"""
Автоматически мигрирует все правила из knowledge_base.json в таблицу ai_knowledge_base SQLite.
Автоматически мигрирует правила из knowledge_base.json и системный промпт в таблицу ai_knowledge_base SQLite.
"""
kb_path = os.path.join(os.path.dirname(DATA_DIR), "knowledge_base.json")
system_ai_prompt = (
"СИСТЕМНЫЙ ПРОМПТ ИИ (MATCHING & VERIFICATION):\n"
"Ты — экспертный ИИ-аудитор СКУД и кадрового учета 1С. Твоя задача — проводить интеллектуальный анализ аномалий "
"и сопоставление (Fuzzy Matching) нераспознанных ФИО из СКУД с эталонным Штатным расписанием 1С.\n\n"
"1. ЗАЩИТА ОТ ПОЛНЫХ ТЁЗОК (FULL NAME COLLISION):\n"
" - Если в эталонном штате 1С присутствуют два или более сотрудников с ОДИНАКОВЫМИ ФИО (полные тёзки), "
"ты ОБЯЗАН учитывать контекст: Подразделение и Должность.\n"
" - Ни в коем случае не объединяй записи разных людей! Если контекст не позволяет однозначно определить, кто именно из тёзок совершил проход, ПОМЕТЬ СЛУЧАЙ КАК НЕОДНОЗНАЧНЫЙ и НЕ делай автоматическую подмену.\n\n"
"2. ПРАВИЛА СОПОСТАВЛЕНИЯ (MATCHING RULES):\n"
" - Допускаются исправления мелких опечаток, перестановки порядка (Имя Фамилия -> Фамилия Имя), замена букв Е/Ё, И/Й.\n"
" - Учитывай сокращения инициалов (например, 'Иванов И.В.' -> 'Иванов Иван Васильевич').\n"
" - Если уверенность в совпадении меньше 85% — НЕ сопоставляй, оставляй запись как нераспознанную.\n\n"
"3. ФОРМАТ ВЫХОДНЫХ ДАННЫХ:\n"
" Верни строго JSON-объект без вводных слов и Markdown-разметки: {\"mappings\": [...], \"unmatched\": [...]}"
)
rules_to_insert = [
system_ai_prompt,
"ГЛОБАЛЬНОЕ ПРАВИЛО 1: Любое физическое присутствие сотрудника по СКУД при наличии в 1С документа отсутствия (отпуск, больничный, уход за ребенком и др.), КРОМЕ командировок и служебных поездок — ЯВЛЯЕТСЯ СТРОГОЙ ИСТИННОЙ АНОМАЛИЕЙ.",
"ГЛОБАЛЬНОЕ ПРАВИЛО 2: Документы отсутствий из 1С:ЗУП имеют 100% высший приоритет над статическим реестром удалёнщиков.",
"ГЛОБАЛЬНОЕ ПРАВИЛО 3: Удалённые сотрудники имеют право беспрепятственно работать в офисе по СКУД в любое время (это не является аномалией)."
]
kb_path = os.path.join(os.path.dirname(DATA_DIR), "knowledge_base.json")
if os.path.exists(kb_path):
try:
with open(kb_path, "r", encoding="utf-8") as f:
@@ -124,11 +142,12 @@ def sync_knowledge_base_to_db():
cursor = conn.cursor()
for rule in rules_to_insert:
cursor.execute(
"INSERT OR IGNORE INTO ai_knowledge_base (rule_text) VALUES (?)",
(rule.strip(),)
"INSERT OR IGNORE INTO ai_knowledge_base (rule_text, added_by) VALUES (?, ?)",
(rule.strip(), "System_AI")
)
conn.commit()
def save_scud_to_db(df_scud, date_str):
"""Оптимизированное сохранение логов СКУД за дату."""
if df_scud.empty:
modules/zup_extractor.py
+50 -52
View File
@@ -1,40 +1,39 @@
import os
import logging
from datetime import date
from datetime import date, datetime
import pyodbc
import pandas as pd
from config import DATA_DIR, normalize_fio
from config import DATA_DIR, normalize_fio, ZUP_SQL_CONFIG
logger = logging.getLogger(__name__)
# Конфигурация MS SQL Server 1С:ЗУП 3.1
ZUP_SQL_CONFIG = {
"driver": "{ODBC Driver 18 for SQL Server}",
"server": os.getenv("ZUP_MS_SQL_SERVER", "192.168.1.10"), # Укажите IP MS SQL 1С
"database": os.getenv("ZUP_MS_SQL_DB", "zup_db"),
"user": os.getenv("ZUP_MS_SQL_USER", "sa"),
"password": os.getenv("ZUP_MS_SQL_PASS", "your_password"),
}
def get_zup_connection_string() -> str:
"""Формирует строку подключения pyodbc к MS SQL Server."""
"""Формирует строку подключения pyodbc к MS SQL Server из config.py."""
return (
f"DRIVER={ZUP_SQL_CONFIG['driver']};"
f"SERVER={ZUP_SQL_CONFIG['server']};"
f"DATABASE={ZUP_SQL_CONFIG['database']};"
f"UID={ZUP_SQL_CONFIG['user']};"
f"PWD={ZUP_SQL_CONFIG['password']};"
f"TrustServerCertificate=yes;"
f"Encrypt=no;"
f"TrustServerCertificate={ZUP_SQL_CONFIG.get('trust_server_certificate', 'yes')};"
f"Encrypt={ZUP_SQL_CONFIG.get('encrypt', 'no')};"
)
def fetch_zup_absences_from_sql(target_date: date) -> pd.DataFrame:
def fetch_zup_absences_from_sql(target_date) -> pd.DataFrame:
"""
Извлекает оперативные отсутствия и действующие декреты из MS SQL 1С:ЗУП 3.1
на указанную дату.
на указанную дату (принимает как datetime.date, так и строку 'DD.MM.YYYY' / 'DD_MM_YYYY').
"""
if isinstance(target_date, str):
clean_date_str = target_date.replace('_', '.')
try:
target_date = datetime.strptime(clean_date_str, "%d.%m.%Y").date()
except ValueError:
logger.error(f"[❌] Неверный формат даты для SQL-запроса: {target_date}. Ожидался DD.MM.YYYY")
return pd.DataFrame()
query = """
DECLARE @TargetDate DATE = ?;
@@ -86,17 +85,21 @@ def fetch_zup_absences_from_sql(target_date: date) -> pd.DataFrame:
ORDER BY [ФИО] ASC;
"""
conn_str = get_zup_connection_string()
with pyodbc.connect(conn_str) as conn:
df = pd.read_sql(query, conn, params=[target_date])
if not df.empty:
df['fio_clean'] = df['ФИО'].apply(normalize_fio)
return df
try:
conn_str = get_zup_connection_string()
with pyodbc.connect(conn_str, timeout=5) as conn:
df = pd.read_sql(query, conn, params=[target_date])
if not df.empty:
df['fio_clean'] = df['ФИО'].apply(normalize_fio)
return df
except Exception as e:
logger.error(f"[❌] Ошибка SQL-выгрузки отсутствий за {target_date}: {e}")
return pd.DataFrame()
def fetch_zup_staff_from_sql() -> pd.DataFrame:
"""Извлекает весь актуальный штат из MS SQL 1С:ЗУП."""
"""Резервная выгрузка штата из MS SQL."""
query = """
SELECT DISTINCT
LTRIM(RTRIM(ref_emp._Description)) AS [ФИО],
@@ -107,53 +110,48 @@ def fetch_zup_staff_from_sql() -> pd.DataFrame:
AND ref_emp._Marked = 0x00
ORDER BY [ФИО] ASC;
"""
conn_str = get_zup_connection_string()
with pyodbc.connect(conn_str) as conn:
df = pd.read_sql(query, conn)
if not df.empty:
df['fio_clean'] = df['ФИО'].apply(normalize_fio)
return df
def sync_zup_to_excel(target_date: date) -> bool:
"""
Извлекает данные из MS SQL и автоматически создает локальные Excel-файлы 1С:
data/1c/Штат_DD_MM_YYYY.xlsx
data/1c/Отсутствия_DD_MM_YYYY.xlsx
"""
try:
date_str_file = target_date.strftime("%d_%m_%Y")
conn_str = get_zup_connection_string()
with pyodbc.connect(conn_str, timeout=5) as conn:
df = pd.read_sql(query, conn)
if not df.empty:
df['fio_clean'] = df['ФИО'].apply(normalize_fio)
return df
except Exception as e:
logger.error(f"[❌] Ошибка выгрузки штата из MS SQL: {e}")
return pd.DataFrame()
def sync_zup_to_excel(target_date) -> bool:
"""Создает дамп в Excel при необходимости."""
try:
if isinstance(target_date, str):
clean_date_str = target_date.replace('_', '.')
target_date_obj = datetime.strptime(clean_date_str, "%d.%m.%Y").date()
else:
target_date_obj = target_date
date_str_file = target_date_obj.strftime("%d_%m_%Y")
c_1c_dir = os.path.join(DATA_DIR, "1c")
os.makedirs(c_1c_dir, exist_ok=True)
df_staff = fetch_zup_staff_from_sql()
df_absences = fetch_zup_absences_from_sql(target_date)
df_absences = fetch_zup_absences_from_sql(target_date_obj)
logger.info(f"[1С Direct SQL] Выгрузка за {target_date.strftime('%d.%m.%Y')}: Штат ({len(df_staff)} чел.), Отсутствия ({len(df_absences)} чел.)")
# Сохранение в формате, который ожидает data_loader.py
staff_excel_path = os.path.join(c_1c_dir, f"Штат_{date_str_file}.xlsx")
absences_excel_path = os.path.join(c_1c_dir, f"Отсутствия_{date_str_file}.xlsx")
# Формируем структуру Excel с нужным количеством пропущенных строк (skiprows)
with pd.ExcelWriter(staff_excel_path, engine='openpyxl') as writer:
# Пустой заголовок (8 строк) для эмуляции отчета 1С
dummy_headers = pd.DataFrame([[""] * 13] * 8)
dummy_headers.to_excel(writer, index=False, header=False)
# Данные
df_staff[['ФИО', 'Подразделение', 'Должность']].to_excel(writer, startrow=8, index=False)
with pd.ExcelWriter(absences_excel_path, engine='openpyxl') as writer:
# Пустой заголовок (3 строки) для эмуляции отчета 1С
dummy_headers = pd.DataFrame([[""] * 3] * 3)
dummy_headers.to_excel(writer, index=False, header=False)
# Данные
df_absences[['ФИО', 'Вид_отсутствия']].to_excel(writer, startrow=3, index=False)
logger.info(f"[✓] Успешно созданы файлы 1С за {target_date.strftime('%d.%m.%Y')}: {staff_excel_path}, {absences_excel_path}")
return True
except Exception as e:
scripts/db_cli.py
+9 -3
View File
@@ -76,10 +76,16 @@ if __name__ == "__main__":
print(df.to_string(index=False) if not df.empty else "Аномалий в базе пока нет.")
elif cmd == "rules":
df = query_db("SELECT * FROM ai_knowledge_base ORDER BY id ASC")
if df is not None:
df = query_db("SELECT id, added_by, created_at, rule_text FROM ai_knowledge_base ORDER BY id ASC")
if df is not None and not df.empty:
print("\n🧠 ПРАВИЛА БАЗЫ ЗНАНИЙ ИИ (ОБУЧЕНИЕ):")
print(df.to_string(index=False) if not df.empty else "База знаний пуста.")
print("=" * 80)
for _, row in df.iterrows():
print(f"📌 [ID: {row['id']}] | Добавил: {row['added_by']} | Дата: {row['created_at']}")
print(f"{row['rule_text']}")
print("-" * 80)
else:
print("\n🧠 База знаний пуста.")
elif cmd == "scud" and len(sys.argv) > 2:
date_str = sys.argv[2]
test_zup_conn.py
+56
View File
@@ -0,0 +1,56 @@
import sys
import pyodbc
from config import ZUP_SQL_CONFIG
def test_connection():
print("=" * 60)
print("🧪 ПРОВЕРКА ПОДКЛЮЧЕНИЯ К MS SQL SERVER (1С:ЗУП)")
print("=" * 60)
print(f" • Сервер: {ZUP_SQL_CONFIG['server']}")
print(f" • База: {ZUP_SQL_CONFIG['database']}")
print(f" • Логин: {ZUP_SQL_CONFIG['user']}")
print(f" • Драйвер: {ZUP_SQL_CONFIG['driver']}")
print("-" * 60)
conn_str = (
f"DRIVER={ZUP_SQL_CONFIG['driver']};"
f"SERVER={ZUP_SQL_CONFIG['server']};"
f"DATABASE={ZUP_SQL_CONFIG['database']};"
f"UID={ZUP_SQL_CONFIG['user']};"
f"PWD={ZUP_SQL_CONFIG['password']};"
f"TrustServerCertificate={ZUP_SQL_CONFIG['trust_server_certificate']};"
f"Encrypt={ZUP_SQL_CONFIG['encrypt']};"
)
try:
print("⏳ Установка соединения...")
with pyodbc.connect(conn_str, timeout=10) as conn:
cursor = conn.cursor()
# 1. Проверка версии SQL Сервера
cursor.execute("SELECT @@VERSION;")
db_version = cursor.fetchone()[0]
print("✅ Подключение успешно установлено!\n")
print(f"📋 Версия MS SQL Server:\n{db_version.splitlines()[0]}\n")
# 2. Проверка доступа к таблице физических лиц _Reference299
cursor.execute("SELECT COUNT(*) FROM dbo._Reference299 WITH (NOLOCK);")
count_phys = cursor.fetchone()[0]
print(f"📊 Записей в таблице физлиц (_Reference299): {count_phys} шт.")
# 3. Проверка доступа к регистру кадровых состояний _InfoRg16921
cursor.execute("SELECT COUNT(*) FROM dbo._InfoRg16921 WITH (NOLOCK);")
count_states = cursor.fetchone()[0]
print(f"📊 Записей в регистре состояний (_InfoRg16921): {count_states} шт.")
print("\n🎉 Все проверки успешно пройдены!")
except pyodbc.Error as ex:
sqlstate = ex.args[0] if ex.args else "UNKNOWN"
print("\n❌ ОШИБКА ПОДКЛЮЧЕНИЯ!")
print(f"Код ошибки: {sqlstate}")
print(f"Детали:\n{ex}")
sys.exit(1)
if __name__ == "__main__":
test_connection()
update_kb.py
+45
View File
@@ -0,0 +1,45 @@
import sqlite3
import os
from config import DATA_DIR
DB_PATH = os.path.join(DATA_DIR, "scud_orion_ai.db")
system_ai_prompt = (
"СИСТЕМНЫЙ ПРОМПТ ИИ (MATCHING & VERIFICATION):\n"
"Ты — экспертный ИИ-аудитор СКУД и кадрового учета 1С. Твоя задача — проводить интеллектуальный анализ аномалий "
"и сопоставление (Fuzzy Matching) нераспознанных ФИО из СКУД с эталонным Штатным расписанием 1С.\n\n"
"1. ЗАЩИТА ОТ ПОЛНЫХ ТЁЗОК (FULL NAME COLLISION):\n"
" - Если в эталонном штате 1С присутствуют два или более сотрудников с ОДИНАКОВЫМИ ФИО (полные тёзки), "
"ты ОБЯЗАН учитывать контекст: Подразделение и Должность.\n"
" - Ни в коем случае не объединяй записи разных людей! Если контекст не позволяет однозначно определить, кто именно из тёзок совершил проход, ПОМЕТЬ СЛУЧАЙ КАК НЕОДНОЗНАЧНЫЙ и НЕ делай автоматическую подмену.\n\n"
"2. ПРАВИЛА СОПОСТАВЛЕНИЯ (MATCHING RULES):\n"
" - Допускаются исправления мелких опечаток, перестановки порядка (Имя Фамилия -> Фамилия Имя), замена букв Е/Ё, И/Й.\n"
" - Учитывай сокращения инициалов (например, 'Иванов И.В.' -> 'Иванов Иван Васильевич').\n"
" - Если уверенность в совпадении меньше 85% — НЕ сопоставляй, оставляй запись как нераспознанную.\n\n"
"3. ФОРМАТ ВЫХОДНЫХ ДАННЫХ:\n"
" Верни строго JSON-объект без вводных слов и Markdown-разметки: {\"mappings\": [...], \"unmatched\": [...]}"
)
rules = [
(system_ai_prompt, "System_AI"),
("ГЛОБАЛЬНОЕ ПРАВИЛО 1: Любое физическое присутствие сотрудника по СКУД при наличии в 1С документа отсутствия (отпуск, больничный, уход за ребенком и др.), КРОМЕ командировок и служебных поездок — ЯВЛЯЕТСЯ СТРОГОЙ ИСТИННОЙ АНОМАЛИЕЙ.", "Human"),
("ГЛОБАЛЬНОЕ ПРАВИЛО 2: Документы отсутствий из 1С:ЗУП имеют 100% высший приоритет над статическим реестром удалёнщиков.", "Human"),
("ГЛОБАЛЬНОЕ ПРАВИЛО 3: Удалённые сотрудники имеют право беспрепятственно работать в офисе по СКУД в любое время (это не является аномалией).", "Human")
]
conn = sqlite3.connect(DB_PATH)
cursor = conn.cursor()
# Очищаем таблицу перед записью, чтобы избежать дублей и UNIQUE constraint error
cursor.execute("DELETE FROM ai_knowledge_base;")
for rule_text, added_by in rules:
cursor.execute(
"INSERT INTO ai_knowledge_base (rule_text, added_by) VALUES (?, ?)",
(rule_text, added_by)
)
conn.commit()
conn.close()
print("✅ База знаний успешно перезаписана с включением системного промпта ИИ!")
zup_detective.py
+63
View File
@@ -0,0 +1,63 @@
import pyodbc
from config import ZUP_SQL_CONFIG
print("="*60)
print("🕵️‍♂️ ZUP DETECTIVE: ПОИСК СТАТУСА 'УВОЛЕН'")
print("="*60)
conn_str = (
f"DRIVER={ZUP_SQL_CONFIG['driver']};"
f"SERVER={ZUP_SQL_CONFIG['server']};"
f"DATABASE={ZUP_SQL_CONFIG['database']};"
f"UID={ZUP_SQL_CONFIG['user']};"
f"PWD={ZUP_SQL_CONFIG['password']};"
f"TrustServerCertificate=yes;"
)
with pyodbc.connect(conn_str) as conn:
cursor = conn.cursor()
# 1. Получаем ID наших подопытных
cursor.execute("SELECT TOP 1 _IDRRef FROM dbo._Reference299 WHERE _Description LIKE '%Шумский Алексей Анатольевич%'")
id_work = cursor.fetchone()[0]
cursor.execute("SELECT TOP 1 _IDRRef FROM dbo._Reference299 WHERE _Description LIKE '%Щелканов Дмитрий Викторович%'")
id_fired = cursor.fetchone()[0]
# 2. Сканируем два самых больших кадровых регистра, которые нашел наш сканер
for table in ['_InfoRg16684', '_InfoRg32796']:
print(f"\n🔍 АНАЛИЗ РЕГИСТРА: {table}")
# Ищем все колонки-ссылки
cursor.execute(f"SELECT name FROM sys.columns WHERE object_id = OBJECT_ID('{table}') AND system_type_id = 173 AND max_length = 16")
cols_bin = [r[0] for r in cursor.fetchall()]
found = False
for col in cols_bin:
cursor.execute(f"SELECT COUNT(*) FROM dbo.{table} WITH (NOLOCK) WHERE {col} = ?", (id_work,))
if cursor.fetchone()[0] > 0:
print(f"✅ Найдена колонка связи с Сотрудником: {col}")
# Забираем полные строки обоих людей
cursor.execute(f"SELECT * FROM dbo.{table} WITH (NOLOCK) WHERE {col} = ?", (id_work,))
data_work = cursor.fetchone()
all_cols = [c[0] for c in cursor.description]
cursor.execute(f"SELECT * FROM dbo.{table} WITH (NOLOCK) WHERE {col} = ?", (id_fired,))
data_fired = cursor.fetchone()
if data_work and data_fired:
print("\n⚖️ ОТЛИЧИЯ В ДАННЫХ (Работает vs Уволен):")
for i, c in enumerate(all_cols):
# Выводим только те поля, где есть отличия
if data_work[i] != data_fired[i]:
print(f"Поле [{c}]:")
print(f" 🟢 Шумский: {data_work[i]}")
print(f" 🔴 Щелканов: {data_fired[i]}\n")
found = True
break
if found:
break
print("✅ Расследование завершено!")
zup_scanner.py
+60
View File
@@ -0,0 +1,60 @@
import pyodbc
from config import ZUP_SQL_CONFIG
print("="*60)
print("🔍 ЗАПУСК СКАНЕРА СТРУКТУРЫ 1С:ЗУП 3.1")
print("="*60)
conn_str = (
f"DRIVER={ZUP_SQL_CONFIG['driver']};"
f"SERVER={ZUP_SQL_CONFIG['server']};"
f"DATABASE={ZUP_SQL_CONFIG['database']};"
f"UID={ZUP_SQL_CONFIG['user']};"
f"PWD={ZUP_SQL_CONFIG['password']};"
f"TrustServerCertificate=yes;"
f"Encrypt=no;"
)
with pyodbc.connect(conn_str) as conn:
cursor = conn.cursor()
print("⏳ Анализ справочников...")
cursor.execute("SELECT name FROM sys.tables WHERE name LIKE '_Reference%'")
for row in cursor.fetchall():
t = row[0]
try:
cursor.execute(f"SELECT TOP 5 CAST(_Description AS NVARCHAR(255)) FROM {t} WITH (NOLOCK) WHERE _Description <> ''")
vals = [r[0].strip() for r in cursor.fetchall() if r[0]]
if not vals: continue
text = " ".join(vals).lower()
# Ищем сотрудников (исключая физлиц _Reference299)
if t != '_Reference299' and any(w in text for w in ["александров", "иванов", "смирнов", "петрова", "сергеев"]):
print(f"👤 СОТРУДНИКИ ({t}): {vals[:3]}")
# Ищем должности
elif any(w in text for w in ["инженер", "специалист", "директор", "уборщик"]):
print(f"💼 ДОЛЖНОСТИ ({t}): {vals[:3]}")
# Ищем подразделения
elif any(w in text for w in ["отдел", "управление", "служба"]):
print(f"🏢 ПОДРАЗДЕЛЕНИЯ ({t}): {vals[:3]}")
except:
pass
print("\n⏳ Анализ регистров кадровой истории...")
cursor.execute("SELECT name FROM sys.tables WHERE name LIKE '_InfoRg%'")
for row in cursor.fetchall():
t = row[0]
try:
# Ищем широкие регистры (Текущие кадровые данные имеют много колонок)
cursor.execute(f"SELECT COUNT(*) FROM sys.columns WHERE object_id = OBJECT_ID('{t}')")
col_count = cursor.fetchone()[0]
cursor.execute(f"SELECT COUNT(*) FROM {t} WITH (NOLOCK)")
row_count = cursor.fetchone()[0]
if col_count > 15 and row_count > 500:
print(f"📋 ШИРОКИЙ РЕГИСТР КАДРОВ ({t}) -> Колонок: {col_count}, Строк: {row_count}")
except:
pass
print("\n✅ Сканирование завершено!")