Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
127 changes: 36 additions & 91 deletions src/luanma/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -42,54 +42,21 @@ def _build_parser() -> argparse.ArgumentParser:
help="待处理的 zip 压缩包或已解压出乱码文件名的目录",
)
action = parser.add_argument_group("动作(默认: 预览)")
action.add_argument(
"-x", "--extract", action="store_true", help="按检测到的编码解压"
)
action.add_argument(
"--fix",
action="store_true",
help="生成文件名为 UTF-8 的新压缩包(默认命名 *_utf8.zip)",
)
action.add_argument(
"--rename",
action="store_true",
help="重命名目录里的乱码文件名(输入为目录时)",
)
parser.add_argument(
"-e",
"--encoding",
help=f"手动指定文件名编码(可选: {', '.join(CANDIDATES)} 等)",
)
action.add_argument("-x", "--extract", action="store_true", help="按检测到的编码解压")
action.add_argument("--fix", action="store_true", help="生成文件名为 UTF-8 的新压缩包(默认命名 *_utf8.zip)")
action.add_argument("--rename", action="store_true", help="重命名目录里的乱码文件名(输入为目录时)")
parser.add_argument("-e", "--encoding", help=f"手动指定文件名编码(可选: {', '.join(CANDIDATES)} 等)")
parser.add_argument("-d", "--dest", help="解压目标目录(配合 -x)")
parser.add_argument(
"-o", "--output", help="输出压缩包路径(配合 --fix, 仅限单个输入)"
)
parser.add_argument(
"-p", "--password", help="加密压缩包的密码(ZipCrypto)"
)
parser.add_argument(
"--keep-junk",
action="store_true",
help="保留 __MACOSX、.DS_Store、Thumbs.db 等系统垃圾文件",
)
parser.add_argument(
"--json", action="store_true", help="以 JSON 输出(供脚本使用)"
)
parser.add_argument(
"-V", "--version", action="version", version=f"luanma {__version__}"
)
parser.add_argument("-o", "--output", help="输出压缩包路径(配合 --fix, 仅限单个输入)")
parser.add_argument("-p", "--password", help="加密压缩包的密码(ZipCrypto)")
parser.add_argument("--keep-junk", action="store_true", help="保留 __MACOSX、.DS_Store、Thumbs.db 等系统垃圾文件")
parser.add_argument("--json", action="store_true", help="以 JSON 输出(供脚本使用)")
parser.add_argument("-V", "--version", action="version", version=f"luanma {__version__}")
return parser


def _display(text: str) -> str:
"""Neutralize control characters before echoing untrusted names.

Archive members are attacker-controlled; a name embedding ESC could
otherwise inject terminal escape sequences into the preview.
"""
return "".join(
ch if ch == " " or ch.isprintable() else "·" for ch in text
)
return "".join(ch if ch == " " or ch.isprintable() else "·" for ch in text)


def _shorten(text: str, width: int = 58) -> str:
Expand All @@ -104,10 +71,7 @@ def _print_preview(path: str, det, previews) -> None:
if encrypted:
print(f" (含 {encrypted} 个加密条目, 解压需 -p 提供密码)")
return
print(
f"{path}: 检测到文件名编码 {det.encoding.upper()}"
f"(置信度: {det.confidence_label})"
)
print(f"{path}: 检测到文件名编码 {det.encoding.upper()}(置信度: {det.confidence_label})")
if det.confidence == "low":
ranking = ", ".join(det.ranked()[:3])
print(f" 注意: 置信度较低, 候选编码依次为 {ranking}, 可用 -e 指定")
Expand All @@ -123,45 +87,32 @@ def _print_preview(path: str, det, previews) -> None:
print(f" (另有 {junk} 个系统垃圾文件将被跳过, --keep-junk 可保留)")
if encrypted:
print(f" (含 {encrypted} 个加密条目, 解压/转换需 -p 提供密码)")
print(
"提示: 加 -x 按此编码解压, 或加 --fix 生成 UTF-8 压缩包; "
"-e <编码> 可手动指定"
)
print("提示: 加 -x 按此编码解压, 或加 --fix 生成 UTF-8 压缩包; -e <编码> 可手动指定")


