winddrescue.py

#!/usr/bin/env python3
"""
winddrescue: a small Windows-oriented rescue copier.

This is a from-scratch implementation inspired by GNU ddrescue's documented
recovery flow: rescue large good areas first, keep a mapfile for resume, then
trim/scrape failed ranges at sector granularity and optionally retry bad
sectors. It does not contain GNU ddrescue source code.
"""

from __future__ import annotations

import argparse
import ctypes
from ctypes import wintypes
from dataclasses import dataclass
import os
import re
import shutil
import signal
import sys
import time
from typing import Iterable, List, Optional, Sequence, Tuple

from winddrescue_metadata import APP_VERSION


if os.name != "nt":
    raise SystemExit("winddrescue currently runs only on Windows.")


kernel32 = ctypes.WinDLL("kernel32", use_last_error=True)

GENERIC_READ = 0x80000000
GENERIC_WRITE = 0x40000000
FILE_SHARE_READ = 0x00000001
FILE_SHARE_WRITE = 0x00000002
FILE_SHARE_DELETE = 0x00000004
CREATE_ALWAYS = 2
OPEN_EXISTING = 3
OPEN_ALWAYS = 4
FILE_ATTRIBUTE_NORMAL = 0x00000080
FILE_FLAG_SEQUENTIAL_SCAN = 0x08000000
FILE_BEGIN = 0
INVALID_HANDLE_VALUE = ctypes.c_void_p(-1).value


kernel32.CreateFileW.argtypes = [
    wintypes.LPCWSTR,
    wintypes.DWORD,
    wintypes.DWORD,
    wintypes.LPVOID,
    wintypes.DWORD,
    wintypes.DWORD,
    wintypes.HANDLE,
]
kernel32.CreateFileW.restype = wintypes.HANDLE
kernel32.ReadFile.argtypes = [
    wintypes.HANDLE,
    wintypes.LPVOID,
    wintypes.DWORD,
    ctypes.POINTER(wintypes.DWORD),
    wintypes.LPVOID,
]
kernel32.ReadFile.restype = wintypes.BOOL
kernel32.WriteFile.argtypes = [
    wintypes.HANDLE,
    wintypes.LPCVOID,
    wintypes.DWORD,
    ctypes.POINTER(wintypes.DWORD),
    wintypes.LPVOID,
]
kernel32.WriteFile.restype = wintypes.BOOL
kernel32.SetFilePointerEx.argtypes = [
    wintypes.HANDLE,
    ctypes.c_longlong,
    ctypes.POINTER(ctypes.c_longlong),
    wintypes.DWORD,
]
kernel32.SetFilePointerEx.restype = wintypes.BOOL
kernel32.GetFileSizeEx.argtypes = [wintypes.HANDLE, ctypes.POINTER(ctypes.c_longlong)]
kernel32.GetFileSizeEx.restype = wintypes.BOOL
kernel32.SetEndOfFile.argtypes = [wintypes.HANDLE]
kernel32.SetEndOfFile.restype = wintypes.BOOL
kernel32.FlushFileBuffers.argtypes = [wintypes.HANDLE]
kernel32.FlushFileBuffers.restype = wintypes.BOOL
kernel32.CloseHandle.argtypes = [wintypes.HANDLE]
kernel32.CloseHandle.restype = wintypes.BOOL
kernel32.FormatMessageW.argtypes = [
    wintypes.DWORD,
    wintypes.LPVOID,
    wintypes.DWORD,
    wintypes.DWORD,
    wintypes.LPWSTR,
    wintypes.DWORD,
    wintypes.LPVOID,
]
kernel32.FormatMessageW.restype = wintypes.DWORD


STATUS_LABEL = {
    "?": "non-tried",
    "+": "rescued",
    "*": "non-trimmed",
    "/": "non-scraped",
    "-": "bad-sector",
}


class WinApiError(OSError):
    def __init__(self, action: str, code: Optional[int] = None):
        self.code = ctypes.get_last_error() if code is None else code
        super().__init__(self.code, f"{action}: {format_win_error(self.code)}")


class ReadFailure(Exception):
    def __init__(self, pos: int, size: int, code: int, message: str):
        self.pos = pos
        self.size = size
        self.code = code
        self.message = message
        super().__init__(f"read failed at 0x{pos:X} size 0x{size:X}: {message}")


