Sorting and Metrics¶

def resolve_flags(events: Sequence[tuple[str, str]]) -> DisplayOptions:
    """Resolve an ordered sequence of flag events into
    [`DisplayOptions`][recursivist.flags.DisplayOptions].

    Each event is a ``(mode, metric)`` pair drawn from the registry, in the
    left-to-right order the flags appeared on the command line. The resolution rules are
    described in the module docstring.

    Args:
        events: The flag events, ordered by command-line position.

    Returns:
        The resolved [`DisplayOptions`][recursivist.flags.DisplayOptions].
    """
    sort_key: str | None = None
    sort_locked = False
    combined_winner: str | None = None
    display_only_metrics: list[str] = []
    display_only_git = False

    for mode, metric in events:
        if mode in (MODE_SORT_ONLY, MODE_COMBINED):
            if not sort_locked:
                sort_locked = True
                sort_key = metric
                if mode == MODE_COMBINED:
                    combined_winner = metric
        elif metric == METRIC_GIT:
            display_only_git = True
        elif metric not in display_only_metrics:
            display_only_metrics.append(metric)

    metrics: list[str] = []
    if combined_winner in NUMERIC_METRICS:
        metrics.append(combined_winner)
    for metric in display_only_metrics:
        if metric not in metrics:
            metrics.append(metric)

    show_git = display_only_git or combined_winner == METRIC_GIT
    return DisplayOptions(
        sort_key=sort_key,
        metrics=tuple(metrics),
        show_git_status=show_git,
    )

def resolve_display_options(
    *,
    sort_loc: bool = False,
    sort_size: bool = False,
    sort_mtime: bool = False,
    sort_similarity: bool = False,
    sort_git: bool = False,
    disp_loc: bool = False,
    disp_size: bool = False,
    disp_mtime: bool = False,
    disp_git: bool = False,
    order: Sequence[str] = (),
) -> DisplayOptions:
    """Resolve the raw per-flag booleans into
    [`DisplayOptions`][recursivist.flags.DisplayOptions].

    The boolean arguments say *which* flags are active and *order* says in what order
    they were given; the CLI parser supplies both. *order* only arranges the active set
    — a flag listed in it but not active is ignored, and an active flag missing from it
    is placed after the listed ones, in registry order, so resolution is deterministic.

    Args:
        sort_loc: Whether ``--sort-by-loc`` was given.
        sort_size: Whether ``--sort-by-size`` was given.
        sort_mtime: Whether ``--sort-by-mtime`` was given.
        sort_similarity: Whether ``--sort-by-similarity`` was given.
        sort_git: Whether ``--sort-by-git-status`` was given.
        disp_loc: Whether ``--loc`` was given.
        disp_size: Whether ``--size`` was given.
        disp_mtime: Whether ``--mtime`` was given.
        disp_git: Whether ``--git-status`` was given.
        order: The ids of the flags (see `FLAG_SPECS`) in the order the parser
            encountered them on the command line. A repeated id counts at its first
            position. When omitted, every active flag falls back to its registry
            position.

    Returns:
        The resolved [`DisplayOptions`][recursivist.flags.DisplayOptions].

    Raises:
        ValueError: If *order* contains an id that is not in the registry.
    """
    active = {
        "sort_similarity": sort_similarity,
        "sort_loc": sort_loc,
        "sort_size": sort_size,
        "sort_mtime": sort_mtime,
        "sort_git": sort_git,
        "disp_loc": disp_loc,
        "disp_size": disp_size,
        "disp_mtime": disp_mtime,
        "disp_git": disp_git,
    }
    position: dict[str, int] = {}
    for index, flag_id in enumerate(order):
        if flag_id not in active:
            raise ValueError(f"Unknown flag id in order: {flag_id!r}")
        position.setdefault(flag_id, index)

    active_specs = [spec for spec in FLAG_SPECS if active[spec.id]]
    ordered = sorted(active_specs, key=lambda spec: position.get(spec.id, len(order)))
    events = [(spec.mode, spec.metric) for spec in ordered]
    return resolve_flags(events)

