youfu
3261a461cb
- Add a '✨ Rename' button and live preview to the Output Settings section - Convert BiliBili Live recording stems of the form '[YYYY-MM-DD HH-MM-SS][streamer][title]' into 'YYYY年MM月DD日【直播回放】title' - Preview shows the converted result and enables/disables the button based on whether the current name matches the pattern - On auto-fill, keep the raw stem (no '_concat' suffix) so the pattern can match; strip a trailing '_concat' before matching as well Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
1097 lines
44 KiB
Python
Executable File
1097 lines
44 KiB
Python
Executable File
"""
|
||
Video Concatenator — Production Rewrite
|
||
========================================
|
||
Key improvements over the original:
|
||
|
||
[CRITICAL FIX]
|
||
1. No console/bash windows ever appear — subprocess.CREATE_NO_WINDOW applied
|
||
to ALL subprocess calls (ffmpeg, ffprobe), including duration scanning.
|
||
|
||
[PROGRESS FIX]
|
||
2. Progress bar no longer "catches up" at the end:
|
||
- Bar is capped at 99% while ffmpeg is still writing.
|
||
- The moment ffmpeg emits "progress=end" the bar snaps to 100% and status
|
||
reads "Finalising…" — the user sees an honest state with zero lag.
|
||
|
||
[UX]
|
||
3. Drag-and-drop support (tkinterdnd2 if installed; graceful fallback).
|
||
4. Per-file duration shown in the list, fetched in background threads.
|
||
5. Row index (#) column so order is immediately obvious.
|
||
6. Separate elapsed / ETA / speed label so progress row isn't cluttered.
|
||
7. Output file already-exists guard with overwrite prompt.
|
||
8. Single-file concat warning (copy semantics).
|
||
9. Partial output file removed automatically on Stop.
|
||
10. Light/Dark theme toggle button in title bar.
|
||
|
||
[PERFORMANCE]
|
||
11. Duration values are cached per item after the first ffprobe scan, so
|
||
the total_duration calculation before concat is near-instant.
|
||
12. ffprobe calls for the list are parallelised (one thread per file).
|
||
|
||
[RESPONSIVE UI]
|
||
13. DPI-aware: calls SetProcessDpiAwareness(2) before any Tk window is
|
||
created; minsize scales with DPI so the window is usable on 4K displays.
|
||
"""
|
||
|
||
import os
|
||
import sys
|
||
import json
|
||
import re
|
||
import shutil
|
||
import subprocess
|
||
import tempfile
|
||
import threading
|
||
import time
|
||
from pathlib import Path
|
||
|
||
import tkinter as tk
|
||
from tkinter import filedialog, messagebox
|
||
import customtkinter as ctk
|
||
|
||
# ── DPI awareness (must happen before any Tk/CTk window) ──────────────────────
|
||
if sys.platform == "win32":
|
||
try:
|
||
import ctypes
|
||
ctypes.windll.shcore.SetProcessDpiAwareness(2) # Per-monitor V2
|
||
except Exception:
|
||
try:
|
||
ctypes.windll.user32.SetProcessDPIAware()
|
||
except Exception:
|
||
pass
|
||
|
||
# ── Optional drag-and-drop ─────────────────────────────────────────────────────
|
||
try:
|
||
from tkinterdnd2 import TkinterDnD, DND_FILES # type: ignore
|
||
_DND = True
|
||
except ImportError:
|
||
_DND = False
|
||
|
||
# ── Appearance ─────────────────────────────────────────────────────────────────
|
||
ctk.set_appearance_mode("system")
|
||
ctk.set_default_color_theme("blue")
|
||
|
||
# ── Suppress console windows on Windows for ALL child processes ────────────────
|
||
_NO_WINDOW: dict = {}
|
||
if sys.platform == "win32":
|
||
_NO_WINDOW["creationflags"] = subprocess.CREATE_NO_WINDOW
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════════════════════════
|
||
# Utility helpers
|
||
# ══════════════════════════════════════════════════════════════════════════════
|
||
|
||
def _app_dir() -> str:
|
||
if getattr(sys, "frozen", False):
|
||
return os.path.dirname(sys.executable)
|
||
return os.path.dirname(os.path.abspath(__file__))
|
||
|
||
|
||
def _find_tool(name: str) -> str | None:
|
||
"""Locate ffmpeg / ffprobe next to the exe / script, then on PATH."""
|
||
base = _app_dir()
|
||
for sub in ["_internal", ""]:
|
||
for variant in [name + ".exe", name]:
|
||
p = os.path.join(base, sub, variant)
|
||
if os.path.exists(p):
|
||
return p
|
||
return shutil.which(name + ".exe") or shutil.which(name)
|
||
|
||
|
||
def _fmt_dur(seconds: float) -> str:
|
||
"""Format a duration as H:MM:SS or M:SS."""
|
||
if seconds < 0:
|
||
return "…"
|
||
if seconds == 0:
|
||
return "—"
|
||
s = int(round(seconds))
|
||
h, rem = divmod(s, 3600)
|
||
m, sec = divmod(rem, 60)
|
||
return f"{h}:{m:02d}:{sec:02d}" if h else f"{m}:{sec:02d}"
|
||
|
||
|
||
def _fmt_elapsed(s: float) -> str:
|
||
if s < 60:
|
||
return f"{int(s)}s"
|
||
m, sec = divmod(int(s), 60)
|
||
return f"{m}m {sec:02d}s"
|
||
|
||
|
||
def _dpi_scale(widget) -> float:
|
||
try:
|
||
return max(1.0, widget.winfo_fpixels("1i") / 96.0)
|
||
except Exception:
|
||
return 1.0
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════════════════════════
|
||
# Application
|
||
# ══════════════════════════════════════════════════════════════════════════════
|
||
|
||
_BaseClass = TkinterDnD.Tk if _DND else ctk.CTk # type: ignore[misc]
|
||
|
||
|
||
class App(_BaseClass): # type: ignore[valid-type]
|
||
|
||
VIDEO_EXTS = {".mp4", ".mov", ".mkv", ".avi", ".flv",
|
||
".ts", ".wmv", ".m4v", ".webm", ".mts", ".m2ts"}
|
||
|
||
def __init__(self):
|
||
super().__init__()
|
||
self.title("Video Concatenator")
|
||
self.geometry("820x860")
|
||
self.minsize(700, 660)
|
||
|
||
# ── State ─────────────────────────────────────────────────────────────
|
||
# Each item: {path:str, checked:BooleanVar, duration:float, dur_label:CTkLabel|None, row_widget}
|
||
self.items: list[dict] = []
|
||
self.active_index: int | None = None
|
||
self._dur_lock = threading.Lock()
|
||
|
||
self.out_name_var = tk.StringVar(value="output_concat")
|
||
self.out_ext_var = tk.StringVar(value=".mp4")
|
||
self.out_dir_var = tk.StringVar(value="")
|
||
self.del_orig_var = tk.BooleanVar(value=False)
|
||
|
||
self._proc: subprocess.Popen | None = None
|
||
self._stop_req = False
|
||
|
||
# ── Tools ─────────────────────────────────────────────────────────────
|
||
self.ffmpeg = _find_tool("ffmpeg")
|
||
self.ffprobe = _find_tool("ffprobe")
|
||
|
||
# ── Build UI ──────────────────────────────────────────────────────────
|
||
self._build_ui()
|
||
self.update_idletasks()
|
||
scale = _dpi_scale(self)
|
||
self.minsize(int(700 * scale), int(660 * scale))
|
||
|
||
# ── Initial status ─────────────────────────────────────────────────────
|
||
if not self.ffmpeg:
|
||
self._set_status(
|
||
"ffmpeg not found — place ffmpeg.exe next to this app or add to PATH.", True
|
||
)
|
||
self.start_btn.configure(state="disabled")
|
||
messagebox.showerror(
|
||
"Missing dependency",
|
||
"ffmpeg could not be found.\n\n"
|
||
"Please place ffmpeg.exe (and ffprobe.exe) next to this program,\n"
|
||
"or install ffmpeg and ensure it is on your PATH."
|
||
)
|
||
else:
|
||
hint = ("Ready. Drag & drop videos or click ＋ Add Videos."
|
||
if _DND else "Ready. Click ＋ Add Videos to get started.")
|
||
self._set_status(hint)
|
||
|
||
self._refresh_start()
|
||
|
||
# ══════════════════════════════════════════════════════════════════════════
|
||
# UI Construction
|
||
# ══════════════════════════════════════════════════════════════════════════
|
||
|
||
def _build_ui(self):
|
||
card = ctk.CTkFrame(self, corner_radius=16)
|
||
card.pack(fill="both", expand=True, padx=20, pady=20)
|
||
|
||
# Title bar
|
||
tbar = ctk.CTkFrame(card, fg_color="transparent")
|
||
tbar.pack(fill="x", padx=16, pady=(16, 2))
|
||
ctk.CTkLabel(tbar, text="🎬 Video Concatenator",
|
||
font=ctk.CTkFont(size=22, weight="bold")).pack(side="left")
|
||
ctk.CTkButton(
|
||
tbar, text="☀ / ☾", width=68,
|
||
fg_color="transparent",
|
||
text_color=("gray30", "gray70"),
|
||
hover_color=("gray85", "gray25"),
|
||
command=self._toggle_theme
|
||
).pack(side="right")
|
||
|
||
# Scrollable body
|
||
scroll = ctk.CTkScrollableFrame(card, corner_radius=12)
|
||
scroll.pack(fill="both", expand=True, padx=14, pady=(8, 14))
|
||
|
||
self._build_sec_add(scroll)
|
||
self._build_sec_list(scroll)
|
||
self._build_sec_output(scroll)
|
||
self._build_sec_options(scroll)
|
||
self._build_sec_run(scroll)
|
||
|
||
# ── Section: Add Videos ───────────────────────────────────────────────────
|
||
|
||
def _build_sec_add(self, parent):
|
||
sec = self._section(parent, "1 · Add Videos")
|
||
sec.pack(fill="x", pady=(0, 12))
|
||
|
||
row = ctk.CTkFrame(sec, fg_color="transparent")
|
||
row.pack(fill="x", padx=14, pady=(10, 4))
|
||
|
||
ctk.CTkButton(row, text="＋ Add Videos", width=170,
|
||
command=self.pick_files).pack(side="left")
|
||
|
||
ctk.CTkButton(
|
||
row, text="Clear All", width=110,
|
||
fg_color=("gray80", "gray25"),
|
||
text_color=("gray10", "gray90"),
|
||
hover_color=("gray70", "gray35"),
|
||
command=self.clear_all
|
||
).pack(side="left", padx=(10, 0))
|
||
|
||
self.count_lbl = ctk.CTkLabel(
|
||
row, text="0 files",
|
||
font=ctk.CTkFont(size=12), text_color=("gray45", "gray55")
|
||
)
|
||
self.count_lbl.pack(side="right")
|
||
|
||
ctk.CTkLabel(
|
||
sec,
|
||
text=("📂 Drag & drop videos here, or use the button above."
|
||
if _DND else
|
||
"Tip: you can add videos multiple times to repeat a clip."),
|
||
font=ctk.CTkFont(size=12), text_color=("gray50", "gray55")
|
||
).pack(anchor="w", padx=14, pady=(0, 12))
|
||
|
||
if _DND:
|
||
self.drop_target_register(DND_FILES)
|
||
self.dnd_bind("<<Drop>>", self._on_drop)
|
||
|
||
# ── Section: List ─────────────────────────────────────────────────────────
|
||
|
||
def _build_sec_list(self, parent):
|
||
sec = self._section(parent, "2 · Review & Reorder")
|
||
sec.pack(fill="x", pady=(0, 12))
|
||
|
||
body = ctk.CTkFrame(sec, fg_color="transparent")
|
||
body.pack(fill="x", padx=14, pady=(10, 4))
|
||
|
||
# Left — list panel
|
||
left = ctk.CTkFrame(body, corner_radius=10)
|
||
left.pack(side="left", fill="both", expand=True, padx=(0, 10))
|
||
|
||
# Column header row
|
||
hdr = ctk.CTkFrame(left, fg_color=("gray88", "gray20"), corner_radius=8)
|
||
hdr.pack(fill="x", padx=8, pady=(8, 2))
|
||
for text, w, anchor in [
|
||
("#", 30, "center"),
|
||
("✓", 30, "center"),
|
||
("File name", 0, "w"),
|
||
("Duration", 76, "center"),
|
||
]:
|
||
lbl = ctk.CTkLabel(hdr, text=text,
|
||
font=ctk.CTkFont(size=11, weight="bold"),
|
||
anchor=anchor)
|
||
if w:
|
||
lbl.configure(width=w)
|
||
lbl.pack(side="left" if text != "Duration" else "right",
|
||
padx=(6 if text == "#" else 0, 8 if text == "Duration" else 0),
|
||
pady=4)
|
||
else:
|
||
lbl.pack(side="left", fill="x", expand=True, padx=4, pady=4)
|
||
|
||
self.list_area = ctk.CTkScrollableFrame(left, height=190, corner_radius=8)
|
||
self.list_area.pack(fill="both", expand=True, padx=8, pady=(0, 8))
|
||
|
||
# Right — buttons
|
||
right = ctk.CTkFrame(body, fg_color="transparent")
|
||
right.pack(side="right", fill="y")
|
||
self.btn_up = self._side_btn(right, "↑ Move Up", self.move_up)
|
||
self.btn_down = self._side_btn(right, "↓ Move Down", self.move_down)
|
||
self.btn_remove = self._side_btn(right, "✕ Remove", self.remove_active)
|
||
ctk.CTkFrame(right, height=10, fg_color="transparent").pack()
|
||
self.btn_all = self._side_btn(right, "Select All", self.select_all)
|
||
self.btn_none = self._side_btn(right, "Select None", self.select_none)
|
||
|
||
self.total_lbl = ctk.CTkLabel(
|
||
sec, text="Total: —",
|
||
font=ctk.CTkFont(size=12), text_color=("gray45", "gray55")
|
||
)
|
||
self.total_lbl.pack(anchor="e", padx=14, pady=(0, 10))
|
||
|
||
# ── Section: Output ───────────────────────────────────────────────────────
|
||
|
||
def _build_sec_output(self, parent):
|
||
sec = self._section(parent, "3 · Output Settings")
|
||
sec.pack(fill="x", pady=(0, 12))
|
||
|
||
body = ctk.CTkFrame(sec, fg_color="transparent")
|
||
body.pack(fill="x", padx=14, pady=(10, 4))
|
||
|
||
# Name row
|
||
r1 = ctk.CTkFrame(body, fg_color="transparent")
|
||
r1.pack(fill="x", pady=(0, 4))
|
||
ctk.CTkLabel(r1, text="File name:", width=96, anchor="w").pack(side="left")
|
||
ctk.CTkEntry(r1, textvariable=self.out_name_var).pack(side="left", fill="x", expand=True)
|
||
ctk.CTkOptionMenu(
|
||
r1, values=[".mp4", ".mkv", ".mov", ".avi", ".ts"],
|
||
variable=self.out_ext_var, width=84
|
||
).pack(side="left", padx=(8, 0))
|
||
|
||
# Auto-rename button — converts BiliBili Live filename to readable format
|
||
self.rename_btn = ctk.CTkButton(
|
||
r1, text="✨ Rename", width=110,
|
||
fg_color=("#EDE9FE", "#3B1F6E"),
|
||
text_color=("#5B21B6", "#C4B5FD"),
|
||
hover_color=("#DDD6FE", "#4C2889"),
|
||
command=self._apply_bililive_rename
|
||
)
|
||
self.rename_btn.pack(side="left", padx=(8, 0))
|
||
|
||
# Preview label: shows what the rename will produce before applying
|
||
self.rename_preview_lbl = ctk.CTkLabel(
|
||
body, text="",
|
||
font=ctk.CTkFont(size=11), text_color=("#7C3AED", "#A78BFA"),
|
||
anchor="w"
|
||
)
|
||
self.rename_preview_lbl.pack(fill="x", pady=(0, 4))
|
||
|
||
# Directory row
|
||
r2 = ctk.CTkFrame(body, fg_color="transparent")
|
||
r2.pack(fill="x")
|
||
ctk.CTkLabel(r2, text="Save to:", width=96, anchor="w").pack(side="left")
|
||
ctk.CTkEntry(r2, textvariable=self.out_dir_var).pack(side="left", fill="x", expand=True)
|
||
ctk.CTkButton(
|
||
r2, text="Browse…", width=100,
|
||
fg_color=("gray80", "gray25"),
|
||
text_color=("gray10", "gray90"),
|
||
hover_color=("gray70", "gray35"),
|
||
command=self.pick_dir
|
||
).pack(side="left", padx=(8, 0))
|
||
|
||
self.out_path_lbl = ctk.CTkLabel(
|
||
sec, text="Output path: —",
|
||
font=ctk.CTkFont(size=11), text_color=("gray45", "gray55"),
|
||
wraplength=620, anchor="w", justify="left"
|
||
)
|
||
self.out_path_lbl.pack(anchor="w", padx=14, pady=(6, 10))
|
||
|
||
for v in (self.out_name_var, self.out_ext_var, self.out_dir_var):
|
||
v.trace_add("write", lambda *_: self._refresh_out_label())
|
||
self._refresh_out_label()
|
||
|
||
# ── Section: Options ──────────────────────────────────────────────────────
|
||
|
||
def _build_sec_options(self, parent):
|
||
sec = self._section(parent, "4 · Options")
|
||
sec.pack(fill="x", pady=(0, 12))
|
||
|
||
ctk.CTkCheckBox(
|
||
sec, text="Delete original files after successful concatenation",
|
||
variable=self.del_orig_var
|
||
).pack(anchor="w", padx=14, pady=(10, 2))
|
||
|
||
ctk.CTkLabel(
|
||
sec,
|
||
text="⚠ Deletion is permanent — originals cannot be recovered.",
|
||
text_color="#F97316", font=ctk.CTkFont(size=11)
|
||
).pack(anchor="w", padx=40, pady=(0, 10))
|
||
|
||
# ── Section: Run ──────────────────────────────────────────────────────────
|
||
|
||
def _build_sec_run(self, parent):
|
||
sec = self._section(parent, "5 · Run")
|
||
sec.pack(fill="x", pady=(0, 14))
|
||
|
||
body = ctk.CTkFrame(sec, fg_color="transparent")
|
||
body.pack(fill="x", padx=14, pady=(12, 6))
|
||
|
||
# Buttons
|
||
btn_row = ctk.CTkFrame(body, fg_color="transparent")
|
||
btn_row.pack(fill="x", pady=(0, 10))
|
||
|
||
self.start_btn = ctk.CTkButton(
|
||
btn_row, text="▶ Start Concatenation",
|
||
width=230, height=42,
|
||
font=ctk.CTkFont(size=14, weight="bold"),
|
||
command=self.start_concat
|
||
)
|
||
self.start_btn.pack(side="left")
|
||
|
||
self.stop_btn = ctk.CTkButton(
|
||
btn_row, text="■ Stop", width=106, height=42,
|
||
fg_color=("gray80", "gray25"),
|
||
text_color=("gray10", "gray90"),
|
||
hover_color=("#FCA5A5", "#7F1D1D"),
|
||
command=self.stop_concat, state="disabled"
|
||
)
|
||
self.stop_btn.pack(side="left", padx=(12, 0))
|
||
|
||
# Progress bar row
|
||
prog_row = ctk.CTkFrame(body, fg_color="transparent")
|
||
prog_row.pack(fill="x", pady=(0, 6))
|
||
|
||
self.progress = ctk.CTkProgressBar(prog_row, height=16, corner_radius=8)
|
||
self.progress.set(0)
|
||
self.progress.pack(side="left", fill="x", expand=True, padx=(0, 10))
|
||
|
||
self.pct_lbl = ctk.CTkLabel(
|
||
prog_row, text="—", width=46, anchor="e",
|
||
font=ctk.CTkFont(size=12, weight="bold")
|
||
)
|
||
self.pct_lbl.pack(side="right")
|
||
|
||
# Status + timing row
|
||
info_row = ctk.CTkFrame(body, fg_color="transparent")
|
||
info_row.pack(fill="x")
|
||
|
||
self.status_lbl = ctk.CTkLabel(info_row, text="", anchor="w",
|
||
font=ctk.CTkFont(size=12))
|
||
self.status_lbl.pack(side="left", fill="x", expand=True)
|
||
|
||
self.timing_lbl = ctk.CTkLabel(
|
||
info_row, text="", anchor="e",
|
||
font=ctk.CTkFont(size=11), text_color=("gray45", "gray55")
|
||
)
|
||
self.timing_lbl.pack(side="right")
|
||
|
||
# Collapsible log
|
||
self._log_open = False
|
||
self._log_btn = ctk.CTkButton(
|
||
sec, text="Log ▾", width=80,
|
||
fg_color="transparent",
|
||
text_color=("gray40", "gray60"),
|
||
hover_color=("gray88", "gray22"),
|
||
font=ctk.CTkFont(size=12),
|
||
command=self._toggle_log
|
||
)
|
||
self._log_btn.pack(anchor="w", padx=14, pady=(4, 0))
|
||
|
||
self._log_box = ctk.CTkTextbox(
|
||
sec, height=110, corner_radius=8,
|
||
font=ctk.CTkFont(size=11, family="Courier")
|
||
)
|
||
self._log_box.insert("1.0", "")
|
||
self._log_box.configure(state="disabled")
|
||
self._log_box.pack_forget()
|
||
ctk.CTkFrame(sec, height=6, fg_color="transparent").pack()
|
||
|
||
# ══════════════════════════════════════════════════════════════════════════
|
||
# UI helpers
|
||
# ══════════════════════════════════════════════════════════════════════════
|
||
|
||
def _section(self, parent, title: str) -> ctk.CTkFrame:
|
||
f = ctk.CTkFrame(parent, corner_radius=12)
|
||
hdr = ctk.CTkFrame(f, fg_color="transparent")
|
||
hdr.pack(fill="x", padx=14, pady=(10, 0))
|
||
ctk.CTkLabel(hdr, text=title,
|
||
font=ctk.CTkFont(size=13, weight="bold")).pack(side="left")
|
||
ctk.CTkFrame(f, height=1, fg_color=("gray78", "gray32")).pack(
|
||
fill="x", padx=14, pady=(6, 0)
|
||
)
|
||
return f
|
||
|
||
def _side_btn(self, parent, text: str, cmd) -> ctk.CTkButton:
|
||
b = ctk.CTkButton(
|
||
parent, text=text, width=152, height=32,
|
||
fg_color=("gray82", "gray22"),
|
||
text_color=("gray10", "gray90"),
|
||
hover_color=("gray72", "gray32"),
|
||
font=ctk.CTkFont(size=12),
|
||
command=cmd
|
||
)
|
||
b.pack(pady=4)
|
||
return b
|
||
|
||
def _set_status(self, text: str, error: bool = False):
|
||
color = "#DC2626" if error else ("gray10", "gray90")
|
||
self.status_lbl.configure(text=text, text_color=color)
|
||
|
||
def _set_timing(self, text: str):
|
||
self.timing_lbl.configure(text=text)
|
||
|
||
def _toggle_theme(self):
|
||
mode = ctk.get_appearance_mode().lower()
|
||
ctk.set_appearance_mode("light" if mode == "dark" else "dark")
|
||
|
||
def _toggle_log(self):
|
||
self._log_open = not self._log_open
|
||
if self._log_open:
|
||
self._log_btn.configure(text="Log ▴")
|
||
self._log_box.pack(fill="x", padx=14, pady=(0, 6))
|
||
else:
|
||
self._log_btn.configure(text="Log ▾")
|
||
self._log_box.pack_forget()
|
||
|
||
def _append_log(self, text: str):
|
||
if not self._log_open:
|
||
return
|
||
self._log_box.configure(state="normal")
|
||
self._log_box.insert("end", text + "\n")
|
||
self._log_box.see("end")
|
||
self._log_box.configure(state="disabled")
|
||
|
||
def _refresh_out_label(self):
|
||
out = self._out_path()
|
||
self.out_path_lbl.configure(
|
||
text=f"Output path: {out}" if out else "Output path: —"
|
||
)
|
||
self._refresh_rename_preview()
|
||
self._refresh_start()
|
||
|
||
def _refresh_rename_preview(self):
|
||
"""Show a live preview of what the rename button will produce."""
|
||
if not hasattr(self, "rename_preview_lbl"):
|
||
return
|
||
current = self.out_name_var.get().strip()
|
||
renamed = self._bililive_rename(current)
|
||
if renamed and renamed != current:
|
||
self.rename_preview_lbl.configure(
|
||
text=f" → {renamed}"
|
||
)
|
||
self.rename_btn.configure(state="normal")
|
||
else:
|
||
self.rename_preview_lbl.configure(text="")
|
||
# Only disable if the name cannot be converted
|
||
self.rename_btn.configure(state="disabled" if not renamed else "normal")
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════════════════════
|
||
# BiliBili Live filename auto-rename
|
||
# ══════════════════════════════════════════════════════════════════════════
|
||
|
||
# Pattern: [2025-06-28 20-31-15][纳豆nado][title content]
|
||
_BILILIVE_RE = re.compile(
|
||
r"^\[(\d{4})-(\d{2})-(\d{2}) \d{2}-\d{2}-\d{2}\]\[[^\]]*\]\[(.+)\]$"
|
||
)
|
||
|
||
@staticmethod
|
||
def _bililive_rename(stem: str) -> str:
|
||
"""
|
||
Convert a BiliBili Live recording stem to a readable name.
|
||
Input: [2026-03-14 22-46-37][纳豆nado][【点歌专场】今晚九点来捏]
|
||
Output: 2026年03月14日【直播回放】【点歌专场】今晚九点来捏
|
||
Returns the converted string, or "" if the stem does not match.
|
||
"""
|
||
# Strip any trailing _concat suffix before matching
|
||
test = stem.removesuffix("_concat").strip()
|
||
m = App._BILILIVE_RE.match(test)
|
||
if not m:
|
||
return ""
|
||
year, month, day, title = m.groups()
|
||
return f"{year}年{month}月{day}日【直播回放】{title}"
|
||
|
||
def _apply_bililive_rename(self):
|
||
"""Apply the rename conversion to the current output name."""
|
||
current = self.out_name_var.get().strip()
|
||
renamed = self._bililive_rename(current)
|
||
if not renamed:
|
||
return
|
||
self.out_name_var.set(renamed)
|
||
self.rename_preview_lbl.configure(text=" ✓ Renamed!")
|
||
self.rename_btn.configure(state="disabled")
|
||
|
||
def _refresh_start(self):
|
||
if not hasattr(self, "start_btn"):
|
||
return # UI not fully built yet
|
||
ok = bool(
|
||
self.ffmpeg
|
||
and not self._is_running()
|
||
and self._checked_paths()
|
||
and self.out_dir_var.get().strip()
|
||
and self.out_name_var.get().strip()
|
||
)
|
||
self.start_btn.configure(state="normal" if ok else "disabled")
|
||
|
||
def _is_running(self) -> bool:
|
||
return self._proc is not None and self._proc.poll() is None
|
||
|
||
def _out_path(self) -> str:
|
||
d = self.out_dir_var.get().strip()
|
||
name = self.out_name_var.get().strip()
|
||
ext = self.out_ext_var.get().strip()
|
||
if not d or not name:
|
||
return ""
|
||
if not ext.startswith("."):
|
||
ext = "." + ext
|
||
return str(Path(d) / f"{name}{ext}")
|
||
|
||
def _checked_paths(self) -> list[str]:
|
||
return [it["path"] for it in self.items if it["checked"].get()]
|
||
|
||
# ══════════════════════════════════════════════════════════════════════════
|
||
# Drag-and-drop
|
||
# ══════════════════════════════════════════════════════════════════════════
|
||
|
||
def _on_drop(self, event):
|
||
# tkinterdnd2 wraps paths that contain spaces in {braces}
|
||
raw = event.data
|
||
paths = [a or b for a, b in re.findall(r'\{([^}]+)\}|(\S+)', raw)]
|
||
self._add_files(paths)
|
||
|
||
# ══════════════════════════════════════════════════════════════════════════
|
||
# Data model
|
||
# ══════════════════════════════════════════════════════════════════════════
|
||
|
||
def _add_files(self, paths: list[str]):
|
||
existing = {it["path"] for it in self.items}
|
||
new: list[dict] = []
|
||
for p in paths:
|
||
p = p.strip()
|
||
if not p or Path(p).suffix.lower() not in self.VIDEO_EXTS:
|
||
continue
|
||
if p not in existing:
|
||
item: dict = {
|
||
"path": p, "checked": tk.BooleanVar(value=True),
|
||
"duration": -1.0, "dur_label": None, "row_widget": None
|
||
}
|
||
self.items.append(item)
|
||
new.append(item)
|
||
|
||
if not new:
|
||
return
|
||
|
||
self.active_index = 0
|
||
|
||
# Auto-fill output dir / name
|
||
if not self.out_dir_var.get().strip():
|
||
self.out_dir_var.set(str(Path(self.items[0]["path"]).parent))
|
||
if self.out_name_var.get() in ("", "output_concat"):
|
||
# Keep the raw BiliBili stem so the rename pattern can match it
|
||
self.out_name_var.set(Path(self.items[0]["path"]).stem)
|
||
|
||
self._rebuild_list()
|
||
self._update_count()
|
||
self._refresh_out_label()
|
||
|
||
# Background duration scanning (new items only)
|
||
for item in new:
|
||
threading.Thread(target=self._scan_dur, args=(item,), daemon=True).start()
|
||
|
||
def clear_all(self):
|
||
self.items.clear()
|
||
self.active_index = None
|
||
self._rebuild_list()
|
||
self._update_count()
|
||
self._update_total()
|
||
self._refresh_start()
|
||
|
||
def _update_count(self):
|
||
n = len(self.items)
|
||
self.count_lbl.configure(text=f"{n} file{'s' if n != 1 else ''}")
|
||
|
||
def _update_total(self):
|
||
with self._dur_lock:
|
||
known = [it["duration"] for it in self.items if it["duration"] >= 0]
|
||
pending = sum(1 for it in self.items if it["duration"] < 0)
|
||
total = sum(known)
|
||
if not self.items:
|
||
self.total_lbl.configure(text="Total: —")
|
||
elif pending:
|
||
self.total_lbl.configure(text=f"Total: {_fmt_dur(total)} (scanning {pending}…)")
|
||
else:
|
||
self.total_lbl.configure(text=f"Total: {_fmt_dur(total)}")
|
||
|
||
# ── Duration scanning ─────────────────────────────────────────────────────
|
||
|
||
def _scan_dur(self, item: dict):
|
||
dur = self._probe_dur(item["path"])
|
||
with self._dur_lock:
|
||
item["duration"] = dur
|
||
self.after(0, lambda: self._paint_dur(item))
|
||
self.after(0, self._update_total)
|
||
|
||
def _paint_dur(self, item: dict):
|
||
lbl = item.get("dur_label")
|
||
if lbl:
|
||
try:
|
||
lbl.configure(text=_fmt_dur(item["duration"]))
|
||
except Exception:
|
||
pass
|
||
|
||
# ── List rendering ────────────────────────────────────────────────────────
|
||
|
||
def _rebuild_list(self):
|
||
for it in self.items:
|
||
it["dur_label"] = None
|
||
it["row_widget"] = None
|
||
for w in list(self.list_area.winfo_children()):
|
||
w.destroy()
|
||
for idx, it in enumerate(self.items):
|
||
self._make_row(idx, it)
|
||
self._paint_highlights()
|
||
self._refresh_start()
|
||
|
||
def _make_row(self, idx: int, item: dict):
|
||
row = ctk.CTkFrame(self.list_area, corner_radius=7, height=36)
|
||
row.pack(fill="x", pady=3)
|
||
row.pack_propagate(False)
|
||
|
||
def activate(_=None, i=idx):
|
||
self.active_index = i
|
||
self._paint_highlights()
|
||
|
||
# Index
|
||
ctk.CTkLabel(row, text=str(idx + 1), width=30, anchor="center",
|
||
font=ctk.CTkFont(size=11),
|
||
text_color=("gray50", "gray50")).pack(side="left", padx=(8, 0))
|
||
|
||
# Checkbox
|
||
cb = ctk.CTkCheckBox(row, text="", variable=item["checked"], width=26)
|
||
cb.pack(side="left", padx=(4, 2))
|
||
cb.configure(command=lambda: (self._refresh_start(), self._update_total()))
|
||
cb.bind("<Button-1>", lambda e, i=idx: activate())
|
||
|
||
# File name label
|
||
name_lbl = ctk.CTkLabel(row, text=Path(item["path"]).name,
|
||
anchor="w", font=ctk.CTkFont(size=12))
|
||
name_lbl.pack(side="left", fill="x", expand=True, padx=(2, 4))
|
||
|
||
# Duration label
|
||
dur_text = _fmt_dur(item["duration"]) if item["duration"] >= 0 else "…"
|
||
dur_lbl = ctk.CTkLabel(row, text=dur_text, width=72, anchor="center",
|
||
font=ctk.CTkFont(size=11),
|
||
text_color=("gray45", "gray55"))
|
||
dur_lbl.pack(side="right", padx=(0, 10))
|
||
item["dur_label"] = dur_lbl
|
||
|
||
for widget in (row, name_lbl):
|
||
widget.bind("<Button-1>", activate)
|
||
item["row_widget"] = row
|
||
|
||
def _paint_highlights(self):
|
||
active_bg = ("#DBEAFE", "#1E3A5F")
|
||
normal_bg = ("transparent", "transparent")
|
||
for i, it in enumerate(self.items):
|
||
row = it.get("row_widget")
|
||
if not row:
|
||
continue
|
||
try:
|
||
row.configure(fg_color=active_bg if self.active_index == i else normal_bg)
|
||
except Exception:
|
||
pass
|
||
|
||
# ══════════════════════════════════════════════════════════════════════════
|
||
# File actions
|
||
# ══════════════════════════════════════════════════════════════════════════
|
||
|
||
def pick_files(self):
|
||
files = filedialog.askopenfilenames(
|
||
title="Select video files",
|
||
filetypes=[
|
||
("Video files", "*.mp4 *.mov *.mkv *.avi *.flv *.ts *.wmv *.m4v *.webm *.mts"),
|
||
("All files", "*.*"),
|
||
]
|
||
)
|
||
if files:
|
||
self._add_files(list(files))
|
||
|
||
def pick_dir(self):
|
||
d = filedialog.askdirectory(title="Choose output folder")
|
||
if d:
|
||
self.out_dir_var.set(d)
|
||
|
||
def move_up(self):
|
||
if self.active_index is None or self.active_index <= 0:
|
||
return
|
||
i = self.active_index
|
||
self.items[i - 1], self.items[i] = self.items[i], self.items[i - 1]
|
||
self.active_index = i - 1
|
||
self._rebuild_list()
|
||
|
||
def move_down(self):
|
||
if self.active_index is None or self.active_index >= len(self.items) - 1:
|
||
return
|
||
i = self.active_index
|
||
self.items[i + 1], self.items[i] = self.items[i], self.items[i + 1]
|
||
self.active_index = i + 1
|
||
self._rebuild_list()
|
||
|
||
def remove_active(self):
|
||
if self.active_index is None or not self.items:
|
||
return
|
||
i = self.active_index
|
||
self.items.pop(i)
|
||
self.active_index = min(i, len(self.items) - 1) if self.items else None
|
||
self._rebuild_list()
|
||
self._update_count()
|
||
self._update_total()
|
||
|
||
def select_all(self):
|
||
for it in self.items:
|
||
it["checked"].set(True)
|
||
self._refresh_start()
|
||
|
||
def select_none(self):
|
||
for it in self.items:
|
||
it["checked"].set(False)
|
||
self._refresh_start()
|
||
|
||
# ══════════════════════════════════════════════════════════════════════════
|
||
# ffprobe (always CREATE_NO_WINDOW)
|
||
# ══════════════════════════════════════════════════════════════════════════
|
||
|
||
def _probe_dur(self, path: str) -> float:
|
||
if not self.ffprobe:
|
||
return 0.0
|
||
try:
|
||
r = subprocess.run(
|
||
[self.ffprobe, "-v", "error",
|
||
"-show_entries", "format=duration",
|
||
"-of", "json", path],
|
||
stdout=subprocess.PIPE, stderr=subprocess.PIPE,
|
||
text=True, timeout=20,
|
||
**_NO_WINDOW
|
||
)
|
||
if r.returncode != 0:
|
||
return 0.0
|
||
data = json.loads(r.stdout or "{}")
|
||
return max(0.0, float(data.get("format", {}).get("duration", 0) or 0))
|
||
except Exception:
|
||
return 0.0
|
||
|
||
def _total_dur(self, files: list[str]) -> float:
|
||
total = 0.0
|
||
for path in files:
|
||
item = next((it for it in self.items if it["path"] == path), None)
|
||
if item and item["duration"] >= 0:
|
||
total += item["duration"]
|
||
else:
|
||
total += self._probe_dur(path)
|
||
return total
|
||
|
||
# ══════════════════════════════════════════════════════════════════════════
|
||
# Concatenation
|
||
# ══════════════════════════════════════════════════════════════════════════
|
||
|
||
def start_concat(self):
|
||
files = self._checked_paths()
|
||
if not files:
|
||
messagebox.showerror("No files selected", "Please check at least one video.")
|
||
return
|
||
|
||
if len(files) == 1:
|
||
if not messagebox.askyesno(
|
||
"Only one file",
|
||
"Only one file is selected.\nThis will effectively copy it. Continue?"
|
||
):
|
||
return
|
||
|
||
out = self._out_path()
|
||
if not out:
|
||
messagebox.showerror("Missing output", "Set an output name and folder.")
|
||
return
|
||
|
||
# Name conflict guard
|
||
out_stem = Path(out).stem
|
||
if any(Path(f).stem == out_stem for f in files):
|
||
messagebox.showerror(
|
||
"Name conflict",
|
||
"Output name matches a source file.\nChoose a different output name."
|
||
)
|
||
return
|
||
|
||
# Overwrite guard
|
||
if Path(out).exists():
|
||
if not messagebox.askyesno("Overwrite?",
|
||
f"{Path(out).name} already exists. Overwrite?"):
|
||
return
|
||
|
||
# Reset UI
|
||
self._stop_req = False
|
||
self.start_btn.configure(state="disabled")
|
||
self.stop_btn.configure(state="normal")
|
||
self.progress.set(0)
|
||
self.pct_lbl.configure(text="0%")
|
||
self._set_status("Starting…")
|
||
self._set_timing("")
|
||
|
||
if self._log_open:
|
||
self._log_box.configure(state="normal")
|
||
self._log_box.delete("1.0", "end")
|
||
self._log_box.configure(state="disabled")
|
||
|
||
threading.Thread(target=self._worker, args=(files, out), daemon=True).start()
|
||
|
||
def stop_concat(self):
|
||
self._stop_req = True
|
||
if self._proc and self._is_running():
|
||
try:
|
||
self._proc.terminate()
|
||
except Exception:
|
||
pass
|
||
self._set_status("Stopping…")
|
||
|
||
# ── Worker thread ─────────────────────────────────────────────────────────
|
||
|
||
def _worker(self, files: list[str], out_file: str):
|
||
out_dir = str(Path(out_file).parent)
|
||
|
||
# Write concat list file
|
||
try:
|
||
tmp = tempfile.NamedTemporaryFile(
|
||
mode="w", encoding="utf-8", newline="\n",
|
||
delete=False, dir=out_dir, suffix=".txt"
|
||
)
|
||
list_path = tmp.name
|
||
for f in files:
|
||
s = str(Path(f).resolve()).replace("\\", "/").replace("'", r"'\''")
|
||
tmp.write(f"file '{s}'\n")
|
||
tmp.close()
|
||
except Exception as e:
|
||
self.after(0, lambda: messagebox.showerror("Error", f"Could not write temp file:\n{e}"))
|
||
self.after(0, lambda: self._finish(False))
|
||
return
|
||
|
||
# Get total duration (uses cache; may do live probes for uncached files)
|
||
self.after(0, lambda: self._set_status("Scanning durations…"))
|
||
total_secs = self._total_dur(files)
|
||
|
||
# Build ffmpeg command
|
||
cmd = [
|
||
self.ffmpeg,
|
||
"-y",
|
||
"-f", "concat", "-safe", "0",
|
||
"-i", list_path,
|
||
"-c:v", "copy", "-c:a", "copy",
|
||
"-movflags", "+faststart",
|
||
"-nostdin", "-hide_banner",
|
||
"-loglevel", "error",
|
||
"-progress", "pipe:1",
|
||
out_file,
|
||
]
|
||
self.after(0, lambda: self._append_log("CMD: " + " ".join(str(x) for x in cmd)))
|
||
self.after(0, lambda: self._set_status("Concatenating…"))
|
||
|
||
t0 = time.time()
|
||
rc = -1
|
||
stderr_out = ""
|
||
|
||
try:
|
||
proc = subprocess.Popen(
|
||
cmd,
|
||
stdout=subprocess.PIPE,
|
||
stderr=subprocess.PIPE,
|
||
text=True,
|
||
cwd=out_dir,
|
||
bufsize=1,
|
||
**_NO_WINDOW # ← THE FIX: no console window ever
|
||
)
|
||
self._proc = proc
|
||
|
||
cur_ms = 0
|
||
|
||
while True:
|
||
if self._stop_req:
|
||
break
|
||
|
||
line = proc.stdout.readline() if proc.stdout else ""
|
||
if not line:
|
||
if proc.poll() is not None:
|
||
break
|
||
time.sleep(0.04)
|
||
continue
|
||
|
||
line = line.strip()
|
||
|
||
if line.startswith("out_time_ms="):
|
||
try:
|
||
cur_ms = int(line.split("=", 1)[1])
|
||
except ValueError:
|
||
pass
|
||
|
||
elif line == "progress=end":
|
||
# ffmpeg finished writing — snap to 100% RIGHT NOW
|
||
elapsed = time.time() - t0
|
||
self.after(0, lambda e=elapsed: (
|
||
self.progress.set(1.0),
|
||
self.pct_lbl.configure(text="100%"),
|
||
self._set_status("Finalising output file…"),
|
||
self._set_timing(f"Elapsed: {_fmt_elapsed(e)}")
|
||
))
|
||
|
||
elif line.startswith("progress=continue"):
|
||
if total_secs > 0 and cur_ms > 0:
|
||
played = cur_ms / 1_000_000.0
|
||
pct = max(0.0, min(played / total_secs, 0.99)) # cap at 99 until end
|
||
elapsed = time.time() - t0
|
||
speed = played / max(elapsed, 1e-9)
|
||
eta = (total_secs - played) / max(speed, 1e-9)
|
||
|
||
def _ui(p=pct, e=elapsed, r=eta, sp=speed):
|
||
self.progress.set(p)
|
||
self.pct_lbl.configure(text=f"{p*100:.0f}%")
|
||
self._set_timing(
|
||
f"Elapsed {_fmt_elapsed(e)} · ETA {_fmt_elapsed(r)} · {sp:.1f}×"
|
||
)
|
||
self.after(0, _ui)
|
||
|
||
try:
|
||
_, stderr_out = proc.communicate(timeout=10)
|
||
except Exception:
|
||
stderr_out = ""
|
||
rc = proc.returncode
|
||
|
||
except Exception as exc:
|
||
rc = -1
|
||
stderr_out = str(exc)
|
||
|
||
# Cleanup temp file
|
||
try:
|
||
os.remove(list_path)
|
||
except Exception:
|
||
pass
|
||
|
||
# ── Handle result ──────────────────────────────────────────────────────
|
||
if self._stop_req:
|
||
try:
|
||
Path(out_file).unlink(missing_ok=True)
|
||
except Exception:
|
||
pass
|
||
self.after(0, lambda: self._set_status("Stopped by user."))
|
||
self.after(0, lambda: self._finish(False))
|
||
return
|
||
|
||
if rc != 0:
|
||
err = (stderr_out or "").strip() or "Unknown ffmpeg error."
|
||
self.after(0, lambda: self._set_status("Concatenation failed.", True))
|
||
self.after(0, lambda: messagebox.showerror("FFmpeg error", err))
|
||
self.after(0, lambda: self._finish(False))
|
||
return
|
||
|
||
# Success
|
||
elapsed_total = time.time() - t0
|
||
try:
|
||
size_mb = Path(out_file).stat().st_size / (1024 * 1024)
|
||
except Exception:
|
||
size_mb = 0.0
|
||
|
||
self.after(0, lambda: self.progress.set(1.0))
|
||
self.after(0, lambda: self.pct_lbl.configure(text="100%"))
|
||
self.after(0, lambda: self._set_status("✓ Done!"))
|
||
self.after(0, lambda: self._set_timing(
|
||
f"Finished in {_fmt_elapsed(elapsed_total)} · {size_mb:.1f} MB"
|
||
))
|
||
self.after(0, lambda: messagebox.showinfo(
|
||
"Concatenation complete",
|
||
f"Saved to:\n{out_file}\n\n"
|
||
f"Size: {size_mb:.1f} MB | Time taken: {_fmt_elapsed(elapsed_total)}"
|
||
))
|
||
|
||
if self.del_orig_var.get():
|
||
deleted = sum(1 for f in files if self._try_delete(f))
|
||
if deleted:
|
||
self.after(0, lambda d=deleted: messagebox.showinfo(
|
||
"Originals deleted", f"{d} original file(s) removed."
|
||
))
|
||
|
||
self.after(0, lambda: self._finish(True))
|
||
|
||
@staticmethod
|
||
def _try_delete(path: str) -> bool:
|
||
try:
|
||
os.remove(path)
|
||
return True
|
||
except OSError:
|
||
return False
|
||
|
||
def _finish(self, _ok: bool):
|
||
self._proc = None
|
||
self._stop_req = False
|
||
self.stop_btn.configure(state="disabled")
|
||
self._refresh_out_label()
|
||
self._refresh_start()
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════════════════════════
|
||
|
||
if __name__ == "__main__":
|
||
app = App()
|
||
app.mainloop()
|