[Python] 纯文本查看 复制代码
# -*- coding=utf-8 -*-
import json
import os
import shutil
import signal
import subprocess
import sys
import threading
import time
from collections import deque
from queue import Queue, Empty
from PyQt5.QtCore import Qt, QThread, pyqtSignal, QTimer
from PyQt5.QtWidgets import (
QApplication, QMainWindow, QWidget, QVBoxLayout, QHBoxLayout, QGridLayout,
QPushButton, QListWidget, QFileDialog, QGroupBox, QCheckBox, QLabel,
QProgressBar, QPlainTextEdit, QLineEdit, QMessageBox, QComboBox,
QSlider, QFormLayout,
)
# 路径配置
REALESRGAN_PATH = r"c:\realesrgan-ncnn-vulkan\realesrgan-ncnn-vulkan.exe"
FFMPEG_PATH = r"C:\ffmpeg\bin\ffmpeg.exe"
FFPROBE_PATH = r"C:\ffmpeg\bin\ffprobe.exe"
DEFAULT_OUTPUT_DIR = r"C:\Restored_Videos"
# 模型
MODELS = [
("realesr-animevideov3", "RealESRGAN AnimeVideoV3"),
("realesrgan-x4plus", "RealESRGAN x4+"),
("realesrgan-x4plus-anime", "RealESRGAN x4+ Anime"),
]
# 参数
DEFAULT_POST_PARAMS = {
"dn_spatial": 4, "dn_temporal": 3,
"deblock_mode": "strong", "deblock_block": 8,
"contrast": 1.08, "saturation": 1.15, "brightness": 0.02,
"sharp_luma": 1.5, "sharp_chroma": 1.0, "volume": 1.0,
}
CHUNK_SIZE = 512 # 分块大小
N_WORKERS = max(1, (os.cpu_count() or 4) // 4) # 并行进程数
def _resolve_tool(path):
if not path: return ""
for p in (path, path + ".exe"):
if os.path.exists(p): return p
return shutil.which(os.path.basename(path)) or path
FFMPEG_PATH = _resolve_tool(FFMPEG_PATH)
FFPROBE_PATH = _resolve_tool(FFPROBE_PATH)
REALESRGAN_PATH = _resolve_tool(REALESRGAN_PATH)
def startupinfo():
if os.name == "nt":
si = subprocess.STARTUPINFO()
si.dwFlags |= subprocess.STARTF_USESHOWWINDOW
si.wShowWindow = 0
return si
return None
class UtilityWorker(QThread):
log_signal = pyqtSignal(str)
finished_signal = pyqtSignal(bool, str)
def __init__(self, func, parent=None):
super().__init__(parent)
self.func = func
def run(self):
try:
msg = self.func() or "完成"
self.log_signal.emit(str(msg))
self.finished_signal.emit(True, str(msg))
except Exception as e:
self.log_signal.emit(f"❌ {e}")
self.finished_signal.emit(False, str(e))
class WorkerThread(QThread):
progress_signal = pyqtSignal(int, str)
log_signal = pyqtSignal(str)
finished_signal = pyqtSignal(bool, str)
def __init__(self, file_list, output_dir, scale, model, options, post_params=None):
super().__init__()
self.file_list = list(file_list)
self.output_dir = output_dir
self.scale = max(1, int(scale))
self.model = model or ""
self.options = options or {}
self.post_params = post_params or dict(DEFAULT_POST_PARAMS)
self._stop = threading.Event()
self._procs = []
self._lock = threading.Lock()
self._finished_emitted = False
def _emit_finished(self, ok, msg):
if self._finished_emitted: return
self._finished_emitted = True
self.finished_signal.emit(ok, msg)
def stop(self):
if self._stop.is_set(): return
self._stop.set()
self.log_signal.emit("⚠️ 正在终止所有子进程...")
self._kill_all(wait=True)
def _launch(self, cmd, capture=False):
kwargs = {}
if os.name == "nt":
si = startupinfo()
if si is not None: kwargs["startupinfo"] = si
else:
kwargs["preexec_fn"] = os.setsid
if capture:
proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
text=True, encoding="utf-8", errors="ignore", **kwargs)
else:
proc = subprocess.Popen(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, **kwargs)
with self._lock:
self._procs.append(proc)
return proc
def _untrack(self, proc):
with self._lock:
if proc in self._procs: self._procs.remove(proc)
def _kill_proc(self, proc, wait=True):
if proc is None or proc.poll() is not None: return
try:
if os.name == "nt":
subprocess.Popen(["taskkill", "/PID", str(proc.pid), "/T", "/F"],
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
else:
os.killpg(os.getpgid(proc.pid), signal.SIGTERM)
except Exception:
try:
proc.terminate()
except Exception:
pass
if not wait: return
try:
proc.wait(timeout=2)
except subprocess.TimeoutExpired:
try:
proc.kill()
proc.wait(timeout=2)
except Exception:
pass
def _kill_all(self, wait=True):
with self._lock:
procs = list(self._procs)
for p in procs: self._kill_proc(p, wait=False)
if wait:
for p in procs:
try:
p.wait(timeout=3)
except subprocess.TimeoutExpired:
self._kill_proc(p, wait=True)
with self._lock:
self._procs.clear()
@staticmethod
def _list_png(directory):
if not os.path.isdir(directory): return []
try:
return sorted(f for f in os.listdir(directory) if f.lower().endswith(".png"))
except OSError:
return []
@staticmethod
def _count_png(directory):
if not os.path.isdir(directory): return 0
try:
return sum(1 for e in os.scandir(directory) if e.is_file() and e.name.lower().endswith(".png"))
except OSError:
return 0
def get_video_info(self, path):
fps, w, h, nf = 30.0, 0, 0, 0
try:
out = subprocess.check_output(
[FFPROBE_PATH, "-v", "error", "-select_streams", "v:0",
"-show_entries", "stream=r_frame_rate,width,height,nb_frames",
"-show_entries", "format=duration", "-of", "json", path],
stderr=subprocess.DEVNULL, text=True, encoding="utf-8", errors="ignore", timeout=30)
d = json.loads(out)
st = (d.get("streams") or [{}])[0]
fr = str(st.get("r_frame_rate") or st.get("avg_frame_rate") or "30/1")
if "/" in fr:
n, dn = fr.split("/")
if float(dn) != 0: fps = round(float(n) / float(dn), 3)
else:
try:
fps = float(fr)
except Exception:
pass
if fps <= 0: fps = 30.0
w = int(st.get("width") or 0)
h = int(st.get("height") or 0)
nb = st.get("nb_frames")
if nb and nb != "N/A":
nf = int(nb)
else:
dur = float(d.get("format", {}).get("duration", 0) or 0)
nf = int(dur * fps) if dur > 0 else 0
except Exception as e:
self.log_signal.emit(f"⚠️ 获取视频信息失败: {e}")
return fps, w, h, nf
def _has_audio(self, path):
try:
out = subprocess.check_output(
[FFPROBE_PATH, "-v", "error", "-select_streams", "a",
"-show_entries", "stream=index", "-of", "csv=p=0", path],
stderr=subprocess.DEVNULL, text=True, encoding="utf-8", errors="ignore", timeout=30)
return bool(out.strip())
except Exception:
return False
def _build_pre_filters(self):
p = self.post_params
vf_parts = []
if self.options.get("denoise"):
s, t = float(p.get("dn_spatial", 4)), float(p.get("dn_temporal", 3))
vf_parts.append(f"hqdn3d={s:g}:{s:g}:{t:g}:{t:g}")
if self.options.get("artifact"):
mode = str(p.get("deblock_mode", "strong")).lower()
if mode not in ("weak", "medium", "strong"): mode = "strong"
block = min((4, 8, 16), key=lambda x: abs(x - int(p.get("deblock_block", 8))))
vf_parts.append(f"deblock=filter={mode}:block={block}")
return ",".join(vf_parts) if vf_parts else ""
def _build_post_filters(self):
p = self.post_params
vf_parts = []
if self.options.get("color"):
c = float(p.get("contrast", 1.08))
s = float(p.get("saturation", 1.15))
b = float(p.get("brightness", 0.02))
vf_parts.append(f"eq=contrast={c:.2f}:saturation={s:.2f}:brightness={b:.2f}")
if self.options.get("sharpen"):
sl = float(p.get("sharp_luma", 1.5))
sc = float(p.get("sharp_chroma", 1.0))
vf_parts.append(f"unsharp=5:5:{sl:.1f}:5:5:{sc:.1f}")
return ",".join(vf_parts) if vf_parts else ""
def _run_cmd(self, cmd, silent=False, out_dir=None, total=0, label=""):
proc = self._launch(cmd, capture=True)
q = Queue(maxsize=2048)
def reader():
try:
for line in proc.stdout: q.put(line)
except Exception:
pass
finally:
q.put(None)
reader_thread = threading.Thread(target=reader, daemon=True)
reader_thread.start()
lines = deque(maxlen=100)
last_progress_time = 0.0
try:
while True:
if self._stop.is_set(): raise InterruptedError("用户停止")
if out_dir and total > 0 and time.time() - last_progress_time >= 0.5:
cnt = self._count_png(out_dir)
pct = min(99, int(cnt / max(total, 1) * 100))
self.progress_signal.emit(pct, f"{label}: {cnt}/{total} ({pct}%)")
last_progress_time = time.time()
try:
line = q.get(timeout=0.2)
except Empty:
if proc.poll() is not None: break
continue
if line is None: break
txt = line.strip()
if txt and not silent: lines.append(txt)
reader_thread.join(timeout=2)
rc = proc.wait()
if self._stop.is_set(): raise InterruptedError("用户停止")
return rc, "\n".join(lines)
except InterruptedError:
self._kill_proc(proc, wait=True);
raise
except Exception:
self._kill_proc(proc, wait=True);
raise
finally:
self._untrack(proc)
# 分块
def _split_into_chunks(self, src_dir):
files = self._list_png(src_dir)
total = len(files)
if total == 0:
return []
chunks_root = os.path.join(os.path.dirname(src_dir), "chunks")
os.makedirs(chunks_root, exist_ok=True)
chunk_dirs = []
for i in range(0, total, CHUNK_SIZE):
idx = i // CHUNK_SIZE
dst = os.path.join(chunks_root, f"chunk_{idx:04d}")
os.makedirs(dst, exist_ok=True)
for f in files[i:min(i + CHUNK_SIZE, total)]:
src_path = os.path.join(src_dir, f)
dst_path = os.path.join(dst, f)
try:
os.rename(src_path, dst_path)
except OSError:
shutil.move(src_path, dst_path)
chunk_dirs.append(dst)
return chunk_dirs
# AI增强
def _run_ai_chunks(self, work_dir, chunk_dirs, ai_seq_dir, out_w, out_h, tile):
total_frames = sum(self._count_png(d) for d in chunk_dirs)
if total_frames == 0:
raise RuntimeError("AI 输入帧为空")
self.log_signal.emit(
f"🚀 AI修复: {len(chunk_dirs)}块 | {total_frames}帧 | "
f"模型:{self.model} | Tile={tile}"
)
max_parallel = min(N_WORKERS, len(chunk_dirs))
running = []
next_idx = 0
t0 = time.time()
last_progress_time = 0.0
PROGRESS_INTERVAL = 0.5
while next_idx < len(chunk_dirs) or running:
while len(running) < max_parallel and next_idx < len(chunk_dirs):
cin = chunk_dirs[next_idx]
cmd = [
REALESRGAN_PATH, "-i", cin, "-o", ai_seq_dir,
"-n", self.model, "-t", str(tile),
"-j", "1:2:2", "-f", "png", "-g", "auto"
]
if self.scale > 1:
cmd += ["-s", str(self.scale)]
proc = self._launch(cmd, capture=False)
running.append((next_idx, proc, cin))
self.log_signal.emit(
f" ▶ Chunk {next_idx + 1}/{len(chunk_dirs)} PID:{proc.pid}"
)
next_idx += 1
time.sleep(0.2 if running else 0.05)
still = []
fail_msg = None
for idx, proc, chunk_path in running:
if proc.poll() is not None:
rc = proc.returncode
self._untrack(proc)
if rc != 0:
fail_msg = f"Chunk {idx + 1} 失败(退出码{rc})"
break
try:
shutil.rmtree(chunk_path, ignore_errors=True)
except Exception as e:
self.log_signal.emit(f" ⚠️ 清理chunk {idx + 1}失败: {e}")
self.log_signal.emit(
f" ✅ Chunk {idx + 1}/{len(chunk_dirs)} 完成并已清理"
)
else:
still.append((idx, proc, chunk_path))
if fail_msg:
raise RuntimeError(fail_msg)
running = still
now = time.time()
if now - last_progress_time >= PROGRESS_INTERVAL or not running:
done = self._count_png(ai_seq_dir)
done_safe = min(done, total_frames)
pct = min(99, int(done_safe / max(total_frames, 1) * 100))
elapsed = max(now - t0, 0.01)
speed = done_safe / elapsed
remaining = total_frames - done_safe
eta = remaining / max(speed, 0.001) if speed > 0 else 0
self.progress_signal.emit(
pct,
f"AI增强: {done_safe}/{total_frames}帧 ({pct}%) | "
f"{speed:.1f}fps | ETA {int(eta)}s"
)
last_progress_time = now
cost = time.time() - t0
final_done = min(self._count_png(ai_seq_dir), total_frames)
self.log_signal.emit(
f"✅ AI完成: {final_done}/{total_frames}帧 | {cost:.1f}s | "
f"{final_done / max(cost, 0.01):.1f}fps"
)
def run(self):
total_videos = len(self.file_list)
need_ai = bool(self.options.get("ai"))
need_pre = bool(self.options.get("denoise") or self.options.get("artifact"))
ai_scale = self.scale if (need_ai and self.scale > 1) else (2 if need_ai else 1)
eff_scale = ai_scale
steps = ["拆帧" + ("+预处理" if need_pre else "")]
if need_ai:
steps.append("AI 修复")
steps.append("合成与增强")
try:
os.makedirs(self.output_dir, exist_ok=True)
except Exception as e:
self._emit_finished(False, f"输出目录不可写: {e}")
return
for vid_idx, path in enumerate(self.file_list):
if self._stop.is_set():
break
base = os.path.basename(path)
self.log_signal.emit(f"🎬 [{vid_idx + 1}/{total_videos}] {base}")
if not os.path.exists(path):
self._emit_finished(False, f"{base} 不存在")
return
work_dir = os.path.join(
self.output_dir,
f"_work_{vid_idx}_{os.getpid()}_{int(time.time())}"
)
try:
fps, w, h, total_frames = self.get_video_info(path)
if w <= 0 or h <= 0:
raise RuntimeError("无法获取视频分辨率")
total_frames = int(max(0, total_frames))
out_w, out_h = w * eff_scale, h * eff_scale
# Step 1: 拆帧
step_idx = 1
self.progress_signal.emit(0, f"Step {step_idx}/{len(steps)}: {steps[0]}...")
safe_dir = os.path.join(work_dir, "safe")
os.makedirs(safe_dir, exist_ok=True)
cmd_extract = [FFMPEG_PATH, "-y", "-i", path]
pre_vf = self._build_pre_filters()
if pre_vf:
cmd_extract += ["-vf", pre_vf]
self.log_signal.emit(f"🎨 预处理滤镜: {pre_vf}")
cmd_extract += [
"-pix_fmt", "rgb24", "-compression_level", "1",
os.path.join(safe_dir, "frame_%08d.png")
]
ret, _ = self._run_cmd(
cmd_extract, out_dir=safe_dir,
total=total_frames, label="拆帧"
)
if ret != 0:
raise RuntimeError("拆帧失败")
actual_frames = self._count_png(safe_dir)
if actual_frames == 0:
raise RuntimeError("未提取到帧")
self.log_signal.emit(f"✅ 拆帧完成: {actual_frames} 帧")
chunk_dirs = self._split_into_chunks(safe_dir)
self.log_signal.emit(f"🔀 已切分为 {len(chunk_dirs)} 个块")
shutil.rmtree(safe_dir, ignore_errors=True)
# Step 2: AI增强
ai_seq_dir = os.path.join(work_dir, "ai_seq")
os.makedirs(ai_seq_dir, exist_ok=True)
if need_ai:
step_idx += 1
self.progress_signal.emit(
0, f"Step {step_idx}/{len(steps)}: AI 修复..."
)
self._run_ai_chunks(
work_dir, chunk_dirs, ai_seq_dir, out_w, out_h, tile=0
)
shutil.rmtree(
os.path.join(work_dir, "chunks"), ignore_errors=True
)
else:
step_idx += 1
self.progress_signal.emit(
0, f"Step {step_idx}/{len(steps)}: 整理帧序列..."
)
for cd in chunk_dirs:
for f in self._list_png(cd):
src = os.path.join(cd, f)
dst = os.path.join(ai_seq_dir, f)
try:
os.rename(src, dst)
except OSError:
shutil.move(src, dst)
shutil.rmtree(cd, ignore_errors=True)
shutil.rmtree(
os.path.join(work_dir, "chunks"), ignore_errors=True
)
self.log_signal.emit(
f"✅ 帧序列就绪: {self._count_png(ai_seq_dir)} 帧"
)
# Step 3: 合成与增强
step_idx += 1
self.progress_signal.emit(
0, f"Step {step_idx}/{len(steps)}: 合成与增强..."
)
out_mp4 = os.path.join(
self.output_dir,
os.path.splitext(base)[0] + "_restored.mp4"
)
has_audio = self._has_audio(path)
cmd_merge = [
FFMPEG_PATH, "-y", "-framerate", str(fps),
"-i", os.path.join(ai_seq_dir, "frame_%08d.png")
]
if has_audio:
cmd_merge += ["-i", path]
vol = max(
0.0, min(3.0, float(self.post_params.get("volume", 1.0)))
)
if abs(vol - 1.0) > 0.01:
cmd_merge += [
"-filter_complex",
f"[1:a]volume={vol:.2f}[aout]",
"-map", "0:v:0", "-map", "[aout]"
]
else:
cmd_merge += [
"-map", "0:v:0", "-map", "1:a:0", "-c:a", "copy"
]
else:
cmd_merge += ["-map", "0:v:0"]
post_vf = self._build_post_filters()
if post_vf:
self.log_signal.emit(f"🎨 合成期滤镜: {post_vf}")
cmd_merge += ["-vf", post_vf]
cmd_merge += [
"-c:v", "libx264", "-crf", "20", "-preset", "fast",
"-pix_fmt", "yuv420p", "-movflags", "+faststart"
]
if has_audio:
cmd_merge += ["-c:a", "aac", "-b:a", "192k"]
cmd_merge += [out_mp4]
ret, merge_output = self._run_cmd(cmd_merge)
if ret != 0:
raise RuntimeError(
f"视频合并失败:\n{merge_output[-500:]}"
)
if not os.path.exists(out_mp4):
raise RuntimeError("输出文件不存在")
sz = os.path.getsize(out_mp4) / (1024 * 1024)
if sz < 0.01:
raise RuntimeError(f"输出文件异常小({sz:.2f}MB)")
self.log_signal.emit(
f"🎬 完成: {os.path.basename(out_mp4)} ({sz:.1f}MB)"
)
self.progress_signal.emit(
int((vid_idx + 1) / max(total_videos, 1) * 100),
f"已完成: {base}"
)
except InterruptedError:
self._emit_finished(False, "用户停止")
return
except Exception as e:
self.log_signal.emit(f"❌ {base} 失败: {e}")
self._emit_finished(False, f"{base} 失败")
return
finally:
self._kill_all(wait=True)
if os.path.exists(work_dir):
shutil.rmtree(work_dir, ignore_errors=True)
if self._stop.is_set():
self._emit_finished(False, "用户停止")
else:
self.progress_signal.emit(100, "全部完成")
self._emit_finished(True, f"✅ {total_videos}个视频处理完成!")
# ==================== GUI ====================
class VideoRepairTool(QMainWindow):
def __init__(self):
super().__init__()
self.worker = None
self.util_worker = None
self.output_dir = DEFAULT_OUTPUT_DIR
self.init_ui()
self.check_tools_startup()
def init_ui(self):
self.setWindowTitle("视频增强工具")
self.resize(860, 700)
central = QWidget()
self.setCentralWidget(central)
root = QVBoxLayout(central)
root.setContentsMargins(14, 14, 14, 14)
root.setSpacing(10)
title = QLabel("🎬 视频增强工具")
title.setAlignment(Qt.AlignCenter)
title.setStyleSheet("font-size:26px; font-weight:bold; color:#e65100;")
root.addWidget(title)
content = QGridLayout()
content.setColumnStretch(0, 5)
content.setColumnStretch(1, 4)
root.addLayout(content, 1)
left, right = QVBoxLayout(), QVBoxLayout()
# 文件列表
file_group = QGroupBox("待处理视频列表")
fl = QVBoxLayout(file_group)
btn_row = QHBoxLayout()
self.btn_choose_file = QPushButton("🔍 选择文件")
self.btn_clear_list = QPushButton("🗑 清空列表")
self.btn_choose_file.setStyleSheet("background:#43a047;color:white;border:none;padding:8px;border-radius:4px;")
self.btn_clear_list.setStyleSheet("background:#fb8c00;color:white;border:none;padding:8px;border-radius:4px;")
self.btn_choose_file.clicked.connect(self.add_files)
self.btn_clear_list.clicked.connect(self.clear_list)
btn_row.addWidget(self.btn_choose_file)
btn_row.addWidget(self.btn_clear_list)
fl.addLayout(btn_row)
fl.addWidget(QLabel("待处理文件:"))
self.list_files = QListWidget()
fl.addWidget(self.list_files, 1)
out_row = QHBoxLayout()
out_row.addWidget(QLabel("输出文件夹:"))
self.txt_out = QLineEdit(self.output_dir)
self.btn_browse = QPushButton("浏览")
self.btn_browse.setStyleSheet(
"background:#4a90d9;color:white;border:none;padding:10px;font-weight:bold;border-radius:4px;")
self.btn_browse.clicked.connect(self.choose_output_dir)
out_row.addWidget(self.txt_out, 1)
out_row.addWidget(self.btn_browse)
fl.addLayout(out_row)
left.addWidget(file_group, 3)
# 功能选项
func_group = QGroupBox("功能选项")
fg = QGridLayout(func_group)
self.chk_denoise = QCheckBox("去噪")
self.chk_artifact = QCheckBox("去伪影")
self.chk_color = QCheckBox("色彩增强")
self.chk_sharpen = QCheckBox("锐化")
self.chk_volume = QCheckBox("音频")
self.chk_ai = QCheckBox("AI增强")
self.chk_superres = QCheckBox("超分辨率")
fg.addWidget(self.chk_denoise, 0, 0)
fg.addWidget(self.chk_artifact, 1, 0)
fg.addWidget(self.chk_color, 2, 0)
fg.addWidget(self.chk_sharpen, 3, 0)
fg.addWidget(self.chk_volume, 4, 0)
fg.addWidget(self.chk_ai, 0, 1)
fg.addWidget(self.chk_superres, 1, 1)
scale_row = QHBoxLayout()
scale_row.addWidget(QLabel("超分倍数:"))
self.cmb_scale = QComboBox()
self.cmb_scale.addItems(["2", "3", "4"])
self.cmb_scale.setCurrentIndex(0)
self.cmb_scale.setFixedWidth(80)
self.lbl_scale_hint = QLabel("(AI增强后可用)")
self.lbl_scale_hint.setStyleSheet("color:#888; font-size:11px;")
scale_row.addWidget(self.cmb_scale)
scale_row.addWidget(self.lbl_scale_hint, 1)
fg.addLayout(scale_row, 2, 1)
model_row = QHBoxLayout()
model_row.addWidget(QLabel("AI模型:"))
self.cmb_model = QComboBox()
for mid, mlabel in MODELS:
self.cmb_model.addItem(mlabel, mid)
self.cmb_model.setCurrentIndex(0)
model_row.addWidget(self.cmb_model, 1)
fg.addLayout(model_row, 3, 1)
for chk in [self.chk_ai, self.chk_superres, self.chk_denoise,
self.chk_artifact, self.chk_color, self.chk_sharpen, self.chk_volume]:
chk.toggled.connect(self._update_controls)
left.addWidget(func_group, 2)
# 参数调节
grp_params = QGroupBox("参数调节")
params_layout = QFormLayout()
params_layout.setSpacing(6)
dn_row = QHBoxLayout()
self.sld_dn_spatial = QSlider(Qt.Horizontal);
self.sld_dn_spatial.setRange(0, 10)
self.sld_dn_spatial.setValue(DEFAULT_POST_PARAMS["dn_spatial"])
self.lbl_dn_spatial = QLabel(str(DEFAULT_POST_PARAMS["dn_spatial"]))
self.sld_dn_spatial.valueChanged.connect(lambda v: self.lbl_dn_spatial.setText(str(v)))
self.sld_dn_temporal = QSlider(Qt.Horizontal);
self.sld_dn_temporal.setRange(0, 8)
self.sld_dn_temporal.setValue(DEFAULT_POST_PARAMS["dn_temporal"])
self.lbl_dn_temporal = QLabel(str(DEFAULT_POST_PARAMS["dn_temporal"]))
self.sld_dn_temporal.valueChanged.connect(lambda v: self.lbl_dn_temporal.setText(str(v)))
dn_row.addWidget(QLabel("空间:"));
dn_row.addWidget(self.sld_dn_spatial, 1);
dn_row.addWidget(self.lbl_dn_spatial)
dn_row.addSpacing(16)
dn_row.addWidget(QLabel("时间:"));
dn_row.addWidget(self.sld_dn_temporal, 1);
dn_row.addWidget(self.lbl_dn_temporal)
params_layout.addRow("去噪强度:", dn_row)
db_row = QHBoxLayout()
self.cmb_deblock_mode = QComboBox()
self.cmb_deblock_mode.addItems(["weak", "strong"])
self.cmb_deblock_mode.setCurrentText(DEFAULT_POST_PARAMS["deblock_mode"])
self.sld_deblock_block = QSlider(Qt.Horizontal);
self.sld_deblock_block.setRange(4, 16)
self.sld_deblock_block.setValue(DEFAULT_POST_PARAMS["deblock_block"])
self.lbl_deblock_block = QLabel(str(DEFAULT_POST_PARAMS["deblock_block"]))
self.sld_deblock_block.valueChanged.connect(lambda v: self.lbl_deblock_block.setText(str(v)))
db_row.addWidget(QLabel("模式:"));
db_row.addWidget(self.cmb_deblock_mode)
db_row.addSpacing(16)
db_row.addWidget(QLabel("块大小:"));
db_row.addWidget(self.sld_deblock_block, 1);
db_row.addWidget(self.lbl_deblock_block)
params_layout.addRow("去伪影:", db_row)
col_row = QHBoxLayout()
self.sld_contrast = QSlider(Qt.Horizontal);
self.sld_contrast.setRange(50, 200)
self.sld_contrast.setValue(int(DEFAULT_POST_PARAMS["contrast"] * 100))
self.lbl_contrast = QLabel(f"{DEFAULT_POST_PARAMS['contrast']:.2f}")
self.sld_contrast.valueChanged.connect(lambda v: self.lbl_contrast.setText(f"{v / 100:.2f}"))
self.sld_saturation = QSlider(Qt.Horizontal);
self.sld_saturation.setRange(50, 250)
self.sld_saturation.setValue(int(DEFAULT_POST_PARAMS["saturation"] * 100))
self.lbl_saturation = QLabel(f"{DEFAULT_POST_PARAMS['saturation']:.2f}")
self.sld_saturation.valueChanged.connect(lambda v: self.lbl_saturation.setText(f"{v / 100:.2f}"))
self.sld_brightness = QSlider(Qt.Horizontal);
self.sld_brightness.setRange(-50, 50)
self.sld_brightness.setValue(int(DEFAULT_POST_PARAMS["brightness"] * 100))
self.lbl_brightness = QLabel(f"{DEFAULT_POST_PARAMS['brightness']:.2f}")
self.sld_brightness.valueChanged.connect(lambda v: self.lbl_brightness.setText(f"{v / 100:.2f}"))
col_row.addWidget(QLabel("对比:"));
col_row.addWidget(self.sld_contrast, 1);
col_row.addWidget(self.lbl_contrast)
col_row.addSpacing(8)
col_row.addWidget(QLabel("饱和:"));
col_row.addWidget(self.sld_saturation, 1);
col_row.addWidget(self.lbl_saturation)
col_row.addSpacing(8)
col_row.addWidget(QLabel("亮度:"));
col_row.addWidget(self.sld_brightness, 1);
col_row.addWidget(self.lbl_brightness)
params_layout.addRow("色彩:", col_row)
sh_row = QHBoxLayout()
self.sld_sharp_luma = QSlider(Qt.Horizontal);
self.sld_sharp_luma.setRange(0, 30)
self.sld_sharp_luma.setValue(int(DEFAULT_POST_PARAMS["sharp_luma"] * 10))
self.lbl_sharp_luma = QLabel(f"{DEFAULT_POST_PARAMS['sharp_luma']:.1f}")
self.sld_sharp_luma.valueChanged.connect(lambda v: self.lbl_sharp_luma.setText(f"{v / 10:.1f}"))
self.sld_sharp_chroma = QSlider(Qt.Horizontal);
self.sld_sharp_chroma.setRange(0, 30)
self.sld_sharp_chroma.setValue(int(DEFAULT_POST_PARAMS["sharp_chroma"] * 10))
self.lbl_sharp_chroma = QLabel(f"{DEFAULT_POST_PARAMS['sharp_chroma']:.1f}")
self.sld_sharp_chroma.valueChanged.connect(lambda v: self.lbl_sharp_chroma.setText(f"{v / 10:.1f}"))
sh_row.addWidget(QLabel("亮度:"))
sh_row.addWidget(self.sld_sharp_luma, 1)
sh_row.addWidget(self.lbl_sharp_luma)
sh_row.addSpacing(16)
sh_row.addWidget(QLabel("色度:"));
sh_row.addWidget(self.sld_sharp_chroma, 1)
sh_row.addWidget(self.lbl_sharp_chroma)
params_layout.addRow("锐化:", sh_row)
vol_row = QHBoxLayout()
self.sld_volume = QSlider(Qt.Horizontal);
self.sld_volume.setRange(0, 30)
self.sld_volume.setValue(int(DEFAULT_POST_PARAMS["volume"] * 10))
self.lbl_volume = QLabel(f"{DEFAULT_POST_PARAMS['volume']:.1f}")
self.sld_volume.valueChanged.connect(lambda v: self.lbl_volume.setText(f"{v / 10:.1f}"))
vol_row.addWidget(QLabel("音量:"))
vol_row.addWidget(self.sld_volume, 1)
vol_row.addWidget(self.lbl_volume)
vol_row.addStretch()
params_layout.addRow("音频:", vol_row)
self.btn_reset_params = QPushButton("↩ 恢复缺省值")
self.btn_reset_params.setStyleSheet(
"background:#78909c;color:white;border:none;padding:6px;border-radius:4px;font-size:12px;")
self.btn_reset_params.clicked.connect(self._reset_post_params)
params_layout.addRow("", self.btn_reset_params)
grp_params.setLayout(params_layout)
left.addWidget(grp_params, 2)
ctrl_row = QHBoxLayout()
self.btn_start = QPushButton("🚀 开始处理")
self.btn_stop = QPushButton("⏹ 停止")
self.btn_start.setStyleSheet(
"QPushButton{background:#43a047;color:white;border:none;padding:10px;font-weight:bold;border-radius:4px;}QPushButton:disabled{background:#bdbdbd;color:#757575;}")
self.btn_stop.setStyleSheet(
"QPushButton{background:#e53935;color:white;border:none;padding:10px;font-weight:bold;border-radius:4px;}QPushButton:disabled{background:#bdbdbd;color:#757575;}")
self.btn_start.clicked.connect(self.start_process)
self.btn_stop.clicked.connect(self.stop_process)
self.btn_stop.setEnabled(False)
ctrl_row.addWidget(self.btn_start, 1)
ctrl_row.addWidget(self.btn_stop, 1)
left.addLayout(ctrl_row)
hint = QLabel(f"💡 CPU {N_WORKERS} 进程并行 | GPU自动选择 | 关闭窗口自动结束子进程")
hint.setStyleSheet("color:#e65100; font-size:12px; font-weight:bold;")
hint.setAlignment(Qt.AlignCenter)
left.addWidget(hint)
content.addLayout(left, 0, 0)
# 右侧
progress_group = QGroupBox("处理进度")
pl = QVBoxLayout(progress_group)
self.lbl_status = QLabel("状态: 就绪")
self.lbl_status.setStyleSheet("font-weight:bold; color:#2979ff; font-size:10px;")
pl.addWidget(self.lbl_status)
self.progress_bar = QProgressBar()
self.progress_bar.setRange(0, 100)
pl.addWidget(self.progress_bar)
right.addWidget(progress_group, 1)
util_group = QGroupBox("辅助功能")
ug = QGridLayout(util_group)
self.btn_open_out = QPushButton("打开输出文件夹")
self.btn_check_ffmpeg = QPushButton("检查 FFmpeg")
self.btn_open_out.clicked.connect(self.open_output_dir)
self.btn_check_ffmpeg.clicked.connect(lambda: self.run_utility(self.check_ffmpeg_status, "检查FFmpeg"))
ug.addWidget(self.btn_open_out, 0, 0)
ug.addWidget(self.btn_check_ffmpeg, 0, 1)
right.addWidget(util_group, 1)
log_group = QGroupBox("处理日志")
ll = QVBoxLayout(log_group)
self.log_text = QPlainTextEdit()
self.log_text.setReadOnly(True)
ll.addWidget(self.log_text, 1)
btn_clear_log = QPushButton("清空日志")
btn_clear_log.clicked.connect(lambda: self.log_text.clear())
ll.addWidget(btn_clear_log)
right.addWidget(log_group, 5)
content.addLayout(right, 0, 1)
self._freezable_widgets = [
self.chk_denoise, self.chk_artifact, self.chk_color, self.chk_sharpen, self.chk_volume,
self.chk_ai, self.chk_superres, self.cmb_scale, self.cmb_model,
self.sld_dn_spatial, self.lbl_dn_spatial, self.sld_dn_temporal, self.lbl_dn_temporal,
self.cmb_deblock_mode, self.sld_deblock_block, self.lbl_deblock_block,
self.sld_contrast, self.lbl_contrast, self.sld_saturation, self.lbl_saturation,
self.sld_brightness, self.lbl_brightness,
self.sld_sharp_luma, self.lbl_sharp_luma, self.sld_sharp_chroma, self.lbl_sharp_chroma,
self.list_files, self.txt_out, self.btn_choose_file, self.btn_clear_list, self.btn_browse,
self.sld_volume, self.lbl_volume, self.btn_reset_params,
]
self._update_controls()
self.setStyleSheet("""
QGroupBox{font-weight:bold; border:1px solid #e0e0e0; border-radius:6px; margin-top:12px; padding-top:8px;}
QGroupBox::title{subcontrol-origin:margin; left:10px; padding:0 4px; color:#2979ff;}
QListWidget, QPlainTextEdit, QLineEdit, QComboBox{border:1px solid #ccc; border-radius:4px; padding:4px;}
""")
def _update_controls(self):
def _se(name, enabled):
w = getattr(self, name, None)
if w is not None: w.setEnabled(enabled)
ai_on = self.chk_ai.isChecked()
sr_on = self.chk_superres.isChecked()
if not ai_on and sr_on:
self.chk_superres.blockSignals(True)
self.chk_superres.setChecked(False)
self.chk_superres.blockSignals(False)
_se("chk_superres", ai_on)
_se("cmb_scale", ai_on and self.chk_superres.isChecked())
_se("cmb_model", ai_on)
_se("sld_dn_spatial", self.chk_denoise.isChecked())
_se("sld_dn_temporal", self.chk_denoise.isChecked())
_se("lbl_dn_spatial", self.chk_denoise.isChecked())
_se("lbl_dn_temporal", self.chk_denoise.isChecked())
_se("cmb_deblock_mode", self.chk_artifact.isChecked())
_se("sld_deblock_block", self.chk_artifact.isChecked())
_se("lbl_deblock_block", self.chk_artifact.isChecked())
_se("sld_contrast", self.chk_color.isChecked())
_se("sld_saturation", self.chk_color.isChecked())
_se("sld_volume", self.chk_volume.isChecked())
_se("sld_brightness", self.chk_color.isChecked())
_se("lbl_contrast", self.chk_color.isChecked())
_se("lbl_saturation", self.chk_color.isChecked())
_se("lbl_brightness", self.chk_color.isChecked())
_se("sld_sharp_luma", self.chk_sharpen.isChecked())
_se("sld_sharp_chroma", self.chk_sharpen.isChecked())
_se("lbl_sharp_luma", self.chk_sharpen.isChecked())
_se("lbl_sharp_chroma", self.chk_sharpen.isChecked())
_se("lbl_volume", self.chk_volume.isChecked())
def _set_controls_enabled(self, enabled):
for w in self._freezable_widgets: w.setEnabled(enabled)
if enabled: self._update_controls()
def _get_post_params(self):
return {
"dn_spatial": self.sld_dn_spatial.value(), "dn_temporal": self.sld_dn_temporal.value(),
"deblock_mode": self.cmb_deblock_mode.currentText(), "deblock_block": self.sld_deblock_block.value(),
"contrast": self.sld_contrast.value() / 100.0, "saturation": self.sld_saturation.value() / 100.0,
"brightness": self.sld_brightness.value() / 100.0,
"sharp_luma": self.sld_sharp_luma.value() / 10.0, "sharp_chroma": self.sld_sharp_chroma.value() / 10.0,
"volume": self.sld_volume.value() / 10.0,
}
def _reset_post_params(self):
d = DEFAULT_POST_PARAMS
self.sld_dn_spatial.setValue(d["dn_spatial"]);
self.sld_dn_temporal.setValue(d["dn_temporal"])
self.cmb_deblock_mode.setCurrentText(d["deblock_mode"]);
self.sld_deblock_block.setValue(d["deblock_block"])
self.sld_contrast.setValue(int(d["contrast"] * 100));
self.sld_saturation.setValue(int(d["saturation"] * 100))
self.sld_brightness.setValue(int(d["brightness"] * 100))
self.sld_sharp_luma.setValue(int(d["sharp_luma"] * 10));
self.sld_sharp_chroma.setValue(int(d["sharp_chroma"] * 10))
self.sld_volume.setValue(int(d["volume"] * 10))
self.append_log("↩ 参数已恢复缺省值")
def check_tools_startup(self):
for p, name in [(REALESRGAN_PATH, "Real‑ESRGAN"), (FFMPEG_PATH, "FFmpeg"), (FFPROBE_PATH, "FFprobe")]:
if not p or not os.path.exists(p): self.append_log(f"⚠️ 未找到 {name}: {p}")
def add_files(self):
files, _ = QFileDialog.getOpenFileNames(self, "选择视频", "",
"视频文件 (*.mp4 *.avi *.mkv *.mov *.flv *.wmv *.ts)")
for f in files:
if f not in [self.list_files.item(i).text() for i in range(self.list_files.count())]:
self.list_files.addItem(f)
def clear_list(self):
self.list_files.clear()
def choose_output_dir(self):
d = QFileDialog.getExistingDirectory(self, "选择输出目录", self.output_dir)
if d: self.output_dir = d; self.txt_out.setText(d)
def open_output_dir(self):
d = self.txt_out.text().strip() or DEFAULT_OUTPUT_DIR
os.makedirs(d, exist_ok=True)
try:
if os.name == "nt":
os.startfile(d)
elif sys.platform == "darwin":
subprocess.Popen(["open", d])
else:
subprocess.Popen(["xdg-open", d])
except Exception as e:
QMessageBox.warning(self, "提示", f"无法打开: {e}")
def start_process(self):
if self.worker and self.worker.isRunning():
QMessageBox.warning(self, "警告", "上一个任务尚未结束");
return
files = [self.list_files.item(i).text() for i in range(self.list_files.count())]
if not files: QMessageBox.warning(self, "提示", "请先添加至少 1 个视频文件！"); return
for p, name in [(FFMPEG_PATH, "FFmpeg"), (FFPROBE_PATH, "FFprobe")]:
if not p or not os.path.exists(p): QMessageBox.warning(self, "警告", f"未找到 {name}: {p}"); return
use_ai = self.chk_ai.isChecked()
if use_ai and (not REALESRGAN_PATH or not os.path.exists(REALESRGAN_PATH)):
QMessageBox.warning(self, "警告", f"未找到 Real‑ESRGAN: {REALESRGAN_PATH}");
return
if not any([use_ai, self.chk_denoise.isChecked(), self.chk_artifact.isChecked(),
self.chk_color.isChecked(), self.chk_sharpen.isChecked()]):
QMessageBox.warning(self, "提示", "请至少选择一个处理功能！");
return
self.output_dir = self.txt_out.text().strip() or DEFAULT_OUTPUT_DIR
os.makedirs(self.output_dir, exist_ok=True)
use_sr = self.chk_superres.isChecked() and use_ai
scale = int(self.cmb_scale.currentText()) if use_sr else 1
model = self.cmb_model.currentData() if use_ai else ""
options = {"ai": use_ai, "denoise": self.chk_denoise.isChecked(),
"artifact": self.chk_artifact.isChecked(), "color": self.chk_color.isChecked(),
"sharpen": self.chk_sharpen.isChecked()}
self.worker = WorkerThread(files, self.output_dir, scale=scale, model=model,
options=options, post_params=self._get_post_params())
self.worker.progress_signal.connect(self.update_progress, Qt.QueuedConnection)
self.worker.log_signal.connect(self.append_log, Qt.QueuedConnection)
self.worker.finished_signal.connect(self.process_finished, Qt.QueuedConnection)
self.progress_bar.setValue(0)
self.btn_start.setEnabled(False);
self.btn_stop.setEnabled(True)
self._set_controls_enabled(False)
feats = []
if self.chk_denoise.isChecked(): feats.append("去噪")
if self.chk_artifact.isChecked(): feats.append("去伪影")
if self.chk_color.isChecked(): feats.append("色彩")
if self.chk_sharpen.isChecked(): feats.append("锐化")
if use_ai: feats.append(f"AI({model})")
if use_sr: feats.append(f"{scale}x超分")
self.append_log(f"🚀 启动 | CPU×{N_WORKERS} | {' + '.join(feats)}")
self.worker.start()
def stop_process(self):
if self.worker:
self.btn_stop.setEnabled(False);
self.btn_start.setEnabled(True)
self.lbl_status.setText("状态: 正在停止…");
self.worker.stop()
def update_progress(self, value, text):
if value >= 0: self.progress_bar.setValue(value)
if text: self.lbl_status.setText(f"状态: {text}")
def append_log(self, text):
self.log_text.appendPlainText(f"[{time.strftime('%H:%M:%S')}] {text}")
self.log_text.ensureCursorVisible()
def process_finished(self, success, msg):
is_stop = "用户停止" in msg
self._set_controls_enabled(True)
if is_stop:
self.btn_stop.setEnabled(False);
self.lbl_status.setText("状态: 已停止 🛑")
self.append_log("\n--- 处理已停止 ---")
elif success:
self.btn_start.setEnabled(True);
self.btn_stop.setEnabled(False)
self.lbl_status.setText("状态: 全部完成 ✅");
self.progress_bar.setValue(100)
self.append_log(f"\n🎉 {msg}")
else:
self.btn_start.setEnabled(True);
self.btn_stop.setEnabled(False)
self.lbl_status.setText(f"状态: {msg}")
self.append_log(f"\n--- 处理结束: {msg} ---")
def _cleanup():
if self.worker is not None:
try:
self.worker.blockSignals(True); self.worker.quit(); self.worker.wait(3000)
except RuntimeError:
pass
self.worker.deleteLater();
self.worker = None
if success and not is_stop: QMessageBox.information(self, "处理完成", msg)
QTimer.singleShot(0, _cleanup)
def run_utility(self, func, title):
if self.util_worker and self.util_worker.isRunning(): return
self.append_log(f"🔎 {title}…")
self.util_worker = UtilityWorker(func, self)
self.util_worker.log_signal.connect(self.append_log, Qt.QueuedConnection)
self.util_worker.finished_signal.connect(lambda ok, msg: setattr(self, "util_worker", None),
Qt.QueuedConnection)
self.util_worker.start()
def check_ffmpeg_status(self):
if not FFMPEG_PATH or not os.path.exists(FFMPEG_PATH): return f"❌ 未找到: {FFMPEG_PATH}"
try:
r = subprocess.run([FFMPEG_PATH, "-version"], stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
text=True, encoding="utf-8", errors="ignore", timeout=10)
first = (r.stdout or "").strip().splitlines()[0] if r.stdout else "?"
return f"✅ FFmpeg: {first}"
except subprocess.TimeoutExpired:
return "❌ FFmpeg 版本查询超时"
except Exception as e:
return f"❌ FFmpeg 检查失败: {e}"
def closeEvent(self, event):
if self.worker and self.worker.isRunning():
self.worker.stop();
self.worker.wait(5000)
if self.worker.isRunning(): self.worker.terminate(); self.worker.wait(2000)
if self.worker is not None:
self.worker.blockSignals(True);
self.worker.deleteLater();
self.worker = None
if self.util_worker and self.util_worker.isRunning():
self.util_worker.wait(1000);
self.util_worker.deleteLater();
self.util_worker = None
event.accept()
def main():
QApplication.setAttribute(Qt.AA_EnableHighDpiScaling, True)
app = QApplication(sys.argv)
win = VideoRepairTool()
win.show()
sys.exit(app.exec_())
if __name__ == '__main__':
main()
[复制链接]
发表于 2026-8-26 20:01
|
发表于 2026-8-26 20:13