def sort_files_by_similarity(files: Sequence[FileEntry]) -> list[FileEntry]:
    """Order files so that similarly named files sit next to each other.

    Unlike the LOC/size/mtime metrics, name similarity is *relational*: it depends on
    how a file's name compares to the others rather than on any single measured value,
    so it cannot be expressed as a ``sorted`` key. Instead this builds a greedy
    nearest-neighbour chain:

    1. Entries are seeded in case-insensitive name order and the alphabetically-first
       name becomes the start of the chain. Using a fixed, name-derived anchor makes the
       result deterministic and stable across runs (the same directory always yields the
       same order).
    2. Repeatedly, the not-yet-placed entry whose name is most similar to the most
       recently placed name is appended. Similarity is the `difflib.SequenceMatcher`
       ratio computed case-insensitively on the full filename (extension included), so
       ``main.py``/``main.js`` and ``test_api.py``/``test_api.js`` naturally cluster.
    3. Ratio ties are broken by case-insensitive name order: because the candidates are
       kept alphabetically sorted and the best match is only replaced on a strictly
       greater ratio, the alphabetically-first of any tied group wins.

    This is a heuristic (locally greedy) ordering rather than a globally optimal
    grouping, which is the appropriate trade-off for a directory listing: each directory
    holds relatively few files, so the ``O(n^2)`` pairwise comparisons are cheap, and
    the result reliably places obvious name-siblings adjacent to one another.

    Only the name is used, so the metric fields do not affect the result.

    Args:
        files: The [`FileEntry`][recursivist._models.FileEntry] items to order.

    Returns:
        Reordered list of [`FileEntry`][recursivist._models.FileEntry] with name-similar
        files adjacent.
    """
    if not files:
        return []
    entries = list(files)
    if len(entries) < 2:
        return entries
    remaining = sorted(entries, key=_name_key)
    ordered: list[FileEntry] = [remaining.pop(0)]
    matcher = SequenceMatcher(autojunk=False)
    while remaining:
        matcher.set_seq2(ordered[-1].name.lower())
        best_idx = 0
        best_ratio = -1.0
        for idx, candidate in enumerate(remaining):
            matcher.set_seq1(candidate.name.lower())
            ratio = matcher.ratio()
            if ratio > best_ratio:
                best_ratio = ratio
                best_idx = idx
        ordered.append(remaining.pop(best_idx))
    return ordered

def sort_files_by_type(
    files: Sequence[FileEntry],
    sort_key: str | None = None,
    git_markers: Mapping[str, str] | None = None,
) -> list[FileEntry]:
    """Order files by a single resolved sort key.

    Exactly one ordering is applied, chosen by *sort_key*:

    - ``None``: the default — by extension, then case-insensitive name.
    - ``"loc"`` / ``"size"`` / ``"mtime"``: by that metric, largest/newest first, and
      by case-insensitive name among files with equal values.
    - ``"git_status"``: grouped by Git status (modified, added, deleted, untracked, then
      clean), and by case-insensitive name within each group. Requires *git_markers*.
    - ``"similarity"``: by name similarity, via
      [`sort_files_by_similarity`][recursivist.sorting.sort_files_by_similarity].

    Every ordering ends in a name tie-breaker, so the result is independent of the
    order of *files* (which the scanner takes from ``os.listdir``).

    This mirrors the resolution in [`recursivist.flags`][recursivist.flags], where only
    the first sorting flag on the command line takes effect, so there is never more than
    one active metric to combine.

    Args:
        files: The [`FileEntry`][recursivist._models.FileEntry] items to order.
        sort_key: The single metric to order by, or ``None`` for the default.
        git_markers: ``{filename: status_char}`` mapping, required when *sort_key* is
            ``"git_status"``.

    Returns:
        Sorted list of [`FileEntry`][recursivist._models.FileEntry].
    """
    if not files:
        return []

    entries = list(files)

    if sort_key == METRIC_LOC:
        return sorted(entries, key=lambda e: (-e.loc, *_name_key(e)))
    if sort_key == METRIC_SIZE:
        return sorted(entries, key=lambda e: (-e.size, *_name_key(e)))
    if sort_key == METRIC_MTIME:
        return sorted(entries, key=lambda e: (-e.mtime, *_name_key(e)))
    if sort_key == METRIC_GIT:
        markers = git_markers or {}
        return sorted(
            entries,
            key=lambda e: (
                _GIT_SORT_RANK.get(markers.get(e.name, ""), _GIT_SORT_CLEAN),
                *_name_key(e),
            ),
        )
    if sort_key == METRIC_SIMILARITY:
        return sort_files_by_similarity(entries)
    return sorted(
        entries,
        key=lambda e: (os.path.splitext(e.name)[1].lower(), *_name_key(e)),
    )