def format_win_error(code: int) -> str:
    flags = 0x00001000 | 0x00000200  # FORMAT_MESSAGE_FROM_SYSTEM | IGNORE_INSERTS
    buf = ctypes.create_unicode_buffer(512)
    length = kernel32.FormatMessageW(flags, None, code, 0, buf, len(buf), None)
    if length:
        return buf.value.strip()
    return f"Windows error {code}"


def parse_size(value: str) -> int:
    match = re.fullmatch(r"\s*(0x[0-9a-fA-F]+|\d+)\s*([KMGTPE]?i?B?|[kmgtpe]?)?\s*", value)
    if not match:
        raise argparse.ArgumentTypeError(f"invalid size: {value!r}")
    number_text, suffix = match.groups()
    number = int(number_text, 0)
    suffix = (suffix or "").upper()
    suffix = suffix[:-1] if suffix.endswith("B") else suffix
    suffix = suffix[:-1] if suffix.endswith("I") else suffix
    multiplier = {
        "": 1,
        "K": 1024,
        "M": 1024**2,
        "G": 1024**3,
        "T": 1024**4,
        "P": 1024**5,
        "E": 1024**6,
    }.get(suffix)
    if multiplier is None:
        raise argparse.ArgumentTypeError(f"invalid size suffix: {value!r}")
    return number * multiplier


def parse_bad_ranges(value: str) -> List[Tuple[int, int]]:
    ranges: List[Tuple[int, int]] = []
    if not value:
        return ranges
    for part in value.split(","):
        fields = part.split(":")
        if len(fields) != 2:
            raise argparse.ArgumentTypeError("bad ranges must use pos:size[,pos:size]")
        pos = parse_size(fields[0])
        size = parse_size(fields[1])
        if size <= 0:
            raise argparse.ArgumentTypeError("bad range size must be positive")
        ranges.append((pos, pos + size))
    return ranges


def human_size(value: int) -> str:
    units = ["B", "KiB", "MiB", "GiB", "TiB"]
    amount = float(value)
    for unit in units:
        if amount < 1024.0 or unit == units[-1]:
            if unit == "B":
                return f"{value} B"
            return f"{amount:.1f} {unit}"
        amount /= 1024.0
    return f"{value} B"


class WinFile:
    def __init__(self, path: str, access: int, creation: int):
        self.path = path
        self.access = access
        self.creation = creation
        self.handle = self._open()

    def _open(self):
        handle = kernel32.CreateFileW(
            self.path,
            self.access,
            FILE_SHARE_READ | FILE_SHARE_WRITE | FILE_SHARE_DELETE,
            None,
            self.creation,
            FILE_ATTRIBUTE_NORMAL | FILE_FLAG_SEQUENTIAL_SCAN,
            None,
        )
        if handle == INVALID_HANDLE_VALUE:
            raise WinApiError(f"open {self.path!r}")
        return handle

    def reopen(self) -> None:
        self.close()
        self.handle = self._open()

    def seek(self, pos: int) -> None:
        new_pos = ctypes.c_longlong()
        if not kernel32.SetFilePointerEx(self.handle, ctypes.c_longlong(pos), ctypes.byref(new_pos), FILE_BEGIN):
            raise WinApiError(f"seek {self.path!r} to 0x{pos:X}")

    def size(self) -> int:
        value = ctypes.c_longlong()
        if not kernel32.GetFileSizeEx(self.handle, ctypes.byref(value)):
            raise WinApiError(f"get size for {self.path!r}")
        return int(value.value)

    def set_size(self, size: int) -> None:
        self.seek(size)
        if not kernel32.SetEndOfFile(self.handle):
            raise WinApiError(f"set size for {self.path!r} to 0x{size:X}")

    def read_at(self, pos: int, size: int) -> bytes:
        if size <= 0:
            return b""
        self.seek(pos)
        buf = ctypes.create_string_buffer(size)
        read = wintypes.DWORD()
        ok = kernel32.ReadFile(self.handle, buf, size, ctypes.byref(read), None)
        if not ok:
            code = ctypes.get_last_error()
            raise ReadFailure(pos, size, code, format_win_error(code))
        return buf.raw[: read.value]

    def write_at(self, pos: int, data: bytes) -> None:
        view = memoryview(data)
        offset = 0
        while offset < len(view):
            chunk = view[offset : offset + 0x7FFF0000]
            self.seek(pos + offset)
            written = wintypes.DWORD()
            ok = kernel32.WriteFile(self.handle, chunk.tobytes(), len(chunk), ctypes.byref(written), None)
            if not ok:
                raise WinApiError(f"write {self.path!r} at 0x{pos + offset:X}")
            if written.value == 0:
                raise OSError(f"write {self.path!r} at 0x{pos + offset:X}: wrote zero bytes")
            offset += written.value

    def flush(self) -> None:
        if not kernel32.FlushFileBuffers(self.handle):
            raise WinApiError(f"flush {self.path!r}")

    def close(self) -> None:
        handle = getattr(self, "handle", None)
        if handle and handle != INVALID_HANDLE_VALUE:
            kernel32.CloseHandle(handle)
            self.handle = None

    def __enter__(self) -> "WinFile":
        return self

    def __exit__(self, exc_type, exc, tb) -> None:
        self.close()


