"""
resolve-symlinks.py — 并行扫描目录下所有符号链接，将指向普通文件的链接原地替换为真实文件。
替换时不会删除符号链接本身，而是将其重命名为 原名.link-bak 作为备份；
后可通过 --restore 扫描所有 .link-bak 并恢复原始符号链接。
支持目录和单个文件两种模式：
- 传入目录 => 递归扫描其下所有符号链接并处理
- 传入单个文件 => 仅处理该符号链接
用法：ohos-resolve-symlinks.py <目录|文件> [选项]
ohos-resolve-symlinks.py <目录|文件> --restore
采用 ohos-sign-elf 风格的并行调度：
- Master 遍历目录收集任务
- Worker 池并行执行
- 进度汇报
"""
import argparse
import os
import shutil
import signal
import sys
from concurrent.futures import ProcessPoolExecutor, as_completed
from pathlib import Path
from typing import Generator, Tuple
Result = Tuple[str, str]
def backup_then_replace_one(link_path: str) -> Result:
"""将符号链接重命名为 .link-bak，再复制真实文件到原路径。
安全策略：
1. 先备份链接，再复制真实文件 —— 任何一步失败都可手动恢复。
2. 只处理指向 *普通文件* 的符号链接（跳过指向目录、设备、不存在的目标）。
3. 保留目标文件的 metadata（mtime、权限），owner 沿用原链接的 uid/gid。
"""
try:
link = Path(link_path)
if not link.is_symlink():
return (link_path, "skip-not-symlink")
target = link.resolve()
if not target.exists():
return (link_path, f"skip-broken -> {target}")
if not target.is_file():
return (link_path, f"skip-not-file -> {target}")
while target.is_symlink():
target = target.resolve()
bak_path = link.parent / f"{link.name}.link-bak"
if bak_path.exists():
return (link_path, f"error: 备份文件已存在 {bak_path}")
link.rename(bak_path)
try:
shutil.copy2(str(target), str(link), follow_symlinks=True)
orig_stat = bak_path.lstat()
os.chown(str(link), orig_stat.st_uid, orig_stat.st_gid, follow_symlinks=False)
except Exception:
try:
os.unlink(str(link))
except OSError:
pass
bak_path.rename(link_path)
raise
return (link_path, f"ok ({os.path.getsize(link)} B), backup at {bak_path.name}")
except Exception as exc:
return (link_path, f"error: {exc}")
def backup_then_replace_dir_one(link_path: str) -> Result:
"""将指向目录的符号链接重命名为 .link-bak，再复制目标目录的全部内容到原路径。
复制时解析所有内部符号连接到实际文件/目录（copytree(symlinks=False)），
替换后原路径变为一个不含任何符号连接的纯普通目录树。
"""
try:
link = Path(link_path)
if not link.is_symlink():
return (link_path, "skip-not-symlink")
target = link.resolve()
if not target.is_dir():
return (link_path, f"skip-not-dir -> {target}")
bak_path = link.parent / f"{link.name}.link-bak"
if bak_path.exists():
return (link_path, f"error: 备份已存在 {bak_path}")
link.rename(bak_path)
try:
shutil.copytree(
str(target), str(link),
symlinks=False,
ignore_dangling_symlinks=True,
)
except Exception:
try:
shutil.rmtree(str(link), ignore_errors=True)
except Exception:
pass
bak_path.rename(link_path)
raise
return (link_path, f"ok (dir), backup at {bak_path.name}")
except Exception as exc:
return (link_path, f"error: {exc}")
def restore_one(bak_path_str: str) -> Result:
"""将 .link-bak 符号链接恢复回原名，并删除之前替换出来的普通文件。"""
try:
bak_path = Path(bak_path_str)
if not bak_path.name.endswith('.link-bak') or not bak_path.is_symlink():
return (bak_path_str, "skip-not-link-bak")
original_path_str = bak_path_str[:-len('.link-bak')]
original_path = Path(original_path_str)
if original_path.is_dir():
return (bak_path_str, "skip-original-is-dir")
orig_exists = original_path.exists() or original_path.is_symlink()
if orig_exists:
original_path.unlink()
bak_path.rename(original_path_str)
if orig_exists:
return (bak_path_str, f"restored (deleted replaced file)")
else:
return (bak_path_str, f"restored (original path was already absent)")
except Exception as exc:
return (bak_path_str, f"error: {exc}")
def restore_dir_one(bak_path_str: str) -> Result:
"""删除整个复制的真实目录树，再将 .link-bak 符号链接恢复回原名。"""
try:
bak_path = Path(bak_path_str)
if not bak_path.name.endswith('.link-bak') or not bak_path.is_symlink():
return (bak_path_str, "skip-not-link-bak")
original_path_str = bak_path_str[:-len('.link-bak')]
original_path = Path(original_path_str)
if not original_path.is_dir():
return (bak_path_str, "skip-not-real-dir")
shutil.rmtree(str(original_path))
bak_path.rename(original_path_str)
return (bak_path_str, "restored (deleted copied directory)")
except Exception as exc:
return (bak_path_str, f"error: {exc}")
def collect_symlinks(root: str) -> Generator[str, None, None]:
"""递归枚举 root 下所有符号链接（排除 .link-bak 文件），避免无限递归跟随。"""
try:
with os.scandir(root) as it:
for entry in it:
if entry.is_symlink():
if not entry.name.endswith('.link-bak'):
yield entry.path
if entry.is_dir() and not entry.is_symlink():
yield from collect_symlinks(entry.path)
except PermissionError as exc:
print(f"[warn] 权限不足，跳过目录: {exc}", file=sys.stderr)
except OSError as exc:
print(f"[warn] 扫描失败: {exc}", file=sys.stderr)
def collect_link_bak(root: str) -> Generator[str, None, None]:
"""递归枚举 root 下所有 .link-bak 符号链接（还原时使用）。"""
try:
with os.scandir(root) as it:
for entry in it:
if entry.is_symlink() and entry.name.endswith('.link-bak'):
yield entry.path
if entry.is_dir() and not entry.is_symlink():
yield from collect_link_bak(entry.path)
except PermissionError as exc:
print(f"[warn] 权限不足，跳过目录: {exc}", file=sys.stderr)
except OSError as exc:
print(f"[warn] 扫描失败: {exc}", file=sys.stderr)
def main():
parser = argparse.ArgumentParser(
description="并行将目录下的符号链接替换为实际文件（可还原）。"
)
parser.add_argument(
"root",
help="扫描路径（目录或单个文件）",
)
parser.add_argument(
"--restore",
action="store_true",
help="还原模式：扫描 .link-bak 备份文件并恢复原始符号链接",
)
parser.add_argument(
"-j", "--jobs",
type=int,
default=os.cpu_count() or 4,
help="并行 Worker 数（默认 = CPU 核数）",
)
parser.add_argument(
"--skip-names",
default="brew",
help="替换时要跳过的文件名黑名单，逗号分隔，不区分路径（默认 brew）",
)
parser.add_argument(
"--include-dir",
action="store_true",
help="也替换/还原指向目录的符号链接（默认仅处理文件符号链接）",
)
parser.add_argument(
"--dry-run",
action="store_true",
help="仅列出待处理的条目，不实际执行操作",
)
args = parser.parse_args()
target = os.path.abspath(os.path.expanduser(args.root))
if not os.path.lexists(target):
print(f"错误：路径不存在: {target}", file=sys.stderr)
sys.exit(1)
is_dir = os.path.isdir(target) and not os.path.islink(target)
if args.restore:
if is_dir:
bak_candidate = target + '.link-bak'
if os.path.islink(bak_candidate):
bak_links = [bak_candidate]
is_dir = False
print(f"还原单个目录备份: {target}")
else:
print(f"还原模式 - 扫描目录: {target}")
bak_links = list(collect_link_bak(target))
else:
if target.endswith('.link-bak') and os.path.islink(target):
bak_links = [target]
else:
bak_candidate = target + '.link-bak'
if os.path.islink(bak_candidate):
bak_links = [bak_candidate]
else:
print(f"错误：指定的文件不是符号链接也不是 .link-bak: {target}", file=sys.stderr)
sys.exit(1)
total = len(bak_links)
print(f"找到 {total} 个 .link-bak 备份")
if total == 0:
return
file_baks = []
dir_baks = []
for p in bak_links:
original = p[:-len('.link-bak')]
if os.path.isdir(original):
dir_baks.append(p)
else:
file_baks.append(p)
skipped_dirs = 0
if not args.include_dir and dir_baks:
skipped_dirs = len(dir_baks)
print(f"跳过 {skipped_dirs} 个目录备份（使用 --include-dir 可处理目录）")
dir_baks = []
if args.dry_run:
for p in bak_links:
bak_target = os.readlink(p)
original = p[:-len('.link-bak')] if p.endswith('.link-bak') else p
kind = " [目录]" if p in dir_baks else ""
print(f" 将恢复: {original}")
print(f" 备份: {p} -> {bak_target}{kind}")
return
def _run_restore(items, worker_fn, label):
if not items:
print(f" （无 {label} 需要处理）")
return (0, 0, 0)
n = len(items)
print(f" {label}: {n} 个, {args.jobs if n > 1 else '单任务'} workers")
restored_cnt = err_cnt = skip_cnt = 0
done = 0
with ProcessPoolExecutor(max_workers=args.jobs if n > 1 else 1) as pool:
fut_map = {pool.submit(worker_fn, p): p for p in items}
try:
for fut in as_completed(fut_map):
done += 1
path, status = fut.result()
pct = done * 100 // n
if status.startswith("restored"):
restored_cnt += 1
elif status.startswith("skip"):
skip_cnt += 1
else:
err_cnt += 1
print(f" [{done}/{n} {pct:3d}%] {path}")
if not status.startswith("restored"):
print(f" {status}")
except KeyboardInterrupt:
print("\n收到中断，等待正在进行的 worker 完成…", file=sys.stderr)
for fut in fut_map:
fut.cancel()
pool.shutdown(wait=False)
print(f"已取消。已处理 {done}/{n}，剩余已跳过。", file=sys.stderr)
sys.exit(130)
return (restored_cnt, skip_cnt, err_cnt)
signal.signal(signal.SIGINT, signal.SIG_IGN)
print()
print(f"阶段 1/2: 恢复文件备份")
r_files, s_files, e_files = _run_restore(file_baks, restore_one, "文件")
print()
print(f"阶段 2/2: 恢复目录备份")
r_dirs, s_dirs, e_dirs = _run_restore(dir_baks, restore_dir_one, "目录")
print()
r_total = r_files + r_dirs
s_total = s_files + s_dirs
e_total = e_files + e_dirs
print(f"还原完成: 已处理 {total}，恢复 {r_total}，跳过 {s_total}，错误 {e_total}"
f"{'，目录备份跳过 ' + str(skipped_dirs) if skipped_dirs else ''}")
return
skip_set = {n.strip() for n in args.skip_names.split(",")} if args.skip_names else set()
if is_dir:
print(f"扫描阶段: {target}")
links = list(collect_symlinks(target))
else:
links = [target]
total = len(links)
print(f"找到 {total} 条符号链接")
if total == 0:
return
to_replace = []
skipped_blacklist = 0
for p in links:
name = os.path.basename(p)
if name in skip_set:
skipped_blacklist += 1
else:
to_replace.append(p)
if skipped_blacklist:
print(f"黑名单过滤: 跳过 {skipped_blacklist} 条（位于 {args.skip_names} 黑名单）")
file_links = []
dir_links = []
for p in to_replace:
if os.path.islink(p) and os.path.isdir(p):
dir_links.append(p)
else:
file_links.append(p)
skipped_dirs = 0
if not args.include_dir and dir_links:
skipped_dirs = len(dir_links)
print(f"跳过 {skipped_dirs} 条目录符号链接（使用 --include-dir 可处理目录）")
dir_links = []
if args.dry_run:
for p in links:
if os.path.islink(p):
link_target = os.readlink(p)
else:
link_target = "(not a symlink)"
name = os.path.basename(p)
tag = " [blacklisted]" if name in skip_set else ""
kind = " [目录]" if p in dir_links else " [目录](已跳过)" if (os.path.islink(p) and os.path.isdir(p)) else ""
print(f" {p} -> {link_target}{tag}{kind}")
print(f" 其中: {len(file_links)} 条文件链接, {len(dir_links)} 条目录链接"
f"{'（跳过 ' + str(skipped_dirs) + '）' if skipped_dirs else ''}")
return
def _run_replace(items, worker_fn, label):
if not items:
print(f" （无 {label} 需要处理）")
return (0, 0, 0)
n = len(items)
print(f" {label}: {n} 个, {args.jobs if n > 1 else '单任务'} workers（备份为 .link-bak）")
ok_cnt = err_cnt = skip_cnt = 0
done = 0
with ProcessPoolExecutor(max_workers=args.jobs if n > 1 else 1) as pool:
fut_map = {pool.submit(worker_fn, p): p for p in items}
try:
for fut in as_completed(fut_map):
done += 1
path, status = fut.result()
pct = done * 100 // n
if status.startswith("ok"):
ok_cnt += 1
elif status.startswith("skip"):
skip_cnt += 1
else:
err_cnt += 1
print(f" [{done}/{n} {pct:3d}%] {path}")
if not status.startswith("ok"):
print(f" {status}")
except KeyboardInterrupt:
print("\n收到中断，等待正在进行的 worker 完成…", file=sys.stderr)
for fut in fut_map:
fut.cancel()
pool.shutdown(wait=False)
print(f"已取消。已处理 {done}/{n}，剩余已跳过。", file=sys.stderr)
sys.exit(130)
return (ok_cnt, skip_cnt, err_cnt)
if not file_links and not dir_links:
print("所有符号链接均被跳过，无操作。")
return
signal.signal(signal.SIGINT, signal.SIG_IGN)
print()
phase_file = "1/1" if not dir_links else "1/2"
print(f"阶段 {phase_file}: 替换文件符号链接")
ok_files, skip_files, err_files = _run_replace(file_links, backup_then_replace_one, "文件符号链接")
ok_dirs = skip_dirs = err_dirs = 0
if dir_links:
print()
print(f"阶段 2/2: 替换目录符号链接")
ok_dirs, skip_dirs, err_dirs = _run_replace(dir_links, backup_then_replace_dir_one, "目录符号链接")
print()
ok_total = ok_files + ok_dirs
skip_total = skip_files + skip_dirs
err_total = err_files + err_dirs
extras = []
if skipped_blacklist:
extras.append(f"黑名单跳过 {skipped_blacklist}")
if skipped_dirs:
extras.append(f"目录跳过 {skipped_dirs}")
extras_str = f"，{'，'.join(extras)}" if extras else ""
print(f"完成: 总 {len(links)}，替换 {ok_total}，跳过（条件）{skip_total}，错误 {err_total}{extras_str}")
if __name__ == "__main__":
main()