def count_lines_of_code(file_path: str) -> int:
    """Count the number of lines in a text file.

    The encoding is inferred from the first 4 KiB: UTF-16 files are recognized by their
    byte-order mark or by a regular pattern of null bytes, while files containing null
    bytes that are not UTF-16 are treated as binary and skipped. Everything else is read
    as UTF-8. Undecodable bytes are replaced rather than rejected, which never changes
    the line count, so the file is opened once and read in a single pass.

    Only regular files are read. Named pipes, sockets and devices are opened without
    blocking and then skipped, because reading one can wait forever or never end.

    Args:
        file_path: Path to the file.

    Returns:
        The number of lines, or ``0`` if the file is empty, binary, not a regular file,
        or cannot be read.
    """
    try:
        with open(file_path, "rb", opener=_open_nonblocking) as binary_file:
            if not stat.S_ISREG(os.fstat(binary_file.fileno()).st_mode):
                logger.debug("Not a regular file, skipping: %s", file_path)
                return 0
            sample = binary_file.read(4096)
            encoding = _detect_text_encoding(sample)
            if encoding is None:
                return 0
            binary_file.seek(0)
            with io.TextIOWrapper(
                binary_file, encoding=encoding, errors="replace"
            ) as text_file:
                return sum(1 for _ in text_file)
    except (OSError, ValueError) as e:
        logger.debug("Could not read file: %s: %s", file_path, e)
        return 0

def get_file_size(file_path: str) -> int:
    """Return the size of a file in bytes.

    Args:
        file_path: Path to the file whose size should be retrieved.

    Returns:
        Size of the file in bytes, or ``0`` when the file cannot be accessed (e.g.,
        permission error or the path no longer exists).
    """
    try:
        return os.path.getsize(file_path)
    except Exception as e:
        logger.debug("Could not get size for %s: %s", file_path, e)
        return 0

def format_size(size_in_bytes: int) -> str:
    """Format a byte count as a human-readable size string.

    Scales the value to bytes, KB, MB, or GB and formats it with one decimal place for
    every unit above bytes. The unit is chosen after rounding, so a value that rounds up
    to 1024 moves to the next unit (``1048575`` is ``"1.0 MB"``, not ``"1024.0 KB"``).
    GB is the largest unit, so it is never promoted.

    Args:
        size_in_bytes: Size in bytes.

    Returns:
        A human-readable size string (e.g. ``"512 B"`` or ``"4.2 MB"``).
    """
    if size_in_bytes < 1024:
        return f"{size_in_bytes} B"
    value = size_in_bytes / 1024
    for unit in ("KB", "MB"):
        text = f"{value:.1f}"
        if float(text) < 1024:
            return f"{text} {unit}"
        value /= 1024
    return f"{value:.1f} GB"

def get_file_mtime(file_path: str) -> float:
    """Return a file's modification time in seconds since the epoch.

    Args:
        file_path: Path to the file.

    Returns:
        The modification time as a float, or ``0.0`` if the file cannot be accessed.
    """
    try:
        return os.path.getmtime(file_path)
    except Exception as e:
        logger.debug("Could not get modification time for %s: %s", file_path, e)
        return 0.0

