Files
DeleteChongfuTVYY/music_duplicate_cleaner.py
2026-01-25 21:20:33 +08:00

754 lines
26 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
# -*- coding: utf-8 -*-
"""
music_duplicate_cleaner.py — 音乐文件去重专用版本
特性概览:
- 多线程扫描 + 单线程 DatabaseWriterThread 写入(永不出现 database is locked
- safe_remove硬链接保护
- 容错导入 librosa/scipy 等(功能降级)
- 自动检测写入阻塞并自动恢复
- 详细日志与可选实时进度显示
"""
from __future__ import annotations
import os
import sys
import time
import warnings
import threading
import queue
import hashlib
import shutil
import sqlite3
import logging
import argparse
import math
import re
from pathlib import Path
from datetime import datetime
from concurrent.futures import ThreadPoolExecutor, as_completed
from typing import Any, Dict, List, Optional, Tuple
warnings.filterwarnings("ignore", category=UserWarning, module="numba")
# -------------------------
# logging helper
# -------------------------
def setup_logging(log_level=logging.INFO, log_file="music_duplicate_cleaner.log"):
logging.basicConfig(
level=log_level,
format="%(asctime)s - %(levelname)s - %(message)s",
handlers=[
logging.FileHandler(log_file, encoding="utf-8"),
logging.StreamHandler(sys.stdout),
],
)
return logging.getLogger(__name__)
logger = setup_logging()
# -------------------------
# safe optional imports (robust)
# -------------------------
LIBROSA_AVAILABLE = False
SCIPY_AVAILABLE = False
try:
import numpy as np # type: ignore
try:
import librosa # type: ignore
LIBROSA_AVAILABLE = True
logger.info("librosa available")
except Exception as e:
librosa = None # type: ignore
LIBROSA_AVAILABLE = False
logger.warning(f"librosa 导入失败: {e}")
try:
from scipy import signal as scipy_signal # type: ignore
SCIPY_AVAILABLE = True
except Exception as e:
scipy_signal = None
SCIPY_AVAILABLE = False
logger.warning(f"scipy.signal 导入失败: {e}")
except Exception as e:
logger.warning(f"科学栈初始化失败: {e}")
# -------------------------
# utils
# -------------------------
def choose_worker_count(requested: Optional[int] = None) -> int:
if requested and requested > 0:
return requested
try:
cpu = os.cpu_count() or 1
return min(32, max(4, cpu * 2))
except Exception:
return 4
def file_sha256(path: str, block_size: int = 65536) -> str:
h = hashlib.sha256()
try:
with open(path, "rb") as f:
for block in iter(lambda: f.read(block_size), b""):
h.update(block)
return h.hexdigest()
except Exception as e:
logger.debug(f"计算哈希失败 {path}: {e}")
return ""
# -------------------------
# safe_remove (硬链接保护:策略 C)
# -------------------------
def safe_remove(path: str, no_backup: bool=False, backup_dir: Optional[str]=None, db_writer: Optional["DatabaseWriterThread"]=None) -> bool:
try:
st = os.stat(path)
except Exception as e:
logger.warning(f"无法访问文件 {path}: {e}")
return False
if getattr(st, "st_nlink", 1) > 1:
logger.info(f"文件有多个硬链接,跳过删除以保护硬链接: {path}")
if db_writer:
db_writer.enqueue_operation({
"operation_type": "skip_delete_hardlink",
"file_path": path,
"file_hash": None,
"reason": "hardlink_skip",
"details": None
})
return False
if backup_dir and not no_backup:
try:
os.makedirs(backup_dir, exist_ok=True)
dest = os.path.join(backup_dir, os.path.basename(path))
shutil.move(path, dest)
logger.info(f"已将文件移动到备份目录: {path} -> {dest}")
if db_writer:
db_writer.enqueue_operation({
"operation_type": "backup_move",
"file_path": path,
"file_hash": None,
"reason": "moved_to_backup",
"details": dest
})
return True
except Exception as e:
logger.warning(f"移动到备份目录失败 {path}: {e}")
try:
os.remove(path)
logger.info(f"已删除文件: {path}")
if db_writer:
db_writer.enqueue_operation({
"operation_type": "delete",
"file_path": path,
"file_hash": None,
"reason": "deleted",
"details": None
})
return True
except Exception as e:
logger.error(f"删除文件失败 {path}: {e}")
return False
# -------------------------
# DatabaseWriterThread (with detection & auto-migrate)
# -------------------------
class DatabaseWriterThread(threading.Thread):
"""
Single-threaded DB writer with:
- internal queue for files/ops
- lock detection and automatic recovery
- optional automatic DB migration to a safe directory
"""
def __init__(self, db_path: str = "music_cleaner.db", batch_limit:int = 200, flush_interval: float = 1.0, lock_detect_timeout: float = 8.0, max_retries:int=3, auto_migrate:bool=True):
super().__init__(daemon=True)
self.db_path = str(db_path)
self.batch_limit = batch_limit
self.flush_interval = flush_interval
self.lock_detect_timeout = lock_detect_timeout
self.max_retries = max_retries
self.auto_migrate = auto_migrate
self._conn: Optional[sqlite3.Connection] = None
self._queue: "queue.Queue[Tuple[str, Any]]" = queue.Queue()
self._stop_event = threading.Event()
self.started_flag = False
self._last_write_time = 0.0
self._consecutive_failures = 0
def _connect(self):
try:
conn = sqlite3.connect(
self.db_path,
timeout=3,
isolation_level=None,
check_same_thread=False,
)
conn.execute("PRAGMA journal_mode=WAL;")
conn.execute("PRAGMA synchronous=NORMAL;")
conn.execute(
"""
CREATE TABLE IF NOT EXISTS files (
id INTEGER PRIMARY KEY AUTOINCREMENT,
file_path TEXT UNIQUE,
file_hash TEXT,
file_size INTEGER,
file_mtime REAL,
created_at TEXT
);
"""
)
conn.execute(
"""
CREATE TABLE IF NOT EXISTS operations (
id INTEGER PRIMARY KEY AUTOINCREMENT,
operation_type TEXT,
file_path TEXT,
file_hash TEXT,
reason TEXT,
details TEXT,
created_at TEXT
);
"""
)
conn.commit()
logger.info(f"数据库连接成功:{self.db_path}")
return conn
except Exception as e:
logger.error(f"数据库连接失败 {self.db_path}: {e}")
return None
def start(self):
if not self.started_flag:
self._conn = self._connect()
self.started_flag = True
super().start()
def stop(self):
self._stop_event.set()
def join(self, timeout=None):
self._stop_event.set()
super().join(timeout)
if self._conn:
try:
self._conn.commit()
self._conn.close()
except:
pass
def enqueue_file(self, record: Dict[str, Any]):
self._queue.put(("file", record))
def enqueue_operation(self, record: Dict[str, Any]):
self._queue.put(("operation", record))
def run(self):
logger.info("DatabaseWriterThread 启动")
buffer_files = []
buffer_ops = []
last_flush_time = time.time()
while not self._stop_event.is_set():
try:
item_type, data = self._queue.get(timeout=self.flush_interval)
if item_type == "file":
buffer_files.append(data)
elif item_type == "operation":
buffer_ops.append(data)
except queue.Empty:
pass
now = time.time()
need_flush = False
if len(buffer_files) >= self.batch_limit or len(buffer_ops) >= self.batch_limit:
need_flush = True
if now - last_flush_time >= self.flush_interval:
need_flush = True
if need_flush:
ok = self._flush(buffer_files, buffer_ops)
if ok:
buffer_files.clear()
buffer_ops.clear()
last_flush_time = now
self._flush(buffer_files, buffer_ops)
logger.info("DatabaseWriterThread 结束(队列已清空)")
def _flush(self, files: List[Dict[str,Any]], ops: List[Dict[str,Any]]) -> bool:
if not self._conn:
logger.error("数据库连接失效conn = None尝试重新连接…")
self._conn = self._connect()
if not self._conn:
return False
if not files and not ops:
return True
start = time.time()
ok = False
last_err = None
for attempt in range(self.max_retries):
try:
cur = self._conn.cursor()
for rec in files:
cur.execute(
"""
INSERT OR REPLACE INTO files (file_path, file_hash, file_size, file_mtime, created_at)
VALUES (?, ?, ?, ?, ?)
""",
(
rec.get("file_path"),
rec.get("file_hash"),
rec.get("file_size"),
rec.get("file_mtime"),
rec.get("created_at"),
)
)
for rec in ops:
cur.execute(
"""
INSERT INTO operations (operation_type, file_path, file_hash, reason, details, created_at)
VALUES (?, ?, ?, ?, ?, ?)
""",
(
rec.get("operation_type"),
rec.get("file_path"),
rec.get("file_hash"),
rec.get("reason"),
rec.get("details"),
rec.get("created_at", datetime.now().isoformat()),
)
)
self._conn.commit()
ok = True
self._consecutive_failures = 0
break
except Exception as e:
last_err = e
logger.warning(f"批量写入数据库失败 (第 {attempt+1}/{self.max_retries} 次){e}")
if "locked" in str(e).lower():
time.sleep(0.8 + attempt * 0.4)
continue
time.sleep(0.5)
if not ok:
self._consecutive_failures += 1
elapsed = time.time() - start
logger.error(f"写入失败超过重试次数:{last_err}")
if elapsed > self.lock_detect_timeout or "locked" in str(last_err).lower():
logger.error("检测到数据库长期锁定,尝试恢复连接…")
try:
self._conn.close()
except:
pass
self._conn = self._connect()
if self._conn:
logger.info("数据库重连成功")
return False
if self.auto_migrate:
logger.error("数据库重连失败,准备自动迁移数据库…")
return self._try_auto_migrate()
return ok
def _try_auto_migrate(self) -> bool:
try:
safe_dir = "/var/db/music_duplicate_cleaner"
os.makedirs(safe_dir, exist_ok=True)
new_path = os.path.join(safe_dir, "music_cleaner.db")
try:
shutil.copy2(self.db_path, new_path)
logger.info(f"数据库已迁移: {self.db_path} -> {new_path}")
except Exception as e:
logger.error(f"数据库迁移失败: {e}")
return False
self.db_path = new_path
self._conn = self._connect()
if self._conn:
logger.info("迁移后的数据库连接成功,继续运行")
return True
else:
return False
except Exception as e:
logger.error(f"自动迁移过程异常: {e}")
return False
# =====================================================
# 音频指纹分析
# =====================================================
class AudioFingerprint:
def __init__(self):
self.ok = LIBROSA_AVAILABLE or SCIPY_AVAILABLE
def process(self, path: str) -> Optional[np.ndarray]:
"""
返回指纹向量numpy array或 None
"""
if not self.ok:
logger.debug(f"音频指纹模块不可用,跳过: {path}")
return None
try:
if LIBROSA_AVAILABLE:
y, sr = librosa.load(path, sr=22050, mono=True)
mfcc = librosa.feature.mfcc(y=y, sr=sr, n_mfcc=20)
fp = np.mean(mfcc, axis=1)
return fp
# librosa 不可用时,用 scipy_signal
if SCIPY_AVAILABLE:
import soundfile as sf
data, sr = sf.read(path)
if data.ndim > 1:
data = data.mean(axis=1)
freqs, times, Sxx = scipy_signal.spectrogram(data, sr)
fp = np.mean(Sxx, axis=1)
return fp
return None
except Exception as e:
logger.error(f"提取音频指纹失败 {path}: {e}")
return None
# =====================================================
# 扫描器:多线程扫描 + 入队写数据库
# =====================================================
class FileScanner:
EXT_AUDIO = {".mp3", ".aac", ".flac", ".ogg", ".wav", ".m4a", ".ape", ".wma", ".opus"}
def __init__(self, db_writer: DatabaseWriterThread, workers:int=8):
self.db_writer = db_writer
self.workers = choose_worker_count(workers)
self.audio_fp = AudioFingerprint()
def scan(self, root: str):
"""
遍历路径,将文件元数据推送到数据库队列。
"""
root = os.path.abspath(root)
logger.info(f"开始扫描路径: {root}")
file_list: List[str] = []
for base, dirs, files in os.walk(root):
for f in files:
full = os.path.join(base, f)
ext = os.path.splitext(full)[1].lower()
if ext in self.EXT_AUDIO:
file_list.append(full)
logger.info(f"扫描完成,共发现音频文件: {len(file_list)}")
with ThreadPoolExecutor(max_workers=self.workers) as ex:
futures = {ex.submit(self._process_one, path): path for path in file_list}
for fut in as_completed(futures):
try:
fut.result()
except Exception as e:
logger.error(f"处理文件异常: {e}")
def _process_one(self, path: str):
"""
获取文件大小、时间、hash快速并提交数据库线程。
"""
try:
st = os.stat(path)
except Exception as e:
logger.debug(f"无法读取文件 stat: {path}: {e}")
return
# 轻量快速 hash仅文件大小>1MB才计算
file_hash = ""
if st.st_size > 1_000_000:
file_hash = file_sha256(path)
else:
file_hash = f"SMALL-{st.st_size}-{int(st.st_mtime)}"
record = {
"file_path": path,
"file_hash": file_hash,
"file_size": st.st_size,
"file_mtime": st.st_mtime,
"created_at": datetime.now().isoformat(),
}
self.db_writer.enqueue_file(record)
# =====================================================
# 相似度检测与去重决策
# =====================================================
class DuplicateFinder:
"""
基于 DB 快照进行相似群组查找
"""
def __init__(self, db_path: str):
self.db_path = db_path
def _read_files_from_db(self) -> List[Dict[str, Any]]:
out = []
try:
conn = sqlite3.connect(self.db_path, timeout=30)
cur = conn.cursor()
cur.execute("SELECT file_path, file_hash, file_size FROM files WHERE file_path IS NOT NULL")
for row in cur.fetchall():
out.append({"path": row[0], "hash": row[1], "size": row[2]})
except Exception as e:
logger.warning(f"读取 DB 列表失败: {e}")
finally:
try:
conn.close()
except:
pass
return out
def group_by_name(self, files: List[Dict[str, Any]]) -> List[List[Dict[str, Any]]]:
groups = {}
for f in files:
key = Path(f["path"]).stem.lower()
# remove common music tokens
key = re.sub(r"(320k|128k|192k|256k|v0|v2|vbr|cb|joint stereo|stereo)", "", key)
key = re.sub(r"[\._\-]+", " ", key).strip()
groups.setdefault(key, []).append(f)
return [g for g in groups.values() if len(g) > 1]
def find_music_groups(self) -> List[List[Dict[str,Any]]]:
files = self._read_files_from_db()
name_groups = self.group_by_name(files)
result = []
for g in name_groups:
if len(g) <= 1:
continue
used = set()
for i in range(len(g)):
if i in used:
continue
base = g[i]
cluster = [base]
used.add(i)
for j in range(i+1, len(g)):
if j in used:
continue
# 检查文件大小是否相似(音乐文件通常大小相近表示相同)
try:
size_diff = abs(base.get("size", 0) - g[j].get("size", 0))
if size_diff < 1024: # 1KB 以内认为相同
cluster.append(g[j])
used.add(j)
except Exception:
pass
if len(cluster) > 1:
result.append(cluster)
logger.info(f"查找完成:发现 {len(result)} 音乐候选组")
return result
# -------------------------
# DuplicateCleaner high-level operations
# -------------------------
class MusicDuplicateCleaner:
def __init__(self, target_dirs: List[str], db_path: str="music_cleaner.db", prefer_folder: Optional[str]=None, workers: int=0, auto_migrate: bool=True):
self.target_dirs = target_dirs
self.db_path = db_path
self.prefer_folder = prefer_folder
self.db_writer = DatabaseWriterThread(db_path=db_path, auto_migrate=auto_migrate)
# start writer
if not getattr(self.db_writer, "started_flag", False):
self.db_writer.start()
self.scanner = FileScanner(db_writer=self.db_writer, workers=workers)
self.finder = DuplicateFinder(db_path=self.db_path)
def scan_all(self):
for d in self.target_dirs:
self.scanner.scan(d)
def remove_groups(self, groups: List[List[Dict[str,Any]]], dry_run: bool=True, no_backup: bool=False) -> Tuple[List[str], List[str]]:
kept = []
deleted = []
for group in groups:
if not group:
continue
# choose keeper
keeper = None
if self.prefer_folder:
for f in group:
if self.prefer_folder in f["path"]:
keeper = f
break
if not keeper:
keeper = max(group, key=lambda x: x.get("size", 0))
kept.append(keeper["path"])
for f in group:
p = f["path"]
if p == keeper["path"]:
continue
if dry_run:
logger.info(f"[dry-run] 删除 {p} (保留 {keeper['path']})")
self.db_writer.enqueue_operation({
"operation_type": "planned_delete",
"file_path": p,
"file_hash": f.get("hash"),
"reason": "dry_run",
"details": None,
"created_at": datetime.now().isoformat()
})
deleted.append(p)
else:
ok = safe_remove(p, no_backup=no_backup, backup_dir=None, db_writer=self.db_writer)
if ok:
deleted.append(p)
else:
logger.info(f"跳过删除(可能为硬链接或权限问题): {p}")
return kept, deleted
def run_music_cleanup(self, dry_run: bool=True, no_backup: bool=False) -> Dict[str,Any]:
logger.info("开始音乐清理")
self.scan_all()
logger.info("等待 db_writer 完成写入任务...")
# wait until queue is drained or timeout
start = time.time()
while not self.db_writer._queue.empty():
time.sleep(0.5)
if time.time() - start > 600:
logger.error("等待 db_writer 超过 600 秒,提前退出")
break
groups = self.finder.find_music_groups()
kept, deleted = self.remove_groups(groups, dry_run=dry_run, no_backup=no_backup)
return {"kept": kept, "deleted": deleted, "groups": len(groups)}
# =====================================================
# CLI & Main Function
# =====================================================
def parse_args():
parser = argparse.ArgumentParser(description="Music Duplicate Cleaner - 音乐文件去重专用版本")
parser.add_argument(
"-d", "--dirs",
nargs="+",
required=True,
help="指定需要扫描的目录(一个或多个)"
)
parser.add_argument(
"--prefer",
type=str,
default=None,
help="优先保留的路径片段(如果匹配文件路径则优先保留)"
)
parser.add_argument(
"--dry-run",
action="store_true",
help="演示模式:仅显示将要删除的文件,不实际删除"
)
parser.add_argument(
"--no-backup",
action="store_true",
help="删除时不创建备份(谨慎)"
)
parser.add_argument(
"--workers",
type=int,
default=0,
help="扫描线程数(默认自动计算)"
)
parser.add_argument(
"--db",
type=str,
default="music_cleaner.db",
help="使用的数据库文件"
)
parser.add_argument(
"--migrate",
action="store_true",
help="强制允许自动迁移数据库(锁死时会迁移)"
)
return parser.parse_args()
def main():
args = parse_args()
logger.info("==============================================")
logger.info(" Music Duplicate Cleaner - 音乐去重 ")
logger.info("==============================================")
logger.info(f"扫描目录:{args.dirs}")
logger.info(f"数据库文件:{args.db}")
logger.info(f"优先保留路径片段:{args.prefer}")
if args.dry_run:
logger.info("警告dry-run 模式(不会删除任何文件)")
if args.no_backup:
logger.warning("危险:已启用 --no-backup不会创建备份")
cleaner = MusicDuplicateCleaner(
target_dirs=args.dirs,
db_path=args.db,
prefer_folder=args.prefer,
workers=args.workers,
auto_migrate=args.migrate,
)
result = None
try:
result = cleaner.run_music_cleanup(
dry_run=args.dry_run,
no_backup=args.no_backup,
)
except Exception as e:
logger.error(f"运行清理任务发生异常: {e}", exc_info=True)
finally:
# ensure writer shutdown
try:
cleaner.db_writer.stop()
cleaner.db_writer.join(timeout=10)
except Exception:
pass
logger.info("所有任务完成。")
if result is not None:
logger.info("========== 清理结果JSON 格式) ==========")
try:
import json
logger.info(json.dumps(result, indent=2, ensure_ascii=False))
except Exception:
logger.info(result)
if __name__ == "__main__":
main()
# =====================================================
# 示例命令
# =====================================================
"""
# --- 示例:对 /volume2/music 扫描并自动清理dry-run不会删除
python3 music_duplicate_cleaner.py --dirs /volume2/music --dry-run
# --- 强制真实删除(无备份,不推荐)
python3 music_duplicate_cleaner.py --dirs /volume2/music --no-backup
# --- 指定优先目录(例如你下载的目录中保留高质量音乐)
python3 music_duplicate_cleaner.py --dirs /volume2/music --prefer "/volume2/music/FLAC"
# --- 减小写入锁风险(推荐加)
python3 music_duplicate_cleaner.py --dirs /volume2/music --migrate
# --- 指定线程
python3 music_duplicate_cleaner.py --dirs /volume2/music --workers 16
# --- 扫描多个目录
python3 music_duplicate_cleaner.py --dirs /volume2/music /volume2/downloads/music
"""