@dataclass
class Extent:
    pos: int
    size: int
    status: str

    @property
    def end(self) -> int:
        return self.pos + self.size


class RescueMap:
    def __init__(self, extents: Sequence[Extent], current_pos: int = 0, phase: str = "?", current_pass: int = 1):
        self.extents = list(extents)
        self.current_pos = current_pos
        self.phase = phase
        self.current_pass = current_pass
        self.compact()

    @classmethod
    def initial(cls, size: int) -> "RescueMap":
        return cls([Extent(0, size, "?")])

    @classmethod
    def load(cls, path: str, size: int) -> "RescueMap":
        if not os.path.exists(path) or os.path.getsize(path) == 0:
            return cls.initial(size)
        non_comment: List[str] = []
        with open(path, "r", encoding="utf-8") as handle:
            for raw in handle:
                line = raw.split("#", 1)[0].strip()
                if line:
                    non_comment.append(line)
        if not non_comment:
            return cls.initial(size)
        status_fields = non_comment[0].split()
        if len(status_fields) < 3:
            raise ValueError(f"invalid map status line in {path}")
        current_pos = int(status_fields[0], 0)
        phase = status_fields[1]
        current_pass = int(status_fields[2], 0)
        extents: List[Extent] = []
        for line in non_comment[1:]:
            fields = line.split()
            if len(fields) != 3:
                raise ValueError(f"invalid map extent line: {line}")
            pos = int(fields[0], 0)
            extent_size = int(fields[1], 0)
            status = fields[2]
            if status not in STATUS_LABEL:
                raise ValueError(f"invalid map status {status!r}")
            if extent_size > 0:
                extents.append(Extent(pos, extent_size, status))
        result = cls(extents, current_pos, phase, current_pass)
        result.normalize(size)
        return result

    def save(self, path: str, operation: str = "") -> None:
        directory = os.path.dirname(os.path.abspath(path)) or "."
        os.makedirs(directory, exist_ok=True)
        tmp_path = path + ".tmp"
        bak_path = path + ".bak"
        with open(tmp_path, "w", encoding="utf-8", newline="\n") as handle:
            now = time.strftime("%Y-%m-%d %H:%M:%S")
            handle.write("# Mapfile. Created by winddrescue\n")
            handle.write("# Format follows GNU ddrescue-style status characters.\n")
            handle.write(f"# Current time: {now}\n")
            if operation:
                handle.write(f"# {operation}\n")
            handle.write("# current_pos current_status current_pass\n")
            handle.write(f"0x{self.current_pos:016X} {self.phase} {self.current_pass}\n")
            handle.write("# pos size status\n")
            for extent in self.extents:
                handle.write(f"0x{extent.pos:016X} 0x{extent.size:016X} {extent.status}\n")
            handle.flush()
            os.fsync(handle.fileno())
        if os.path.exists(path):
            try:
                os.replace(path, bak_path)
            except OSError:
                pass
        os.replace(tmp_path, path)

    def compact(self) -> None:
        self.extents = [e for e in sorted(self.extents, key=lambda item: item.pos) if e.size > 0]
        merged: List[Extent] = []
        for extent in self.extents:
            if merged and merged[-1].end == extent.pos and merged[-1].status == extent.status:
                merged[-1].size += extent.size
            else:
                merged.append(Extent(extent.pos, extent.size, extent.status))
        self.extents = merged

    def normalize(self, size: int) -> None:
        normalized: List[Extent] = []
        cursor = 0
        for extent in sorted(self.extents, key=lambda item: item.pos):
            if extent.end <= 0 or extent.pos >= size:
                continue
            pos = max(0, extent.pos)
            end = min(size, extent.end)
            if pos > cursor:
                normalized.append(Extent(cursor, pos - cursor, "?"))
            if end > cursor:
                normalized.append(Extent(max(pos, cursor), end - max(pos, cursor), extent.status))
                cursor = end
        if cursor < size:
            normalized.append(Extent(cursor, size - cursor, "?"))
        self.extents = normalized
        self.compact()

    def set_range(self, pos: int, size: int, status: str) -> None:
        if size <= 0:
            return
        end = pos + size
        updated: List[Extent] = []
        for extent in self.extents:
            if extent.end <= pos or extent.pos >= end:
                updated.append(extent)
                continue
            if extent.pos < pos:
                updated.append(Extent(extent.pos, pos - extent.pos, extent.status))
            middle_start = max(extent.pos, pos)
            middle_end = min(extent.end, end)
            updated.append(Extent(middle_start, middle_end - middle_start, status))
            if extent.end > end:
                updated.append(Extent(end, extent.end - end, extent.status))
        self.extents = updated
        self.compact()

    def ranges(self, statuses: Iterable[str], reverse: bool = False) -> List[Extent]:
        wanted = set(statuses)
        result = [Extent(e.pos, e.size, e.status) for e in self.extents if e.status in wanted]
        return list(reversed(result)) if reverse else result

    def stats(self) -> dict:
        result = {status: 0 for status in STATUS_LABEL}
        for extent in self.extents:
            result[extent.status] += extent.size
        return result


