import os import re import argparse import shutil from datetime import datetime # ====================== 配置区 ====================== RULES_FILE = [ (r"^[A-Za-z] ", ""), (r"\[.*?\]", ""), (r"【.*?】", ""), (r"丨", ""), (r"\b(2160p|1080p|720p|HDR|DV|DoVi|HEVC|x265|x264|DDP|AAC|Atmos|TrueHD|中英双语|双语字幕)\b", "", re.IGNORECASE), (r"\.+", ""), (r"\s", ""), (r"[^\u4e00-\u9fa5a-zA-Z0-9().:+-]", ""), ] RULES_DIR = [ (r"^[A-Za-z] ", ""), (r"丨", ""), (r"\b(2160p|1080p|720p|HDR|DV|DoVi|HEVC|x265|x264|DDP|AAC|Atmos|TrueHD|中英双语|双语字幕)\b", "", re.IGNORECASE), (r"\.+", ""), (r"\s", ""), (r"[^\u4e00-\u9fa5a-zA-Z0-9().:+-]", ""), ] ALLOW_EXT = {".mkv", ".mp4", ".mov", ".srt", ".ass", ".sub", ".strm",".flv",".avi",".wmv",".ts",".rmvb",".webm",".mp3",".flac",".aac",".wav",".ogg",".m4a",".wma",".alac",".rm",".iso"} # ==================================================== def full2half(s): result = [] for char in s: code = ord(char) if code == 12288: char = chr(32) elif 65281 <= code <= 65374: char = chr(code - 65248) result.append(char) return "".join(result) def clean_file_name(name: str) -> str: name_no_ext, ext = os.path.splitext(name) name_no_ext = full2half(name_no_ext) for item in RULES_FILE: if len(item) == 3: pat, repl, flag = item name_no_ext = re.sub(pat, repl, name_no_ext, flags=flag) else: pat, repl = item name_no_ext = re.sub(pat, repl, name_no_ext) name_no_ext = name_no_ext.strip() return name_no_ext + ext def clean_dir_name(name: str) -> str: name = full2half(name) year_match = re.search(r"\[(\d{4})\]", name) year_str = "" if year_match: year_str = year_match.group(1) name = re.sub(r"\[.*?\]", "", name) for item in RULES_DIR: if len(item) == 3: pat, repl, flag = item name = re.sub(pat, repl, name, flags=flag) else: pat, repl = item name = re.sub(pat, repl, name) name = name.strip() if year_str and name: name = f"{name}({year_str})" return name def get_quality_tag(filename:str): m = re.search(r"\b(2160p|1080p|720p)\b", filename, re.IGNORECASE) if m: return m.group(1).lower() return None def resolve_conflict(dst_dir:str, base_name:str, ext:str, orig_filename:str): """解决输出目录同名冲突：优先追加画质，无画质则追加(1)(2)""" candidate = base_name + ext dst_path = os.path.join(dst_dir, candidate) if not os.path.exists(dst_path): return candidate qual = get_quality_tag(orig_filename) if qual: candidate = f"{base_name}({qual}){ext}" dst_path = os.path.join(dst_dir, candidate) if not os.path.exists(dst_path): return candidate idx = 1 while True: candidate = f"{base_name}({idx}){ext}" dst_path = os.path.join(dst_dir, candidate) if not os.path.exists(dst_path): return candidate idx +=1 def write_log(msg: str, log_file: str): now = datetime.now().strftime("%Y-%m-%d %H:%M:%S") line = f"[{now}] {msg}" print(line) with open(log_file, "a", encoding="utf-8") as f: f.write(line + "\n") f.flush() def process_tree(src_root:str, dst_root:str, dry_run:bool, log_path:str): stat = { "scan_dir_total":0, "scan_file_total":0, "media_file_total":0, "file_success":0, "file_fail":0, "dir_success":0, "dir_fail":0 } for src_dir, dirs, files in os.walk(src_root): stat["scan_dir_total"] +=1 rel = os.path.relpath(src_dir, src_root) if rel == ".": dst_current_dir = dst_root else: dst_current_dir = os.path.join(dst_root, clean_dir_name(os.path.basename(src_dir))) # 创建输出目录 if not dry_run: try: os.makedirs(dst_current_dir, exist_ok=True) stat["dir_success"] +=1 except Exception as e: stat["dir_fail"] +=1 write_log(f"【目录创建失败】{dst_current_dir} | {e}", log_path) else: write_log(f"【预览-输出目录】{dst_current_dir}", log_path) # 处理本目录全部文件 for fname in files: stat["scan_file_total"] += 1 src_file = os.path.join(src_dir, fname) ext = os.path.splitext(fname)[1].lower() if ext in ALLOW_EXT: stat["media_file_total"] +=1 new_fname = clean_file_name(fname) base, ext_out = os.path.splitext(new_fname) final_fname = resolve_conflict(dst_current_dir, base, ext_out, fname) dst_file = os.path.join(dst_current_dir, final_fname) write_log(f"待复制文件-> {fname} → {final_fname}", log_path) if not dry_run: try: shutil.copy2(src_file, dst_file) stat["file_success"] +=1 except Exception as e: stat["file_fail"] +=1 write_log(f"【文件复制失败】src:{src_file} dst:{dst_file} | {e}", log_path) return stat def main(): parser = argparse.ArgumentParser(description="媒体文件名镜像清洗工具，源目录只读，输出全新镜像目录") parser.add_argument("--dir", required=True, help="原始源媒体根目录") parser.add_argument("--output", required=True, help="输出镜像根目录") parser.add_argument("--apply", action="store_true", help="加上此参数才真正复制生成镜像；默认预览dry‑run") parser.add_argument("--log", default="./rename_log.txt", help="日志文件路径，默认同脚本目录rename_log.txt") args = parser.parse_args() src_dir = args.dir out_dir = args.output dry_run = not args.apply log_path = args.log if not os.path.isdir(src_dir): print(f"错误：源目录不存在 -> {src_dir}") return with open(log_path, "w", encoding="utf-8") as f: f.flush() write_log("===== 媒体文件名镜像清洗【源目录只读，生成镜像副本】 =====", log_path) write_log(f"源目录：{src_dir}", log_path) write_log(f"输出镜像目录：{out_dir}", log_path) write_log(f"预览模式 DRY_RUN = {dry_run}", log_path) stat = process_tree(src_dir, out_dir, dry_run, log_path) write_log("----------任务统计汇总----------", log_path) write_log(f"总扫描目录：{stat['scan_dir_total']}", log_path) write_log(f"总扫描文件：{stat['scan_file_total']}，媒体文件：{stat['media_file_total']}", log_path) write_log(f"目录：成功创建 {stat['dir_success']}，失败 {stat['dir_fail']}", log_path) write_log(f"文件：成功复制 {stat['file_success']}，失败 {stat['file_fail']}", log_path) write_log("============================================================", log_path) if __name__ == "__main__": main()