[Asm] 纯文本查看 复制代码
import os
import re
import hashlib
import csv
import sys
import json
from datetime import datetime
from typing import List, Dict, Tuple, Optional
# 跨平台文件锁兼容
IS_WINDOWS = sys.platform.startswith("win")
if IS_WINDOWS:
import msvcrt
else:
import fcntl
# ====================================== 【用户配置区 仅需修改这里即可】======================================
# 运行模式：
# preview = 预览模式（仅扫描输出结果，不修改任何文件，安全校验用）
# exec = 执行模式（自动将重复文件移动到备份目录，无人工交互，适配定时任务）
RUN_MODE = "exec"
# 音乐根目录（支持子文件夹递归扫描）
MUSIC_ROOT_DIR = "/opt/usb1/webdav/music"
# 重复文件备份目录（按运行日期自动建子文件夹，不会直接删除文件）
BACKUP_DIR = os.path.join(MUSIC_ROOT_DIR, "_deduplicate_backup")
# 日志与报表保存目录
LOG_DIR = os.path.join(MUSIC_ROOT_DIR, "_deduplicate_log")
# 最高优先级歌手（绝对第一优先级，繁简体自动匹配）
TOP_PRIORITY_SINGER = re.compile(r"周杰伦|周杰倫|Jay Chou", re.IGNORECASE)
# 次优先级白名单歌手（除周杰伦外，优先保留的歌手，可自行增删）
WHITELIST_SINGER = re.compile(
r"林俊杰|Beyond|BEYOND|陈奕迅|王菲|邓紫棋|G.E.M.|A-Lin|A-LIN|张杰|单依纯",
re.IGNORECASE
)
# 是否递归扫描子文件夹（True=扫描所有子目录，False=仅扫描根目录）
SCAN_SUBFOLDER = False
# 是否自动标准化重命名保留的文件（统一格式：歌手 - 歌名.后缀）
AUTO_RENAME = True
# 是否自动清理去重后产生的空目录
AUTO_CLEAN_EMPTY_DIR = True
# 音质优先级（越靠前优先级越高）
AUDIO_QUALITY_PRIORITY = ["flac", "wav", "m4a", "mp3", "ogg"]
# 支持的音频后缀
AUDIO_EXTENSIONS = set(AUDIO_QUALITY_PRIORITY)
# -------------------------- 【新增功能配置】--------------------------
# 【新增】是否启用ID3标签解析（优先读取文件内置的歌手/歌名信息，比文件名更靠谱）
ENABLE_ID3_PARSE = True
# 【新增】是否启用音频指纹识别（解决同歌不同文件、乱码、无标签文件的识别，需安装对应依赖）
ENABLE_AUDIO_FINGERPRINT = False
# 【新增】音频指纹识别模式：online=联网识别（获取ISRC唯一标识，精准度最高），offline=离线本地比对
FINGERPRINT_MODE = "online"
# 【新增】AcoustID API Key（联网模式必填，免费申请：https://acoustid.org/api-key）
ACOUSTID_API_KEY = "YOUR_API_KEY_HERE"
# 【新增】指纹缓存文件路径（避免重复计算，大幅提升二次运行速度）
FINGERPRINT_CACHE_FILE = os.path.join(LOG_DIR, "fingerprint_cache.json")
# 【新增】离线指纹相似度阈值（0-1，越高越严格，超过该值判定为同一首歌）
FINGERPRINT_SIMILARITY_THRESHOLD = 0.95
# ==========================================================================================================
# ====================================== 【核心规则区 无特殊需求无需修改】=====================================
# 【文件名标签前缀】剔除开头的【】格式标签
TAG_PREFIX_PATTERN = re.compile(r"^【[^】]+】\s*", re.IGNORECASE)
# 【改编关键词】包含这些词的视为非原版，优先级降低
ADAPTATION_KEYWORDS = r"DJ|Live|Remix|Mix|版|伴奏|消音|降调|加速|慢摇|完整版|feat|with|翻奏|翻唱|混音|重制|0\.\d+x|\d+x|现场|演唱会|Instrumental|Acoustic|Edit|Radio"
# 【原版保留关键词】包含这些词的视为原版标注，不降低优先级
ORIGINAL_TAG_KEYWORDS = r"国语|粤语|专辑版|原版|正式版|电视剧|主题曲|片尾曲|插曲"
# 歌名清洗正则：仅剔除改编相关内容，保留原版标注
CLEAN_PATTERN = re.compile(
rf"[\(\[（【][^\)\]）】]*?(?:{ADAPTATION_KEYWORDS})[^\)\]）】]*?[\)\]）】]",
re.IGNORECASE
)
# 各类版本判断正则
ORIGINAL_CLEAN_PATTERN = re.compile(ADAPTATION_KEYWORDS, re.IGNORECASE)
ORIGINAL_TAG_PATTERN = re.compile(ORIGINAL_TAG_KEYWORDS, re.IGNORECASE)
DJ_REMIX_PATTERN = re.compile(r"DJ|Remix|Mix", re.IGNORECASE)
LIVE_PATTERN = re.compile(r"Live|现场|演唱会", re.IGNORECASE)
ACCOMPANIMENT_PATTERN = re.compile(r"伴奏|消音|Instrumental", re.IGNORECASE)
# 歌手名分隔符
SINGER_SPLIT_PATTERN = re.compile(r"[&_、和\+/]")
# 繁简体核心映射表（零依赖，覆盖歌名/歌手常用字）
TRAD2SIMP_MAP = {
'倫':'伦', '傑':'杰', '過':'过', '裡':'里', '裏':'里', '麼':'么', '麽':'么',
'國':'国', '會':'会', '愛':'爱', '無':'无', '頭':'头', '發':'发', '現':'现',
'聽':'听', '說':'说', '話':'话', '開':'开', '關':'关', '間':'间', '個':'个',
'這':'这', '裡':'里', '邊':'边', '對':'对', '錯':'错', '學':'学', '師':'师',
'藝':'艺', '術':'术', '夢':'梦', '動':'动', '靜':'静', '電':'电', '腦':'脑',
'龍':'龙', '戰':'战', '騎':'骑', '士':'士', '飛':'飞', '機':'机', '聖':'圣',
'誕':'诞', '響':'响', '遠':'远', '離':'离', '後':'后', '還':'还', '遊':'游',
'戲':'戏', '盡':'尽', '歡':'欢', '點':'点', '號':'号', '淚':'泪', '緣':'缘',
'風':'风', '雲':'云', '戀':'恋', '給':'给', '妳':'你', '妳':'你', '歲':'岁',
'月':'月', '聲':'声', '音':'音', '樂':'乐', '欄':'栏', '殤':'殇', '紅':'红',
'塵':'尘', '緒':'绪', '憶':'忆', '葉':'叶', '書':'书', '寫':'写', '詞':'词',
'曲':'曲', '編':'编', '製':'制', '專':'专', '輯':'辑', '廠':'厂', '牌':'牌'
}
TRAD2SIMP_TABLE = str.maketrans(TRAD2SIMP_MAP)
# 程序锁文件（防止重复运行）
LOCK_FILE = "/tmp/music_deduplicate.lock" if not IS_WINDOWS else os.path.join(os.getenv("TEMP", "C:\\Windows\\Temp"), "music_deduplicate.lock")
# ==========================================================================================================
# ====================================== 【全局变量初始化】======================================
lock_file: Optional[object] = None
logger: Optional[object] = None
fingerprint_cache: Dict[str, Dict] = {}
# 依赖可用性校验
MUTAGEN_AVAILABLE = False
ACOUSTID_AVAILABLE = False
NUMPY_AVAILABLE = False
# ==========================================================================================================
# ====================================== 【依赖可用性预校验】======================================
try:
from mutagen import File as MutagenFile
MUTAGEN_AVAILABLE = True
except ImportError:
pass
try:
import acoustid
ACOUSTID_AVAILABLE = True
except ImportError:
pass
try:
import numpy as np
from scipy.spatial.distance import cosine
NUMPY_AVAILABLE = True
except ImportError:
pass
# ==========================================================================================================
# ====================================== 【工具函数区】======================================
def trad_to_simp(text: str) -> str:
"""繁体转简体，零依赖归一化"""
return text.translate(TRAD2SIMP_TABLE)
def normalize_text(text: str) -> str:
"""文本归一化：繁简转换+大小写统一+特殊符号清洗+空格统一"""
text = trad_to_simp(text)
text = text.lower()
text = re.sub(r"[^\u4e00-\u9fa5a-zA-Z0-9\s]", "", text)
text = re.sub(r"\s+", " ", text).strip()
return text
def get_file_md5(file_path: str, block_size: int = 4096) -> str:
"""分块计算文件MD5，大文件不占内存，强化异常处理"""
if not os.path.exists(file_path) or not os.path.isfile(file_path):
return ""
md5 = hashlib.md5()
try:
with open(file_path, "rb") as f:
chunk = f.read(block_size)
while chunk:
md5.update(chunk)
chunk = f.read(block_size)
return md5.hexdigest()
except PermissionError:
if logger:
logger.warning(f"无权限读取文件，跳过MD5计算：{file_path}")
return ""
except Exception as e:
if logger:
logger.error(f"MD5计算失败：{file_path}，错误：{str(e)}")
return ""
def get_singer_normalized(singer_text: str) -> str:
"""歌手名归一化，提取主歌手"""
singer_text = trad_to_simp(singer_text)
main_singer = SINGER_SPLIT_PATTERN.split(singer_text)[0].strip()
return normalize_text(main_singer)
def get_audio_id3_info(file_path: str) -> Tuple[str, str]:
"""
读取音频文件ID3标签，返回(歌手, 歌名)
无标签/读取失败返回空字符串
支持mp3/flac/m4a/wav/ogg全格式
"""
# 修复：全局变量声明，仅读取无需赋值，兼容作用域规则
global ENABLE_ID3_PARSE, MUTAGEN_AVAILABLE
if not ENABLE_ID3_PARSE or not MUTAGEN_AVAILABLE:
return "", ""
try:
audio = MutagenFile(file_path, easy=True)
if not audio:
return "", ""
# 优先读取标准标签
artist = audio.get("artist", [""])[0].strip()
title = audio.get("title", [""])[0].strip()
# 兜底兼容其他标签格式
if not artist:
artist = audio.get("albumartist", [""])[0].strip()
if not title:
title = audio.get("track", [""])[0].strip()
return artist, title
except Exception as e:
if logger:
logger.debug(f"ID3标签读取失败：{file_path}，错误：{str(e)}")
return "", ""
def load_fingerprint_cache():
"""加载指纹缓存，避免重复计算"""
global fingerprint_cache
if not os.path.exists(FINGERPRINT_CACHE_FILE):
fingerprint_cache = {}
return
try:
with open(FINGERPRINT_CACHE_FILE, "r", encoding="utf-8") as f:
fingerprint_cache = json.load(f)
if logger:
logger.info(f"已加载指纹缓存，共 {len(fingerprint_cache)} 条记录")
except Exception as e:
if logger:
logger.warning(f"指纹缓存加载失败：{str(e)}")
fingerprint_cache = {}
def save_fingerprint_cache():
"""保存指纹缓存到文件"""
global fingerprint_cache
try:
os.makedirs(os.path.dirname(FINGERPRINT_CACHE_FILE), exist_ok=True)
with open(FINGERPRINT_CACHE_FILE, "w", encoding="utf-8") as f:
json.dump(fingerprint_cache, f, ensure_ascii=False, indent=2)
except Exception as e:
if logger:
logger.warning(f"指纹缓存保存失败：{str(e)}")
def get_audio_fingerprint(file_path: str, file_md5: str) -> Tuple[int, List[int], str, str]:
"""
计算音频文件指纹，返回(时长, 指纹数组, ISRC, 标准化歌曲ID)
计算失败返回(0, [], "", "")
"""
# 修复：全局变量声明
global ENABLE_AUDIO_FINGERPRINT, ACOUSTID_AVAILABLE, FINGERPRINT_MODE, ACOUSTID_API_KEY, fingerprint_cache
if not ENABLE_AUDIO_FINGERPRINT or not ACOUSTID_AVAILABLE:
return 0, [], "", ""
# 优先从缓存读取
if file_md5 in fingerprint_cache:
cache_data = fingerprint_cache[file_md5]
return (
cache_data.get("duration", 0),
cache_data.get("fingerprint", []),
cache_data.get("isrc", ""),
cache_data.get("song_id", "")
)
try:
# 计算基础指纹
duration, fingerprint = acoustid.fingerprint_file(file_path)
isrc = ""
song_id = ""
# 联网模式：识别歌曲ISRC和标准信息
if FINGERPRINT_MODE == "online" and ACOUSTID_API_KEY != "YOUR_API_KEY_HERE":
results = acoustid.lookup(ACOUSTID_API_KEY, fingerprint, duration, meta="recordings+isrc")
if results.get("results"):
best_result = results["results"][0]
if best_result.get("recordings"):
recording = best_result["recordings"][0]
isrc = recording.get("isrc", "").strip()
song_id = best_result.get("id", "").strip()
# 写入缓存
fingerprint_cache[file_md5] = {
"duration": duration,
"fingerprint": fingerprint,
"isrc": isrc,
"song_id": song_id,
"file_path": file_path
}
return duration, fingerprint, isrc, song_id
except Exception as e:
if logger:
logger.warning(f"音频指纹计算失败：{file_path}，错误：{str(e)}")
return 0, [], "", ""
def calculate_fingerprint_similarity(fp1: List[int], fp2: List[int]) -> float:
"""计算两个指纹的余弦相似度，返回0-1之间的数值，越接近1越相似"""
global NUMPY_AVAILABLE
if not NUMPY_AVAILABLE or not fp1 or not fp2:
return 0.0
# 统一长度
min_len = min(len(fp1), len(fp2))
fp1_arr = np.array(fp1[:min_len])
fp2_arr = np.array(fp2[:min_len])
# 余弦相似度计算
similarity = 1 - cosine(fp1_arr, fp2_arr)
return max(0.0, similarity)
def parse_audio_file(file_path: str) -> Optional[Dict]:
"""
解析单个音频文件，优先ID3标签，其次文件名解析，新增指纹相关字段
非音频文件返回None
"""
global TOP_PRIORITY_SINGER, WHITELIST_SINGER, TAG_PREFIX_PATTERN, ORIGINAL_CLEAN_PATTERN
global ORIGINAL_TAG_PATTERN, DJ_REMIX_PATTERN, LIVE_PATTERN, ACCOMPANIMENT_PATTERN
global CLEAN_PATTERN, AUDIO_EXTENSIONS, AUDIO_QUALITY_PRIORITY
try:
full_filename = os.path.basename(file_path)
file_name_without_ext, ext = os.path.splitext(full_filename)
ext = ext.lstrip(".").lower()
file_dir = os.path.dirname(file_path)
except Exception as e:
if logger:
logger.error(f"文件名解析失败：{file_path}，错误：{str(e)}")
return None
if ext not in AUDIO_EXTENSIONS:
return None
if full_filename.startswith(".") or os.path.islink(file_path) or not os.path.isfile(file_path):
return None
# ========== 核心升级：优先读取ID3标签 ==========
id3_artist, id3_title = get_audio_id3_info(file_path)
use_id3 = bool(id3_artist and id3_title)
# 文件名清洗（ID3读取失败时使用）
clean_filename = TAG_PREFIX_PATTERN.sub("", file_name_without_ext).strip()
clean_filename = clean_filename if clean_filename else file_name_without_ext
# 歌手&歌名提取：ID3优先，其次文件名标准格式
if use_id3:
singer_part = id3_artist
name_part = id3_title
else:
singer_part = ""
name_part = clean_filename
if " - " in clean_filename:
singer_part, name_part = clean_filename.rsplit(" - ", 1)
singer_part = singer_part.strip()
name_part = name_part.strip()
# ========== 优先级判断逻辑 ==========
is_top_singer = bool(TOP_PRIORITY_SINGER.search(singer_part))
is_whitelist_singer = bool(WHITELIST_SINGER.search(singer_part))
has_adaptation = bool(ORIGINAL_CLEAN_PATTERN.search(name_part))
is_original_clean = not has_adaptation
is_original_tag = bool(ORIGINAL_TAG_PATTERN.search(name_part)) and not has_adaptation
is_dj_remix = bool(DJ_REMIX_PATTERN.search(name_part))
is_live = bool(LIVE_PATTERN.search(name_part))
is_accompaniment = bool(ACCOMPANIMENT_PATTERN.search(name_part))
# 核心歌名提取与归一化（去重核心依据，ID3兜底）
core_name_raw = CLEAN_PATTERN.sub("", name_part).strip()
core_name_raw = core_name_raw if core_name_raw else name_part
core_name_normalized = normalize_text(core_name_raw)
# 歌手归一化
singer_normalized = get_singer_normalized(singer_part)
# 歌名纯净度评分，修复除零异常
original_length = len(name_part)
clean_score = len(core_name_raw) / original_length if original_length > 0 else 0.0
# 音质权重
try:
quality_weight = len(AUDIO_QUALITY_PRIORITY) - AUDIO_QUALITY_PRIORITY.index(ext)
except ValueError:
quality_weight = 0
# 文件大小（同格式下越大音质越好）
try:
file_size = os.path.getsize(file_path)
except Exception:
file_size = 0
# 优先级标签（日志展示用）
id3_tag = "【ID3标签识别】" if use_id3 else "【文件名识别】"
if is_top_singer:
priority_tag = f"{id3_tag}【最高优先级-周杰伦】"
elif is_whitelist_singer:
priority_tag = f"{id3_tag}【白名单歌手】"
elif is_original_tag:
priority_tag = f"{id3_tag}【专辑原版】"
elif is_original_clean:
priority_tag = f"{id3_tag}【纯净原唱版】"
elif is_dj_remix:
priority_tag = f"{id3_tag}【DJ改编版】"
elif is_live:
priority_tag = f"{id3_tag}【Live现场版】"
elif is_accompaniment:
priority_tag = f"{id3_tag}【伴奏/消音版】"
else:
priority_tag = f"{id3_tag}【其他版本】"
# 标准化文件名
standard_filename = f"{singer_part} - {core_name_raw}.{ext}" if singer_part else f"{core_name_raw}.{ext}"
return {
"full_path": file_path,
"file_dir": file_dir,
"original_filename": full_filename,
"standard_filename": standard_filename,
"core_name_raw": core_name_raw,
"core_name_normalized": core_name_normalized,
"singer_raw": singer_part,
"singer_normalized": singer_normalized,
"ext": ext,
"file_size": file_size,
"is_top_singer": is_top_singer,
"is_whitelist_singer": is_whitelist_singer,
"is_original_tag": is_original_tag,
"is_original_clean": is_original_clean,
"is_dj_remix": is_dj_remix,
"is_live": is_live,
"is_accompaniment": is_accompaniment,
"quality_weight": quality_weight,
"clean_score": clean_score,
"priority_tag": priority_tag,
"md5": "",
# 新增指纹相关字段
"fingerprint": [],
"fingerprint_duration": 0,
"isrc": "",
"acoustid_song_id": "",
"use_id3": use_id3
}
def scan_audio_files(root_dir: str, scan_subfolder: bool) -> List[Dict]:
"""扫描目录下所有音频文件，支持递归，优化目录过滤逻辑"""
global BACKUP_DIR, LOG_DIR
audio_files = []
if scan_subfolder:
for root, _, files in os.walk(root_dir):
dir_basename = os.path.basename(root)
if dir_basename in ["_deduplicate_backup", "_deduplicate_log"]:
continue
for filename in files:
file_path = os.path.join(root, filename)
file_info = parse_audio_file(file_path)
if file_info:
audio_files.append(file_info)
else:
for filename in os.listdir(root_dir):
file_path = os.path.join(root_dir, filename)
if not os.path.isfile(file_path):
continue
file_info = parse_audio_file(file_path)
if file_info:
audio_files.append(file_info)
return audio_files
def init_dirs(*dirs):
"""初始化目录，不存在则创建，增加权限校验"""
global logger
for dir_path in dirs:
if not os.path.exists(dir_path):
try:
os.makedirs(dir_path, exist_ok=True)
except Exception as e:
if logger:
logger.error(f"目录创建失败：{dir_path}，错误：{str(e)}")
else:
print(f"目录创建失败：{dir_path}，错误：{str(e)}")
release_program_lock()
sys.exit(1)
if not os.access(dir_path, os.R_OK | os.W_OK):
if logger:
logger.error(f"目录无读写权限：{dir_path}")
else:
print(f"目录无读写权限：{dir_path}")
release_program_lock()
sys.exit(1)
def get_program_lock() -> bool:
"""跨平台获取程序排他锁，防止重复运行"""
global lock_file, LOCK_FILE, IS_WINDOWS
try:
lock_file = open(LOCK_FILE, "w")
if IS_WINDOWS:
msvcrt.locking(lock_file.fileno(), msvcrt.LK_NBLCK, 1)
else:
fcntl.flock(lock_file, fcntl.LOCK_EX | fcntl.LOCK_NB)
return True
except BlockingIOError:
return False
except Exception as e:
if logger:
logger.error(f"程序锁获取失败：{str(e)}")
else:
print(f"程序锁获取失败：{str(e)}")
return False
def release_program_lock():
"""跨平台释放程序锁，强制清理锁文件"""
global lock_file, LOCK_FILE, IS_WINDOWS
try:
if lock_file:
if IS_WINDOWS:
msvcrt.locking(lock_file.fileno(), msvcrt.LK_UNLCK, 1)
else:
fcntl.flock(lock_file, fcntl.LOCK_UN)
lock_file.close()
if os.path.exists(LOCK_FILE):
os.remove(LOCK_FILE)
except Exception:
pass
def clean_empty_dirs(root_dir: str):
"""递归清理空目录，跳过备份和日志目录"""
global logger, BACKUP_DIR, LOG_DIR
if not logger:
return
for root, dirs, files in os.walk(root_dir, topdown=False):
dir_basename = os.path.basename(root)
if dir_basename in ["_deduplicate_backup", "_deduplicate_log"]:
continue
for dir_name in dirs:
dir_path = os.path.join(root, dir_name)
try:
if not os.listdir(dir_path):
os.rmdir(dir_path)
logger.info(f"已清理空目录：{dir_path}")
except Exception as e:
logger.warning(f"空目录清理失败：{dir_path}，错误：{str(e)}")
# ====================================== 【日志系统】======================================
def init_logger(log_dir: str, run_time: str):
"""初始化日志系统，同时输出到控制台和文件，修复重复打印问题"""
import logging
global logger
log_file = os.path.join(log_dir, f"music_deduplicate_{run_time}.log")
logger = logging.getLogger("music_deduplicate")
logger.setLevel(logging.INFO)
if logger.handlers:
logger.handlers.clear()
file_handler = logging.FileHandler(log_file, encoding="utf-8")
file_handler.setFormatter(logging.Formatter("%(asctime)s - %(levelname)s - %(message)s"))
console_handler = logging.StreamHandler()
console_handler.setFormatter(logging.Formatter("%(message)s"))
logger.addHandler(file_handler)
logger.addHandler(console_handler)
def export_csv_report(report_data: List[Dict], csv_path: str):
"""导出处理结果CSV报表，强化异常处理"""
global logger
if not logger:
return
headers = ["优先级标签", "原始文件名", "核心歌名", "歌手", "格式", "文件大小(KB)", "处理结果", "ISRC", "文件路径", "MD5值"]
try:
with open(csv_path, "w", encoding="utf-8-sig", newline="") as f:
writer = csv.DictWriter(f, fieldnames=headers)
writer.writeheader()
for row in report_data:
writer.writerow({
"优先级标签": row["priority_tag"],
"原始文件名": row["original_filename"],
"核心歌名": row["core_name_raw"],
"歌手": row["singer_raw"],
"格式": row["ext"],
"文件大小(KB)": round(row["file_size"]/1024, 2),
"处理结果": row["handle_result"],
"ISRC": row["isrc"],
"文件路径": row["full_path"],
"MD5值": row["md5"]
})
logger.info(f"✅ 报表已导出：{csv_path}")
except Exception as e:
logger.error(f"❌ 报表导出失败：{str(e)}")
# ====================================== 【歌曲分组核心逻辑】======================================
def group_songs_by_identification(audio_files: List[Dict]) -> Dict[str, List[Dict]]:
"""
多级识别分组：优先ISRC > 离线指纹相似度 > ID3歌手+歌名归一化 > 文件名核心名归一化
"""
# 修复：全局变量声明
global ENABLE_AUDIO_FINGERPRINT, FINGERPRINT_MODE, NUMPY_AVAILABLE, FINGERPRINT_SIMILARITY_THRESHOLD, logger
song_groups: Dict[str, List[Dict]] = {}
ungrouped_files: List[Dict] = []
# 第一级：ISRC唯一标识分组（联网模式最高优先级，100%精准）
logger.info("第一级分组：ISRC唯一标识识别...")
for file_info in audio_files:
isrc = file_info.get("isrc", "")
if isrc:
group_key = f"ISRC_{isrc}"
if group_key not in song_groups:
song_groups[group_key] = []
song_groups[group_key].append(file_info)
else:
ungrouped_files.append(file_info)
logger.info(f"ISRC分组完成，共识别 {len(song_groups)} 首歌曲，剩余 {len(ungrouped_files)} 个文件待分组")
# 第二级：离线指纹相似度分组（仅离线模式/无ISRC文件）
if ENABLE_AUDIO_FINGERPRINT and FINGERPRINT_MODE == "offline" and NUMPY_AVAILABLE and ungrouped_files:
logger.info("第二级分组：离线音频指纹相似度比对...")
fingerprint_groups: List[List[Dict]] = []
for file_info in ungrouped_files:
fp = file_info.get("fingerprint", [])
if not fp:
continue
# 遍历已有分组，计算相似度
matched = False
for group in fingerprint_groups:
group_sample_fp = group[0]["fingerprint"]
similarity = calculate_fingerprint_similarity(fp, group_sample_fp)
if similarity >= FINGERPRINT_SIMILARITY_THRESHOLD:
group.append(file_info)
matched = True
break
if not matched:
fingerprint_groups.append([file_info])
# 合并到主分组
for idx, group in enumerate(fingerprint_groups):
group_key = f"FINGERPRINT_{idx}"
song_groups[group_key] = group
# 从待分组列表中移除已匹配的文件
for file in group:
if file in ungrouped_files:
ungrouped_files.remove(file)
logger.info(f"指纹分组完成，新增 {len(fingerprint_groups)} 首歌曲，剩余 {len(ungrouped_files)} 个文件待分组")
# 第三级：ID3歌手+歌名归一化分组
logger.info("第三级分组：ID3标签歌手+歌名归一化识别...")
id3_ungrouped = []
for file_info in ungrouped_files:
if file_info.get("use_id3", False):
singer_norm = file_info["singer_normalized"]
name_norm = file_info["core_name_normalized"]
if singer_norm and name_norm:
group_key = f"ID3_{singer_norm}_{name_norm}"
if group_key not in song_groups:
song_groups[group_key] = []
song_groups[group_key].append(file_info)
continue
id3_ungrouped.append(file_info)
ungrouped_files = id3_ungrouped
logger.info(f"ID3分组完成，剩余 {len(ungrouped_files)} 个文件待分组")
# 第四级：文件名核心名归一化分组（原逻辑兜底）
logger.info("第四级分组：文件名核心名归一化识别...")
for file_info in ungrouped_files:
core_key = file_info["core_name_normalized"]
if not core_key:
logger.warning(f"跳过无效文件，无法识别核心歌名：{file_info['original_filename']}")
continue
group_key = f"FILENAME_{core_key}"
if group_key not in song_groups:
song_groups[group_key] = []
song_groups[group_key].append(file_info)
logger.info(f"文件名分组完成，全部分组结束")
return song_groups
# ====================================== 【主程序】======================================
def main():
# ========== 核心修复：声明需要修改的全局变量 ==========
global ENABLE_ID3_PARSE, ENABLE_AUDIO_FINGERPRINT
# ======================================================
RUN_TIME = datetime.now().strftime("%Y%m%d_%H%M%S")
# 先获取锁，再初始化目录和日志，确保所有异常都能释放锁
lock_success = get_program_lock()
if not lock_success:
print("【错误】程序已在运行中，请勿重复执行")
release_program_lock()
sys.exit(1)
try:
init_dirs(BACKUP_DIR, LOG_DIR)
init_logger(LOG_DIR, RUN_TIME)
current_backup_dir = os.path.join(BACKUP_DIR, RUN_TIME)
# 配置合法性校验
if RUN_MODE not in ["preview", "exec"]:
logger.error(f"【错误】无效的运行模式：{RUN_MODE}，仅支持 'preview' 或 'exec'")
sys.exit(1)
if not os.path.isdir(MUSIC_ROOT_DIR):
logger.error(f"【错误】音乐目录不存在：{MUSIC_ROOT_DIR}")
sys.exit(1)
if not os.access(MUSIC_ROOT_DIR, os.R_OK):
logger.error(f"【错误】无权限读取音乐目录：{MUSIC_ROOT_DIR}")
sys.exit(1)
# 依赖可用性提示（已修复作用域问题）
if ENABLE_ID3_PARSE and not MUTAGEN_AVAILABLE:
logger.warning("【警告】已启用ID3解析，但未安装mutagen库，已自动禁用ID3解析")
ENABLE_ID3_PARSE = False
if ENABLE_AUDIO_FINGERPRINT and not ACOUSTID_AVAILABLE:
logger.warning("【警告】已启用音频指纹，但未安装pyacoustid库，已自动禁用音频指纹")
ENABLE_AUDIO_FINGERPRINT = False
if FINGERPRINT_MODE == "online" and ACOUSTID_API_KEY == "YOUR_API_KEY_HERE":
logger.warning("【警告】联网指纹模式未配置API Key，将仅计算本地指纹，无法获取ISRC")
logger.info("=" * 100)
logger.info(f"音乐去重任务启动 | 运行模式：{RUN_MODE} | 扫描目录：{MUSIC_ROOT_DIR}")
logger.info(f"功能状态：ID3解析={'✅ 已启用' if ENABLE_ID3_PARSE else '❌ 已禁用'} | 音频指纹={'✅ 已启用' if ENABLE_AUDIO_FINGERPRINT else '❌ 已禁用'}")
logger.info("=" * 100)
# 加载指纹缓存
if ENABLE_AUDIO_FINGERPRINT:
load_fingerprint_cache()
# 扫描音频文件
logger.info("开始扫描音频文件...")
audio_files = scan_audio_files(MUSIC_ROOT_DIR, SCAN_SUBFOLDER)
if not audio_files:
logger.info("未扫描到任何音频文件，任务结束")
sys.exit(0)
logger.info(f"扫描完成，共发现 {len(audio_files)} 个有效音频文件")
# 第一步：MD5内容去重（优先执行，避免无效计算）
logger.info("开始MD5内容去重...")
md5_unique_map: Dict[str, Dict] = {}
md5_dup_files: List[Dict] = []
for file_info in audio_files:
file_md5 = get_file_md5(file_info["full_path"])
file_info["md5"] = file_md5
if not file_md5:
continue
if file_md5 not in md5_unique_map:
md5_unique_map[file_md5] = file_info
else:
file_info["handle_result"] = "待移动(内容重复)"
md5_dup_files.append(file_info)
unique_audio_files = list(md5_unique_map.values())
logger.info(f"MD5内容去重完成，过滤掉 {len(md5_dup_files)} 个内容完全一致的重复文件，剩余 {len(unique_audio_files)} 个唯一内容文件")
# 第二步：音频指纹计算（仅唯一内容文件）
if ENABLE_AUDIO_FINGERPRINT:
logger.info("开始计算音频指纹...")
total_files = len(unique_audio_files)
for idx, file_info in enumerate(unique_audio_files, 1):
if idx % 10 == 0:
logger.info(f"指纹计算进度：{idx}/{total_files}")
duration, fingerprint, isrc, song_id = get_audio_fingerprint(file_info["full_path"], file_info["md5"])
file_info["fingerprint"] = fingerprint
file_info["fingerprint_duration"] = duration
file_info["isrc"] = isrc
file_info["acoustid_song_id"] = song_id
# 保存指纹缓存
save_fingerprint_cache()
logger.info("音频指纹计算完成，缓存已保存")
# 第三步：多级识别歌曲分组
logger.info("开始多级歌曲识别与分组...")
song_groups = group_songs_by_identification(unique_audio_files)
logger.info(f"歌曲识别完成，共识别 {len(song_groups)} 首独立歌曲")
# 第四步：同歌曲不同版本优先级排序
keep_files: List[Dict] = []
version_dup_files: List[Dict] = []
report_data: List[Dict] = []
for group_key, group in song_groups.items():
# 原有优先级排序规则完全兼容
sorted_group = sorted(
group,
key=lambda x: (
x["is_top_singer"],
x["is_whitelist_singer"],
x["is_original_tag"],
x["is_original_clean"],
-x["is_dj_remix"],
-x["is_live"],
-x["is_accompaniment"],
x["quality_weight"],
x["clean_score"],
x["file_size"]
),
reverse=True
)
# 最优文件保留，其余加入待移动列表
keep_file = sorted_group[0]
keep_file["handle_result"] = "保留"
keep_files.append(keep_file)
report_data.append(keep_file)
if len(sorted_group) > 1:
for dup_file in sorted_group[1:]:
dup_file["handle_result"] = "待移动(版本重复)"
version_dup_files.append(dup_file)
report_data.append(dup_file)
# 合并所有待移动文件
move_files = md5_dup_files + version_dup_files
# 结果统计与输出
logger.info("=" * 100)
logger.info(f"【最终结果统计】")
logger.info(f"✅ 需保留的最优文件：{len(keep_files)} 个")
logger.info(f"❌ 待处理的重复文件：{len(move_files)} 个")
logger.info(f"📂 歌曲识别：{len(audio_files)} 个文件 → {len(song_groups)} 首独立歌曲")
logger.info(f"🔍 MD5内容去重：识别到 {len(md5_dup_files)} 个内容完全一致的重复文件")
logger.info("=" * 100)
# 输出保留文件列表
logger.info("\n【保留的最优文件列表】")
for idx, file in enumerate(keep_files, 1):
logger.info(f"{idx}. {file['priority_tag']} {file['original_filename']}")
logger.info(f" 核心歌名：{file['core_name_raw']} | 歌手：{file['singer_raw']} | 格式：{file['ext']} | ISRC：{file['isrc']} | 路径：{file['file_dir']}")
# 输出待移动文件列表
if move_files:
logger.info("\n" + "=" * 100)
logger.info("\n【待处理的重复文件列表】")
for idx, file in enumerate(move_files, 1):
logger.info(f"{idx}. {file['priority_tag']} {file['original_filename']} | 原因：{file['handle_result']}")
logger.info(f" 核心歌名：{file['core_name_raw']} | 歌手：{file['singer_raw']} | 格式：{file['ext']} | ISRC：{file['isrc']} | 路径：{file['file_dir']}")
logger.info("\n" + "=" * 100)
# 执行模式处理
if RUN_MODE == "exec":
success_count = 0
fail_count = 0
if move_files:
logger.info("【执行模式】开始处理重复文件...")
init_dirs(current_backup_dir)
for file in move_files:
try:
relative_path = os.path.relpath(file["file_dir"], MUSIC_ROOT_DIR)
backup_sub_dir = os.path.join(current_backup_dir, relative_path)
init_dirs(backup_sub_dir)
backup_file_path = os.path.join(backup_sub_dir, file["original_filename"])
os.rename(file["full_path"], backup_file_path)
logger.info(f"✅ 已移动：{file['original_filename']} → 备份目录：{backup_sub_dir}")
success_count += 1
except Exception as e:
logger.error(f"❌ 移动失败：{file['original_filename']}，错误信息：{str(e)}")
fail_count += 1
logger.info("\n" + "=" * 100)
logger.info(f"【重复文件处理完成】成功移动 {success_count} 个重复文件，失败 {fail_count} 个")
logger.info(f"📦 重复文件备份目录：{current_backup_dir}")
else:
logger.info("【执行模式】未发现重复文件，无需移动处理")
# 自动重命名保留的文件
if AUTO_RENAME:
logger.info("\n【自动重命名】开始标准化重命名保留文件...")
rename_count = 0
rename_fail_count = 0
for file in keep_files:
if file["original_filename"] == file["standard_filename"]:
continue
old_path = file["full_path"]
new_path = os.path.join(file["file_dir"], file["standard_filename"])
try:
if not os.path.exists(new_path):
os.rename(old_path, new_path)
logger.info(f"✅ 已重命名：{file['original_filename']} → {file['standard_filename']}")
rename_count += 1
else:
logger.warning(f"⚠️ 重命名跳过，目标文件已存在：{file['standard_filename']}")
except Exception as e:
logger.error(f"❌ 重命名失败：{file['original_filename']}，错误信息：{str(e)}")
rename_fail_count += 1
logger.info(f"【自动重命名完成】成功重命名 {rename_count} 个文件，失败 {rename_fail_count} 个")
# 自动清理空目录
if AUTO_CLEAN_EMPTY_DIR:
logger.info("\n【空目录清理】开始清理空目录...")
clean_empty_dirs(MUSIC_ROOT_DIR)
logger.info("【空目录清理完成】")
else:
logger.info("【预览模式】仅展示结果，未修改任何文件")
logger.info("如需执行去重，请将配置项中的 RUN_MODE 修改为 'exec'")
# 导出CSV报表
csv_path = os.path.join(LOG_DIR, f"music_deduplicate_report_{RUN_TIME}.csv")
export_csv_report(report_data, csv_path)
logger.info("=" * 100)
logger.info(f"任务全部执行结束 | 日志与报表目录：{LOG_DIR}")
logger.info("=" * 100)
finally:
release_program_lock()
if __name__ == "__main__":
main()
[复制链接]
发表于 2026-4-15 19:35
