[Asm] 纯文本查看 复制代码
import os
import sys
import re
import json
import hashlib
import logging
import threading
import subprocess
import time
import urllib
from pathlib import Path
from queue import Queue, Empty
from difflib import SequenceMatcher
import tkinter as tk
from tkinter import ttk, filedialog, messagebox
import acoustid
from mutagen import File as MutagenFile
from send2trash import send2trash
# ========== ffmpeg 先找，再 import pydub ==========
HAS_FFMPEG = False
def find_ffmpeg():
global HAS_FFMPEG
# ★ 打包后 exe 所在目录（这是关键！）
if getattr(sys, 'frozen', False):
# PyInstaller 打包后的路径
base_dir = Path(sys.executable).parent
else:
# 开发环境
base_dir = Path(__file__).parent
possible = [
base_dir / "fpcalc.exe", # ★ exe同级目录（打包后首选）
base_dir / "ffmpeg.exe", # ffmpeg 也可能放这里
Path(__file__).parent / "fpcalc.exe",
Path("H:/测试文件夹/ffmpeg.exe"),
Path("H:/测试文件夹2/ffmpeg.exe"),
Path("C:/ffmpeg/bin/ffmpeg.exe"),
Path.home() / "ffmpeg" / "bin" / "ffmpeg.exe",
]
for p in possible:
if p.exists():
HAS_FFMPEG = True
AudioSegment.converter = str(p)
return str(p)
try:
subprocess.run(["ffmpeg", "-version"], capture_output=True, timeout=2)
HAS_FFMPEG = True
return "ffmpeg"
except:
HAS_FFMPEG = False
return None
from pydub import AudioSegment
# ========== 屏蔽无关警告 ==========
logging.getLogger('mutagen').setLevel(logging.ERROR)
logging.getLogger('pydub').setLevel(logging.ERROR)
import warnings
warnings.filterwarnings('ignore', category=RuntimeWarning)
# ========== pygame ==========
import pygame
pygame.mixer.init(frequency=44100, size=-16, channels=2, buffer=4096)
# ==================== 常量 ====================
SUPPORTED_FORMATS = {'.mp3', '.flac', '.wav', '.aac', '.m4a', '.ogg', '.ape', '.wma'}
CACHE_FILE = Path.home() / '.music_dedup_cache.json'
FINGERPRINT_THRESHOLD = 0.85
FILENAME_SIM_THRESHOLD = 0.7
AI_CONFIDENCE_THRESHOLD = 0.98 # ★ 98%以上才采纳AI结果
# ==================== 工具函数 ====================
def calc_file_hash(filepath: Path, algo='sha256') -> str:
h = hashlib.new(algo)
with open(filepath, 'rb') as f:
while chunk := f.read(8192):
h.update(chunk)
return h.hexdigest()
def extract_real_song_name(filename: str) -> str:
name = Path(filename).stem
for pattern in [
r'\s*-\s*(副本|复件|copy|duplicate|copie)\s*$',
r'\s*[(_\[]\s*\d+\s*[)\]]\s*$',
r'\s*-\s*\d+\s*$',
r'\s*\(live\)\s*$', r'\s*\(Live\)\s*$',
]:
name = re.sub(pattern, '', name, flags=re.IGNORECASE)
name = re.sub(r'^\d+[\.\-_\s]+', '', name)
name = re.sub(r'[＿\s]+', '-', name)
name = re.sub(r'[－—]', '-', name)
name = re.sub(r'\s*\(DJ版\)\s*$', ' (DJ版)', name, flags=re.IGNORECASE)
name = re.sub(r'\s*\(Remix\)\s*$', ' (Remix)', name, flags=re.IGNORECASE)
name = re.sub(r'\s*\(翻唱\)\s*$', ' (翻唱)', name, flags=re.IGNORECASE)
parts = [p.strip() for p in re.split(r'[-\s_]+', name) if p.strip()]
if len(parts) <= 1:
return parts[0] if parts else name
if len(parts) == 2:
a, b = parts[0], parts[1]
def is_artist(s):
return bool(re.fullmatch(r'^[\u4e00-\u9fa5]{1,4}$', s)) or bool(re.fullmatch(r'^[A-Za-z]{2,15}$', s))
if is_artist(a) and not is_artist(b):
return b
if is_artist(b) and not is_artist(a):
return a
return a if len(a) >= len(b) else b
rem = parts[1:]
if len(rem) == 2:
a, b = rem[0], rem[1]
if len(a) <= 4 and len(b) > 4:
return b
if len(b) <= 4 and len(a) > 4:
return a
return a if len(a) >= len(b) else b
return max(rem, key=len)
def get_audio_info(filepath: Path) -> dict:
info = {'path': str(filepath),'size': filepath.stat().st_size,'format': filepath.suffix.lower(),
'duration':0,'bitrate':0,'title':'','artist':'','real_song_name':'','ai_song_name':'',
'ai_confidence':0.0,'ai_reason':'','ai_used':False}
try:
audio = MutagenFile(filepath, easy=True)
if audio is not None:
info['duration'] = int(audio.info.length) if hasattr(audio.info,'length') else 0
info['bitrate'] = getattr(audio.info,'bitrate',0) or 0
info['title'] = audio.get('title',[''])[0] if audio.get('title') else ''
info['artist'] = audio.get('artist',[''])[0] if audio.get('artist') else ''
except:
pass
# 先用规则引擎提取一个兜底歌名
info['real_song_name'] = info['title'] if info['title'] else extract_real_song_name(filepath.name)
return info
def filename_similarity(a,b):
return SequenceMatcher(None,a.lower(),b.lower()).ratio()
def format_time(s):
return f"{s//60}:{s%60:02d}"
def play_external(p):
try:
if sys.platform == 'win32':
os.startfile(p)
elif sys.platform == 'darwin':
subprocess.run(['open', p])
else:
subprocess.run(['xdg-open', p])
except:
pass
# ==================== AI 歌名识别（重识别逻辑）====================
class AISongNameChecker:
"""
1. AI 直接从文件名中识别纯净歌名（不含歌手、编号等）
2. 返回歌名 + 置信度
3. 置信度 ≥ 0.98 才采纳AI结果，否则保持原规则引擎结果
4. 兼容标准模型和推理模型（content / reasoning 双字段）
"""
SYSTEM_PROMPT = """你是一个专业的音乐文件名解析专家。你的任务只有一个：
**从给定的文件名中，识别出纯净的歌曲名称。**
规则：
1. 只输出歌曲名，不要包含歌手、乐队、编号、专辑、比特率、年份等任何额外信息
2. 不要包含"副本""copy""- 副本"等文件标记
3. 保留歌曲名本身的修饰，如"(DJ版)""(Remix)""(Live)""(翻唱)"
4. 如果文件名明显包含多个信息（如"编号-歌名-歌手"），只取歌名部分
5. 如果无法确定歌名，返回空字符串 ""
输出格式（严格JSON，不要任何多余文字）：
{"song_name": "歌名", "confidence": 0.99, "reason": "判断理由"}
confidence 说明：
- 0.98~1.0：非常确定，文件名结构清晰，歌名明确
- 0.90~0.97：比较确定，但有轻微歧义
- 0.70~0.89：不太确定，文件名混乱
- <0.70：无法确定
示例：
文件名：0001-38度6-黑龙.wav
→ {"song_name": "38度6", "confidence": 0.99, "reason": "编号-歌名-歌手结构清晰"}
文件名：阿果吉曲-海来阿木 - 副本.flac
→ {"song_name": "阿果吉曲", "confidence": 0.98, "reason": "歌名-歌手结构，去除副本标记"}
文件名：乱七八糟的未知文件.mp3
→ {"song_name": "", "confidence": 0.30, "reason": "文件名无明确歌名结构"}"""
def __init__(self):
self.config_file = Path(__file__).parent / 'config.json'
self.config = self._load()
def _load(self):
default = {
"api_key": "",
"base_url": "https://api.openai.com/v1",
"model": "gpt-3.5-turbo",
"timeout": 60,
"max_tokens": 1000,
"temperature": 0.0, # ★ 歌名识别用 0，最稳定
"reasoning_effort": "low"
}
if self.config_file.exists():
try:
default.update(json.loads(self.config_file.read_text(encoding='utf-8')))
except:
pass
return default
def save(self):
self.config_file.write_text(
json.dumps(self.config, ensure_ascii=False, indent=2),
encoding='utf-8'
)
def is_configured(self):
return bool(self.config.get("api_key"))
def _build_payload(self, messages):
payload = {
"model": self.config["model"],
"messages": messages,
"temperature": float(self.config.get("temperature", 0.0)),
"max_tokens": int(self.config.get("max_tokens", 1000)),
}
if self.config.get("reasoning_effort"):
payload["reasoning_effort"] = self.config["reasoning_effort"]
payload["response_format"] = {"type": "json_object"}
return payload
def _make_request(self, messages):
payload = self._build_payload(messages)
req = urllib.request.Request(
f"{self.config['base_url'].rstrip('/')}/chat/completions",
data=json.dumps(payload).encode('utf-8'),
headers={
"Content-Type": "application/json",
"Authorization": f"Bearer {self.config['api_key']}"
}
)
try:
with urllib.request.urlopen(req, timeout=int(self.config.get("timeout", 60))) as resp:
return True, json.loads(resp.read().decode('utf-8'))
except urllib.error.HTTPError as e:
try:
body = json.loads(e.read().decode('utf-8', errors='ignore'))
err_msg = body.get('error', {})
if isinstance(err_msg, dict):
err_msg = err_msg.get('message', str(body))
return False, f"HTTP {e.code}: {err_msg}"
except:
return False, f"HTTP {e.code}: {e.reason}"
except Exception as e:
return False, str(e)
def _extract_text(self, message_dict):
if not isinstance(message_dict, dict):
return None
text = message_dict.get("content")
if text and text.strip():
return text.strip()
text = message_dict.get("reasoning")
if text and text.strip():
return text.strip()
text = message_dict.get("text")
if text and text.strip():
return text.strip()
return None
def _parse_json(self, text):
if not text:
return None
# 直接解析
try:
return json.loads(text)
except:
pass
# 提取 ```json ``` 块
m = re.search(r'```(?:json)?\s*(\{.*?\})\s*```', text, re.DOTALL)
if m:
try:
return json.loads(m.group(1))
except:
pass
# 找第一个 { 到最后一个 }
m = re.search(r'\{.*\}', text, re.DOTALL)
if m:
try:
return json.loads(m.group(0))
except:
pass
return None
def test_connection(self):
if not self.is_configured():
return False, "API Key 未配置"
messages = [
{"role": "system", "content": self.SYSTEM_PROMPT},
{"role": "user", "content": "文件名：0001-38度6-黑龙.wav\n\n请按JSON格式返回。"}
]
success, result = self._make_request(messages)
if not success:
return False, f"连接失败：{result}"
try:
choice = result["choices"][0]["message"]
text = self._extract_text(choice)
if not text:
return False, f"返回无文本。原始响应：{json.dumps(result, ensure_ascii=False)[:500]}"
parsed = self._parse_json(text)
if not parsed or "song_name" not in parsed:
# 非JSON，看能不能当歌名
clean = text.strip().strip('"\'')
return True, f"✅ 连接成功（非JSON模式）\n模型返回：{clean[:50]}"
return True, (f"✅ 连接成功！\n"
f"识别歌名：{parsed.get('song_name','')}\n"
f"置信度：{parsed.get('confidence','')}\n"
f"原因：{parsed.get('reason','')}")
except Exception as e:
return False, f"解析失败：{e}"
def recognize_song_name(self, filenames):
"""
核心方法：给一组文件名，让AI识别歌名
返回：(song_name, confidence, reason) 或 None
"""
if not self.is_configured():
return None
lines = ["请从以下文件名中识别纯净的歌曲名称：", ""]
for i, fname in enumerate(filenames, 1):
lines.append(f"{i}. {fname}")
lines.append("")
lines.append("这些文件名指向同一首歌，请只输出一个统一的歌名（JSON格式）。")
user_content = "\n".join(lines)
messages = [
{"role": "system", "content": self.SYSTEM_PROMPT},
{"role": "user", "content": user_content}
]
success, result = self._make_request(messages)
if not success:
print(f"AI识别失败: {result}")
return None
try:
choice = result["choices"][0]["message"]
text = self._extract_text(choice)
if not text:
print("AI返回空文本")
return None
parsed = self._parse_json(text)
if not parsed:
print(f"AI返回无法解析的JSON: {text[:200]}")
return None
song_name = str(parsed.get("song_name", "")).strip()
try:
confidence = float(parsed.get("confidence", 0))
except:
confidence = 0.0
reason = str(parsed.get("reason", "")).strip()
if not song_name:
print(f"AI认为无法确定歌名: {reason}")
return None
return song_name, confidence, reason
except Exception as e:
print(f"AI识别异常: {e}")
return None
# ==================== 播放器引擎 ====================
class AudioPlayer:
def __init__(self):
self.current_file = None
self.full_audio = None
self.total_duration = 0
self.offset_ms = 0
self.play_start_time = 0
self.is_playing = False
self.is_paused = False
self.on_progress = None
self.on_finish = None
self._monitor = None
self._stop_evt = threading.Event()
def get_current_pos(self) -> int:
if self.is_playing and not self.is_paused:
elapsed = (time.time() - self.play_start_time) * 1000
return int(self.offset_ms + elapsed)
return int(self.offset_ms)
def can_play_internal(self, p: Path) -> bool:
ext = p.suffix.lower()
if ext in ('.mp3','.wav','.ogg'):
return True
if ext == '.flac' and HAS_FFMPEG:
return True
return False
def load(self, filepath: Path) -> bool:
self.stop()
self.current_file = filepath
self.offset_ms = 0
try:
ext = filepath.suffix.lower()
if ext in ('.mp3','.wav','.ogg'):
pygame.mixer.music.load(str(filepath))
info = get_audio_info(filepath)
self.total_duration = info['duration'] * 1000
return True
elif ext == '.flac' and HAS_FFMPEG:
self.full_audio = AudioSegment.from_file(str(filepath))
self.total_duration = len(self.full_audio)
chunk = self.full_audio[:10000]
tmp = Path.home() / ".music_dedup_temp.wav"
chunk.export(str(tmp), format="wav")
pygame.mixer.music.load(str(tmp))
return True
return False
except:
return False
def play(self, start_ms: int = 0):
if self.is_paused and start_ms == 0:
pygame.mixer.music.unpause()
self.is_paused = False
self.play_start_time = time.time()
self._start_monitor()
return True
self.stop()
self.offset_ms = start_ms
self.play_start_time = time.time()
try:
ext = self.current_file.suffix.lower()
if ext in ('.mp3','.wav','.ogg'):
pygame.mixer.music.play(start=start_ms/1000.0)
elif ext == '.flac' and self.full_audio is not None:
chunk = self.full_audio[start_ms:]
tmp = Path.home() / ".music_dedup_temp.wav"
chunk.export(str(tmp), format="wav")
pygame.mixer.music.load(str(tmp))
pygame.mixer.music.play()
self.is_playing = True
self.is_paused = False
self._start_monitor()
return True
except:
return False
def pause(self):
if self.is_playing and not self.is_paused:
pygame.mixer.music.pause()
self.offset_ms = self.get_current_pos()
self.is_paused = True
return True
return False
def resume(self):
if self.is_paused:
pygame.mixer.music.unpause()
self.is_paused = False
self.play_start_time = time.time()
self._start_monitor()
return True
return False
def stop(self):
self._stop_evt.set()
pygame.mixer.music.stop()
self.is_playing = False
self.is_paused = False
def toggle(self):
if self.is_playing and not self.is_paused:
return self.pause()
elif self.is_paused:
return self.resume()
else:
return self.play()
def seek(self, ms: int):
ms = max(0, min(ms, self.total_duration))
self.play(start_ms=ms)
def _start_monitor(self):
self._stop_evt.clear()
if self._monitor and self._monitor.is_alive():
return
def loop():
finish_fired = False
while not self._stop_evt.is_set() and self.is_playing:
if not self.is_paused:
cur = self.get_current_pos()
tot = self.total_duration
if self.on_progress:
self.on_progress(cur, tot)
if cur >= tot - 500:
if not finish_fired and self.on_finish:
finish_fired = True
self.on_finish()
break
time.sleep(0.1)
self._monitor = threading.Thread(target=loop, daemon=True)
self._monitor.start()
# ==================== 去重引擎（核心逻辑修改）====================
class DedupEngine:
def __init__(self, cache_file=CACHE_FILE):
self.cache_file = cache_file
self.cache = self._load()
self.file_infos = {}
self.dupe_groups = []
self.fpcalc_path = None
self.ai_checker = AISongNameChecker()
def _load(self):
if self.cache_file.exists():
try:
return json.loads(self.cache_file.read_text(encoding='utf-8'))
except:
pass
return {}
def _save(self):
self.cache_file.write_text(json.dumps(self.cache, ensure_ascii=False, indent=2), encoding='utf-8')
def find_fpcalc(self, d=None):
possible = [Path(__file__).parent / "fpcalc.exe"]
if d:
possible.append(d / "fpcalc.exe")
try:
r = subprocess.run(['where','fpcalc'], capture_output=True, text=True, shell=True, timeout=2)
if r.returncode == 0:
possible += [Path(x.strip()) for x in r.stdout.split('\n') if x.strip()]
except:
pass
possible.append(Path("C:/Windows/System32/fpcalc.exe"))
for p in possible:
if p.exists():
try:
subprocess.run([str(p),'-version'], capture_output=True, check=True, timeout=2)
return str(p)
except:
continue
return None
def scan(self, folder, cb=None):
files = []
all_files = list(folder.rglob('*'))
total = len(all_files)
for i, p in enumerate(all_files):
if p.is_file() and p.suffix.lower() in SUPPORTED_FORMATS:
files.append(p)
self.file_infos[p.resolve()] = get_audio_info(p)
if cb and i % 50 == 0:
cb(i / total * 0.3, f"扫描 {i}/{total}")
return files
def by_name(self, files, cb=None):
groups, visited = [], set()
total = len(files)
for i, p1 in enumerate(files):
if p1 in visited:
continue
g = [p1]
s1 = self.file_infos[p1.resolve()]['real_song_name']
for p2 in files[i+1:]:
if p2 in visited:
continue
s2 = self.file_infos[p2.resolve()]['real_song_name']
if not s1 or not s2:
continue
if s1 == s2 or filename_similarity(s1, s2) >= FILENAME_SIM_THRESHOLD:
g.append(p2)
visited.add(p2)
if len(g) > 1:
groups.append(g)
visited.add(p1)
if cb:
cb(0.3 + 0.3 * i / total, f"比对 {i}/{total}")
return groups
def fingerprint(self, p: Path, cache: dict) -> str:
st = p.stat()
key = f"{p.resolve()}|{st.st_size}|{st.st_mtime}"
if key in cache:
return cache[key]
if self.fpcalc_path:
try:
env = os.environ.copy()
env['PATH'] = str(Path(self.fpcalc_path).parent) + os.pathsep + env.get('PATH', '')
_, fp = acoustid.fingerprint_file(str(p))
if fp:
cache[key] = fp
return fp
except:
pass
h = calc_file_hash(p)
cache[key] = h
return h
def by_audio(self, files, cb=None):
fps, groups, visited = {}, [], set()
total = len(files)
for i, p in enumerate(files):
fp = self.fingerprint(p.resolve(), self.cache)
if fp:
fps[p] = fp
if cb:
cb(0.6 + 0.3 * i / total, f"指纹 {i}/{total}")
items = list(fps.items())
item_count = len(items)
for i, (p1, fp1) in enumerate(items):
if p1 in visited:
continue
g = [p1]
for j in range(i+1, item_count):
p2, fp2 = items[j]
if p2 in visited:
continue
if len(fp1) < 200 and len(fp2) < 200:
if fp1 == fp2:
g.append(p2)
visited.add(p2)
elif SequenceMatcher(None, fp1, fp2).ratio() >= FINGERPRINT_THRESHOLD:
g.append(p2)
visited.add(p2)
if len(g) > 1:
groups.append(g)
visited.add(p1)
if cb:
cb(0.9 + 0.1 * i / item_count, f"比对 {i}/{item_count}")
self._save()
return groups
def run(self, folder, mode, use_ai=False, cb=None):
self.fpcalc_path = self.find_fpcalc(folder)
cb and cb(0, "✅ fpcalc已找到" if self.fpcalc_path else "⚠️ 用哈希代替")
files = self.scan(folder, cb)
if mode == 'filename':
self.dupe_groups = self.by_name(files, cb)
elif mode == 'audio':
self.dupe_groups = self.by_audio(files, cb)
else:
ng = self.by_name(files, cb)
flat = [p for g in ng for p in g] or files
self.dupe_groups = self.by_audio(flat, cb)
# ★★★ AI 歌名重识别阶段 ★★★
if use_ai and self.ai_checker.is_configured():
total = len(self.dupe_groups)
for i, group in enumerate(self.dupe_groups):
cb and cb(0.95 + 0.05 * (i / total), f"AI识别歌名 {i+1}/{total}")
# 收集组内所有文件名
filenames = [p.name for p in group]
# 调用AI识别
ai_result = self.ai_checker.recognize_song_name(filenames)
if ai_result:
ai_name, confidence, reason = ai_result
# ★ 核心逻辑：只有置信度 ≥ 98% 才采纳
if confidence >= AI_CONFIDENCE_THRESHOLD:
for p in group:
orig_name = self.file_infos[p.resolve()]['real_song_name']
self.file_infos[p.resolve()]['ai_song_name'] = ai_name
self.file_infos[p.resolve()]['real_song_name'] = ai_name # ★ 替换原歌名
self.file_infos[p.resolve()]['ai_confidence'] = confidence
self.file_infos[p.resolve()]['ai_reason'] = reason
self.file_infos[p.resolve()]['ai_used'] = True
print(f"✅ AI替换歌名: {orig_name} → {ai_name} (置信度:{confidence:.0%})")
else:
# 置信度不足，保持原规则引擎结果
for p in group:
self.file_infos[p.resolve()]['ai_song_name'] = ai_name
self.file_infos[p.resolve()]['ai_confidence'] = confidence
self.file_infos[p.resolve()]['ai_reason'] = reason
self.file_infos[p.resolve()]['ai_used'] = False
print(f"⚠️ AI置信度不足({confidence:.0%})，保持原歌名: {filenames[0]}")
cb and cb(1.0, f"完成! {len(self.dupe_groups)}组")
# ==================== UI ====================
class App:
def __init__(self, root):
self.root = root
self.root.title("音乐去重工具 v2.0（AI歌名重识别版）")
self.root.geometry("1380x850")
self.engine = DedupEngine()
self.player = AudioPlayer()
self.selected_group = None
self.group_idx = 0
self.checks = {}
self.scan_mode = tk.StringVar(value="filename")
self.auto_non_flac = tk.BooleanVar(value=False)
self.auto_best = tk.BooleanVar(value=True)
self.use_ai = tk.BooleanVar(value=False)
self.queue = Queue()
self.playing_path = None
self._build()
self._poll()
self.player.on_progress = self._on_progress
self.player.on_finish = self._on_finish
if HAS_FFMPEG:
self.status.config(text="✅ ffmpeg已找到，支持FLAC")
else:
self.status.config(text="⚠️ 无ffmpeg，FLAC用外部播放器")
def _build(self):
tb = ttk.Frame(self.root, padding=5)
tb.pack(fill='x')
ttk.Button(tb, text="选择文件夹", command=self._sel_dir).pack(side='left', padx=5)
self.path_lbl = ttk.Label(tb, text="未选择文件夹", foreground='gray')
self.path_lbl.pack(side='left', padx=5, fill='x', expand=True)
for txt, val in [("文件名去重","filename"),("音频指纹去重","audio"),("两者结合","both")]:
ttk.Radiobutton(tb, text=txt, variable=self.scan_mode, value=val).pack(side='left', padx=5)
ttk.Button(tb, text="开始扫描", command=self._scan).pack(side='left', padx=5)
ttk.Button(tb, text="删除选中", command=self._delete).pack(side='left', padx=5)
ttk.Button(tb, text="清空结果", command=self._clear).pack(side='left', padx=5)
# AI区域
ai_frame = ttk.LabelFrame(tb, text="AI歌名识别(≥98%才采纳)", padding=2)
ai_frame.pack(side='left', padx=10)
ttk.Checkbutton(ai_frame, text="🤖 启用AI识别", variable=self.use_ai).pack(side='left', padx=5)
ttk.Button(ai_frame, text="⚙️ 配置", command=self._ai_config).pack(side='left', padx=2)
ttk.Button(ai_frame, text="🔌 测试", command=self._test_ai).pack(side='left', padx=2)
self.pvar = tk.DoubleVar()
ttk.Progressbar(self.root, variable=self.pvar, maximum=1.0).pack(fill='x', padx=10, pady=5)
self.status = ttk.Label(self.root, text="就绪")
self.status.pack(anchor='w', padx=10)
paned = ttk.PanedWindow(self.root, orient='horizontal')
paned.pack(fill='both', expand=True, padx=10, pady=5)
left = ttk.Frame(paned)
paned.add(left, weight=1)
ttk.Label(left, text="重复组列表", font=('微软雅黑',10,'bold')).pack(anchor='w')
self.gtree = ttk.Treeview(left, columns=('song','n','ai','conf'), show='headings')
self.gtree.heading('song', text='歌曲名')
self.gtree.heading('n', text='重复数')
self.gtree.heading('ai', text='AI')
self.gtree.heading('conf', text='置信度')
self.gtree.column('song', width=160)
self.gtree.column('n', width=55, anchor='center')
self.gtree.column('ai', width=45, anchor='center')
self.gtree.column('conf', width=70, anchor='center')
self.gtree.pack(fill='both', expand=True, pady=5)
self.gtree.bind('<<TreeviewSelect>>', self._on_group)
right = ttk.Frame(paned)
paned.add(right, weight=2)
ttk.Label(right, text="文件详情（双击播放，点击☑️切换删除）", font=('微软雅黑',10,'bold')).pack(anchor='w')
self.ftree = ttk.Treeview(right, columns=('chk','fmt','orig','final','size','dur','reason','path'), show='headings')
for c, t, w in [
('chk','删除',50),('fmt','格式',60),('orig','原歌名',130),
('final','最终歌名',130),('size','大小',70),('dur','时长',60),
('reason','AI判断',140),('path','路径',280)
]:
self.ftree.heading(c, text=t)
self.ftree.column(c, width=w, anchor='center' if c not in ('orig','final','reason','path') else 'w')
self.ftree.pack(fill='both', expand=True, pady=5)
self.ftree.bind('<Double-1>', self._on_dclick)
self.ftree.bind('<Button-1>', self._on_click)
pf = ttk.LabelFrame(self.root, text="🎵 内置播放器", padding=10)
pf.pack(fill='x', padx=10, pady=5)
rc = ttk.Frame(pf)
rc.pack(fill='x', pady=(0,5))
self.now_lbl = ttk.Label(rc, text="未播放", foreground='gray')
self.now_lbl.pack(side='left', padx=5)
ttk.Button(rc, text="⏮ 上一曲", command=self._prev).pack(side='left', padx=2)
self.play_btn = ttk.Button(rc, text="▶️ 播放", command=self._toggle)
self.play_btn.pack(side='left', padx=2)
ttk.Button(rc, text="⏹ 停止", command=self._stop).pack(side='left', padx=2)
ttk.Button(rc, text="⏭ 下一曲", command=self._next).pack(side='left', padx=2)
self.pos_lbl = ttk.Label(rc, text="", foreground='green')
self.pos_lbl.pack(side='left', padx=15)
sc = ttk.Frame(pf)
sc.pack(fill='x')
self.t_cur = ttk.Label(sc, text="00:00")
self.t_cur.pack(side='left', padx=5)
self.scale = ttk.Scale(sc, from_=0, to=100, orient='horizontal')
self.scale.pack(side='left', fill='x', expand=True, padx=5)
self.scale.bind('<Motion>', self._on_drag)
self.scale.bind('<ButtonRelease-1>', self._on_release)
self.t_tot = ttk.Label(sc, text="00:00")
self.t_tot.pack(side='left', padx=5)
rule = ttk.LabelFrame(self.root, text="自动删除规则", padding=10)
rule.pack(fill='x', padx=10, pady=5)
ttk.Checkbutton(rule, text="自动勾选非FLAC", variable=self.auto_non_flac, command=self._rules).pack(side='left', padx=10)
ttk.Checkbutton(rule, text="自动保留最大文件", variable=self.auto_best, command=self._rules).pack(side='left', padx=10)
ttk.Label(rule, text="💡 AI置信度≥98%才替换原歌名，否则保持规则引擎结果", foreground='blue').pack(side='left', padx=20)
ttk.Label(rule, text="⚠️ 删除的文件会移到回收站", foreground='orange').pack(side='right')
def _poll(self):
try:
while True:
p, s = self.queue.get_nowait()
self.pvar.set(p)
self.status.config(text=s)
if p >= 1.0:
self._on_done()
except Empty:
pass
self.root.after(100, self._poll)
def _sel_dir(self):
f = filedialog.askdirectory()
if f:
self.folder = Path(f)
self.path_lbl.config(text=str(f), foreground='black')
# ---------- AI配置 ----------
def _ai_config(self):
dlg = tk.Toplevel(self.root)
dlg.title("AI歌名识别配置")
dlg.geometry("560x340")
dlg.resizable(False, False)
dlg.transient(self.root)
dlg.grab_set()
frm = ttk.Frame(dlg, padding=15)
frm.pack(fill='both', expand=True)
checker = self.engine.ai_checker
ttk.Label(frm, text="API Key：").grid(row=0, column=0, sticky='w', pady=5)
key_var = tk.StringVar(value=checker.config["api_key"])
ttk.Entry(frm, textvariable=key_var, width=46, show="*").grid(row=0, column=1, pady=5)
ttk.Label(frm, text="Base URL：").grid(row=1, column=0, sticky='w', pady=5)
url_var = tk.StringVar(value=checker.config["base_url"])
ttk.Entry(frm, textvariable=url_var, width=46).grid(row=1, column=1, pady=5)
ttk.Label(frm, text="模型名称：").grid(row=2, column=0, sticky='w', pady=5)
model_var = tk.StringVar(value=checker.config["model"])
ttk.Entry(frm, textvariable=model_var, width=46).grid(row=2, column=1, pady=5)
ttk.Label(frm, text="最大Token：").grid(row=3, column=0, sticky='w', pady=5)
max_var = tk.StringVar(value=str(checker.config.get("max_tokens", 1000)))
ttk.Entry(frm, textvariable=max_var, width=46).grid(row=3, column=1, pady=5)
ttk.Label(frm, text="推理深度：").grid(row=4, column=0, sticky='w', pady=5)
effort_var = tk.StringVar(value=checker.config.get("reasoning_effort", "low"))
ttk.Combobox(frm, textvariable=effort_var, width=43,
values=["none","low","medium","high"]).grid(row=4, column=1, pady=5)
ttk.Label(frm, text="温度(建议0)：").grid(row=5, column=0, sticky='w', pady=5)
temp_var = tk.StringVar(value=str(checker.config.get("temperature", 0.0)))
ttk.Entry(frm, textvariable=temp_var, width=46).grid(row=5, column=1, pady=5)
hint = ("歌名识别建议：temperature=0, reasoning_effort=low\n"
"置信度≥98%才会替换原歌名，低于则保持规则引擎结果")
ttk.Label(frm, text=hint, foreground='gray', justify='left').grid(
row=6, column=0, columnspan=2, sticky='w', pady=10)
btn_frm = ttk.Frame(frm)
btn_frm.grid(row=7, column=0, columnspan=2, pady=10)
ttk.Button(btn_frm, text="保存", command=lambda: self._save_ai_config(
checker, key_var.get(), url_var.get(), model_var.get(),
effort_var.get(), max_var.get(), temp_var.get(), dlg)).pack(side='left', padx=20)
ttk.Button(btn_frm, text="取消", command=dlg.destroy).pack(side='left', padx=20)
def _save_ai_config(self, checker, key, url, model, effort, max_tok, temp, dlg):
checker.config["api_key"] = key.strip()
checker.config["base_url"] = url.strip()
checker.config["model"] = model.strip()
checker.config["reasoning_effort"] = effort.strip() if effort else "none"
try:
checker.config["max_tokens"] = int(max_tok)
except:
checker.config["max_tokens"] = 1000
try:
checker.config["temperature"] = float(temp)
except:
checker.config["temperature"] = 0.0
checker.save()
messagebox.showinfo("成功", "AI配置已保存")
dlg.destroy()
# ---------- 测试AI ----------
def _test_ai(self):
if not self.engine.ai_checker.is_configured():
messagebox.showwarning("提示", "请先配置API Key")
return
def worker():
success, msg = self.engine.ai_checker.test_connection()
self.root.after(0, lambda: self._show_test_result(success, msg))
self.status.config(text="正在测试AI连通性...")
threading.Thread(target=worker, daemon=True).start()
def _show_test_result(self, success, msg):
if success:
messagebox.showinfo("AI测试成功", msg)
self.status.config(text="✅ AI连通正常")
else:
messagebox.showerror("AI测试失败", msg)
self.status.config(text="❌ AI连通失败")
# ---------- 扫描 ----------
def _scan(self):
if not hasattr(self, 'folder'):
messagebox.showwarning("提示", "先选文件夹")
return
if self.use_ai.get() and not self.engine.ai_checker.is_configured():
if messagebox.askyesno("AI未配置", "AI歌名识别需要配置API Key，是否现在配置？"):
self._ai_config()
return
self._clear()
threading.Thread(target=self._worker, daemon=True).start()
def _worker(self):
self.engine.run(
self.folder,
self.scan_mode.get(),
use_ai=self.use_ai.get(),
cb=lambda p, s: self.queue.put((p, s))
)
def _on_done(self):
self._fill_groups()
msg = f"找到 {len(self.engine.dupe_groups)} 个重复组"
if self.use_ai.get() and self.engine.ai_checker.is_configured():
msg += "（AI已参与歌名识别）"
messagebox.showinfo("完成", msg)
# ---------- 左侧列表 ----------
def _fill_groups(self):
self.gtree.delete(*self.gtree.get_children())
for i, g in enumerate(self.engine.dupe_groups):
info = self.engine.file_infos.get(g[0].resolve(), {})
name = info.get('real_song_name', f'组{i+1}')
ai_tag = "✅" if info.get('ai_used') else "❌"
conf = info.get('ai_confidence', 0)
conf_str = f"{conf:.0%}" if conf else "--"
self.gtree.insert('', 'end', iid=f'g{i}', values=(name, len(g), ai_tag, conf_str))
self.gtree.item(f'g{i}', tags=(','.join(str(x) for x in g),))
def _on_group(self, e):
sel = self.gtree.selection()
if not sel:
return
paths = [Path(x) for x in self.gtree.item(sel[0], 'tags')[0].split(',')]
self.selected_group = paths
self.group_idx = 0
self._fill_files(paths)
self._rules()
def _fill_files(self, paths):
self.ftree.delete(*self.ftree.get_children())
self.checks.clear()
for p in paths:
info = self.engine.file_infos.get(p.resolve(), {})
sz = info.get('size', 0) / 1024 / 1024
dur = info.get('duration', 0)
br = info.get('bitrate', 0) // 1000 if info.get('bitrate') else 0
orig = extract_real_song_name(p.name) # 原规则引擎结果
final = info.get('real_song_name', orig) # 最终使用的歌名
reason = info.get('ai_reason', '')
self.ftree.insert('', 'end', iid=str(p), values=(
'☐',
info.get('format', '').upper()[1:],
orig,
final,
f'{sz:.1f}M',
format_time(dur) if dur else '--',
reason,
str(p)
))
self.checks[str(p)] = False
def _on_click(self, e):
r = self.ftree.identify_region(e.x, e.y)
if r != 'cell':
return
col = self.ftree.identify_column(e.x)
iid = self.ftree.identify_row(e.y)
if col == '#1' and iid:
self.checks[iid] = not self.checks.get(iid, False)
self.ftree.set(iid, 'chk', '☑️' if self.checks[iid] else '☐')
def _on_dclick(self, e):
iid = self.ftree.identify_row(e.y)
if not iid:
return
self._stop()
self.playing_path = Path(iid)
self._highlight(iid)
if self.player.can_play_internal(self.playing_path):
if self.player.load(self.playing_path):
info = self.engine.file_infos.get(self.playing_path.resolve(), {})
self.now_lbl.config(text=f"▶️ {info.get('real_song_name','')}", foreground='black')
self.t_tot.config(text=format_time(self.player.total_duration//1000))
self.play_btn.config(text="⏸ 暂停")
self.player.play()
else:
self.status.config(text="⚠️ 内置播放失败→外部")
play_external(self.playing_path)
else:
play_external(self.playing_path)
def _toggle(self):
if not self.playing_path:
return
self.player.toggle()
self.play_btn.config(text="⏸ 暂停" if self.player.is_playing and not self.player.is_paused else "▶️ 播放")
def _stop(self):
self.player.stop()
self.play_btn.config(text="▶️ 播放")
self.scale.set(0)
self.t_cur.config(text="00:00")
self.pos_lbl.config(text="")
def _prev(self):
if not self.selected_group:
return
self.group_idx = (self.group_idx - 1) % len(self.selected_group)
self._play_idx(self.group_idx)
def _next(self):
if not self.selected_group:
return
self.group_idx = (self.group_idx + 1) % len(self.selected_group)
self._play_idx(self.group_idx)
def _play_idx(self, idx):
p = self.selected_group[idx]
self._stop()
self.playing_path = p
self._highlight(str(p))
if self.player.load(p):
info = self.engine.file_infos.get(p.resolve(), {})
self.now_lbl.config(text=f"▶️ {info.get('real_song_name','')}", foreground='black')
self.t_tot.config(text=format_time(self.player.total_duration//1000))
self.play_btn.config(text="⏸ 暂停")
self.player.play()
def _highlight(self, iid):
for x in self.ftree.get_children():
self.ftree.item(x, tags=())
self.ftree.item(iid, tags=('p',))
self.ftree.tag_configure('p', background='#c8e6c9')
def _on_progress(self, cur_ms, tot_ms):
def upd():
if tot_ms > 0:
self.scale.set(min(100, cur_ms / tot_ms * 100))
self.t_cur.config(text=format_time(cur_ms//1000))
self.pos_lbl.config(text=f"{format_time(cur_ms//1000)} / {format_time(tot_ms//1000)}")
self.root.after(0, upd)
def _on_finish(self):
def upd():
self.play_btn.config(text="▶️ 播放")
self.scale.set(100)
self._next()
self.root.after(0, upd)
def _on_drag(self, e):
if self.player.total_duration > 0:
self.t_cur.config(text=format_time(int(self.scale.get()/100*self.player.total_duration)//1000))
def _on_release(self, e):
if self.player.total_duration > 0:
self.player.seek(int(self.scale.get()/100*self.player.total_duration))
def _rules(self):
if not self.selected_group:
return
for p in self.selected_group:
self.checks[str(p)] = False
self.ftree.set(str(p), 'chk', '☐')
if self.auto_non_flac.get():
for p in self.selected_group:
if self.engine.file_infos.get(p.resolve(), {}).get('format') != '.flac':
self.checks[str(p)] = True
self.ftree.set(str(p), 'chk', '☑️')
if self.auto_best.get():
best = max(self.selected_group, key=lambda p: (
1 if self.engine.file_infos.get(p.resolve(), {}).get('format') == '.flac' else 0,
self.engine.file_infos.get(p.resolve(), {}).get('size', 0)
))
self.checks[str(best)] = False
self.ftree.set(str(best), 'chk', '☐')
def _delete(self):
to_del = [p for p, chk in self.checks.items() if chk]
if not to_del:
messagebox.showinfo("提示", "没选文件")
return
if self.playing_path and str(self.playing_path) in to_del:
self._stop()
if not messagebox.askyesno("确认", "移到回收站？"):
return
ok, errs = 0, []
for p in to_del:
try:
send2trash(p)
ok += 1
if self.ftree.exists(p):
self.ftree.delete(p)
except Exception as e:
errs.append(f"{Path(p).name}: {e}")
self._upd_counts()
msg = f"删除 {ok} 个" + (f"，失败 {len(errs)} 个" if errs else "")
messagebox.showinfo("完成", msg)
def _upd_counts(self):
to_del = []
for gid in self.gtree.get_children():
tags = self.gtree.item(gid, 'tags')
if tags:
rem = [p for p in tags[0].split(',') if Path(p).exists()]
if rem:
self.gtree.item(gid, tags=(','.join(rem),))
self.gtree.set(gid, 'n', len(rem))
else:
to_del.append(gid)
for gid in to_del:
self.gtree.delete(gid)
def _clear(self):
self.gtree.delete(*self.gtree.get_children())
self.ftree.delete(*self.ftree.get_children())
self.checks.clear()
self.selected_group = None
self.playing_path = None
self._stop()
self.pvar.set(0)
self.status.config(text="就绪")
# ==================== 主入口 ====================
if __name__ == '__main__':
root = tk.Tk()
app = App(root)
root.mainloop()
[复制链接]
发表于 2026-7-23 14:26
分享到朋友圈