class RescueCopier:
    def __init__(
        self,
        source_path: str,
        dest_path: str,
        map_path: str,
        size: int,
        block_size: int,
        sector_size: int,
        retries: int,
        force: bool,
        no_trim: bool,
        no_scrape: bool,
        reopen_on_error: bool,
        bad_ranges: Sequence[Tuple[int, int]],
    ):
        self.source_path = source_path
        self.dest_path = dest_path
        self.map_path = map_path
        self.size = size
        self.block_size = block_size
        self.sector_size = sector_size
        self.retries = retries
        self.force = force
        self.no_trim = no_trim
        self.no_scrape = no_scrape
        self.reopen_on_error = reopen_on_error
        self.bad_ranges = list(bad_ranges)
        self.read_errors = 0
        self.started = time.monotonic()
        self.last_save = 0.0
        self.last_report = 0.0
        self.map_exists = os.path.exists(map_path)
        self.map = RescueMap.load(map_path, size)
        self.source: Optional[WinFile] = None
        self.dest: Optional[WinFile] = None

    def open_files(self) -> None:
        if not self.map_exists and os.path.exists(self.dest_path) and not self.force:
            raise SystemExit(
                f"destination exists: {self.dest_path}\n"
                "Use --force to start a new rescue and overwrite it, or keep the existing mapfile to resume."
            )
        self.source = WinFile(self.source_path, GENERIC_READ, OPEN_EXISTING)
        creation = OPEN_ALWAYS if self.map_exists else CREATE_ALWAYS
        self.dest = WinFile(self.dest_path, GENERIC_READ | GENERIC_WRITE, creation)
        self.dest.set_size(self.size)

    def close_files(self) -> None:
        if self.dest:
            self.dest.flush()
            self.dest.close()
        if self.source:
            self.source.close()

    def save(self, operation: str = "", force: bool = False) -> None:
        now = time.monotonic()
        if force or now - self.last_save >= 2.0:
            self.map.save(self.map_path, operation)
            self.last_save = now

    def report(self, operation: str, force: bool = False) -> None:
        now = time.monotonic()
        if not force and now - self.last_report < 0.25:
            return
        stats = self.map.stats()
        rescued = stats["+"]
        bad = stats["-"]
        pending = stats["?"] + stats["*"] + stats["/"]
        pct = (rescued / self.size * 100.0) if self.size else 100.0
        elapsed = max(now - self.started, 0.001)
        rate = rescued / elapsed
        line = (
            f"{operation} | rescued {pct:6.2f}% ({human_size(rescued)}) "
            f"bad {human_size(bad)} pending {human_size(pending)} "
            f"errors {self.read_errors} avg {human_size(int(rate))}/s"
        )
        width = shutil.get_terminal_size(fallback=(120, 24)).columns
        sys.stdout.write("\r" + line[: max(0, width - 1)])
        sys.stdout.flush()
        self.last_report = now

    def finish_line(self) -> None:
        sys.stdout.write("\n")
        sys.stdout.flush()

    def simulated_failure(self, pos: int, size: int) -> Optional[ReadFailure]:
        end = pos + size
        for bad_start, bad_end in self.bad_ranges:
            if pos < bad_end and end > bad_start:
                return ReadFailure(pos, size, 0xFFFFFFFF, "simulated read error")
        return None

    def read_input(self, pos: int, size: int) -> bytes:
        simulated = self.simulated_failure(pos, size)
        if simulated:
            raise simulated
        assert self.source is not None
        data = self.source.read_at(pos, size)
        if len(data) != size:
            raise ReadFailure(pos, size, 38, f"short read: got {len(data)} of {size} bytes")
        return data

    def try_copy(self, pos: int, size: int, failure_status: str, operation: str) -> bool:
        self.map.current_pos = pos
        try:
            data = self.read_input(pos, size)
        except ReadFailure:
            self.read_errors += 1
            self.map.set_range(pos, size, failure_status)
            if self.reopen_on_error and self.source:
                self.source.reopen()
            self.save(operation)
            self.report(operation)
            return False
        assert self.dest is not None
        self.dest.write_at(pos, data)
        self.map.set_range(pos, size, "+")
        self.save(operation)
        self.report(operation)
        return True

    def copy_pass(self, reverse: bool, pass_no: int) -> None:
        direction = "backwards" if reverse else "forwards"
        operation = f"Copying non-tried blocks... Pass {pass_no} ({direction})"
        self.map.phase = "?"
        self.map.current_pass = pass_no
        self.save(operation, force=True)
        ranges = self.map.ranges(["?"], reverse=reverse)
        for extent in ranges:
            if reverse:
                cursor = extent.end
                while cursor > extent.pos:
                    size = min(self.block_size, cursor - extent.pos)
                    cursor -= size
                    self.try_copy(cursor, size, "*", operation)
            else:
                cursor = extent.pos
                while cursor < extent.end:
                    size = min(self.block_size, extent.end - cursor)
                    self.try_copy(cursor, size, "*", operation)
                    cursor += size

    def trim(self) -> None:
        if self.no_trim:
            return
        operation = "Trimming failed blocks"
        self.map.phase = "*"
        self.map.current_pass = 1
        self.save(operation, force=True)
        for extent in self.map.ranges(["*"]):
            start = extent.pos
            end = extent.end
            front = start
            while front < end:
                size = min(self.sector_size, end - front)
                if self.try_copy(front, size, "-", operation):
                    front += size
                    continue
                front += size
                break
            back = end
            while back > front:
                size = min(self.sector_size, back - front)
                pos = back - size
                if self.try_copy(pos, size, "-", operation):
                    back = pos
                    continue
                back = pos
                break
            if front < back:
                self.map.set_range(front, back - front, "/")
                self.save(operation)
                self.report(operation)

    def scrape(self) -> None:
        if self.no_scrape:
            return
        operation = "Scraping non-rescued blocks sector by sector"
        self.map.phase = "/"
        self.map.current_pass = 1
        self.save(operation, force=True)
        for extent in self.map.ranges(["/", "*"]):
            cursor = extent.pos
            while cursor < extent.end:
                size = min(self.sector_size, extent.end - cursor)
                self.try_copy(cursor, size, "-", operation)
                cursor += size

    def retry(self) -> None:
        for pass_no in range(1, self.retries + 1):
            reverse = pass_no % 2 == 0
            direction = "backwards" if reverse else "forwards"
            operation = f"Retrying bad sectors... Pass {pass_no} ({direction})"
            self.map.phase = "-"
            self.map.current_pass = pass_no
            self.save(operation, force=True)
            for extent in self.map.ranges(["-"], reverse=reverse):
                if reverse:
                    cursor = extent.end
                    while cursor > extent.pos:
                        size = min(self.sector_size, cursor - extent.pos)
                        cursor -= size
                        self.try_copy(cursor, size, "-", operation)
                else:
                    cursor = extent.pos
                    while cursor < extent.end:
                        size = min(self.sector_size, extent.end - cursor)
                        self.try_copy(cursor, size, "-", operation)
                        cursor += size

    def run(self) -> None:
        self.open_files()
        try:
            self.copy_pass(reverse=False, pass_no=1)
            if self.map.stats()["?"]:
                self.copy_pass(reverse=True, pass_no=2)
            self.trim()
            self.scrape()
            self.retry()
            self.map.phase = "?"
            self.map.current_pass = 0
            self.save("Finished", force=True)
            self.report("Finished", force=True)
            self.finish_line()
        finally:
            self.close_files()


