327 lines
12 KiB
Python
Executable File
327 lines
12 KiB
Python
Executable File
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()
|