449 lines
16 KiB
Python
Executable File
449 lines
16 KiB
Python
Executable File
import os
|
||
import re
|
||
import json
|
||
import argparse
|
||
import shutil
|
||
from datetime import datetime
|
||
|
||
LOG_FILE = "./rename_log.txt"
|
||
ERR_REPORT_JSON = "./report_err_only.json"
|
||
|
||
SKIP_DIRS = [
|
||
"@eaDir",
|
||
]
|
||
SKIP_FILES = {
|
||
".DS_Store",
|
||
"Thumbs.db",
|
||
"desktop.ini",
|
||
}
|
||
|
||
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: str) -> str:
|
||
res = []
|
||
for ch in s:
|
||
oc = ord(ch)
|
||
if oc == 12288:
|
||
res.append(chr(32))
|
||
elif 65281 <= oc <= 65374:
|
||
res.append(chr(oc - 65248))
|
||
else:
|
||
res.append(ch)
|
||
return "".join(res)
|
||
|
||
def remove_emoji(s: str) -> str:
|
||
emoji_pattern = re.compile("["
|
||
u"\U0001F600-\U0001F64F"
|
||
u"\U0001F300-\U0001F5FF"
|
||
u"\U0001F680-\U0001F6FF"
|
||
u"\U0001F1E0-\U0001F1FF"
|
||
u"\U00002700-\U000027BF"
|
||
u"\U0001F900-\U0001F9FF"
|
||
"]+", flags=re.UNICODE)
|
||
return emoji_pattern.sub(r'', s)
|
||
|
||
def post_clean_dot(name: str, is_dir: bool=False) -> str:
|
||
"""后置净化：清除连续点、首尾点、首尾下划线；区分目录/文件，保护扩展名"""
|
||
if is_dir:
|
||
body = name
|
||
ext = ""
|
||
else:
|
||
body, ext = os.path.splitext(name)
|
||
body = re.sub(r"\.+", ".", body)
|
||
body = body.strip("._")
|
||
return body + ext
|
||
|
||
def stage1_light_clean(name: str, is_dir: bool=False) -> str:
|
||
"""阶段一：轻清洗"""
|
||
n = name
|
||
n = remove_emoji(n)
|
||
n = full2half(n)
|
||
n = re.sub(r"^[A-Za-z] +", "", n)
|
||
n = re.sub(r"\s+", "", n)
|
||
n = post_clean_dot(n, is_dir=is_dir)
|
||
return n.strip()
|
||
|
||
def run_rule_list(orig_name: str, rules: list, is_dir: bool=False):
|
||
hit_ids = []
|
||
cur = orig_name
|
||
safeguard_trigger = False
|
||
|
||
for r in sorted(rules, key=lambda x: x.get("order", 9999)):
|
||
rid = r["rule_id"]
|
||
kind = r["kind"]
|
||
repl = r["repl"]
|
||
safe_guard = r.get("safe_guard_empty", False)
|
||
rule_flags = r.get("flags", "")
|
||
prev = cur
|
||
|
||
re_flags = 0
|
||
if rule_flags == "IGNORECASE":
|
||
re_flags = re.IGNORECASE
|
||
|
||
if kind == "string":
|
||
cur = re.sub(r["pattern"], repl, cur, flags=re_flags)
|
||
elif kind == "bracket_pair":
|
||
o_pat = r["open"]
|
||
c_pat = r["close"]
|
||
# bracket_pair额外开启DOTALL，.可匹配中文、各类符号
|
||
while True:
|
||
new_val = re.sub(o_pat + ".*?" + c_pat, repl, cur, flags=re_flags | re.DOTALL)
|
||
if new_val == cur:
|
||
break
|
||
cur = new_val
|
||
if safe_guard:
|
||
body, ext = (cur, "") if is_dir else os.path.splitext(cur)
|
||
if len(body.strip()) == 0:
|
||
cur = prev
|
||
continue
|
||
|
||
cur = cur.strip()
|
||
if safe_guard:
|
||
body, ext = (cur, "") if is_dir else os.path.splitext(cur)
|
||
if len(body.strip()) == 0:
|
||
cur = prev
|
||
continue
|
||
|
||
if cur != prev:
|
||
hit_ids.append(rid)
|
||
|
||
# 全部规则跑完，全局兜底保护
|
||
body_final, ext_final = (cur, "") if is_dir else os.path.splitext(cur)
|
||
if len(body_final.strip()) == 0:
|
||
safeguard_trigger = True
|
||
cur = orig_name
|
||
|
||
# 符号净化
|
||
cur = post_clean_dot(cur, is_dir=is_dir)
|
||
|
||
# 净化之后再次兜底
|
||
body2, ext2 = (cur, "") if is_dir else os.path.splitext(cur)
|
||
if len(body2.strip()) == 0:
|
||
safeguard_trigger = True
|
||
cur = orig_name
|
||
|
||
return cur, hit_ids, safeguard_trigger
|
||
|
||
def assign_unique_name_group(item_list):
|
||
counter = dict()
|
||
out = []
|
||
for it in item_list:
|
||
cand = it["candidate"]
|
||
cnt = counter.get(cand, 0)
|
||
final = cand if cnt == 0 else f"{cand}({cnt})"
|
||
counter[cand] = cnt + 1
|
||
ni = dict(it)
|
||
ni["final_step1"] = final
|
||
out.append(ni)
|
||
return out
|
||
|
||
def assign_dup_suffix(item_list, name_key: str):
|
||
counter = dict()
|
||
for it in item_list:
|
||
nm = it[name_key]
|
||
c = counter.get(nm, 0)
|
||
if c > 0:
|
||
new_nm = f"{nm}_dup({c})"
|
||
it[name_key] = new_nm
|
||
it["step2_dup_warn"] = True
|
||
else:
|
||
it["step2_dup_warn"] = False
|
||
counter[nm] = c + 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 scan_full_tree(root_src, skip_dirs, skip_files, allow_ext):
|
||
dir_entries = []
|
||
file_entries = []
|
||
root_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)
|
||
if src_root_abs != root_abs:
|
||
bn = os.path.basename(src_root_abs)
|
||
c1 = stage1_light_clean(bn, is_dir=True)
|
||
dir_entries.append({
|
||
"src_abs": src_root_abs,
|
||
"src_parent_abs": os.path.dirname(src_root_abs),
|
||
"depth": src_root_abs.count(os.sep),
|
||
"orig_basename": bn,
|
||
"candidate": c1
|
||
})
|
||
for fname in files:
|
||
if fname in skip_files:
|
||
continue
|
||
ext = os.path.splitext(fname)[1].lower()
|
||
if ext not in allow_ext:
|
||
continue
|
||
c1 = stage1_light_clean(fname, is_dir=False)
|
||
file_entries.append({
|
||
"src_parent_abs": src_root_abs,
|
||
"orig_basename": fname,
|
||
"candidate": c1
|
||
})
|
||
return root_abs, dir_entries, file_entries
|
||
|
||
def group_by_parent_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="两阶段媒体清洗【修复括号匹配，DOTALL开启】")
|
||
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("--rules", default=None, help="外部json规则文件路径")
|
||
parser.add_argument("--log", default=LOG_FILE)
|
||
args = parser.parse_args()
|
||
|
||
target_dir = args.dir
|
||
out_dir = args.output
|
||
log_path = args.log
|
||
rules_path = args.rules
|
||
|
||
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}")
|
||
|
||
file_rules = []
|
||
dir_rules = []
|
||
if rules_path is not None and os.path.exists(rules_path):
|
||
try:
|
||
with open(rules_path, "r", encoding="utf-8") as f:
|
||
jdata = json.load(f)
|
||
file_rules = jdata.get("file_rules", [])
|
||
dir_rules = jdata.get("dir_rules", [])
|
||
write_log(f"成功加载外部规则文件 {rules_path}，file_rules={len(file_rules)}，dir_rules={len(dir_rules)}", log_path)
|
||
except json.JSONDecodeError as e:
|
||
write_log(f"规则文件JSON格式错误（第{e.lineno}行第{e.colno}列）：{e.msg}", log_path)
|
||
write_log("提示：键和字符串值必须用英文双引号，正则中的反斜杠必须写成双反斜杠\\\\", log_path)
|
||
return
|
||
except Exception as e:
|
||
write_log(f"加载规则文件失败：{e}", log_path)
|
||
return
|
||
else:
|
||
write_log("未提供--rules或文件不存在，跳过第二阶段深度清洗", log_path)
|
||
|
||
write_log("===== 两阶段媒体清洗【修复括号匹配DOTALL】 =====", log_path)
|
||
write_log(f"源目录:{target_dir}", log_path)
|
||
write_log(f"输出:{out_dir}", log_path)
|
||
write_log(f"模式:{mode_info} 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
|
||
}
|
||
report_entries = []
|
||
|
||
root_src_abs, dir_entries, file_entries = scan_full_tree(target_dir, SKIP_DIRS, SKIP_FILES, ALLOW_EXT)
|
||
stat["src_dir_total"] = len(dir_entries)
|
||
stat["src_file_total"] = len(file_entries)
|
||
write_log(f"采集统计：目录节点 {stat['src_dir_total']}，媒体文件 {stat['src_file_total']}", log_path)
|
||
|
||
dir_groups = group_by_parent_abs(dir_entries, "src_parent_abs")
|
||
dir_final = []
|
||
for p_abs, items in dir_groups.items():
|
||
assigned = assign_unique_name_group(items)
|
||
for it in assigned:
|
||
step1_name = it["final_step1"]
|
||
if len(dir_rules) > 0:
|
||
step2_name, hit_rids, safe_trig = run_rule_list(step1_name, dir_rules, is_dir=True)
|
||
else:
|
||
step2_name = step1_name
|
||
hit_rids = []
|
||
safe_trig = False
|
||
it["step2_name"] = step2_name
|
||
it["hit_rule_ids"] = hit_rids
|
||
it["safe_guard_trigger"] = safe_trig
|
||
dir_final.append(it)
|
||
dir_g2_groups = group_by_parent_abs(dir_final, "src_parent_abs")
|
||
for _, items in dir_g2_groups.items():
|
||
assign_dup_suffix(items, "step2_name")
|
||
|
||
file_groups = group_by_parent_abs(file_entries, "src_parent_abs")
|
||
file_final = []
|
||
for p_abs, items in file_groups.items():
|
||
assigned = assign_unique_name_group(items)
|
||
for it in assigned:
|
||
step1_name = it["final_step1"]
|
||
if len(file_rules) > 0:
|
||
step2_name, hit_rids, safe_trig = run_rule_list(step1_name, file_rules, is_dir=False)
|
||
else:
|
||
step2_name = step1_name
|
||
hit_rids = []
|
||
safe_trig = False
|
||
it["step2_name"] = step2_name
|
||
it["hit_rule_ids"] = hit_rids
|
||
it["safe_guard_trigger"] = safe_trig
|
||
file_final.append(it)
|
||
file_g2_groups = group_by_parent_abs(file_final, "src_parent_abs")
|
||
for _, items in file_g2_groups.items():
|
||
assign_dup_suffix(items, "step2_name")
|
||
|
||
src_to_out = dict()
|
||
src_to_out[root_src_abs] = out_dir
|
||
dir_final_sorted = sorted(dir_final, key=lambda x: x["depth"])
|
||
for d in dir_final_sorted:
|
||
p_abs = d["src_parent_abs"]
|
||
parent_out = src_to_out.get(p_abs)
|
||
if parent_out is None:
|
||
write_log(f"【警告】父目录映射缺失：{p_abs}", log_path)
|
||
continue
|
||
child_out = os.path.join(parent_out, d["step2_name"])
|
||
src_to_out[d["src_abs"]] = child_out
|
||
|
||
for d in dir_final:
|
||
report_entries.append({
|
||
"type":"dir",
|
||
"src_abs":d["src_abs"],
|
||
"orig_basename":d["orig_basename"],
|
||
"step1_name":d["final_step1"],
|
||
"step2_name":d["step2_name"],
|
||
"hit_rule_ids":d["hit_rule_ids"],
|
||
"safe_guard_trigger":d["safe_guard_trigger"],
|
||
"step2_dup_warn":d.get("step2_dup_warn", False),
|
||
"final_out_path": src_to_out.get(d["src_abs"], "")
|
||
})
|
||
for f in file_final:
|
||
src_file = os.path.join(f["src_parent_abs"], f["orig_basename"])
|
||
parent_out = src_to_out.get(f["src_parent_abs"])
|
||
dst_file = os.path.join(parent_out, f["step2_name"]) if parent_out else ""
|
||
report_entries.append({
|
||
"type":"file",
|
||
"src_abs":src_file,
|
||
"orig_basename":f["orig_basename"],
|
||
"step1_name":f["final_step1"],
|
||
"step2_name":f["step2_name"],
|
||
"hit_rule_ids":f["hit_rule_ids"],
|
||
"safe_guard_trigger":f["safe_guard_trigger"],
|
||
"step2_dup_warn":f.get("step2_dup_warn", False),
|
||
"final_out_path": dst_file
|
||
})
|
||
|
||
# 仅输出异常报告，丢弃完整大JSON
|
||
err_list = []
|
||
for entry in report_entries:
|
||
dup = entry.get("step2_dup_warn", False)
|
||
safe = entry.get("safe_guard_trigger", False)
|
||
if dup or safe:
|
||
err_list.append({
|
||
"type": entry.get("type"),
|
||
"orig_basename": entry.get("orig_basename"),
|
||
"step1_name": entry.get("step1_name"),
|
||
"step2_name": entry.get("step2_name"),
|
||
"step2_dup_warn": dup,
|
||
"safe_guard_trigger": safe,
|
||
"hit_rule_ids": entry.get("hit_rule_ids", [])
|
||
})
|
||
try:
|
||
with open(ERR_REPORT_JSON, "w", encoding="utf-8") as fw:
|
||
json.dump(err_list, fw, ensure_ascii=False, indent=2)
|
||
write_log(f"异常报告已输出：{ERR_REPORT_JSON}，异常条目数量：{len(err_list)}", log_path)
|
||
except Exception as e:
|
||
write_log(f"异常报告写入失败 {e}", log_path)
|
||
|
||
for ent in report_entries:
|
||
ot = ent["orig_basename"]
|
||
s1 = ent["step1_name"]
|
||
s2 = ent["step2_name"]
|
||
parts = []
|
||
if ot != s1:
|
||
parts.append(f"原始→step1:{ot} → {s1}")
|
||
if s1 != s2:
|
||
parts.append(f"step1→step2:{s1} → {s2}")
|
||
if ent["step2_dup_warn"]:
|
||
parts.append("[告警：step2重名追加_dup标记]")
|
||
if ent["safe_guard_trigger"]:
|
||
parts.append("[安全保护触发，放弃部分规则]")
|
||
if len(parts) > 0:
|
||
write_log(" | ".join(parts), log_path)
|
||
|
||
del report_entries
|
||
|
||
if DRY_RUN:
|
||
write_log("===== DRY-RUN预览模式结束，未执行磁盘IO =====", log_path)
|
||
return
|
||
|
||
for d in dir_final_sorted:
|
||
full_out = src_to_out.get(d["src_abs"])
|
||
if not full_out:
|
||
continue
|
||
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 f in file_final:
|
||
src_fp = os.path.join(f["src_parent_abs"], f["orig_basename"])
|
||
parent_out = src_to_out.get(f["src_parent_abs"])
|
||
if not parent_out:
|
||
stat["file_fail"] +=1
|
||
write_log(f"【文件复制失败】父目录映射缺失:{src_fp}", log_path)
|
||
continue
|
||
dst_fp = os.path.join(parent_out, f["step2_name"])
|
||
try:
|
||
shutil.copy2(src_fp, dst_fp)
|
||
stat["file_success"] +=1
|
||
except OSError as e:
|
||
stat["file_fail"] +=1
|
||
write_log(f"【文件复制失败】src:{src_fp} dst:{dst_fp} | {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(f"【校验】预期目录输出={stat['src_dir_total']}，实际成功={stat['dir_success']}", log_path)
|
||
write_log(f"【校验】预期文件输出={stat['src_file_total']}，实际成功={stat['file_success']}", log_path)
|
||
write_log("============================================================", log_path)
|
||
|
||
if __name__ == "__main__":
|
||
main() |