# -*- coding: utf-8 -*- """ Утилита распаковки архивов Hinode SOT/SP в структуру коллекции SunWorld. GUI-версия с формой. """ import os import re import sys import shutil import zipfile import logging import threading import tkinter as tk from tkinter import ttk, filedialog, messagebox from pathlib import Path from datetime import datetime # ---------- Константы ---------- DEFAULT_ROOT = Path(r"S:\Z\sot") HEX_ZIP_RE = re.compile(r"^[0-9a-fA-F]{10,}\.zip$", re.IGNORECASE) SESSION_RE = re.compile(r"^(\d{8})_(\d{6})$") # yyyymmdd_HHMMSS # ---------- Логирование ---------- class TextHandler(logging.Handler): """Handler, который пишет логи в tkinter Text-виджет.""" def __init__(self, text_widget: tk.Text): super().__init__() self.text_widget = text_widget def emit(self, record): msg = self.format(record) + "\n" # Запись в главный поток tkinter self.text_widget.after(0, self._append, msg) def _append(self, msg): self.text_widget.configure(state="normal") self.text_widget.insert(tk.END, msg) self.text_widget.see(tk.END) self.text_widget.configure(state="disabled") def setup_logger(log_path: Path, text_widget: tk.Text) -> logging.Logger: logger = logging.getLogger("hinode_unpacker") logger.setLevel(logging.DEBUG) logger.handlers.clear() fmt = logging.Formatter( "%(asctime)s [%(levelname)s] %(message)s", datefmt="%H:%M:%S", ) # Файловый обработчик fh = logging.FileHandler(log_path, encoding="utf-8") fh.setLevel(logging.DEBUG) fh.setFormatter(fmt) logger.addHandler(fh) # Обработчик в Text-виджет th = TextHandler(text_widget) th.setLevel(logging.INFO) th.setFormatter(fmt) logger.addHandler(th) return logger # ---------- Вспомогательные функции ---------- def get_default_downloads_dir() -> "Path | None": """Стандартная папка «Загрузки» Windows.""" try: import winreg key = winreg.OpenKey( winreg.HKEY_CURRENT_USER, r"Software\Microsoft\Windows\CurrentVersion\Explorer\Shell Folders", ) downloads = winreg.QueryValueEx(key, "{374DE290-123F-4565-9164-39C4925E467B}")[0] winreg.CloseKey(key) p = Path(downloads) if p.exists(): return p except Exception: pass home = Path.home() / "Downloads" if home.exists(): return home return None def find_hex_zips_sorted(folder: Path) -> "list[Path]": """ Возвращает список ZIP-файлов с hex-именами (>=10 символов), отсортированный по убыванию времени модификации (свежие сверху). """ if not folder or not folder.exists(): return [] matches = [ p for p in folder.iterdir() if p.is_file() and HEX_ZIP_RE.match(p.name) ] # Сортировка: самый свежий (наибольший mtime) — первый matches.sort(key=lambda p: p.stat().st_mtime, reverse=True) return matches def parse_zip_entry(zip_path_str: str): """Разбор пути внутри ZIP-архива.""" p = Path(zip_path_str.replace("\\", "/")) parts = p.parts if len(parts) < 7: return None level_name = parts[0].lower() if level_name not in ("level1", "level2", "level2.1"): return None session_dir = parts[5] m = SESSION_RE.match(session_dir) if not m: return None yyyymmdd, hhmmss = m.group(1), m.group(2) filename = parts[-1] if filename == "" or zip_path_str.endswith("/"): return None return { "level": level_name, "session_full": f"{yyyymmdd}_{hhmmss}", "session_short": f"{yyyymmdd}_{hhmmss[:4]}", "month_dir": f"{yyyymmdd[:4]}_{yyyymmdd[4:6]}", "filename": filename, } def target_path_for(root: Path, info: dict) -> Path: if info["level"] == "level1": session_folder = info["session_full"] else: session_folder = info["session_short"] return root / info["month_dir"] / session_folder / info["filename"] def process_zip(zip_path: Path, root: Path, logger: logging.Logger) -> dict: stats = {"total": 0, "copied": 0, "skipped_exists": 0, "errors": 0} logger.info(f"=== Обработка архива: {zip_path.name} ===") try: zf = zipfile.ZipFile(zip_path, "r") except zipfile.BadZipFile as e: logger.error(f"Не удалось открыть ZIP: {e}") stats["errors"] += 1 return stats with zf: entries = zf.namelist() logger.info(f"В архиве {len(entries)} записей") for entry in entries: info = parse_zip_entry(entry) if info is None: continue stats["total"] += 1 target = target_path_for(root, info) if target.exists(): logger.warning( f"[ПРОПУСК] Уже существует: {target.relative_to(root)}" ) stats["skipped_exists"] += 1 continue target.parent.mkdir(parents=True, exist_ok=True) try: with zf.open(entry) as src, open(target, "wb") as dst: shutil.copyfileobj(src, dst) logger.info( f"[+OK] {info['level']} -> {target.relative_to(root)}" ) stats["copied"] += 1 except Exception as e: logger.error(f"[ОШИБКА] {entry}: {e}") stats["errors"] += 1 if target.exists(): try: target.unlink() except Exception: pass logger.info( f"Итог по {zip_path.name}: " f"обработано={stats['total']}, " f"скопировано={stats['copied']}, " f"пропущено(есть)={stats['skipped_exists']}, " f"ошибок={stats['errors']}" ) return stats # ---------- GUI ---------- class App(tk.Tk): def __init__(self): super().__init__() self.title("Hinode SOT/SP Unpacker") self.geometry("780x520") self.resizable(True, True) self.downloads_dir = get_default_downloads_dir() self.zip_files: "list[Path]" = [] self._build_ui() self._load_zip_list() # ---------- Построение интерфейса ---------- def _build_ui(self): pad = {"padx": 6, "pady": 4} # --- Строка 1: ZIP-архив --- f1 = ttk.Frame(self) f1.pack(fill="x", **pad) ttk.Label(f1, text="ZIP-архив:").pack(side="left") self.combo_zip = ttk.Combobox(f1, state="readonly") self.combo_zip.pack(side="left", fill="x", expand=True, padx=(6, 0)) ttk.Button(f1, text="Обновить список", command=self._load_zip_list).pack( side="left", padx=(6, 0) ) # --- Строка 2: Корень коллекции --- f2 = ttk.Frame(self) f2.pack(fill="x", **pad) ttk.Label(f2, text="Корень коллекции:").pack(side="left") self.entry_root = ttk.Entry(f2) self.entry_root.insert(0, str(DEFAULT_ROOT)) self.entry_root.pack(side="left", fill="x", expand=True, padx=(6, 0)) ttk.Button(f2, text="...", width=3, command=self._choose_root).pack( side="left", padx=(6, 0) ) # --- Строка 3: Кнопки GO / Exit --- f3 = ttk.Frame(self) f3.pack(fill="x", **pad) self.btn_go = ttk.Button(f3, text="GO", command=self._on_go) self.btn_go.pack(side="left", padx=(0, 6)) self.btn_exit = ttk.Button(f3, text="Exit", command=self._on_exit) self.btn_exit.pack(side="left") self.lbl_status = ttk.Label(f3, text="", foreground="gray") self.lbl_status.pack(side="left", padx=(20, 0)) # --- Область лога --- f_log = ttk.LabelFrame(self, text="Сообщения") f_log.pack(fill="both", expand=True, **pad) self.text_log = tk.Text( f_log, height=18, state="disabled", wrap="word", font=("Consolas", 9), ) sb = ttk.Scrollbar(f_log, orient="vertical", command=self.text_log.yview) self.text_log.configure(yscrollcommand=sb.set) sb.pack(side="right", fill="y") self.text_log.pack(side="left", fill="both", expand=True) # ---------- Загрузка списка ZIP ---------- def _load_zip_list(self): self.zip_files = find_hex_zips_sorted(self.downloads_dir) names = [p.name for p in self.zip_files] self.combo_zip["values"] = names if names: # Самый свежий — первый в списке, он же в edit-поле self.combo_zip.current(0) self._log_print( f"Найдено ZIP в {self.downloads_dir}: {len(names)}" ) else: self.combo_zip.set("") self._log_print( f"Подходящих ZIP не найдено в {self.downloads_dir}" ) # ---------- Выбор корня ---------- def _choose_root(self): current = self.entry_root.get() initial = current if os.path.isdir(current) else str(DEFAULT_ROOT) path = filedialog.askdirectory( title="Выберите корневую папку коллекции SunWorld", initialdir=initial, mustexist=False, ) if path: self.entry_root.delete(0, tk.END) self.entry_root.insert(0, path) # ---------- Вывод в лог-виджет (до запуска распаковки) ---------- def _log_print(self, msg: str): stamp = datetime.now().strftime("%H:%M:%S") self.text_log.configure(state="normal") self.text_log.insert(tk.END, f"{stamp} [INFO] {msg}\n") self.text_log.see(tk.END) self.text_log.configure(state="disabled") # ---------- GO ---------- def _on_go(self): zip_name = self.combo_zip.get().strip() if not zip_name: messagebox.showwarning("Внимание", "Не выбран ZIP-архив.") return # Ищем полный путь по имени zip_path = next((p for p in self.zip_files if p.name == zip_name), None) if zip_path is None: messagebox.showerror("Ошибка", f"Файл не найден: {zip_name}") return root_str = self.entry_root.get().strip() if not root_str: messagebox.showwarning("Внимание", "Не указан корень коллекции.") return root = Path(root_str) # Блокируем кнопки self.btn_go.configure(state="disabled") self.btn_exit.configure(state="disabled") self.lbl_status.configure(text="Работаю...", foreground="blue") # Очищаем текстовый лог self.text_log.configure(state="normal") self.text_log.delete("1.0", tk.END) self.text_log.configure(state="disabled") # Запускаем в отдельном потоке t = threading.Thread( target=self._worker, args=(zip_path, root), daemon=True ) t.start() def _worker(self, zip_path: Path, root: Path): try: root.mkdir(parents=True, exist_ok=True) stamp = datetime.now().strftime("%Y%m%d_%H%M%S") log_path = root / f"unpack_{stamp}.log" logger = setup_logger(log_path, self.text_log) logger.info(f"Корень коллекции: {root}") logger.info(f"Лог-файл: {log_path}") logger.info(f"Архив: {zip_path}") stats = process_zip(zip_path, root, logger) logger.info("=" * 60) logger.info("СВОДКА:") logger.info(f" Обработано файлов : {stats['total']}") logger.info(f" Скопировано новых : {stats['copied']}") logger.info(f" Пропущено (есть) : {stats['skipped_exists']}") logger.info(f" Ошибок : {stats['errors']}") logger.info("=" * 60) # Возврат в GUI-поток self.after(0, self._on_done, stats, log_path) except Exception as e: self.after(0, self._on_done, None, None, str(e)) def _on_done(self, stats, log_path, error=None): self.btn_go.configure(state="normal") self.btn_exit.configure(state="normal") if error: self.lbl_status.configure(text="Ошибка!", foreground="red") messagebox.showerror("Критическая ошибка", error) return self.lbl_status.configure(text="Готово.", foreground="green") msg = ( f"Готово.\n\n" f"Скопировано новых : {stats['copied']}\n" f"Пропущено (есть) : {stats['skipped_exists']}\n" f"Ошибок : {stats['errors']}\n\n" f"Лог сохранён:\n{log_path}" ) messagebox.showinfo("Hinode Unpacker", msg) # ---------- Exit ---------- def _on_exit(self): self.destroy() # ---------- Точка входа ---------- def main(): app = App() app.mainloop() if __name__ == "__main__": try: main() except Exception as e: print(f"Критическая ошибка: {e}") input("Нажмите Enter для выхода...") sys.exit(1)