def determine_source_size(source_path: str, requested_size: Optional[int]) -> int:
    if requested_size is not None:
        return requested_size
    with WinFile(source_path, GENERIC_READ, OPEN_EXISTING) as source:
        try:
            return source.size()
        except WinApiError as exc:
            raise SystemExit(f"could not determine source size: {exc}\nUse --size for raw devices.") from exc


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        description=(
            "Copy a damaged Windows file/device to an output file, skipping unreadable "
            "ranges first and revisiting them later using a resumable mapfile."
        )
    )
    parser.add_argument("source", help="source file or device path, e.g. video.mp4 or \\\\.\\PhysicalDrive1")
    parser.add_argument("destination", help="destination output file")
    parser.add_argument("mapfile", help="rescue mapfile used for resume")
    parser.add_argument("-b", "--block-size", type=parse_size, default=1024 * 1024, help="large copy block size (default: 1M)")
    parser.add_argument("-s", "--sector-size", type=parse_size, default=4096, help="small retry block size (default: 4096)")
    parser.add_argument("-r", "--retries", type=int, default=1, help="bad-sector retry passes after scraping (default: 1)")
    parser.add_argument("--size", type=parse_size, help="copy only this many bytes; required if source size cannot be detected")
    parser.add_argument("-f", "--force", action="store_true", help="overwrite destination when starting without an existing mapfile")
    parser.add_argument("--no-trim", action="store_true", help="skip edge trimming of failed large blocks")
    parser.add_argument("--no-scrape", action="store_true", help="skip sector-by-sector scraping")
    parser.add_argument("--reopen-on-error", action="store_true", help="close and reopen source after each read error")
    parser.add_argument("--version", action="version", version=f"%(prog)s {APP_VERSION}")
    parser.add_argument(
        "--simulate-bad-ranges",
        type=parse_bad_ranges,
        default=[],
        metavar="POS:SIZE[,POS:SIZE]",
        help=argparse.SUPPRESS,
    )
    return parser