def _run_dir(path: str, args) -> dict:
"""Preview or apply mojibake renames inside an extracted directory."""
if args.extract or args.fix:
raise ArchiveError(
f"{path} 是目录; -x/--fix 仅用于压缩包, 目录请用 --rename"
)
raise ArchiveError(f"{path} 是目录; -x/--fix 仅用于压缩包, 目录请用 --rename")
entry: dict = {"path": path}
report = rename_dir(path, encoding=args.encoding, apply=args.rename)
entry["action"] = "rename" if args.rename else "dir-preview"
entry["report"] = report.to_dict()
if args.json:
return entry

if not report.planned and not report.skipped_unsure:
print(f"{path}: 未发现乱码文件名, 无需修复")
return entry
if report.planned:
enc = (report.encoding or "?").upper()
verb = "已重命名" if args.rename else "计划重命名"
print(
f"{path}: 检测到编码 {enc}(置信度: {report.confidence_label}), "
f"{verb} {len(report.planned)} 项"
)
print(f"{path}: 检测到编码 {enc}(置信度: {report.confidence_label}), {verb} {len(report.planned)} 项")
for item in report.planned:
print(f" {_shorten(item.old_name)}")
print(f" -> {_shorten(item.new_name)}")
else:
print(f"{path}: 未发现可确定修复的乱码文件名")
if report.skipped_unsure:
print(
f" (另有 {report.skipped_unsure} 个名字疑似乱码但把握不足, "
"已跳过; 可用 -e 指定编码后重试)"
)
print(f" (另有 {report.skipped_unsure} 个名字疑似乱码但把握不足, 已跳过; 可用 -e 指定编码后重试)")
for err in report.errors:
print(f" 警告: {_display(err)}")
if report.new_root:
Expand All @@ -176,43 +127,33 @@ def _run_one(path: str, args) -> dict:
return _run_dir(path, args)
if args.rename:
raise ArchiveError(f"--rename 仅用于目录, 而 {path} 不是目录")

entry: dict = {"path": path}
det, previews = preview_zip(path, args.encoding)
entry["encoding"] = det.encoding
entry["confidence"] = det.confidence
entry["needs_fix"] = det.needs_fix
entry["entries"] = [p.to_dict() for p in previews]

if args.extract:
report = extract_zip(
path, dest=args.dest, encoding=args.encoding,
keep_junk=args.keep_junk, password=args.password,
)
report = extract_zip(path, dest=args.dest, encoding=args.encoding, keep_junk=args.keep_junk, password=args.password)
entry["action"] = "extract"
entry["report"] = report.to_dict()
if not args.json:
print(
f"{path}: 已解压 {report.extracted} 个文件到 {report.dest}"
f" (编码: {report.encoding or '原样'},"
f" 跳过垃圾文件 {report.junk_skipped},"
f" 重名改名 {report.renamed})"
)
print(f"{path}: 已解压 {report.extracted} 个文件到 {report.dest} (编码: {report.encoding or '原样'}, 跳过垃圾文件 {report.junk_skipped}, 重名改名 {report.renamed})")
for err in report.errors:
print(f" 警告: {_display(err)}")
elif args.fix:
report = convert_zip(
path, output=args.output, encoding=args.encoding,
keep_junk=args.keep_junk, password=args.password,
)
if args.output:
out_parent = os.path.dirname(os.path.abspath(args.output))
if out_parent:
try:
os.makedirs(out_parent, exist_ok=True)
except OSError as exc:
raise ArchiveError(f"无法创建输出目录 {out_parent}: {exc}") from exc
report = convert_zip(path, output=args.output, encoding=args.encoding, keep_junk=args.keep_junk, password=args.password)
entry["action"] = "convert"
entry["report"] = report.to_dict()
if not args.json:
print(
f"{path}: 已生成 {report.output}"
f" (共 {report.converted} 个条目, 文件名已转为 UTF-8,"
f" 跳过垃圾文件 {report.junk_skipped})"
)
print(f"{path}: 已生成 {report.output} (共 {report.converted} 个条目, 文件名已转为 UTF-8, 跳过垃圾文件 {report.junk_skipped})")
for err in report.errors:
print(f" 警告: {_display(err)}")
else:
Expand All @@ -223,15 +164,23 @@ def _run_one(path: str, args) -> dict:


