This commit is contained in:
CrazyJasonwell
2026-10-07 11:07:00 +08:00
parent e88e4de0f5
commit 290ff8984f
96 changed files with 213854 additions and 0 deletions
Binary file not shown.
+124
View File
@@ -0,0 +1,124 @@
import os
import re
from datetime import datetime
# ====================== 【配置区】 ======================
MEDIA_FOLDER = "/Volumes/JoyDisk/myMedia/olstrms/电影/未分类/2024"
DRY_RUN = True # True=预览不修改;False=实际重命名
SKIP_PROCESSED = False
LOG_FILE = "./rename_log.txt"
# 清洗规则
RULES = [
(r"^[A-Za-z] ", ""), #1.移除开头单个字母+空格
(r"\[.*?\]", ""), #2.移除英文方括号及内部内容
(r"【.*?】", ""), #3.移除中文【】及内部内容
(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().:+-]", ""),
]
# 加入strm
ALLOW_EXT = {".mkv", ".mp4", ".mov", ".srt", ".ass", ".sub", ".strm"}
# =======================================================
def clean_filename(name: str) -> str:
name_no_ext, ext = os.path.splitext(name)
# 全角转半角
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)
name_no_ext = full2half(name_no_ext)
for item in RULES:
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 write_log(msg: str):
now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
line = f"[{now}] {msg}"
print(line)
# 增加flush,强制写入磁盘,解决日志文件不更新
with open(LOG_FILE, "a", encoding="utf-8") as f:
f.write(line + "\n")
f.flush()
if __name__ == "__main__":
# 每次运行前清空日志文件
with open(LOG_FILE, "w", encoding="utf-8") as f:
f.flush()
write_log("===== 文件名+文件夹清洗(移除开头A 中文片名前缀) =====")
write_log(f"根目录:{MEDIA_FOLDER}")
write_log(f"预览模式 DRY_RUN = {DRY_RUN}")
file_rename_list = []
dir_rename_list = []
total_scan_dirs = 0
total_all_files = 0 # 全部扫描到的文件(所有后缀)
total_media_files = 0 # 媒体文件(ALLOW_EXT后缀)
for root, dirs, files in os.walk(MEDIA_FOLDER, topdown=False):
total_scan_dirs += 1
for fname in files:
total_all_files +=1 # 任何文件都计入总扫描文件
ext = os.path.splitext(fname)[1].lower()
if ext in ALLOW_EXT:
total_media_files += 1
# 调试:打印所有扫描到的媒体文件(无论是否修改)
write_log(f"[扫描媒体] {fname}")
new_name = clean_filename(fname)
if new_name != fname:
if SKIP_PROCESSED and re.search(r"\(\d{4}\)", new_name):
write_log(f"【跳过文件】{fname},检测年份括号")
continue
old_path = os.path.join(root, fname)
new_path = os.path.join(root, new_name)
file_rename_list.append((old_path, new_path))
write_log(f"待修改文件-> {fname} → {new_name}")
# 处理文件夹
for dname in dirs:
new_name = clean_filename(dname)
if new_name != dname:
if SKIP_PROCESSED and re.search(r"\(\d{4}\)", new_name):
write_log(f"【跳过文件夹】{dname},检测年份括号")
continue
old_path = os.path.join(root, dname)
new_path = os.path.join(root, new_name)
dir_rename_list.append((old_path, new_name))
write_log(f"待修改文件夹-> {dname} → {new_name}")
# ========== 执行阶段 ==========
if not DRY_RUN:
# 先文件
for old, new in file_rename_list:
try:
os.rename(old, new)
except Exception as e:
write_log(f"【文件重命名失败】{old} -> {new} | 错误:{e}")
# 后文件夹
for old, new in dir_rename_list:
try:
os.rename(old, new)
except Exception as e:
write_log(f"【文件夹重命名失败】{old} -> {new} | 错误:{e}")
# 最终汇总输出
write_log(f"任务完成,总扫描目录:{total_scan_dirs},待修改文件夹:{len(dir_rename_list)}")
write_log(f"总扫描文件:{total_all_files},媒体文件:{total_media_files},待修改文件:{len(file_rename_list)}")
write_log("============================================================")
+100
View File
@@ -0,0 +1,100 @@
import os
import re
from datetime import datetime
# ====================== 【配置区,请修改】 ======================
MEDIA_FOLDER = "/Volumes/JoyDisk/myMedia/olstrms/电影/未分类/2025" # Mac挂载NAS路径
# MEDIA_FOLDER = "/media/movies" # 飞牛容器内运行用这个
DRY_RUN = True # True=预览,不修改;确认无误改为False执行改名
SKIP_PROCESSED = False # True=跳过已处理(文件名含(20xx)年份)
LOG_FILE = "./rename_log.txt"
# =================================================================
# 清洗正则规则(文件夹、文件共用一套)
RULES = [
(r"^[A-Za-z] ", ""), # 【新增】移除开头单个英文字母+空格:A 流浪地球 → 流浪地球
(r"(?<=[a-zA-Z0-9])\.(?=[a-zA-Z0-9])", " "), # Dune.Part → Dune Part
(r"\[.*?\]", ""), # 删除全部 [xxx] 方括号内容
(r"\s*(2160p|1080p|720p|HDR|DV|DoVi|HEVC|x265|x264|DDP|AAC|Atmos|TrueHD)\b", ""),
(r"\.+", " "), # 剩下连续多个点替换成空格
]
# 支持的文件后缀
ALLOW_EXT = {".mkv", ".mp4", ".mov", ".srt", ".ass", ".sub"}
def clean_name(name: str) -> str:
"""统一清洗函数,文件夹名、文件名共用"""
name_no_ext, ext = os.path.splitext(name)
# 处理 A . B . C 这种「字母 空格 . 空格 字母」
while re.search(r"(\w) \. (\w)", name_no_ext):
name_no_ext = re.sub(r"(\w) \. (\w)", r"\1\2", name_no_ext)
# 执行所有正则替换规则
for pattern, repl in RULES:
name_no_ext = re.sub(pattern, repl, name_no_ext)
# 合并连续空格,去除首尾空格
name_no_ext = re.sub(r"\s+", " ", name_no_ext).strip()
return name_no_ext + ext
def write_log(msg: str):
time_str = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
line = f"[{time_str}] {msg}"
print(line)
with open(LOG_FILE, "a", encoding="utf-8") as f:
f.write(line + "\n")
if __name__ == "__main__":
write_log("===== 文件名+文件夹清洗(移除开头A 中文片名前缀) =====")
write_log(f"根目录:{MEDIA_FOLDER}")
write_log(f"预览模式 DRY_RUN = {DRY_RUN}")
dir_rename_list = []
file_rename_list = []
# 遍历收集(自底向上,防止目录路径错乱)
for root, dirs, files in os.walk(MEDIA_FOLDER, topdown=False):
# 文件
for fname in files:
ext = os.path.splitext(fname)[1].lower()
if ext not in ALLOW_EXT:
continue
new_name = clean_name(fname)
if new_name != fname:
if SKIP_PROCESSED and re.search(r"\(\d{4}\)", new_name):
write_log(f"跳过文件(疑似已处理): {fname}")
continue
old_path = os.path.join(root, fname)
new_path = os.path.join(root, new_name)
file_rename_list.append((old_path, new_path))
# 文件夹
for dname in dirs:
new_name = clean_name(dname)
if new_name != dname:
if SKIP_PROCESSED and re.search(r"\(\d{4}\)", new_name):
write_log(f"跳过文件夹(疑似已处理): {dname}")
continue
old_path = os.path.join(root, dname)
new_path = os.path.join(root, new_name)
dir_rename_list.append((old_path, new_name))
# 先改文件夹,再改文件
dir_count = 0
for old, new in dir_rename_list:
write_log(f"【文件夹】{os.path.basename(old)} → {os.path.basename(new)}")
dir_count +=1
if not DRY_RUN:
os.rename(old, new)
file_count =0
for old, new in file_rename_list:
write_log(f"【文件】{os.path.basename(old)} → {os.path.basename(new)}")
file_count +=1
if not DRY_RUN:
os.rename(old, new)
write_log(f"任务完成,待修改文件夹:{dir_count},待修改文件:{file_count}")
write_log("="*60 + "\n")
+92
View File
@@ -0,0 +1,92 @@
import os
import re
from datetime import datetime
# ====================== 【配置区,请修改】 ======================
MEDIA_FOLDER = "/Volumes/JoyDisk/myMedia/olstrms/电影/未分类/2024"
DRY_RUN = True # True预览不修改;确认后改为False
SKIP_PROCESSED = False
LOG_FILE = "./rename_log.txt"
# =================================================================
RULES = [
(r"^[A-Za-z] ", ""), #1.开头单个字母+空格
(r"\[.*?\]", ""), #2.英文方括号
(r"【.*?】", ""), #2.中文方括号
(r"\b(2160p|1080p|720p|HDR|DV|DoVi|HEVC|x265|x264|DDP|AAC|Atmos|TrueHD|中英双语)\b", "", re.IGNORECASE),
(r"\.+", ""), #4.连续多点删除
(r"\s", ""), #5.删除全部空格
]
ALLOW_EXT = {".mkv", ".mp4", ".mov", ".srt", ".ass", ".sub"}
def clean_name(name: str) -> str:
name_no_ext, ext = os.path.splitext(name)
for item in RULES:
if len(item) == 3:
pattern, repl, flag = item
name_no_ext = re.sub(pattern, repl, name_no_ext, flags=flag)
else:
pattern, repl = item
name_no_ext = re.sub(pattern, repl, name_no_ext)
name_no_ext = re.sub(r"\s+", " ", name_no_ext).strip()
return name_no_ext + ext
def write_log(msg: str):
time_str = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
line = f"[{time_str}] {msg}"
print(line)
with open(LOG_FILE, "a", encoding="utf-8") as f:
f.write(line + "\n")
if __name__ == "__main__":
write_log("===== 文件名+文件夹清洗【递归处理全部下级子目录】 =====")
write_log(f"根目录:{MEDIA_FOLDER}")
write_log(f"预览模式 DRY_RUN = {DRY_RUN}")
write_log("说明:递归扫描全部子目录,先收集所有待改,执行时【先文件、后文件夹】避免路径失效")
dir_rename_list = []
file_rename_list = []
for root, dirs, files in os.walk(MEDIA_FOLDER, topdown=False):
write_log(f"\n==== 扫描目录:{root} | 当前目录文件数量:{len(files)} ====")
# 处理文件
for fname in files:
ext = os.path.splitext(fname)[1].lower()
if ext not in ALLOW_EXT:
continue
new_name = clean_name(fname)
if new_name != fname:
if SKIP_PROCESSED and re.search(r"\(\d{4}\)", new_name):
write_log(f"【跳过文件】{fname},检测年份括号")
continue
old_path = os.path.join(root, fname)
new_path = os.path.join(root, new_name)
file_rename_list.append((old_path, new_path))
write_log(f"【待改文件】{fname} → {new_name}")
# 处理文件夹
for dname in dirs:
new_name = clean_name(dname)
if new_name != dname:
if SKIP_PROCESSED and re.search(r"\(\d{4}\)", new_name):
write_log(f"【跳过文件夹】{dname},检测年份括号")
continue
old_path = os.path.join(root, dname)
new_path = os.path.join(root, new_name)
dir_rename_list.append((old_path, new_name))
write_log(f"【待改文件夹】{dname} → {new_name}")
# 执行顺序:先全部文件,再全部文件夹
file_count = 0
for old, new in file_rename_list:
write_log(f"【执行文件】{os.path.basename(old)} → {os.path.basename(new)}")
file_count +=1
if not DRY_RUN:
os.rename(old, new)
dir_count = 0
for old, new in dir_rename_list:
write_log(f"【执行文件夹】{os.path.basename(old)} → {os.path.basename(new)}")
dir_count +=1
if not DRY_RUN:
os.rename(old, new)
write_log(f"\n任务完成,待修改文件夹:{dir_count},待修改文件:{file_count}")
write_log("="*60 + "\n")
+326
View File
@@ -0,0 +1,326 @@
import os
import re
import argparse
import shutil
from datetime import datetime
LOG_FILE = "./rename_log.txt"
SKIP_DIRS = [
"@eaDir",
]
SKIP_FILES = {
".DS_Store",
"Thumbs.db",
"desktop.ini",
}
RULES_FILE = [
(r"^[A-Za-z] ", ""),
(r"\[.*?\]", ""),
(r"【.*?】", ""),
(r"丨", ""),
(r"(?<!\w)(2160p|1080p|720p|HDR|DV|DoVi|HEVC|x265|x264|DDP|AAC|Atmos|TrueHD|BD|BluRay|BLURAY|WEBRip|WEB-DL|HD|4K|中英双语|双语字幕|双语|中文|国语|粤语|台语|中英双字|国语中字|繁体中字|简体中字|简繁中字|双语中字|官方中字|特效中字|收藏不迷路|地址发布页|www6v123net|wwwdygangme)(?!\w)", "", re.IGNORECASE),
(r"\.+", ""),
(r"\s", ""),
(r"[^\u4e00-\u9fa5a-zA-Z0-9().:+\-]", ""),
]
RULES_DIR = [
(r"【.*?】", ""),
(r"^\[(.*)\]$", r"\1"),
(r"\]\[", ""),
(r"^[A-Za-z] ", ""),
(r"丨", ""),
(r"(?<!\w)(2160p|1080p|720p|HDR|DV|DoVi|HEVC|x265|x264|DDP|AAC|Atmos|TrueHD|BD|BluRay|BLURAY|WEBRip|WEB-DL|HD|4K|中英双语|双语字幕|双语|中文|国语|粤语|台语|中英双字|国语中字|繁体中字|简体中字|简繁中字|双语中字|官方中字|特效中字|收藏不迷路|地址发布页|www6v123net|wwwdygangme)(?!\w)", "", re.IGNORECASE),
(r"\.+", ""),
(r"\s", ""),
(r"\(\)", ""),
(r"()", ""),
(r"[^\u4e00-\u9fa5a-zA-Z0-9().:+\-[\]]", ""),
]
FULL_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"}
STRM_ONLY_EXT = {".strm", ".srt", ".ass", ".sub"}
MAX_FILENAME_LEN = 245
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)
orig_no_ext = 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()
if len(name_no_ext) == 0:
name_no_ext = orig_no_ext.strip()
if len(name_no_ext) > MAX_FILENAME_LEN:
name_no_ext = name_no_ext[:MAX_FILENAME_LEN]
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)
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})"
if len(name) > MAX_FILENAME_LEN:
name = name[:MAX_FILENAME_LEN]
return name
def assign_unique_names(item_list):
name_counter = dict()
out = []
for it in item_list:
cand = it["candidate"]
cnt = name_counter.get(cand, 0)
if cnt == 0:
final = cand
else:
final = f"{cand}({cnt})"
name_counter[cand] = cnt + 1
new_it = dict(it)
new_it["final_basename"] = final
out.append(new_it)
return out
def write_log(msg: str, log_file: str):
now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
line = f"[{now}] {msg}"
print(line)
try:
with open(log_file, "a", encoding="utf-8") as f:
f.write(line + "\n")
f.flush()
except Exception:
print(f"[LOG-WRITE-FAILED] {line}")
def scan_full_tree(root_src, skip_dirs, skip_files, allow_ext):
"""
返回:
dir_nodes: list,每一个文件夹独立节点,包含:
src_abs: 当前文件夹完整源绝对路径
parent_src_abs: 父文件夹完整源绝对路径
orig_basename: 当前文件夹原始名字
candidate: 清洗后的候选basename
depth:目录深度,用于排序父先于子创建
file_nodes: list,媒体文件节点
root_src_abs:源根目录绝对路径
"""
dir_nodes = []
file_nodes = []
root_src_abs = os.path.abspath(root_src)
for src_root, dirs, files in os.walk(root_src, topdown=True):
remove_idx = []
for idx, dname in enumerate(dirs):
if dname in skip_dirs:
remove_idx.append(idx)
for idx in reversed(remove_idx):
del dirs[idx]
dirs[:] = dirs
src_root_abs = os.path.abspath(src_root)
parent_src_abs = os.path.dirname(src_root_abs)
depth = src_root_abs.count(os.sep)
basename = os.path.basename(src_root_abs)
# 每遍历到一个文件夹,就记录为一条dir_nodes(包含所有层级全部目录节点)
if src_root_abs != root_src_abs:
cand = clean_dir_name(basename)
if len(cand) == 0:
cand = basename
dir_nodes.append({
"src_abs": src_root_abs,
"parent_src_abs": parent_src_abs,
"orig_basename": basename,
"candidate": cand,
"depth": depth
})
#收集本目录下媒体文件
for fname in files:
if fname in skip_files:
continue
ext = os.path.splitext(fname)[1].lower()
if ext not in allow_ext:
continue
cand_f = clean_file_name(fname)
file_nodes.append({
"src_parent_abs": src_root_abs,
"orig_basename": fname,
"candidate": cand_f
})
return root_src_abs, dir_nodes, file_nodes
def group_by_parent_src_abs(item_list, key_name):
groups = dict()
for it in item_list:
k = it[key_name]
if k not in groups:
groups[k] = []
groups[k].append(it)
return groups
def main():
parser = argparse.ArgumentParser(description="镜像重命名【完整节点采集,严格1:1目录文件数量】")
parser.add_argument("--dir", required=True)
parser.add_argument("--output", required=True)
parser.add_argument("--apply", action="store_true")
parser.add_argument("--dry-run", action="store_true")
parser.add_argument("--strm-mode", action="store_true")
parser.add_argument("--log", default="./rename_log.txt")
args = parser.parse_args()
target_dir = args.dir
out_dir = args.output
log_path = args.log
DRY_RUN = True
if args.apply:
DRY_RUN = False
if args.dry_run:
DRY_RUN = True
if args.strm_mode:
ALLOW_EXT = STRM_ONLY_EXT
mode_info = "STRM模式【仅strm+字幕】"
else:
ALLOW_EXT = FULL_ALLOW_EXT
mode_info = "完整媒体模式"
if not os.path.isdir(target_dir):
print(f"源目录不存在 {target_dir}")
return
try:
with open(log_path, "w", encoding="utf-8") as f:
f.flush()
except Exception as e:
print(f"日志初始化警告 {e}")
write_log("===== 重构:全节点采集,源输出条目严格1:1 =====", log_path)
write_log(f"源目录:{target_dir}", log_path)
write_log(f"输出镜像目录:{out_dir}", log_path)
write_log(f"运行模式:{mode_info}", log_path)
write_log(f"预览模式 DRY_RUN = {DRY_RUN}", log_path)
stat = {
"src_dir_total":0,
"src_file_total":0,
"dir_success":0,
"dir_fail":0,
"file_success":0,
"file_fail":0
}
#阶段1:完整扫描,每一个文件夹都是独立节点
root_src_abs, dir_nodes, file_nodes = scan_full_tree(target_dir, SKIP_DIRS, SKIP_FILES, ALLOW_EXT)
stat["src_dir_total"] = len(dir_nodes)
stat["src_file_total"] = len(file_nodes)
write_log(f"【采集统计】源全部目录节点总数:{stat['src_dir_total']},源媒体文件总数:{stat['src_file_total']}", log_path)
#阶段2:按父源绝对路径分组,组内分配唯一basename,处理重名冲突
dir_groups = group_by_parent_src_abs(dir_nodes, "parent_src_abs")
dir_final = []
for p_abs, items in dir_groups.items():
assigned = assign_unique_names(items)
dir_final.extend(assigned)
file_groups = group_by_parent_src_abs(file_nodes, "src_parent_abs")
file_final = []
for p_abs, items in file_groups.items():
assigned = assign_unique_names(items)
file_final.extend(assigned)
#构建源绝对路径 → 输出basename映射
src_to_basename = {d["src_abs"]:d["final_basename"] for d in dir_final}
#构建源绝对路径 → 输出完整路径映射
src_to_out = dict()
src_to_out[root_src_abs] = out_dir
#按depth升序排序:浅目录优先,父一定先处理
dir_final_sorted = sorted(dir_final, key=lambda x:x["depth"])
for d in dir_final_sorted:
p_src = d["parent_src_abs"]
p_out = src_to_out[p_src]
child_out = os.path.join(p_out, d["final_basename"])
src_to_out[d["src_abs"]] = child_out
#打印目录日志
for it in dir_final:
orig = it["orig_basename"]
cand = it["candidate"]
final = it["final_basename"]
if cand != final:
write_log(f"【目录重名冲突】源目录[{orig}] 候选[{cand}] → 分配[{final}]", log_path)
else:
if orig != final:
write_log(f"待修改文件夹-> {orig} → {final}", log_path)
#打印文件日志
for it in file_final:
orig = it["orig_basename"]
cand = it["candidate"]
final = it["final_basename"]
if cand != final:
write_log(f"【文件重名冲突】源文件[{orig}] 候选[{cand}] → 分配[{final}]", log_path)
else:
if orig != final:
write_log(f"待修改文件-> {orig} → {final}", log_path)
#阶段3:按深度顺序创建目录,父目录优先
for d in dir_final_sorted:
full_out = src_to_out[d["src_abs"]]
if not DRY_RUN:
try:
os.makedirs(full_out, exist_ok=True)
stat["dir_success"] += 1
except OSError as e:
stat["dir_fail"] += 1
write_log(f"【目录创建失败】{full_out} | {e}", log_path)
#复制全部媒体文件
for it in file_final:
src_file = os.path.join(it["src_parent_abs"], it["orig_basename"])
dst_parent = src_to_out[it["src_parent_abs"]]
dst_file = os.path.join(dst_parent, it["final_basename"])
if not DRY_RUN:
try:
shutil.copy2(src_file, dst_file)
stat["file_success"] += 1
except OSError as e:
stat["file_fail"] +=1
write_log(f"【文件复制失败】src:{src_file} dst:{dst_file} | {e}", log_path)
write_log("----------任务统计汇总----------", log_path)
write_log(f"源采集目录节点总数量:{stat['src_dir_total']},源媒体文件总数量:{stat['src_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("【校验】预期目录输出={},实际成功={}".format(stat["src_dir_total"], stat["dir_success"]), log_path)
write_log("【校验】预期文件输出={},实际成功={}".format(stat["src_file_total"], stat["file_success"]), log_path)
write_log("============================================================", log_path)
if __name__ == "__main__":
main()
+291
View File
@@ -0,0 +1,291 @@
import os
import re
import argparse
import shutil
from datetime import datetime
# ====================== 【配置区】 ======================
LOG_FILE = "./rename_log.txt"
# 需要完全跳过的目录名称(basename,大小写敏感)
SKIP_DIRS = [
"@eaDir", # 群晖隐含目录
]
# 需要跳过的隐含文件名称(完整文件名匹配)
SKIP_FILES = {
".DS_Store",
"Thumbs.db",
"desktop.ini",
}
# 文件清洗规则
RULES_FILE = [
(r"^[A-Za-z] ", ""),
(r"\[.*?\]", ""),
(r"【.*?】", ""),
(r"丨", ""),
(r"(?<!\w)(2160p|1080p|720p|HDR|DV|DoVi|HEVC|x265|x264|DDP|AAC|Atmos|TrueHD|BD|BluRay|BLURAY|WEBRip|WEB-DL|HD|4K|中英双语|双语字幕|双语|中文|国语|粤语|台语|中英双字|国语中字|繁体中字|简体中字|简繁中字|双语中字|官方中字|特效中字)(?!\w)", "", re.IGNORECASE),
(r"\.+", ""),
(r"\s", ""),
(r"[^\u4e00-\u9fa5a-zA-Z0-9().:+\-]", ""),
]
# 文件夹清洗规则
RULES_DIR = [
(r"【.*?】", ""), # 删除中文全角【】以及内部内容
(r"^\[(.*)\]$", r"\1"), # 整串被[]包裹,只剥外层
(r"\]\[", ""), # 清除中间碎片 ][
(r"^[A-Za-z] ", ""),
(r"丨", ""),
(r"(?<!\w)(2160p|1080p|720p|HDR|DV|DoVi|HEVC|x265|x264|DDP|AAC|Atmos|TrueHD|BD|BluRay|BLURAY|WEBRip|WEB-DL|HD|4K|中英双语|双语字幕|双语|中文|国语|粤语|台语|中英双字|国语中字|繁体中字|简体中字|简繁中字|双语中字|官方中字|特效中字)(?!\w)", "", re.IGNORECASE),
(r"\.+", ""),
(r"\s", ""),
(r"\(\)", ""), # 清除英文空括号 ()
(r"()", ""), # 清除中文全角空括号 ()
(r"[^\u4e00-\u9fa5a-zA-Z0-9().:+\-[\]]", ""), # 允许文件夹内部保留[]
]
#完整全部媒体后缀(不带--strm-mode时使用)
FULL_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"}
# strm模式:仅strm+字幕
STRM_ONLY_EXT = {".strm", ".srt", ".ass", ".sub"}
MAX_FILENAME_LEN = 245 # mac文件名最大安全长度,预留扩展名位置
# =======================================================
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()
if len(name_no_ext) > MAX_FILENAME_LEN:
name_no_ext = name_no_ext[:MAX_FILENAME_LEN]
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)
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})"
if len(name) > MAX_FILENAME_LEN:
name = name[:MAX_FILENAME_LEN]
return name
def resolve_dir_path(parent_out_dir: str, want_dir_name: str, orig_src_basename: str, log_file: str):
"""目录冲突处理:父输出目录下,如果want_dir_name已存在,则追加(1)(2)..."""
candidate = want_dir_name
idx = 1
while True:
full_candidate_path = os.path.join(parent_out_dir, candidate)
if not os.path.exists(full_candidate_path):
return candidate
write_log(f"【目录重名冲突】源目录[{orig_src_basename}],输出目录已存在[{candidate}],自动变更为 {want_dir_name}({idx})", log_file)
candidate = f"{want_dir_name}({idx})"
idx += 1
def get_quality_tag(filename:str):
m = re.search(r"(?<!\w)(2160p|1080p|720p)(?!\w)", 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):
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)
try:
with open(log_file, "a", encoding="utf-8") as f:
f.write(line + "\n")
f.flush()
except Exception:
#日志写入失败降级控制台输出,不打断程序
print(f"[LOG-WRITE-FAILED] {line}")
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="加上此参数才执行真实复制生成镜像")
parser.add_argument("--dry-run", action="store_true", help="预览模式(默认就是预览,该参数为显式简写)")
parser.add_argument("--strm-mode", action="store_true", help="开启strm模式:仅处理 .strm .srt .ass .sub;过滤mkv/mp4等视频文件")
parser.add_argument("--log", default="./rename_log.txt", help="日志文件路径,默认 ./rename_log.txt")
args = parser.parse_args()
target_dir = args.dir
out_dir = args.output
log_path = args.log
DRY_RUN = True
if args.apply:
DRY_RUN = False
if args.dry_run:
DRY_RUN = True
if args.strm_mode:
ALLOW_EXT = STRM_ONLY_EXT
mode_info = "STRM模式【仅strm+字幕】"
else:
ALLOW_EXT = FULL_ALLOW_EXT
mode_info = "完整媒体模式【全部媒体后缀】"
if not os.path.isdir(target_dir):
print(f"错误:源目录不存在 -> {target_dir}")
return
#初始化日志
try:
with open(log_path, "w", encoding="utf-8") as f:
f.flush()
except Exception as e:
print(f"警告:无法初始化日志文件 {log_path},后续仅控制台输出日志,异常:{e}")
write_log("===== 文件名+文件夹清洗【目录重名冲突修复|全部标准短横杠参数】 =====", log_path)
write_log(f"源目录:{target_dir}", log_path)
write_log(f"输出镜像目录:{out_dir}", log_path)
write_log(f"运行模式:{mode_info}", log_path)
write_log(f"预览模式 DRY_RUN = {DRY_RUN}", log_path)
if SKIP_DIRS:
write_log(f"配置跳过目录名列表:{SKIP_DIRS}", log_path)
write_log(f"过滤隐含系统文件:{SKIP_FILES}", log_path)
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
}
try:
for root, dirs, files in os.walk(target_dir, topdown=False):
#过滤跳过目录:@eaDir、配置SKIP_DIRS
remove_idx = []
for idx, dname in enumerate(dirs):
if dname in SKIP_DIRS:
remove_idx.append(idx)
for idx in reversed(remove_idx):
del dirs[idx]
stat["scan_dir_total"] += 1
rel = os.path.relpath(root, target_dir)
if rel == ".":
#根目录
dst_current_dir = out_dir
else:
src_basename = os.path.basename(root)
parent_src_rel = os.path.dirname(rel)
parent_out_dir = os.path.join(out_dir, parent_src_rel)
new_dir_name = clean_dir_name(src_basename)
if len(new_dir_name) == 0:
write_log(f"【警告】文件夹清洗后名称为空,使用原名保留:{src_basename}", log_path)
new_dir_name = src_basename
# 目录冲突检测,获取最终输出目录名
final_dir_name = resolve_dir_path(parent_out_dir, new_dir_name, src_basename, log_path)
dst_current_dir = os.path.join(parent_out_dir, final_dir_name)
# 创建输出目录
if not DRY_RUN:
try:
os.makedirs(dst_current_dir, exist_ok=True)
stat["dir_success"] +=1
except OSError as e:
stat["dir_fail"] +=1
write_log(f"【目录创建失败】{dst_current_dir} | {e}", log_path)
# 处理本目录全部文件
for fname in files:
if fname in SKIP_FILES:
continue
stat["scan_file_total"] +=1
ext = os.path.splitext(fname)[1].lower()
if ext in ALLOW_EXT:
stat["media_file_total"] += 1
new_name = clean_file_name(fname)
old_path = os.path.join(root, fname)
base, ext_out = os.path.splitext(new_name)
final_fname = resolve_conflict(dst_current_dir, base, ext_out, fname)
dst_file = os.path.join(dst_current_dir, final_fname)
if final_fname != fname:
write_log(f"待修改文件-> {fname} → {final_fname}", log_path)
if not DRY_RUN:
try:
shutil.copy2(old_path, dst_file)
stat["file_success"] +=1
except OSError as e:
stat["file_fail"] +=1
write_log(f"【文件复制失败】src:{old_path} dst:{dst_file} | {e}", log_path)
# 处理子文件夹,只打印名称发生变更的文件夹
for dname in dirs:
new_name = clean_dir_name(dname)
if len(new_name) == 0:
write_log(f"【警告】文件夹清洗后名称为空,跳过改名:{dname}", log_path)
continue
if new_name != dname:
write_log(f"待修改文件夹-> {dname} → {new_name}", log_path)
except OSError as walk_err:
write_log(f"【严重】遍历目录发生IO异常,部分目录未扫描:{walk_err}", 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()
+277
View File
@@ -0,0 +1,277 @@
import os
import re
import argparse
import shutil
from datetime import datetime
# ====================== 【配置区】 ======================
# 建议日志放在【本机内置磁盘】,不要放在容易掉线的外接卷,防止盘断开日志写失败
LOG_FILE = "./rename_log.txt"
#需要完全跳过的目录名称(basename,大小写敏感)
SKIP_DIRS = [
"@eaDir", # 群晖隐含目录
]
# 需要跳过的隐含文件名称(完整文件名匹配)
SKIP_FILES = {
".DS_Store",
"Thumbs.db",
"desktop.ini",
}
# 文件清洗规则:依旧删除全部 []【】
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().:+-]", ""),
]
# 1.删除中文全角【】
# 2.剥最外层英文[]
# 3.清除碎片 ][
# 4.最后清理遗留空括号 () ()
RULES_DIR = [
(r"【.*?】", ""), # 删除中文全角【】以及内部内容
(r"^\[(.*)\]$", r"\1"), # 整串被英文[]包裹,只剥外层
(r"\]\[", ""), # 清除中间碎片 ][
(r"^[A-Za-z] ", ""),
(r"丨", ""),
(r"\b(2160p|1080p|720p|HDR|DV|DoVi|HEVC|x265|x264|DDP|AAC|Atmos|TrueHD|中英|双语|国语|国粤|中字|超清|高清|清晰|无水印|HD|BD|字幕)\b", "", re.IGNORECASE),
(r"\.+", ""),
(r"\s", ""),
(r"\(\)", ""), # 清除英文空括号 ()
(r"()", ""), # 清除中文全角空括号 ()
(r"[^\u4e00-\u9fa5a-zA-Z0-9().:+\-[\]]", ""), # 允许文件夹内部保留[]
]
#完整全部媒体后缀(不带--strm‑mode时使用)
FULL_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"}
# strm模式:仅strm+字幕
STRM_ONLY_EXT = {".strm", ".srt", ".ass", ".sub"}
MAX_FILENAME_LEN = 245 # mac文件名最大安全长度,预留扩展名位置
# =======================================================
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()
if len(name_no_ext) > MAX_FILENAME_LEN:
name_no_ext = name_no_ext[:MAX_FILENAME_LEN]
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)
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})"
if len(name) > MAX_FILENAME_LEN:
name = name[:MAX_FILENAME_LEN]
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):
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)
try:
with open(log_file, "a", encoding="utf-8") as f:
f.write(line + "\n")
f.flush()
except Exception:
#日志写入失败降级控制台输出,不打断程序
print(f"[LOG-WRITE-FAILED] {line}")
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="加上此参数才执行真实复制生成镜像")
parser.add_argument("--dry-run", action="store_true", help="预览模式(默认就是预览,该参数为显式简写)")
parser.add_argument("--strm-mode", action="store_true", help="开启strm模式:仅处理 .strm .srt .ass .sub;过滤mkv/mp4等视频文件")
parser.add_argument("--log", default="./rename_log.txt", help="日志文件路径,默认 ./rename_log.txt")
args = parser.parse_args()
target_dir = args.dir
out_dir = args.output
log_path = args.log
DRY_RUN = True
if args.apply:
DRY_RUN = False
if args.strm_mode:
ALLOW_EXT = STRM_ONLY_EXT
mode_info = "STRM模式【仅strm+字幕】"
else:
ALLOW_EXT = FULL_ALLOW_EXT
mode_info = "完整媒体模式【全部媒体后缀】"
if not os.path.isdir(target_dir):
print(f"错误:源目录不存在 -> {target_dir}")
return
#初始化日志
try:
with open(log_path, "w", encoding="utf-8") as f:
f.flush()
except Exception as e:
print(f"警告:无法初始化日志文件 {log_path},后续仅控制台输出日志,异常:{e}")
write_log("===== 文件名+文件夹清洗【增加清除空括号()()】 =====", log_path)
write_log(f"源目录:{target_dir}", log_path)
write_log(f"输出镜像目录:{out_dir}", log_path)
write_log(f"运行模式:{mode_info}", log_path)
write_log(f"预览模式 DRY_RUN = {DRY_RUN}", log_path)
if SKIP_DIRS:
write_log(f"配置跳过目录名列表:{SKIP_DIRS}", log_path)
write_log(f"过滤隐含系统文件:{SKIP_FILES}", log_path)
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
}
try:
for root, dirs, files in os.walk(target_dir, topdown=False):
#过滤跳过目录:@eaDir、配置SKIP_DIRS
remove_idx = []
for idx, dname in enumerate(dirs):
if dname in SKIP_DIRS:
remove_idx.append(idx)
for idx in reversed(remove_idx):
del dirs[idx]
stat["scan_dir_total"] += 1
rel = os.path.relpath(root, target_dir)
if rel == ".":
dst_current_dir = out_dir
else:
src_basename = os.path.basename(root)
new_dir_name = clean_dir_name(src_basename)
if len(new_dir_name) == 0:
write_log(f"【警告】文件夹清洗后名称为空,使用原名保留:{src_basename}", log_path)
new_dir_name = src_basename
dst_current_dir = os.path.join(out_dir, rel.replace(src_basename, new_dir_name))
#创建输出目录
if not DRY_RUN:
try:
os.makedirs(dst_current_dir, exist_ok=True)
stat["dir_success"] +=1
except OSError as e:
stat["dir_fail"] +=1
write_log(f"【目录创建失败】{dst_current_dir} | {e}", log_path)
#处理本目录全部文件
for fname in files:
if fname in SKIP_FILES:
continue
stat["scan_file_total"] +=1
ext = os.path.splitext(fname)[1].lower()
if ext in ALLOW_EXT:
stat["media_file_total"] += 1
new_name = clean_file_name(fname)
old_path = os.path.join(root, fname)
base, ext_out = os.path.splitext(new_name)
final_fname = resolve_conflict(dst_current_dir, base, ext_out, fname)
dst_file = os.path.join(dst_current_dir, final_fname)
if final_fname != fname:
write_log(f"待修改文件-> {fname} → {final_fname}", log_path)
if not DRY_RUN:
try:
shutil.copy2(old_path, dst_file)
stat["file_success"] +=1
except OSError as e:
stat["file_fail"] +=1
write_log(f"【文件复制失败】src:{old_path} dst:{dst_file} | {e}", log_path)
#处理子文件夹,只打印名称发生变更的文件夹
for dname in dirs:
new_name = clean_dir_name(dname)
if len(new_name) == 0:
write_log(f"【警告】文件夹清洗后名称为空,跳过改名:{dname}", log_path)
continue
if new_name != dname:
write_log(f"待修改文件夹-> {dname} → {new_name}", log_path)
except OSError as walk_err:
write_log(f"【严重】遍历目录发生IO异常,部分目录未扫描:{walk_err}", 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()
+274
View File
@@ -0,0 +1,274 @@
import os
import re
import argparse
import shutil
from datetime import datetime
# ====================== 【配置区】 ======================
LOG_FILE = "./rename_log.txt"
# 需要完全跳过的目录名称(basename,大小写敏感)
SKIP_DIRS = [
"@eaDir", # 群晖隐含目录
]
# 需要跳过的隐含文件名称(完整文件名匹配)
SKIP_FILES = {
".DS_Store",
"Thumbs.db",
"desktop.ini",
}
# 文件清洗规则
RULES_FILE = [
(r"^[A-Za-z] ", ""),
(r"\[.*?\]", ""),
(r"【.*?】", ""),
(r"丨", ""),
(r"(?<!\w)(2160p|1080p|720p|HDR|DV|DoVi|HEVC|x265|x264|DDP|AAC|Atmos|TrueHD|BD|BluRay|BLURAY|WEBRip|WEB-DL|HD|4K|中英双语|双语字幕|双语|中文|国语|粤语|台语|韩语|中英双字|国语中字|繁体中字|简体中字|简繁中字|双语中字|官方中字|特效中字)(?!\w)", "", re.IGNORECASE),
(r"\.+", ""),
(r"\s", ""),
(r"[^\u4e00-\u9fa5a-zA-Z0-9().:+\-]", ""),
]
# 文件夹清洗规则
RULES_DIR = [
(r"【.*?】", ""), # 删除中文全角【】以及内部内容
(r"^\[(.*)\]$", r"\1"), # 整串被[]包裹,只剥外层
(r"\]\[", ""), # 清除中间碎片 ][
(r"^[A-Za-z] ", ""),
(r"丨", ""),
(r"(?<!\w)(2160p|1080p|720p|HDR|DV|DoVi|HEVC|x265|x264|DDP|AAC|Atmos|TrueHD|BD|BluRay|BLURAY|WEBRip|WEB-DL|HD|4K|中英双语|双语字幕|双语|中文|国语|粤语|台语|韩语|中英双字|国语中字|繁体中字|简体中字|简繁中字|双语中字|官方中字|特效中字)(?!\w)", "", re.IGNORECASE),
(r"\.+", ""),
(r"\s", ""),
(r"\(\)", ""), # 清除英文空括号 ()
(r"()", ""), # 清除中文全角空括号 ()
(r"[^\u4e00-\u9fa5a-zA-Z0-9().:+\-[\]]", ""), # 允许文件夹内部保留[]
]
#完整全部媒体后缀(不带--strm-mode时使用)
FULL_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"}
# strm模式:仅strm+字幕
STRM_ONLY_EXT = {".strm", ".srt", ".ass", ".sub"}
MAX_FILENAME_LEN = 245 # mac文件名最大安全长度,预留扩展名位置
# =======================================================
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()
if len(name_no_ext) > MAX_FILENAME_LEN:
name_no_ext = name_no_ext[:MAX_FILENAME_LEN]
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)
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})"
if len(name) > MAX_FILENAME_LEN:
name = name[:MAX_FILENAME_LEN]
return name
def get_quality_tag(filename:str):
m = re.search(r"(?<!\w)(2160p|1080p|720p)(?!\w)", 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):
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)
try:
with open(log_file, "a", encoding="utf-8") as f:
f.write(line + "\n")
f.flush()
except Exception:
#日志写入失败降级控制台输出,不打断程序
print(f"[LOG-WRITE-FAILED] {line}")
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="加上此参数才执行真实复制生成镜像")
parser.add_argument("--dry-run", action="store_true", help="预览模式(默认就是预览,该参数为显式简写)")
parser.add_argument("--strm-mode", action="store_true", help="开启strm模式:仅处理 .strm .srt .ass .sub;过滤mkv/mp4等视频文件")
parser.add_argument("--log", default="./rename_log.txt", help="日志文件路径,默认 ./rename_log.txt")
args = parser.parse_args()
target_dir = args.dir
out_dir = args.output
log_path = args.log
DRY_RUN = True
if args.apply:
DRY_RUN = False
if args.dry_run:
DRY_RUN = True
if args.strm_mode:
ALLOW_EXT = STRM_ONLY_EXT
mode_info = "STRM模式【仅strm+字幕】"
else:
ALLOW_EXT = FULL_ALLOW_EXT
mode_info = "完整媒体模式【全部媒体后缀】"
if not os.path.isdir(target_dir):
print(f"错误:源目录不存在 -> {target_dir}")
return
#初始化日志
try:
with open(log_path, "w", encoding="utf-8") as f:
f.flush()
except Exception as e:
print(f"警告:无法初始化日志文件 {log_path},后续仅控制台输出日志,异常:{e}")
write_log("===== 文件名+文件夹清洗【修复参数符号,过滤BD/HD/4K/双语/国语等标记】 =====", log_path)
write_log(f"源目录:{target_dir}", log_path)
write_log(f"输出镜像目录:{out_dir}", log_path)
write_log(f"运行模式:{mode_info}", log_path)
write_log(f"预览模式 DRY_RUN = {DRY_RUN}", log_path)
if SKIP_DIRS:
write_log(f"配置跳过目录名列表:{SKIP_DIRS}", log_path)
write_log(f"过滤隐含系统文件:{SKIP_FILES}", log_path)
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
}
try:
for root, dirs, files in os.walk(target_dir, topdown=False):
#过滤跳过目录:@eaDir、配置SKIP_DIRS
remove_idx = []
for idx, dname in enumerate(dirs):
if dname in SKIP_DIRS:
remove_idx.append(idx)
for idx in reversed(remove_idx):
del dirs[idx]
stat["scan_dir_total"] += 1
rel = os.path.relpath(root, target_dir)
if rel == ".":
dst_current_dir = out_dir
else:
src_basename = os.path.basename(root)
new_dir_name = clean_dir_name(src_basename)
if len(new_dir_name) == 0:
write_log(f"【警告】文件夹清洗后名称为空,使用原名保留:{src_basename}", log_path)
new_dir_name = src_basename
dst_current_dir = os.path.join(out_dir, rel.replace(src_basename, new_dir_name))
# 创建输出目录
if not DRY_RUN:
try:
os.makedirs(dst_current_dir, exist_ok=True)
stat["dir_success"] +=1
except OSError as e:
stat["dir_fail"] +=1
write_log(f"【目录创建失败】{dst_current_dir} | {e}", log_path)
# 处理本目录全部文件
for fname in files:
if fname in SKIP_FILES:
continue
stat["scan_file_total"] +=1
ext = os.path.splitext(fname)[1].lower()
if ext in ALLOW_EXT:
stat["media_file_total"] += 1
new_name = clean_file_name(fname)
old_path = os.path.join(root, fname)
base, ext_out = os.path.splitext(new_name)
final_fname = resolve_conflict(dst_current_dir, base, ext_out, fname)
dst_file = os.path.join(dst_current_dir, final_fname)
if final_fname != fname:
write_log(f"待修改文件-> {fname} → {final_fname}", log_path)
if not DRY_RUN:
try:
shutil.copy2(old_path, dst_file)
stat["file_success"] +=1
except OSError as e:
stat["file_fail"] +=1
write_log(f"【文件复制失败】src:{old_path} dst:{dst_file} | 错误:{e}", log_path)
# 处理子文件夹,只打印名称发生变更的文件夹
for dname in dirs:
new_name = clean_dir_name(dname)
if len(new_name) == 0:
write_log(f"【警告】文件夹清洗后名称为空,跳过改名:{dname}", log_path)
continue
if new_name != dname:
write_log(f"待修改文件夹-> {dname} → {new_name}", log_path)
except OSError as walk_err:
write_log(f"【严重】遍历目录发生IO异常,部分目录未扫描:{walk_err}", 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()
+196
View File
@@ -0,0 +1,196 @@
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()
+131
View File
@@ -0,0 +1,131 @@
import os
import re
import argparse
from datetime import datetime
# ====================== 【配置区,不再写死目录】 ======================
LOG_FILE = "./rename_log.txt"
# 清洗规则
RULES = [
(r"^[A-Za-z] ", ""), #1.移除开头单个字母+空格
(r"\[.*?\]", ""), #2.移除英文方括号及内部内容
(r"【.*?】", ""), #3.移除中文【】及内部内容
(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 clean_filename(name: str) -> str:
name_no_ext, ext = os.path.splitext(name)
# 全角转半角
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)
name_no_ext = full2half(name_no_ext)
for item in RULES:
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 write_log(msg: 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 main():
parser = argparse.ArgumentParser(description="媒体文件名批量清洗重命名工具")
parser.add_argument("--dir", required=True, help="目标根目录路径")
parser.add_argument("--apply", action="store_true", help="加上此参数才执行真实重命名;默认仅预览DryRun")
args = parser.parse_args()
target_dir = args.dir
DRY_RUN = not args.apply
# 校验目录是否存在
if not os.path.isdir(target_dir):
print(f"错误:目录不存在 -> {target_dir}")
return
# 每次运行前清空日志文件
with open(LOG_FILE, "w", encoding="utf-8") as f:
f.flush()
write_log("===== 文件名+文件夹清洗(移除开头A 中文片名前缀) =====")
write_log(f"根目录:{target_dir}")
write_log(f"预览模式 DRY_RUN = {DRY_RUN}")
file_rename_list = []
dir_rename_list = []
total_scan_dirs = 0
total_all_files = 0 # 全部扫描到的文件(所有后缀)
total_media_files = 0 # 媒体文件(ALLOW_EXT后缀)
for root, dirs, files in os.walk(target_dir, topdown=False):
total_scan_dirs += 1
for fname in files:
total_all_files +=1
ext = os.path.splitext(fname)[1].lower()
if ext in ALLOW_EXT:
total_media_files += 1
new_name = clean_filename(fname)
if new_name != fname:
old_path = os.path.join(root, fname)
new_path = os.path.join(root, new_name)
file_rename_list.append((old_path, new_path))
write_log(f"待修改文件-> {fname} → {new_name}")
# 处理文件夹
for dname in dirs:
new_name = clean_filename(dname)
if new_name != dname:
old_path = os.path.join(root, dname)
new_path = os.path.join(root, new_name)
dir_rename_list.append((old_path, new_name))
write_log(f"待修改文件夹-> {dname} → {new_name}")
# ========== 执行阶段 ==========
if not DRY_RUN:
# 先文件
for old, new in file_rename_list:
try:
os.rename(old, new)
except Exception as e:
write_log(f"【文件重命名失败】{old} -> {new} | 错误:{e}")
# 后文件夹
for old, new in dir_rename_list:
try:
os.rename(old, new)
except Exception as e:
write_log(f"【文件夹重命名失败】{old} -> {new} | 错误:{e}")
# 最终汇总输出
write_log(f"任务完成,总扫描目录:{total_scan_dirs},待修改文件夹:{len(dir_rename_list)}")
write_log(f"总扫描文件:{total_all_files},媒体文件:{total_media_files},待修改文件:{len(file_rename_list)}")
write_log("============================================================")
if __name__ == "__main__":
main()