235 lines
9.2 KiB
Python
235 lines
9.2 KiB
Python
"""
|
|
Core logic for fixing mojibake in file names and ZIP archives on macOS.
|
|
"""
|
|
|
|
import os
|
|
import zipfile
|
|
import shutil
|
|
|
|
RUSSIAN_RANGE = set(range(0x0400, 0x0500))
|
|
|
|
def score_cyrillic(text: str) -> int:
|
|
"""Return count of Cyrillic characters in text."""
|
|
return sum(1 for c in text if ord(c) in RUSSIAN_RANGE)
|
|
|
|
def is_suspicious_char(c: str) -> bool:
|
|
"""Return True if character commonly appears in mojibake."""
|
|
# Signs of MacCyrillic/CP437/Latin1 mangling of Cyrillic:
|
|
# ≠ ™ • † Ѓ ѓ Ђ ђ Љ љ Њ њ Ћ ћ Џ џ § © ® ™ ¶
|
|
return c in "≠™•†ЃѓЂђЉљЊњЋћЏџ§©®¶ҐґІіЇїЎў"
|
|
|
|
def fix_filename_string(name: str) -> str:
|
|
"""
|
|
Attempt to convert a corrupted filename back to legible Cyrillic.
|
|
Handles:
|
|
- MacCyrillic -> CP866 (macOS Russian Archive Utility default on OEM DOS zips)
|
|
- MacCyrillic -> CP1251
|
|
- CP437 -> CP866 (ditto, default zipfile, unzip)
|
|
- CP437 -> CP1251
|
|
- Latin1 -> CP1251 / CP866 / UTF-8
|
|
- MacRoman -> CP1251 / CP866
|
|
"""
|
|
orig_score = score_cyrillic(name)
|
|
has_suspicious = any(is_suspicious_char(c) for c in name)
|
|
|
|
# Pairs of (source_encoding, target_encoding)
|
|
transforms = [
|
|
("mac_cyrillic", "cp866"),
|
|
("mac_cyrillic", "cp1251"),
|
|
("cp437", "cp866"),
|
|
("cp437", "cp1251"),
|
|
("latin1", "cp1251"),
|
|
("latin1", "cp866"),
|
|
("latin1", "utf-8"),
|
|
("mac_roman", "cp866"),
|
|
("mac_roman", "cp1251"),
|
|
]
|
|
|
|
best_name = name
|
|
best_score = orig_score
|
|
|
|
for enc_src, enc_dst in transforms:
|
|
try:
|
|
raw = name.encode(enc_src)
|
|
candidate = raw.decode(enc_dst)
|
|
cand_score = score_cyrillic(candidate)
|
|
|
|
# If original had suspicious mojibake characters and candidate eliminated them
|
|
# or increased cyrillic score
|
|
if cand_score > best_score or (has_suspicious and cand_score >= best_score and not any(is_suspicious_char(c) for c in candidate)):
|
|
best_score = cand_score
|
|
best_name = candidate
|
|
except Exception:
|
|
continue
|
|
|
|
return best_name
|
|
|
|
|
|
def process_path(target_path: str, progress_callback=None) -> list[tuple[str, str, str]]:
|
|
"""
|
|
Recursively process a directory or file.
|
|
If it's a .zip file, it extracts it with corrected names into a folder next to it.
|
|
If it's a directory or regular file, it renames files and folders in-place (bottom-up).
|
|
|
|
Returns list of tuples: (type, old_name, new_name)
|
|
"""
|
|
results = []
|
|
|
|
if not os.path.exists(target_path):
|
|
return [("error", target_path, "Файл или папка не существует")]
|
|
|
|
# Case 1: ZIP file
|
|
if os.path.isfile(target_path) and target_path.lower().endswith(".zip"):
|
|
return process_zip(target_path, progress_callback)
|
|
|
|
# Case 2: Single file
|
|
if os.path.isfile(target_path):
|
|
dir_name, base_name = os.path.split(target_path)
|
|
# Skip temporary office lock files
|
|
if base_name.startswith("~$"):
|
|
try:
|
|
os.remove(target_path)
|
|
results.append(("delete", base_name, "Удалён временный файл блокировки Office"))
|
|
except Exception as e:
|
|
results.append(("error", base_name, f"Не удалось удалить: {e}"))
|
|
return results
|
|
|
|
fixed = fix_filename_string(base_name)
|
|
if fixed != base_name:
|
|
new_path = os.path.join(dir_name, fixed)
|
|
try:
|
|
os.rename(target_path, new_path)
|
|
results.append(("rename_file", base_name, fixed))
|
|
except Exception as e:
|
|
results.append(("error", base_name, f"Ошибка: {e}"))
|
|
else:
|
|
results.append(("skip", base_name, "Имя корректно"))
|
|
return results
|
|
|
|
# Case 3: Directory (recursive, bottom-up so renaming children doesn't invalidate parent path)
|
|
for root, dirs, files in os.walk(target_path, topdown=False):
|
|
# 1. Process files
|
|
for fname in files:
|
|
fpath = os.path.join(root, fname)
|
|
# Remove ~$ lock files
|
|
if fname.startswith("~$"):
|
|
try:
|
|
os.remove(fpath)
|
|
results.append(("delete", fname, "Удалён временный файл блокировки Office"))
|
|
except Exception as e:
|
|
results.append(("error", fname, f"Не удалось удалить: {e}"))
|
|
continue
|
|
|
|
fixed = fix_filename_string(fname)
|
|
if fixed != fname:
|
|
new_fpath = os.path.join(root, fixed)
|
|
try:
|
|
os.rename(fpath, new_fpath)
|
|
results.append(("rename_file", fname, fixed))
|
|
if progress_callback:
|
|
progress_callback(fname, fixed)
|
|
except Exception as e:
|
|
results.append(("error", fname, f"Ошибка: {e}"))
|
|
else:
|
|
results.append(("ok", fname, "OK"))
|
|
|
|
# 2. Process subdirectories
|
|
for dname in dirs:
|
|
dpath = os.path.join(root, dname)
|
|
fixed_dir = fix_filename_string(dname)
|
|
if fixed_dir != dname:
|
|
new_dpath = os.path.join(root, fixed_dir)
|
|
try:
|
|
os.rename(dpath, new_dpath)
|
|
results.append(("rename_dir", dname, fixed_dir))
|
|
if progress_callback:
|
|
progress_callback(dname, fixed_dir)
|
|
except Exception as e:
|
|
results.append(("error", dname, f"Ошибка: {e}"))
|
|
|
|
# Also check if root folder itself needs renaming
|
|
root_parent, root_name = os.path.split(os.path.abspath(target_path))
|
|
fixed_root = fix_filename_string(root_name)
|
|
if fixed_root != root_name and root_parent:
|
|
try:
|
|
new_root_path = os.path.join(root_parent, fixed_root)
|
|
os.rename(target_path, new_root_path)
|
|
results.append(("rename_dir", root_name, fixed_root))
|
|
except Exception as e:
|
|
results.append(("error", root_name, f"Ошибка переименования папки: {e}"))
|
|
|
|
return results
|
|
|
|
|
|
def process_zip(zip_path: str, progress_callback=None) -> list[tuple[str, str, str]]:
|
|
"""
|
|
Extract a ZIP archive directly, correcting encodings for all entries.
|
|
Creates a folder next to zip_path with the extracted contents.
|
|
"""
|
|
results = []
|
|
base_dir = os.path.dirname(os.path.abspath(zip_path))
|
|
zip_stem = os.path.splitext(os.path.basename(zip_path))[0]
|
|
fixed_stem = fix_filename_string(zip_stem)
|
|
|
|
out_dir = os.path.join(base_dir, fixed_stem)
|
|
counter = 1
|
|
orig_out_dir = out_dir
|
|
while os.path.exists(out_dir):
|
|
out_dir = f"{orig_out_dir}_{counter}"
|
|
counter += 1
|
|
|
|
os.makedirs(out_dir, exist_ok=True)
|
|
results.append(("extract_start", os.path.basename(zip_path), f"Распаковка в {os.path.basename(out_dir)}"))
|
|
|
|
try:
|
|
with zipfile.ZipFile(zip_path, "r") as zf:
|
|
for item in zf.infolist():
|
|
raw_name = item.filename
|
|
|
|
# Office lock files in zip can be skipped
|
|
if os.path.basename(raw_name).startswith("~$"):
|
|
continue
|
|
|
|
# Determine best fixed name
|
|
# If flag_bits & 0x800 == 0: python decoded with CP437
|
|
best_name = raw_name
|
|
if not (item.flag_bits & 0x800):
|
|
try:
|
|
raw_bytes = raw_name.encode("cp437")
|
|
for enc in ["cp866", "cp1251"]:
|
|
try:
|
|
candidate = raw_bytes.decode(enc)
|
|
if score_cyrillic(candidate) > score_cyrillic(best_name):
|
|
best_name = candidate
|
|
break
|
|
except Exception:
|
|
pass
|
|
except Exception:
|
|
pass
|
|
|
|
# Run standard mojibake fixer on the parts of path as well
|
|
parts = best_name.split("/")
|
|
fixed_parts = [fix_filename_string(p) for p in parts]
|
|
clean_rel_path = "/".join(fixed_parts)
|
|
|
|
# Security check: prevent zip slip
|
|
dest_path = os.path.abspath(os.path.join(out_dir, clean_rel_path))
|
|
if not dest_path.startswith(os.path.abspath(out_dir)):
|
|
continue
|
|
|
|
if item.is_dir() or clean_rel_path.endswith("/"):
|
|
os.makedirs(dest_path, exist_ok=True)
|
|
else:
|
|
os.makedirs(os.path.dirname(dest_path), exist_ok=True)
|
|
with zf.open(item) as src_file, open(dest_path, "wb") as dst_file:
|
|
shutil.copyfileobj(src_file, dst_file)
|
|
|
|
results.append(("extract_file", raw_name, clean_rel_path))
|
|
if progress_callback:
|
|
progress_callback(raw_name, clean_rel_path)
|
|
|
|
except Exception as e:
|
|
results.append(("error", os.path.basename(zip_path), f"Ошибка архива: {e}"))
|
|
|
|
return results
|