284 lines
8.5 KiB
Python
284 lines
8.5 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import pathlib
|
|
import sys
|
|
import tempfile
|
|
import zipfile
|
|
from dataclasses import dataclass
|
|
from typing import Iterable
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class SourceEntry:
|
|
source: pathlib.Path
|
|
entry_name: str
|
|
|
|
|
|
class BundleZipError(Exception):
|
|
pass
|
|
|
|
|
|
def normalize_part(value: str) -> str:
|
|
if os.name == "nt":
|
|
return value.casefold()
|
|
return value
|
|
|
|
|
|
def read_file_list(file_list: pathlib.Path) -> list[str]:
|
|
if not file_list.exists():
|
|
raise BundleZipError(f"File list does not exist: {file_list}")
|
|
if not file_list.is_file():
|
|
raise BundleZipError(f"File list is not a file: {file_list}")
|
|
|
|
lines = file_list.read_text(encoding="utf-8").splitlines()
|
|
return [line.strip() for line in lines if line.strip()]
|
|
|
|
|
|
def resolve_input_path(value: str, cwd: pathlib.Path) -> pathlib.Path:
|
|
path = pathlib.Path(value).expanduser()
|
|
if not path.is_absolute():
|
|
path = cwd / path
|
|
return path.resolve(strict=False)
|
|
|
|
|
|
def entry_name_for(path: pathlib.Path, marker: str) -> str:
|
|
normalized_marker = normalize_part(marker)
|
|
matches = [
|
|
index
|
|
for index, part in enumerate(path.parts)
|
|
if normalize_part(part) == normalized_marker
|
|
]
|
|
if not matches:
|
|
raise BundleZipError(
|
|
f"Source file is not under root marker '{marker}': {path}"
|
|
)
|
|
if len(matches) > 1:
|
|
raise BundleZipError(
|
|
f"Source file contains root marker '{marker}' more than once: {path}"
|
|
)
|
|
|
|
relative_parts = path.parts[matches[0] + 1 :]
|
|
if not relative_parts:
|
|
raise BundleZipError(
|
|
f"Source file cannot be made relative to root marker '{marker}': {path}"
|
|
)
|
|
return "/".join(relative_parts)
|
|
|
|
|
|
def validate_sources(
|
|
file_values: Iterable[str],
|
|
*,
|
|
marker: str,
|
|
output: pathlib.Path,
|
|
cwd: pathlib.Path,
|
|
) -> list[SourceEntry]:
|
|
if not marker:
|
|
raise BundleZipError("Relative root marker must not be empty")
|
|
|
|
output = output.resolve(strict=False)
|
|
entries: list[SourceEntry] = []
|
|
seen: dict[str, pathlib.Path] = {}
|
|
|
|
for value in file_values:
|
|
source = resolve_input_path(value, cwd)
|
|
if not source.exists():
|
|
raise BundleZipError(f"Source file does not exist: {source}")
|
|
if not source.is_file():
|
|
raise BundleZipError(f"Source path is not a file: {source}")
|
|
if source == output:
|
|
raise BundleZipError(f"Output path is the same as a source file: {source}")
|
|
|
|
entry_name = entry_name_for(source, marker)
|
|
if entry_name in seen:
|
|
raise BundleZipError(
|
|
"Duplicate zip entry produced: "
|
|
f"{entry_name} from {seen[entry_name]} and {source}"
|
|
)
|
|
seen[entry_name] = source
|
|
entries.append(SourceEntry(source=source, entry_name=entry_name))
|
|
|
|
if not entries:
|
|
raise BundleZipError("File list did not contain any source files")
|
|
|
|
return entries
|
|
|
|
|
|
def zip_entries(path: pathlib.Path) -> list[str]:
|
|
with zipfile.ZipFile(path) as archive:
|
|
return archive.namelist()
|
|
|
|
|
|
def write_zip(entries: list[SourceEntry], output: pathlib.Path) -> list[str]:
|
|
output_parent = output.parent.resolve(strict=False)
|
|
if not output_parent.exists():
|
|
raise BundleZipError(f"Output parent directory does not exist: {output_parent}")
|
|
if not output_parent.is_dir():
|
|
raise BundleZipError(f"Output parent path is not a directory: {output_parent}")
|
|
|
|
expected_entries = [entry.entry_name for entry in entries]
|
|
temp_name = ""
|
|
try:
|
|
with tempfile.NamedTemporaryFile(
|
|
prefix=f".{output.name}.",
|
|
suffix=".tmp",
|
|
dir=output_parent,
|
|
delete=False,
|
|
) as temp_file:
|
|
temp_name = temp_file.name
|
|
|
|
temp_path = pathlib.Path(temp_name)
|
|
with zipfile.ZipFile(
|
|
temp_path,
|
|
"w",
|
|
compression=zipfile.ZIP_DEFLATED,
|
|
compresslevel=6,
|
|
) as archive:
|
|
for entry in entries:
|
|
archive.write(entry.source, arcname=entry.entry_name)
|
|
|
|
actual_entries = zip_entries(temp_path)
|
|
if actual_entries != expected_entries:
|
|
raise BundleZipError(
|
|
"Zip verification failed: "
|
|
f"expected {expected_entries}, got {actual_entries}"
|
|
)
|
|
|
|
os.replace(temp_path, output)
|
|
final_entries = zip_entries(output)
|
|
if final_entries != expected_entries:
|
|
raise BundleZipError(
|
|
"Final zip verification failed: "
|
|
f"expected {expected_entries}, got {final_entries}"
|
|
)
|
|
return final_entries
|
|
finally:
|
|
if temp_name:
|
|
temp_path = pathlib.Path(temp_name)
|
|
if temp_path.exists():
|
|
temp_path.unlink()
|
|
|
|
|
|
def build_summary(
|
|
*,
|
|
status: str,
|
|
output: pathlib.Path,
|
|
source_file_count: int,
|
|
entries: list[str],
|
|
warnings: list[str] | None = None,
|
|
errors: list[str] | None = None,
|
|
) -> dict[str, object]:
|
|
return {
|
|
"status": status,
|
|
"output_zip": str(output.resolve(strict=False)),
|
|
"source_file_count": source_file_count,
|
|
"zip_entry_count": len(entries),
|
|
"entries": entries,
|
|
"warnings": warnings or [],
|
|
"errors": errors or [],
|
|
}
|
|
|
|
|
|
def print_summary(summary: dict[str, object], *, as_json: bool) -> None:
|
|
if as_json:
|
|
print(json.dumps(summary, indent=2, ensure_ascii=False))
|
|
return
|
|
|
|
print(summary["status"])
|
|
print(f"output_zip: {summary['output_zip']}")
|
|
print(f"source_file_count: {summary['source_file_count']}")
|
|
print(f"zip_entry_count: {summary['zip_entry_count']}")
|
|
errors = summary.get("errors") or []
|
|
warnings = summary.get("warnings") or []
|
|
if warnings:
|
|
print("warnings:")
|
|
for warning in warnings:
|
|
print(f"- {warning}")
|
|
if errors:
|
|
print("errors:")
|
|
for error in errors:
|
|
print(f"- {error}")
|
|
entries = summary.get("entries") or []
|
|
if entries:
|
|
print("entries:")
|
|
for entry in entries:
|
|
print(f"- {entry}")
|
|
|
|
|
|
def write_json_output(path: pathlib.Path, summary: dict[str, object]) -> None:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(json.dumps(summary, indent=2, ensure_ascii=False) + "\n", encoding="utf-8")
|
|
|
|
|
|
def parse_args(argv: list[str]) -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(
|
|
description="Create a source-relative zip from a newline-delimited file list.",
|
|
)
|
|
parser.add_argument("--file-list", required=True, help="Newline-delimited file list.")
|
|
parser.add_argument(
|
|
"--relative-root-marker",
|
|
required=True,
|
|
help="Path segment used as the root for zip entry names.",
|
|
)
|
|
parser.add_argument("--output", required=True, help="Destination zip path.")
|
|
parser.add_argument(
|
|
"--verify",
|
|
action="store_true",
|
|
help="Accepted for explicit workflows; verification is always performed.",
|
|
)
|
|
parser.add_argument(
|
|
"--json",
|
|
action="store_true",
|
|
help="Print the validation summary as JSON.",
|
|
)
|
|
parser.add_argument(
|
|
"--json-output",
|
|
help="Optional path to write the validation summary JSON.",
|
|
)
|
|
return parser.parse_args(argv)
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
args = parse_args(argv or sys.argv[1:])
|
|
cwd = pathlib.Path.cwd().resolve(strict=False)
|
|
file_list_path = resolve_input_path(args.file_list, cwd)
|
|
output = resolve_input_path(args.output, cwd)
|
|
file_values: list[str] = []
|
|
|
|
try:
|
|
file_values = read_file_list(file_list_path)
|
|
source_entries = validate_sources(
|
|
file_values,
|
|
marker=args.relative_root_marker,
|
|
output=output,
|
|
cwd=cwd,
|
|
)
|
|
entries = write_zip(source_entries, output)
|
|
summary = build_summary(
|
|
status="PASS",
|
|
output=output,
|
|
source_file_count=len(source_entries),
|
|
entries=entries,
|
|
)
|
|
exit_code = 0
|
|
except BundleZipError as error:
|
|
summary = build_summary(
|
|
status="FAIL",
|
|
output=output,
|
|
source_file_count=len(file_values),
|
|
entries=[],
|
|
errors=[str(error)],
|
|
)
|
|
exit_code = 1
|
|
|
|
if args.json_output:
|
|
write_json_output(resolve_input_path(args.json_output, cwd), summary)
|
|
print_summary(summary, as_json=args.json)
|
|
return exit_code
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|