import os import re import threading import time import tkinter as tk from tkinter import messagebox, filedialog import pyodbc import uuid import sys # Поиск сеансов по домену — необязательная часть: если pywin32 недоступен, # программа должна продолжать работать в режиме «по профилям». try: import ad_sessions SESSIONS_ERROR = "" except Exception as _exc: ad_sessions = None SESSIONS_ERROR = str(_exc) _BANNER = (""" ....::::::::: ::::::::::-----:::::: ::::------------=--------::.. :::--=--============---==+===-:::. ::---========++=++++======+++++==--:.: :--==++++++++++++++++++========---==-::. --==++++++++++++++++++++=====---:::-==--:: :-==+**++++++++++++++++++====------::-==-:: -==+**++++++++++++++++++++=====--------=-:: -=++*++++++++++++++++++++++=========-----:: -=+++++++++++++++++++++++++===========----: :=++++++++++++++++++++++++++==========---:: -=++++++++++++++++++===+++++===-----====-- -=++++++++++++++++++++==+++++=====-----==- -++++++++++++++++++++++++++++++++====----- ==+++++++*#####*******++++++++++++++++=--= +===++++++****##%%%%##*****++++**###%##*+==- +*++=++++****#%##%@%%%%###***++*#%%%%%###*===- +*++=++++**####*#%@%#%%%##**++*#%%%@@%+*%%*+=- *++=+++++**+****####%%##*++==+#%%#%%#+++*+==- +*+++++++++++****###****++=====*####*++=====- +*=+++++++++++++++++++++++=======+**++=----- +*=+++++++++++++++++***++++==-=====+++==---- ++=+++++++++++++++*****+++==---===========-- +=+++++++++++++*****+++++===---++======--=- ==++++++++++**********+++++==--+++++=====- -==+++++++++*********##*******==++++++===- -==++++++++*****#**##*####**+====++++++== #%%%%%===+++++++++***##*****####**++++++==+++=- *+*#%%%#*+====++++++++**#%%%##****##********+====- ++++++*##%@%#+++**====+++++++*##############***###*+====- ++******###%%@@@@@%####*+=+==+++++**************+++++=+*++===- #***+**#*###%@@@@@@@@@@@@@%###*+=+==++++*****++*********++==++++=-=+=+=- #*#%%%###%%@@@@@@@@@@@@@@@@%###*++++++++**+************+++========*****++==- %%%######%%####%@@@@@@@@@@@@@@%%**+++++++++++++++***+++=========*###******+++++== %%@@@@%%%%%%%%%#%%@@@@@@@@@@@@@@%#**+*+**++++++++++++++++=====+%###########***++++* %@@@@@@@@@%%@@%%%%%%%@@@@@@@@@@@@@@###*****++++++*+++++++===+**#%####%###%####***+*** %%%@@@@@@@@@@@@@@@@@@@%@@@@@@@@@@@@@@@@%####**++++***++++*#####*%##%####%#%%%%##******* %%%%@%@@@@@@@@@@@@@@@@@@@%@@@@@@@@@@@@@@@@@@@%%%%%%##%%%@%%####*%%#%####%#%%%%%######### @@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@%%%##%#*#%#%####%##%%%%%%###%#%## @@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@%%#%%###%######%##%#%%%%%%%%%%%## @@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@%%%%%%%####%%#####%##%%%%%%%%%%%%%### @@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@%%%%####%%#%%##%#%%%%%%%%%%@%@%### @@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@%%%%#####%%%#%%%%#%%%%%%%@@@@@@@%%## @@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@%%%%%%%#%%%%##%%#%%%%%%%%%%%%@@@@@@@%%#** @@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@%%%%@@@%####%%%%%%%%%%%%%%%%%%%%%%%%%%%%%@@@@@@%%%#** @@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@%%%%%%@@@@%%%%%%%#%%%%%%%##%%%%%%%%%%%%%%%%%@@@@@@%%%%##** @@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@%%%%@@@@%%@@@@@@@%%%%%%##%%%%%%%###%%%%%%%%%%%%%%%%%@@%%%%%%%%##*+ @@@@@@@@@@@@@@@@@@@@@@@@@@@@@%%%@@@@@@@@%%%%@@@@@%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%###%%*# @@@@@@@@@@@@@@@@@@@@@@@@@@@@@%%@@@@@@%@@@%@@@@@@@%%%@%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%@%%%% """) # В оконном режиме (сборка без консоли) sys.stdout может быть None — не роняем старт if getattr(sys, "stdout", None): try: print(_BANNER) except Exception: pass class RemoteFolderSelector(tk.Tk): def __init__(self): super().__init__() self.title("Добавление баз 1С пользователям") self.geometry("1180x820") self.minsize(1040, 700) # --- Состояние --- self.ip_address = None # None/"" => локальный ПК, иначе IP удалённого ПК self.active_user = None # пользователь, чьи базы показаны в панели "Подключённые базы" self.user_vars = {} # имя пользователя -> BooleanVar (галочка) self.db_vars = {} # имя базы из SQL -> BooleanVar (галочка) self.conn_vars = {} # имя подключённой базы активного пользователя -> BooleanVar self.session_vars = {} # подпись сеанса -> BooleanVar (галочка) self.sessions_by_key = {} # подпись сеанса -> словарь с данными сеанса self.scanning = False # идёт ли сейчас поиск сеансов self.scan_cache = None # последний снимок сеансов всего домена self.scan_cache_at = 0.0 # время снимка (time.monotonic) self.scan_cache_stats = {} # статистика последнего скана # Компоновка окна: три колонки сверху, кнопки действий снизу self.grid_columnconfigure(0, weight=1) self.grid_columnconfigure(1, weight=1) self.grid_columnconfigure(2, weight=1) self.grid_rowconfigure(0, weight=1) self.frame_left = tk.Frame(self) self.frame_left.grid(row=0, column=0, padx=10, pady=10, sticky="nsew") self.frame_mid = tk.Frame(self) self.frame_mid.grid(row=0, column=1, padx=10, pady=10, sticky="nsew") self.frame_right = tk.Frame(self) self.frame_right.grid(row=0, column=2, padx=10, pady=10, sticky="nsew") self._build_left_panel() self._build_sessions_panel() self._build_right_panel() # Кнопки действий снизу, на всю ширину окна actions = tk.Frame(self) actions.grid(row=1, column=0, columnspan=3, padx=10, pady=(0, 12), sticky="ew") actions.grid_columnconfigure(0, weight=1) actions.grid_columnconfigure(1, weight=1) self.add_session_button = tk.Button( actions, text="Добавить базы в отмеченные сеансы", font=("Segoe UI", 11, "bold"), command=self.add_bases_to_sessions ) self.add_session_button.grid(row=0, column=0, padx=(0, 5), sticky="ew") self.add_button = tk.Button( actions, text="Добавить базы отмеченным пользователям (профили)", font=("Segoe UI", 11), command=self.add_database_to_ibases ) self.add_button.grid(row=0, column=1, padx=(5, 0), sticky="ew") # По умолчанию грузим пользователей локального ПК self.load_users() # ------------------------------------------------------------------ UI def _build_left_panel(self): f = self.frame_left f.grid_columnconfigure(0, weight=1) tk.Label(f, text="IP-адрес (необязательно):").grid(row=0, column=0, padx=5, pady=(5, 0), sticky="w") self.ip_entry = tk.Entry(f) self.ip_entry.grid(row=1, column=0, padx=5, pady=(0, 5), sticky="ew") self.ip_entry.bind("", lambda e: self.load_users()) tk.Button(f, text="Загрузить пользователей", command=self.load_users)\ .grid(row=2, column=0, padx=5, pady=5, sticky="ew") # Поиск по пользователям tk.Label(f, text="Поиск пользователей:").grid(row=3, column=0, padx=5, pady=(5, 0), sticky="w") self.user_search_var = tk.StringVar() self.user_search_var.trace_add("write", lambda *a: self._render_checklist( self.user_inner, self.user_vars, self.user_search_var.get(), self.on_user_click)) tk.Entry(f, textvariable=self.user_search_var)\ .grid(row=4, column=0, padx=5, pady=(0, 5), sticky="ew") # Отметить все / снять все (с учётом фильтра) bar = tk.Frame(f) bar.grid(row=5, column=0, padx=5, pady=(0, 3), sticky="w") tk.Button(bar, text="Отметить все", command=lambda: self._set_all(self.user_vars, self.user_search_var.get(), True))\ .pack(side="left", padx=(0, 4)) tk.Button(bar, text="Снять все", command=lambda: self._set_all(self.user_vars, self.user_search_var.get(), False))\ .pack(side="left") # Список пользователей self.user_container, self.user_inner = self._create_scrollable_checklist(f, height=150) self.user_container.grid(row=6, column=0, padx=5, pady=5, sticky="nsew") f.grid_rowconfigure(6, weight=1) # Панель подключённых баз активного пользователя self.conn_label = tk.Label(f, text="Подключённые базы: (выберите пользователя)") self.conn_label.grid(row=7, column=0, padx=5, pady=(8, 0), sticky="w") bar2 = tk.Frame(f) bar2.grid(row=8, column=0, padx=5, pady=(0, 3), sticky="w") tk.Button(bar2, text="Отметить все", command=lambda: self._set_all(self.conn_vars, "", True)).pack(side="left", padx=(0, 4)) tk.Button(bar2, text="Снять все", command=lambda: self._set_all(self.conn_vars, "", False)).pack(side="left") self.conn_container, self.conn_inner = self._create_scrollable_checklist(f, height=150) self.conn_container.grid(row=9, column=0, padx=5, pady=5, sticky="nsew") f.grid_rowconfigure(9, weight=1) tk.Button(f, text="Удалить отмеченные базы у пользователя", command=self.delete_connected_bases)\ .grid(row=10, column=0, padx=5, pady=5, sticky="ew") # Опция дублирования в ZIP (roaming) профиль self.zip_enabled = tk.BooleanVar(value=False) tk.Checkbutton(f, text="Дублировать изменения в ZIP-профиль пользователя", variable=self.zip_enabled, command=self._toggle_zip)\ .grid(row=11, column=0, padx=5, pady=(8, 0), sticky="w") zf = tk.Frame(f) zf.grid(row=12, column=0, padx=5, pady=(0, 5), sticky="ew") zf.grid_columnconfigure(0, weight=1) self.zip_path_var = tk.StringVar(value=r"Z:\ZIP_User_Profiles") self.zip_path_entry = tk.Entry(zf, textvariable=self.zip_path_var, state="disabled") self.zip_path_entry.grid(row=0, column=0, sticky="ew", padx=(0, 4)) self.zip_browse_btn = tk.Button(zf, text="Обзор...", state="disabled", command=self._browse_zip) self.zip_browse_btn.grid(row=0, column=1) def _build_right_panel(self): f = self.frame_right f.grid_columnconfigure(0, weight=1) tk.Label(f, text="SQL-сервер:").grid(row=0, column=0, padx=5, pady=(5, 0), sticky="w") self.server_entry = tk.Entry(f) self.server_entry.grid(row=1, column=0, padx=5, pady=(0, 5), sticky="ew") self.server_entry.insert(0, "SERVERSQL") tk.Label(f, text="Логин SQL:").grid(row=2, column=0, padx=5, pady=(5, 0), sticky="w") self.login_entry = tk.Entry(f) self.login_entry.grid(row=3, column=0, padx=5, pady=(0, 5), sticky="ew") self.login_entry.insert(0, "sa") tk.Label(f, text="Пароль SQL:").grid(row=4, column=0, padx=5, pady=(5, 0), sticky="w") self.password_entry = tk.Entry(f, show="*") self.password_entry.grid(row=5, column=0, padx=5, pady=(0, 5), sticky="ew") # Enter в любом из полей подключения — сразу загружаем базы for entry in (self.server_entry, self.login_entry, self.password_entry): entry.bind("", lambda e: self.load_databases()) tk.Button(f, text="Загрузить базы", command=self.load_databases)\ .grid(row=6, column=0, padx=5, pady=5, sticky="ew") # Поиск по базам tk.Label(f, text="Поиск баз:").grid(row=7, column=0, padx=5, pady=(5, 0), sticky="w") self.db_search_var = tk.StringVar() self.db_search_var.trace_add("write", lambda *a: self._render_checklist( self.db_inner, self.db_vars, self.db_search_var.get())) tk.Entry(f, textvariable=self.db_search_var)\ .grid(row=8, column=0, padx=5, pady=(0, 5), sticky="ew") bar = tk.Frame(f) bar.grid(row=9, column=0, padx=5, pady=(0, 3), sticky="w") tk.Button(bar, text="Отметить все", command=lambda: self._set_all(self.db_vars, self.db_search_var.get(), True))\ .pack(side="left", padx=(0, 4)) tk.Button(bar, text="Снять все", command=lambda: self._set_all(self.db_vars, self.db_search_var.get(), False))\ .pack(side="left") self.db_container, self.db_inner = self._create_scrollable_checklist(f, height=380) self.db_container.grid(row=10, column=0, padx=5, pady=5, sticky="nsew") f.grid_rowconfigure(10, weight=1) def _build_sessions_panel(self): """Панель поиска активных сеансов пользователя по домену.""" f = self.frame_mid f.grid_columnconfigure(0, weight=1) tk.Label(f, text="Сеансы пользователя в домене", font=("Segoe UI", 10, "bold"))\ .grid(row=0, column=0, padx=5, pady=(5, 0), sticky="w") tk.Label(f, text="Имя пользователя:").grid(row=1, column=0, padx=5, pady=(5, 0), sticky="w") self.session_user_var = tk.StringVar() self.session_user_entry = tk.Entry(f, textvariable=self.session_user_var) self.session_user_entry.grid(row=2, column=0, padx=5, pady=(0, 5), sticky="ew") self.session_user_entry.bind("", lambda e: self.find_sessions()) scan_bar = tk.Frame(f) scan_bar.grid(row=3, column=0, padx=5, pady=5, sticky="ew") scan_bar.grid_columnconfigure(0, weight=3) scan_bar.grid_columnconfigure(1, weight=2) self.scan_button = tk.Button(scan_bar, text="Найти сеансы", command=self.find_sessions) self.scan_button.grid(row=0, column=0, padx=(0, 4), sticky="ew") # Принудительный повторный опрос домена, когда данные скана устарели self.rescan_button = tk.Button(scan_bar, text="Пересканировать", command=lambda: self.find_sessions(force=True)) self.rescan_button.grid(row=0, column=1, sticky="ew") # Отключённый сеанс тоже держит профиль загруженным, поэтому по умолчанию учитываем и его self.include_disconnected = tk.BooleanVar(value=True) tk.Checkbutton(f, text="Учитывать отключённые сеансы", variable=self.include_disconnected)\ .grid(row=4, column=0, padx=5, pady=(0, 3), sticky="w") self.scan_status = tk.Label(f, text="", anchor="w", justify="left", fg="#444", wraplength=320) self.scan_status.grid(row=5, column=0, padx=5, pady=(0, 3), sticky="ew") bar = tk.Frame(f) bar.grid(row=6, column=0, padx=5, pady=(0, 3), sticky="w") tk.Button(bar, text="Отметить все", command=lambda: self._set_all(self.session_vars, "", True)).pack(side="left", padx=(0, 4)) tk.Button(bar, text="Снять все", command=lambda: self._set_all(self.session_vars, "", False)).pack(side="left") self.sess_container, self.sess_inner = self._create_scrollable_checklist(f, height=330) self.sess_container.grid(row=7, column=0, padx=5, pady=5, sticky="nsew") f.grid_rowconfigure(7, weight=1) tk.Label(f, text="Базы добавляются в профиль на той машине,\nгде найден сеанс — " "чтобы правку не затёрло\nпри выходе из перемещаемого профиля.", anchor="w", justify="left", fg="#666")\ .grid(row=8, column=0, padx=5, pady=(3, 0), sticky="ew") if ad_sessions is None: self.scan_button.config(state="disabled") self.session_user_entry.config(state="disabled") self.scan_status.config( text=f"Поиск сеансов недоступен: не загружен модуль ({SESSIONS_ERROR})", fg="#a00") def _create_scrollable_checklist(self, parent, width=220, height=180): """Прокручиваемая область для чекбоксов. Возвращает (container, inner).""" container = tk.Frame(parent, borderwidth=1, relief="sunken") canvas = tk.Canvas(container, width=width, height=height, highlightthickness=0) scrollbar = tk.Scrollbar(container, orient="vertical", command=canvas.yview) inner = tk.Frame(canvas) inner.bind("", lambda e: canvas.configure(scrollregion=canvas.bbox("all"))) canvas.create_window((0, 0), window=inner, anchor="nw") canvas.configure(yscrollcommand=scrollbar.set) canvas.pack(side="left", fill="both", expand=True) scrollbar.pack(side="right", fill="y") def _on_mousewheel(event): canvas.yview_scroll(int(-event.delta / 120), "units") canvas.bind("", lambda e: canvas.bind_all("", _on_mousewheel)) canvas.bind("", lambda e: canvas.unbind_all("")) return container, inner def _load_items(self, vars_dict, items): """Создаём свежий набор BooleanVar по списку items (галочки сбрасываются).""" vars_dict.clear() for item in items: vars_dict[item] = tk.BooleanVar(value=False) def _render_checklist(self, inner, vars_dict, filter_text="", on_click=None): """Рисуем чекбоксы, показывая только элементы, подходящие под фильтр. Переменные (галочки) сохраняются между перерисовками — состояние не теряется.""" for child in inner.winfo_children(): child.destroy() ft = (filter_text or "").lower() for name, var in vars_dict.items(): if ft and ft not in name.lower(): continue cmd = (lambda n=name: on_click(n)) if on_click else None tk.Checkbutton(inner, text=name, variable=var, anchor="w", command=cmd)\ .pack(fill="x", anchor="w") # После перерисовки (например, при вводе в поиск) возвращаем прокрутку наверх, # иначе список остаётся проскроллен вниз и результат поиска не виден. inner.master.yview_moveto(0) def _set_all(self, vars_dict, filter_text, value): """Отмечаем/снимаем все чекбоксы, попадающие под текущий фильтр.""" ft = (filter_text or "").lower() for name, var in vars_dict.items(): if ft and ft not in name.lower(): continue var.set(value) def _get_checked(self, vars_dict): return [name for name, var in vars_dict.items() if var.get()] def _toggle_zip(self): state = "normal" if self.zip_enabled.get() else "disabled" self.zip_path_entry.config(state=state) self.zip_browse_btn.config(state=state) def _browse_zip(self): path = filedialog.askdirectory(title="Каталог с ZIP-профилями пользователей") if path: self.zip_path_var.set(os.path.normpath(path)) # -------------------------------------------------------------- Загрузка def load_users(self): """Загружаем список пользователей с локального ПК или с удалённого по IP.""" self.ip_address = self.ip_entry.get().strip() if not self.ip_address: base = r"C:\Users" else: base = f"\\\\{self.ip_address}\\C$\\Users" try: folders = [f for f in os.listdir(base) if os.path.isdir(os.path.join(base, f))] except Exception as e: messagebox.showerror("Ошибка", f"Не удалось загрузить пользователей:\n{e}") return self._load_items(self.user_vars, folders) self._render_checklist(self.user_inner, self.user_vars, self.user_search_var.get(), self.on_user_click) # Сбрасываем панель подключённых баз — она относилась к прежнему источнику self.active_user = None self.conn_vars.clear() self._render_checklist(self.conn_inner, self.conn_vars, "") self.conn_label.config(text="Подключённые базы: (выберите пользователя)") def load_databases(self): """Подключаемся к SQL-серверу и загружаем список баз.""" server = self.server_entry.get().strip() login = self.login_entry.get() password = self.password_entry.get() try: connection_string = f"DRIVER={{SQL Server}};SERVER={server};UID={login};PWD={password};" connection = pyodbc.connect(connection_string) cursor = connection.cursor() cursor.execute("SELECT name FROM sys.databases") databases = cursor.fetchall() cursor.close() connection.close() except Exception as e: messagebox.showerror("Ошибка", f"Не удалось подключиться к SQL-серверу:\n{e}") return # Служебные базы SQL Server скрываем — это не базы 1С system_dbs = {"master", "tempdb", "model", "msdb"} names = [db[0] for db in databases if db[0].lower() not in system_dbs] self._load_items(self.db_vars, names) self._render_checklist(self.db_inner, self.db_vars, self.db_search_var.get()) def on_user_click(self, user): """Клик по пользователю — делаем его активным и показываем его подключённые базы. Заодно подставляем имя в поле поиска сеансов, чтобы не набирать вручную.""" self.load_connected_bases(user) if ad_sessions is not None: self.session_user_var.set(user) def load_connected_bases(self, user): """Читаем ibases.v8i пользователя и показываем список его баз в панели.""" self.active_user = user self.ip_address = self.ip_entry.get().strip() path = self.get_ibases_path(user) text = self._read_ibases(path) _, sections = self._parse_ibases(text) names = [name for name, _ in sections] self._load_items(self.conn_vars, names) self._render_checklist(self.conn_inner, self.conn_vars, "") suffix = "" if names else " (баз нет)" self.conn_label.config(text=f"Подключённые базы: {user}{suffix}") # --------------------------------------------------------------- Пути def get_ibases_path(self, user): """Путь к ibases.v8i пользователя (локально либо по IP).""" if not self.ip_address: base_path = f"C:\\Users\\{user}\\AppData\\Roaming\\1C\\1CEStart" else: base_path = f"\\\\{self.ip_address}\\C$\\Users\\{user}\\AppData\\Roaming\\1C\\1CEStart" return os.path.join(base_path, "ibases.v8i") def get_zip_ibases_path(self, user): """Путь к ibases.v8i внутри ZIP-профиля пользователя. Папку профиля ищем автопоиском: либо .* (например .V6). Возвращаем None, если корень недоступен или профиль не найден.""" root = self.zip_path_var.get().strip() if not root or not os.path.isdir(root): return None u = user.lower() candidates = [] try: for name in os.listdir(root): if not os.path.isdir(os.path.join(root, name)): continue low = name.lower() if low == u or low.startswith(u + "."): candidates.append(name) except Exception: return None if not candidates: return None # Приоритет: точное .V6, затем любой суффикс .*, затем совпадение без суффикса def score(n): low = n.lower() if low == u + ".v6": return 0 if "." in low: return 1 return 2 candidates.sort(key=score) chosen = candidates[0] return os.path.join(root, chosen, "AppData", "Roaming", "1C", "1CEStart", "ibases.v8i") # ------------------------------------------------------- Чтение/запись v8i def _read_ibases(self, path): """Читаем ibases.v8i, устойчиво к кодировке (UTF-8 с BOM / без / cp1251).""" if not os.path.exists(path): return "" for enc in ("utf-8-sig", "cp1251"): try: with open(path, "r", encoding=enc) as f: return f.read() except UnicodeDecodeError: continue with open(path, "r", encoding="utf-8", errors="ignore") as f: return f.read() def _parse_ibases(self, text): """Разбираем содержимое на (преамбула, [(имя_базы, блок_текста), ...]). Секция начинается со строки вида [ИмяБазы].""" header_re = re.compile(r"^\s*\[(.+?)\]\s*$") preamble = [] sections = [] # список [имя, [строки]] current = None for line in text.splitlines(keepends=True): m = header_re.match(line) if m: current = [m.group(1), [line]] sections.append(current) elif current is None: preamble.append(line) else: current[1].append(line) return "".join(preamble), [(name, "".join(ls)) for name, ls in sections] def build_base_entry(self, base, server): """Формируем запись одной базы для ibases.v8i.""" return ( f"\n[{base}]\n" f'Connect=Srvr="{server}";Ref="{base}";\n' f"ID={uuid.uuid4()}\n" "OrderInList=255\n" "Folder=/\n" "OrderInTree=16640\n" "External=0\n" "ClientConnectionSpeed=Normal\n" "App=Auto\n" "WA=1\n" "Version=8.3\n" "DefaultApp=ThickClient\n" "DisableLocalSpeechToText=0\n" ) def _add_entries_to_file(self, ibases_path, bases, server): """Добавляем базы в один файл ibases.v8i (с фильтром дублей). Возвращаем (добавлено, пропущено).""" base_dir = os.path.dirname(ibases_path) if not os.path.exists(base_dir): os.makedirs(base_dir) existing = self._read_ibases(ibases_path) entries = [] added = skipped = 0 for base in bases: connect_line = f'Connect=Srvr="{server}";Ref="{base}";' if connect_line in existing or any(connect_line in e for e in entries): skipped += 1 continue entries.append(self.build_base_entry(base, server)) added += 1 if entries: with open(ibases_path, "a", encoding="utf-8") as f: f.write("".join(entries)) return added, skipped def _delete_entries_from_file(self, ibases_path, base_names): """Удаляем указанные базы из одного файла ibases.v8i. Возвращаем число удалённых.""" if not os.path.exists(ibases_path): return 0 text = self._read_ibases(ibases_path) preamble, sections = self._parse_ibases(text) target = {b.lower() for b in base_names} kept = [blk for (name, blk) in sections if name.lower() not in target] removed = len(sections) - len(kept) if removed: with open(ibases_path, "w", encoding="utf-8") as f: f.write(preamble + "".join(kept)) return removed # ----------------------------------------------------------- Действия def add_database_to_ibases(self): """Добавляем отмеченные базы всем отмеченным пользователям (+ ZIP по галочке).""" self.ip_address = self.ip_entry.get().strip() users = self._get_checked(self.user_vars) bases = self._get_checked(self.db_vars) server = self.server_entry.get().strip() if not users: messagebox.showerror("Ошибка", "Не отмечен ни один пользователь!") return if not bases: messagebox.showerror("Ошибка", "Не отмечена ни одна база!") return if not server: messagebox.showerror("Ошибка", "Не указан SQL-сервер!") return use_zip = self.zip_enabled.get() added = skipped = zadded = zskipped = 0 errors = [] for user in users: try: a, s = self._add_entries_to_file(self.get_ibases_path(user), bases, server) added += a skipped += s except Exception as e: errors.append(f"{user}: {e}") if use_zip: zp = self.get_zip_ibases_path(user) if zp is None: errors.append(f"{user}: ZIP-профиль не найден") else: try: a, s = self._add_entries_to_file(zp, bases, server) zadded += a zskipped += s except Exception as e: errors.append(f"{user} (ZIP): {e}") # Обновим панель подключённых баз, если активный пользователь среди затронутых if self.active_user in users: self.load_connected_bases(self.active_user) summary = ( f"Пользователей: {len(users)}, баз отмечено: {len(bases)}\n" f"Локально — добавлено: {added}, пропущено (уже есть): {skipped}" ) if use_zip: summary += f"\nZIP-профили — добавлено: {zadded}, пропущено: {zskipped}" self._report(summary, errors) def delete_connected_bases(self): """Удаляем отмеченные подключённые базы у активного пользователя (+ ZIP по галочке).""" user = self.active_user if not user: messagebox.showerror("Ошибка", "Сначала выберите пользователя слева.") return bases = self._get_checked(self.conn_vars) if not bases: messagebox.showerror("Ошибка", "Не отмечена ни одна база для удаления!") return if not messagebox.askyesno( "Подтверждение", f"Удалить {len(bases)} баз(ы) у пользователя «{user}»?\n\n" + "\n".join(bases) ): return self.ip_address = self.ip_entry.get().strip() use_zip = self.zip_enabled.get() removed_local = removed_zip = 0 errors = [] try: removed_local = self._delete_entries_from_file(self.get_ibases_path(user), bases) except Exception as e: errors.append(f"{user}: {e}") if use_zip: zp = self.get_zip_ibases_path(user) if zp is None: errors.append(f"{user}: ZIP-профиль не найден") else: try: removed_zip = self._delete_entries_from_file(zp, bases) except Exception as e: errors.append(f"{user} (ZIP): {e}") self.load_connected_bases(user) summary = f"Пользователь: {user}\nУдалено локально: {removed_local}" if use_zip: summary += f"\nУдалено в ZIP-профиле: {removed_zip}" self._report(summary, errors) # ------------------------------------------------------- Сеансы в домене # Сколько секунд снимок домена считается свежим. За это время поиск другого # пользователя отрабатывает мгновенно, без повторного опроса машин. SCAN_CACHE_TTL = 120 def find_sessions(self, force=False): """Ищем сеансы пользователя. Если снимок домена свежий — фильтруем его мгновенно; иначе опрашиваем домен в фоне, чтобы окно не подвисало.""" if ad_sessions is None or self.scanning: return user = self.session_user_var.get().strip() if not user: messagebox.showerror("Ошибка", "Укажите имя пользователя для поиска сеансов.") return domain = ad_sessions.current_domain() include_disc = self.include_disconnected.get() age = time.monotonic() - self.scan_cache_at if not force and self.scan_cache is not None and age < self.SCAN_CACHE_TTL: matched = ad_sessions.filter_sessions( self.scan_cache, user=user, domain=domain, include_disconnected=include_disc) stats = dict(self.scan_cache_stats) stats["matched"] = len(matched) stats["from_cache"] = True stats["cache_age"] = int(age) self._on_sessions_found(user, matched, [], stats, None) return self.scanning = True self.scan_button.config(state="disabled", text="Идёт поиск...") self.rescan_button.config(state="disabled") self.scan_status.config(text="Получаем список компьютеров из AD...", fg="#444") self.session_vars.clear() self.sessions_by_key.clear() self._render_checklist(self.sess_inner, self.session_vars, "") def progress(stage, done, total): text = f"{stage} {done}/{total}" if total else stage self.after(0, lambda: self.scan_status.config(text=text)) def worker(): try: # Снимаем сеансы всех пользователей разом и кладём в кэш — # следующий поиск по другому имени будет мгновенным everyone, errors, stats = ad_sessions.scan_domain(progress=progress) matched = ad_sessions.filter_sessions( everyone, user=user, domain=domain, include_disconnected=include_disc) stats["matched"] = len(matched) self.after(0, self._store_scan_cache, everyone, stats) self.after(0, self._on_sessions_found, user, matched, errors, stats, None) except Exception as exc: self.after(0, self._on_sessions_found, user, [], [], {}, f"{type(exc).__name__}: {exc}") threading.Thread(target=worker, daemon=True).start() def _store_scan_cache(self, sessions, stats): self.scan_cache = sessions self.scan_cache_at = time.monotonic() self.scan_cache_stats = dict(stats) def _on_sessions_found(self, user, sessions, errors, stats, fatal): """Обработка результата фонового сканирования (вызывается в потоке GUI).""" self.scanning = False self.scan_button.config(state="normal", text="Найти сеансы") self.rescan_button.config(state="normal") if fatal: self.scan_status.config(text=f"Ошибка поиска: {fatal}", fg="#a00") messagebox.showerror("Ошибка", f"Не удалось выполнить поиск сеансов:\n{fatal}") return # Чистим здесь, а не только перед сканированием: поиск по кэшу приходит # сюда напрямую, и без этого результаты прошлого поиска накапливались бы self.session_vars.clear() self.sessions_by_key.clear() for sess in sessions: machine = ad_sessions.short_name(sess["machine"]) key = f"{machine} (сеанс {sess['session_id']}, {sess['state_name']})" if sess.get("client"): key += f" ← {sess['client']}" self.sessions_by_key[key] = sess self.session_vars[key] = tk.BooleanVar(value=True) # найденное сразу отмечено self._render_checklist(self.sess_inner, self.session_vars, "") total = stats.get("total", 0) alive = stats.get("alive", 0) failed = stats.get("failed", 0) unfinished = stats.get("unfinished", 0) if sessions: head = f"Найдено сеансов: {len(sessions)} у «{user}»." color = "#060" else: head = f"Сеансов пользователя «{user}» не найдено." color = "#a60" if stats.get("from_cache"): tail = (f"Данные скана {stats.get('cache_age', 0)} с назад " f"(машин: {total}, с сеансами: {alive}). " f"Кнопка «Пересканировать» обновит.") else: tail = (f"Машин: {total}, ответили: {alive}, " f"выключено: {stats.get('offline', 0)}. " f"Время: {stats.get('scan_seconds', 0)} с " f"(AD {stats.get('ad_seconds', 0)} с).") if failed: tail += f" Ошибок: {failed}." if unfinished: tail += f" Не успели за отведённое время: {unfinished}." self.scan_status.config(text=head + " " + tail, fg=color) # Если сеансов не нашлось, а часть машин не опрошена — предупреждаем явно. # Иначе админ решит, что пользователь нигде не залогинен, добавит базу # в перемещаемый профиль, и её затрёт при выходе из пропущенного сеанса. if not sessions and (failed or unfinished) and not stats.get("from_cache"): details = "" if errors: details = "\n\n" + "\n".join(f"{host}: {err}" for host, err in errors[:8]) messagebox.showwarning( "Опрошены не все машины", f"Сеансов пользователя «{user}» не найдено, но " f"{failed + unfinished} машин(ы) не ответили.\n\n" "Пользователь может сидеть на одной из них. Прежде чем писать " "в перемещаемый профиль, нажмите «Пересканировать»." + details) def add_bases_to_sessions(self): """Добавляем отмеченные базы в профили на машинах отмеченных сеансов.""" if ad_sessions is None: messagebox.showerror("Ошибка", "Модуль поиска сеансов недоступен.") return keys = self._get_checked(self.session_vars) bases = self._get_checked(self.db_vars) server = self.server_entry.get().strip() if not keys: messagebox.showerror("Ошибка", "Не отмечен ни один сеанс!") return if not bases: messagebox.showerror("Ошибка", "Не отмечена ни одна база!") return if not server: messagebox.showerror("Ошибка", "Не указан SQL-сервер!") return chosen = [self.sessions_by_key[key] for key in keys] # Список сеансов мог быть снят пару минут назад — убеждаемся, что они ещё живы. # Машин тут единицы, проверка почти мгновенная. chosen, gone = ad_sessions.verify_sessions(chosen) if gone: names = "\n".join( f"{ad_sessions.short_name(s['machine'])} (сеанс {s['session_id']})" for s in gone) if not chosen: messagebox.showerror( "Сеансы завершены", "Выбранные сеансы уже завершены — пользователь вышел:\n\n" f"{names}\n\nНажмите «Пересканировать» и попробуйте снова.") return if not messagebox.askyesno( "Часть сеансов завершена", f"Эти сеансы уже завершены и будут пропущены:\n\n{names}\n\n" "Продолжить с остальными?"): return # На одной машине у пользователя может быть несколько сеансов (например, # активный и отключённый), а профиль у них общий — пишем в машину один раз. by_machine = {} for sess in chosen: by_machine.setdefault(sess["machine"], sess) added = skipped = 0 done_machines = [] errors = [] for machine, sess in by_machine.items(): try: path = ad_sessions.ibases_path_for_session(sess) a, s = self._add_entries_to_file(path, bases, server) added += a skipped += s done_machines.append(f"{ad_sessions.short_name(machine)}: +{a}, пропущено {s}") except Exception as exc: errors.append(f"{ad_sessions.short_name(machine)}: {exc}") summary = ( f"Пользователь: {self.session_user_var.get().strip()}\n" f"Машин обработано: {len(done_machines)} из {len(by_machine)}, " f"баз отмечено: {len(bases)}\n" f"Добавлено: {added}, пропущено (уже есть): {skipped}" ) if done_machines: summary += "\n\n" + "\n".join(done_machines) self._report(summary, errors) def _report(self, summary, errors): if errors: messagebox.showwarning("Готово с ошибками", summary + "\n\nОшибки:\n" + "\n".join(errors)) else: messagebox.showinfo("Готово", summary) if __name__ == "__main__": app = RemoteFolderSelector() app.mainloop()