strukto-ai--mirage/python/mirage/commands/builtin/generic/cmp.py

from collections.abc import Awaitable, Callable

from mirage.accessor.base import Accessor
from mirage.commands.builtin.utils.output import format_records
from mirage.commands.errors import UsageError
from mirage.commands.spec.types import CommandName
from mirage.commands.spec.usage import extra_operand_error
from mirage.io.types import ByteSource, IOResult
from mirage.types import PathSpec


async def cmp_cmd(
    paths: list[PathSpec],
    *,
    read_bytes: Callable[..., Awaitable[bytes]],
    accessor: Accessor | None = None,
    silent: bool = False,
    verbose: bool = False,
    limit: int | None = None,
    print_bytes: bool = False,
    skip: int | None = None,
) -> tuple[ByteSource | None, IOResult]:
    if len(paths) > 2:
        raise extra_operand_error(CommandName.CMP, paths[2].raw_path)
    if len(paths) < 2:
        raise UsageError("cmp: requires two paths")
    p0, p1 = paths[0], paths[1]
    data1 = await read_bytes(accessor, p0)
    data2 = await read_bytes(accessor, p1)
    if skip is not None:
        data1 = data1[skip:]
        data2 = data2[skip:]
    if limit is not None:
        data1 = data1[:limit]
        data2 = data2[:limit]
    if data1 == data2:
        return None, IOResult()
    if silent:
        return None, IOResult(exit_code=1)
    if verbose:
        out_lines: list[str] = []
        for idx in range(min(len(data1), len(data2))):
            if data1[idx] != data2[idx]:
                out_lines.append(f"{idx + 1} {data1[idx]:o} {data2[idx]:o}")
        return format_records(out_lines), IOResult(exit_code=1)
    for idx in range(min(len(data1), len(data2))):
        if data1[idx] != data2[idx]:
            line = 1 + data1[:idx].count(ord(b"\n"))
            msg = (f"{p0.virtual} {p1.virtual}"
                   f" differ: char {idx + 1}, line {line}")
            if print_bytes:
                msg += (f" is {data1[idx]:o} {chr(data1[idx])}"
                        f" {data2[idx]:o} {chr(data2[idx])}")
            return format_records([msg]), IOResult(exit_code=1)
    shorter = p0.virtual if len(data1) < len(data2) else p1.virtual
    msg = f"cmp: EOF on {shorter}"
    return format_records([msg]), IOResult(exit_code=1)


__all__ = ["cmp_cmd"]