# -*- coding: utf-8 -*- """导入向导：文件夹 / 文件 / JSON 备份三种来源，带预览与进度。 导入在 QThread 中执行，UI 不冻结；完成后显示结果报告。 """ from __future__ import annotations import os from pathlib import Path from PySide6.QtCore import QRunnable, Qt, QThreadPool, Signal from PySide6.QtWidgets import ( QButtonGroup, QCheckBox, QComboBox, QDialog, QDialogButtonBox, QFileDialog, QGridLayout, QHBoxLayout, QLabel, QLineEdit, QListWidget, QListWidgetItem, QMessageBox, QProgressBar, QPushButton, QRadioButton, QVBoxLayout, QWidget, ) from .. import library as lib from ..db import Database from ..models import ASSET_TYPES_ORDERED, ASSET_TYPE_META, AssetType, Series from .theme import BG_HOVER, BG_INPUT, BORDER, DANGER, FG, FG_DIM, FG_FAINT, SUCCESS, WARN _FILE_FILTERS = ( "所有支持格式 (*.png *.jpg *.jpeg *.webp *.gif *.bmp *.tiff *.pdf *.txt *.md " "*.mdx *.csv *.json *.docx *.mp3 *.wav *.m4a *.aac *.ogg *.flac);" "图片 (*.png *.jpg *.jpeg *.webp *.gif *.bmp *.tiff *.pdf);" "音频 (*.mp3 *.wav *.m4a *.aac *.ogg *.flac);" "文本 (*.txt *.md *.mdx *.csv *.json *.docx);" "所有文件 (*)" ) JSON_FILTER = "资产库备份 (*.json);;所有文件 (*)" class _ImportWorker(QRunnable): """后台执行导入任务，通过信号汇报进度。""" progress = Signal(int, int, str) # done, total, label finished = Signal(object) # ImportReport or Exception started = Signal() def __init__(self, fn, *args): super().__init__() self.fn = fn self.args = args self.setAutoDelete(True) def run(self): try: self.started.emit() self.progress.emit(0, 1, "正在准备…") result = self.fn(*self.args, progress=self._progress) self.finished.emit(result) except Exception as e: self.finished.emit(e) def _progress(self, done, total, label=""): self.progress.emit(done, total, label) class ImportDialog(QDialog): """资产导入向导。 用法:: dlg = ImportDialog(db, parent) if dlg.exec(): dlg.report() # ImportReport """ def __init__(self, db: Database, parent=None): super().__init__(parent) self.db = db self.setWindowTitle("导入资产") self.resize(720, 560) self._worker = None self._report = None lay = QVBoxLayout(self) lay.setContentsMargins(18, 14, 18, 12) lay.setSpacing(12) # ── 步骤 1：来源 ── lay.addWidget(self._title("① 选择来源")) src_row = QWidget() srcl = QHBoxLayout(src_row) srcl.setContentsMargins(0, 0, 0, 0) srcl.setSpacing(10) self.bg = QButtonGroup(self) self.r_folder = QRadioButton("📁 文件夹（批量导入漫画页 / 设定图）") self.r_file = QRadioButton("📄 文件（逐个添加）") self.r_json = QRadioButton("🗂️ JSON 备份（迁移资产库）") self.r_folder.setChecked(True) for r in (self.r_folder, self.r_file, self.r_json): self.bg.addButton(r) srcl.addWidget(r) srcl.addStretch(1) lay.addWidget(src_row) # 路径选择行 self.path_row = QHBoxLayout() self.path_edit = QLineEdit() self.path_edit.setPlaceholderText("选择文件夹…") self.path_edit.setReadOnly(True) self.btn_pick = QPushButton("浏览…") self.btn_pick.setObjectName("Ghost") self.path_row.addWidget(self.path_edit, 1) self.path_row.addWidget(self.btn_pick) lay.addLayout(self.path_row) self.r_folder.toggled.connect(self._on_source_changed) self.r_file.toggled.connect(self._on_source_changed) self.r_json.toggled.connect(self._on_source_changed) self.btn_pick.clicked.connect(self._pick_path) # ── 步骤 2：预览 ── lay.addWidget(self._title("② 预览")) self.preview_list = QListWidget() self.preview_list.setStyleSheet(f""" QListWidget {{ background: {BG_INPUT}; border: 1px solid {BORDER}; border-radius: 6px; }} QListWidget::item {{ padding: 4px 8px; color: {FG}; font-size: 12px; }} QListWidget::item:selected {{ background: {BG_HOVER}; }} """) self.preview_list.setMaximumHeight(130) lay.addWidget(self.preview_list) self.preview_hint = QLabel("选择路径后自动扫描") self.preview_hint.setStyleSheet(f"color: {FG_FAINT}; font-size: 11px;") lay.addWidget(self.preview_hint) # ── 步骤 3：设置 ── lay.addWidget(self._title("③ 导入设置")) grid = QGridLayout() grid.setHorizontalSpacing(12) grid.setVerticalSpacing(8) grid.addWidget(QLabel("目标系列"), 0, 0) self.series_combo = QComboBox() self.btn_new_series = QPushButton("＋ 新建") self.btn_new_series.setFixedWidth(72) self.btn_new_series.setObjectName("Ghost") self.btn_new_series.clicked.connect(self._quick_new_series) grid.addWidget(self.series_combo, 0, 1) grid.addWidget(self.btn_new_series, 0, 2) grid.addWidget(QLabel("资产类型"), 1, 0) self.type_combo = QComboBox() self.type_combo.addItem("自动推断（按扩展名）", None) for t in ASSET_TYPES_ORDERED: m = ASSET_TYPE_META[t] self.type_combo.addItem(f"{m.icon} {m.label}", t.value) self.recurse_chk = QCheckBox("包含子目录") self.recurse_chk.setChecked(True) grid.addWidget(self.type_combo, 1, 1) grid.addWidget(self.recurse_chk, 1, 2) grid.addWidget(QLabel("存放策略"), 2, 0) self.copy_chk = QCheckBox("复制文件到资产库（推荐，原文件可自由移动）") self.copy_chk.setChecked(True) grid.addWidget(self.copy_chk, 2, 1, 1, 2) lay.addLayout(grid) # ── 步骤 4：执行 ── lay.addWidget(self._title("④ 执行导入")) prog_row = QHBoxLayout() self.progress = QProgressBar() self.progress.setRange(0, 1) self.progress.setFormat("就绪") self.status_lbl = QLabel("") self.status_lbl.setStyleSheet(f"color: {FG_DIM}; font-size: 12px;") prog_row.addWidget(self.progress, 1) prog_row.addWidget(self.status_lbl) lay.addLayout(prog_row) # ── 按钮 ── bb = QDialogButtonBox(QDialogButtonBox.StandardButton.Close) self.btn_import = QPushButton("🚀 开始导入") self.btn_import.setMinimumHeight(34) self.btn_import.clicked.connect(self._run_import) bb.addButton(self.btn_import, QDialogButtonBox.ButtonRole.ActionRole) bb.clicked.connect(self._on_bb_clicked) lay.addWidget(bb) self._on_source_changed() # ------------------------------------------------------------------ def _title(self, text: str) -> QLabel: lbl = QLabel(text) lbl.setStyleSheet(f"color: {FG_DIM}; font-size: 12px; font-weight: 700;" f" margin-top: 4px;") return lbl # ------------------------------------------------------------------ def _on_source_changed(self): folder = self.r_folder.isChecked() self.path_edit.setPlaceholderText("选择文件夹…" if folder else "选择文件…") self.path_edit.clear() self.preview_list.clear() self.preview_hint.setText("选择路径后自动扫描") # 类型选择仅在文件夹模式下有意义 self.type_combo.setEnabled(folder) self.recurse_chk.setEnabled(folder) self.copy_chk.setEnabled(not self.r_json.isChecked()) def _pick_path(self): if self.r_folder.isChecked(): p = QFileDialog.getExistingDirectory(self, "选择文件夹", os.path.expanduser("~")) elif self.r_file.isChecked(): p = QFileDialog.getOpenFileName(self, "选择文件", os.path.expanduser("~"), _FILE_FILTERS)[0] else: p = QFileDialog.getOpenFileName(self, "选择备份文件", os.path.expanduser("~"), JSON_FILTER)[0] if p: self.path_edit.setText(p) self._scan() def _scan(self): p = self.path_edit.text().strip() if not p or not os.path.exists(p): return self.preview_list.clear() try: if self.r_folder.isChecked(): at = AssetType.from_string(self.type_combo.currentData()) if self.type_combo.currentData() else None files = lib.scan_files(p, recursive=self.recurse_chk.isChecked(), asset_type=at) if not files: self.preview_hint.setText("⚠️ 未找到可识别的文件") return self.preview_hint.setText(f"📦 将导入 {len(files)} 个文件") for fp, at2 in files[:200]: item = QListWidgetItem(f"{ASSET_TYPE_META[at2].icon} {fp.name} · {ASSET_TYPE_META[at2].label}") item.setData(Qt.ItemDataRole.UserRole, str(fp)) self.preview_list.addItem(item) if len(files) > 200: self.preview_list.addItem(QListWidgetItem(f"… 以及另外 {len(files) - 200} 个文件")) elif self.r_file.isChecked(): ext = Path(p).suffix.lower() at = lib.EXT_TO_TYPE.get(ext, AssetType.REFERENCE) self.preview_hint.setText(f"📄 1 个文件 · 类型: {ASSET_TYPE_META[at].label}") item = QListWidgetItem(f"{ASSET_TYPE_META[at].icon} {Path(p).name}") item.setData(Qt.ItemDataRole.UserRole, p) self.preview_list.addItem(item) else: import json with open(p, encoding="utf-8") as f: data = json.load(f) n = len(data.get("assets", [])) self.preview_hint.setText(f"🗂️ 备份含 {n} 项资产 · {len(data.get('series', []))} 系列") for s in data.get("series", [])[:20]: self.preview_list.addItem(QListWidgetItem(f"📚 {s.get('name', '?')}")) for a in data.get("assets", [])[:50]: at = AssetType.from_string(a.get("type")) self.preview_list.addItem(QListWidgetItem( f"{ASSET_TYPE_META[at].icon} [{ASSET_TYPE_META[at].label}] {a.get('title', '')}")) if n > 50: self.preview_list.addItem(QListWidgetItem(f"… 以及另外 {n - 50} 项")) except Exception as e: self.preview_hint.setText(f"⚠️ 扫描失败: {e}") # ------------------------------------------------------------------ def _quick_new_series(self): from .series_dialog import QuickSeriesDialog name, author, status = QuickSeriesDialog.ask("新建系列", self) if not name: return s = self.db.create_series(Series(name=name, author=author, status=status or "ongoing")) self._reload_series(s.id) def _reload_series(self, select_id=None): cur = self.series_combo.currentData() self.series_combo.clear() self.series_combo.addItem("新建（用文件夹名）", None) for s in self.db.list_series(): self.series_combo.addItem(s.name, s.id) idx = self.series_combo.findData(select_id or cur) self.series_combo.setCurrentIndex(idx if idx >= 0 else 0) # ------------------------------------------------------------------ def _run_import(self): p = self.path_edit.text().strip() if not p: QMessageBox.warning(self, "提示", "请先选择路径。") return if not os.path.exists(p): QMessageBox.warning(self, "提示", "路径不存在。") return self.btn_import.setEnabled(False) self.progress.setRange(0, 1) self.progress.setFormat("导入中…") self.status_lbl.setText("正在处理…") self.status_lbl.setStyleSheet(f"color: {FG_DIM}; font-size: 12px;") mode = "folder" if self.r_folder.isChecked() else ("file" if self.r_file.isChecked() else "json") sid = self.series_combo.currentData() at_val = self.type_combo.currentData() at = AssetType.from_string(at_val) if at_val else None recursive = self.recurse_chk.isChecked() copy_files = self.copy_chk.isChecked() base_dir = str(Path(p).parent) def task(): if mode == "folder": return lib.import_folder( self.db, p, series_id=sid, asset_type=at, recursive=recursive, copy_files=copy_files, ) if mode == "file": a = lib.import_file(self.db, p, series_id=sid, asset_type=at, copy_files=copy_files) return lib.ImportReport(imported=1 if a else 0, failed=[] if a else [str(p)]) return lib.import_library(self.db, p, base_dir=base_dir) self._worker = _ImportWorker(task) self._worker.progress.connect(self._on_progress) self._worker.finished.connect(self._on_done) QThreadPool.globalInstance().start(self._worker) def _on_progress(self, done, total, label): if total: self.progress.setRange(0, total) self.progress.setValue(done) if label: self.status_lbl.setText(label) def _on_done(self, result): self.btn_import.setEnabled(True) if isinstance(result, Exception): self.progress.setFormat("失败") self.status_lbl.setText(f"❌ {result}") self.status_lbl.setStyleSheet(f"color: {DANGER}; font-size: 12px;") QMessageBox.critical(self, "导入失败", str(result)) return self._report = result self.progress.setRange(0, 100) self.progress.setValue(100) ok = result.imported > 0 color = SUCCESS if ok else WARN self.progress.setFormat("完成") icon = "✅" if ok else "⚠️" msg = f"{icon} 已导入 {result.imported} 项 · 缩略图 {result.thumb_generated} 张" if result.failed: msg += f" · 失败 {len(result.failed)}" self.status_lbl.setText(msg) self.status_lbl.setStyleSheet(f"color: {color}; font-size: 12px;") self.accept() def _on_bb_clicked(self, btn): if btn is self.btn_import: return # 导入按钮自行处理 self.reject() def report(self): return self._report