def format_timestamp(timestamp: float) -> str:
    """Format a Unix timestamp as a human-readable, recency-aware string.

    The representation becomes coarser as the timestamp gets older:

    - Today: ``"Today HH:MM"``
    - Yesterday: ``"Yesterday HH:MM"``
    - Within the last week: abbreviated weekday and time (e.g. ``"Mon 14:30"``)
    - Earlier this year: abbreviated month and day (e.g. ``"Mar 15"``)
    - Older: ``"YYYY-MM-DD"``

    Args:
        timestamp: Seconds since the epoch.

    Returns:
        The formatted date/time string, or ``"-"`` when *timestamp* is zero or falls
        outside the representable range.
    """
    if not timestamp:
        return "-"
    try:
        dt_object = datetime.fromtimestamp(timestamp)
    except (OSError, OverflowError, ValueError):
        return "-"
    current_dt = datetime.now()
    current_date = current_dt.date()
    if dt_object.date() == current_date:
        return f"Today {dt_object.strftime('%H:%M')}"
    if dt_object.date() == current_date - timedelta(days=1):
        return f"Yesterday {dt_object.strftime('%H:%M')}"
    if dt_object.date() > current_date:
        return dt_object.strftime("%Y-%m-%d")
    if current_date - dt_object.date() < timedelta(days=7):
        return dt_object.strftime("%a %H:%M")
    if dt_object.year == current_dt.year:
        return dt_object.strftime("%b %d")
    return dt_object.strftime("%Y-%m-%d")

def format_metrics(
    loc: int = 0,
    size: int = 0,
    mtime: float = 0.0,
    metrics: Sequence[str] = (),
) -> str:
    """Build the parenthetical metrics annotation for a file or directory.

    Includes exactly the metrics named in *metrics*, in that order — e.g.
    ``metrics=("size", "loc")`` yields ``"(4.2 KB, 120 lines)"``. The metric names are
    those defined in [`recursivist.flags`][recursivist.flags]: ``"loc"``, ``"size"`` and
    ``"mtime"``.

    Args:
        loc: Lines-of-code count.
        size: Size in bytes.
        mtime: Modification time (seconds since epoch).
        metrics: The metrics to include, in display order.

    Returns:
        The annotation string including the surrounding parentheses, or an empty string
        when *metrics* is empty.
    """
    renderers = {
        "loc": lambda: f"{loc} line" if loc == 1 else f"{loc} lines",
        "size": lambda: format_size(size),
        "mtime": lambda: format_timestamp(mtime),
    }
    parts = [renderers[m]() for m in metrics if m in renderers]
    return f"({', '.join(parts)})" if parts else ""

def format_metrics_suffix(
    loc: int = 0,
    size: int = 0,
    mtime: float = 0.0,
    metrics: Sequence[str] = (),
) -> str:
    """Like [`format_metrics`][recursivist.metrics.format_metrics] but prefixed with a
    single space.

    Convenient for appending directly after a file or directory name. Returns an empty
    string (no leading space) when *metrics* is empty.
    """
    annotation = format_metrics(loc, size, mtime, metrics)
    return f" {annotation}" if annotation else ""

def recorded_dir_metrics(
    directory: Directory, metrics: Sequence[str] = ()
) -> list[str]:
    """Return the entries of *metrics* that *directory* holds a total for.

    A directory carries a total only for the metrics its scan collected, and none at all
    when it was not traversed. The order of *metrics* is preserved.

    Args:
        directory: The directory whose totals are consulted.
        metrics: The metrics to display, in order.

    Returns:
        The displayable subset of *metrics*, in the same order.
    """
    totals: dict[str, int | float | None] = {
        "loc": directory.loc,
        "size": directory.size,
        "mtime": directory.mtime,
    }
    return [m for m in metrics if totals.get(m) is not None]

def format_dir_metrics(directory: Directory, metrics: Sequence[str] = ()) -> str:
    """Return the space-prefixed metrics suffix for a directory.

    Wraps [`format_metrics_suffix`][recursivist.metrics.format_metrics_suffix], reading
    the totals from *directory* and keeping only the requested metrics that it actually
    holds a total for (see
    [`recorded_dir_metrics`][recursivist.metrics.recorded_dir_metrics]) — while
    preserving the requested display order.

    Args:
        directory: The directory whose totals are formatted.
        metrics: The metrics to display, in order.

    Returns:
        The metrics suffix (with a leading space) or an empty string.
    """
    return format_metrics_suffix(
        directory.loc or 0,
        directory.size or 0,
        directory.mtime or 0.0,
        recorded_dir_metrics(directory, metrics),
    )