def _expand_archives(patterns: List[str]) -> List[str]:
"""Expand glob patterns ourselves: PowerShell/cmd do not expand ``*``."""
"""Expand glob patterns ourselves: PowerShell/cmd do not expand ``*``.

A path that already exists is used as-is. Otherwise ``foo[bar].zip``
would be treated as a glob character class and silently match
``foob.zip`` / ``foor.zip`` instead of the literal file.
"""
paths: List[str] = []
for pattern in patterns:
if os.path.exists(pattern):
paths.append(pattern)
continue
if any(ch in pattern for ch in "*?["):
matches = sorted(glob.glob(pattern, recursive=True))
if matches:
paths.extend(matches)
else:
paths.append(pattern) # keep it: reported as missing later
paths.append(pattern)
else:
paths.append(pattern)
return paths
Expand All @@ -246,7 +195,6 @@ def main(argv: Optional[List[str]] = None) -> int:
pass
parser = _build_parser()
args = parser.parse_args(argv)

actions = [args.extract, args.fix, args.rename]
if sum(actions) > 1:
parser.error("-x、--fix、--rename 只能选择一个")
Expand All @@ -259,11 +207,9 @@ def main(argv: Optional[List[str]] = None) -> int:
"".encode(args.encoding)
except LookupError:
parser.error(f"未知编码: {args.encoding}")

archives = _expand_archives(args.archives)
if args.output and len(archives) > 1:
parser.error("-o 只能用于单个压缩包")

results = []
failed = False
partial = False
Expand All @@ -279,7 +225,6 @@ def main(argv: Optional[List[str]] = None) -> int:
results.append({"path": path, "error": str(exc)})
else:
print(f"错误: {exc}", file=sys.stderr)

if args.json:
print(json.dumps({"archives": results}, ensure_ascii=False, indent=2))
if failed:
Expand Down
18 changes: 14 additions & 4 deletions tests/test_cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -186,11 +186,21 @@ def test_rename_conflicts_with_fix(tmp_path):
main([str(d), "--rename", "--fix"])


def test_fix_output_missing_dir_exit2(tmp_path, capsys):
def test_fix_output_creates_missing_parent_dir(tmp_path):
p = make_bad_zip(tmp_path / "gbk.zip", GBK_NAMES, "gbk")
out = tmp_path / "no_such_dir" / "o.zip"
assert main([str(p), "--fix", "-o", str(out)]) == 2
assert "无法写入" in capsys.readouterr().err
out = tmp_path / "no_such_dir" / "nested" / "o.zip"
assert main([str(p), "--fix", "-o", str(out)]) == 0
assert out.is_file()
with zipfile.ZipFile(out) as zf:
assert "课程设计说明.txt" in zf.namelist()


def test_fix_output_parent_is_file_exit2(tmp_path, capsys):
p = make_bad_zip(tmp_path / "gbk.zip", GBK_NAMES, "gbk")
blocker = tmp_path / "blocker"
blocker.write_bytes(b"file")
assert main([str(p), "--fix", "-o", str(blocker / "o.zip")]) == 2
assert "无法" in capsys.readouterr().err


def test_extract_dest_under_file_exit2(tmp_path, capsys):
Expand Down
13 changes: 13 additions & 0 deletions tests/test_literal_glob.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,13 @@
from luanma.cli import main

from helpers import make_utf8_zip


def test_literal_bracket_filename_is_not_globbed(tmp_path, capsys):
literal = tmp_path / "foo[bar].zip"
p = make_utf8_zip(literal, {"ok.txt": b"x"})
decoy = make_utf8_zip(tmp_path / "foob.zip", {"other.txt": b"y"})
assert main([str(p)]) == 0
out = capsys.readouterr().out
assert "foo[bar].zip" in out
assert str(decoy) not in out
Loading