def validate_args(args: argparse.Namespace) -> None:
    if args.block_size <= 0:
        raise SystemExit("--block-size must be positive")
    if args.sector_size <= 0:
        raise SystemExit("--sector-size must be positive")
    if args.retries < 0:
        raise SystemExit("--retries must be zero or greater")
    if args.block_size < args.sector_size:
        raise SystemExit("--block-size must be at least --sector-size")


def main(argv: Optional[Sequence[str]] = None) -> int:
    parser = build_parser()
    args = parser.parse_args(argv)
    validate_args(args)
    size = determine_source_size(args.source, args.size)
    if size < 0:
        raise SystemExit("--size must be zero or greater")

    copier = RescueCopier(
        source_path=args.source,
        dest_path=args.destination,
        map_path=args.mapfile,
        size=size,
        block_size=args.block_size,
        sector_size=args.sector_size,
        retries=args.retries,
        force=args.force,
        no_trim=args.no_trim,
        no_scrape=args.no_scrape,
        reopen_on_error=args.reopen_on_error,
        bad_ranges=args.simulate_bad_ranges,
    )

    interrupted = False

    def handle_interrupt(signum, frame):
        nonlocal interrupted
        interrupted = True
        raise KeyboardInterrupt

    signal.signal(signal.SIGINT, handle_interrupt)
    try:
        copier.run()
    except KeyboardInterrupt:
        copier.save("Interrupted", force=True)
        copier.finish_line()
        print(f"Interrupted. Progress saved to {args.mapfile}")
        return 130
    stats = copier.map.stats()
    if stats["-"] > 0 or stats["/"] > 0 or stats["*"] > 0 or stats["?"] > 0:
        return 2
    return 0 if not interrupted else 130


if __name__ == "__main__":
    raise SystemExit(main())