| 1 | from __future__ import annotations |
| 2 | |
| 3 | from collections import deque |
| 4 | from dataclasses import dataclass |
| 5 | from datetime import datetime |
| 6 | import os |
| 7 | from typing import Any, Callable, Iterable, Literal, Optional, Sequence |
| 8 | |
| 9 | from pathspec import PathSpec |
| 10 | |
| 11 | from helpers import files as files_helper |
| 12 | from helpers.localization import Localization |
| 13 | |
| 14 | SORT_BY_NAME = "name" |
| 15 | SORT_BY_CREATED = "created" |
| 16 | SORT_BY_MODIFIED = "modified" |
| 17 | |
| 18 | SORT_ASC = "asc" |
| 19 | SORT_DESC = "desc" |
| 20 | |
| 21 | OUTPUT_MODE_STRING = "string" |
| 22 | OUTPUT_MODE_FLAT = "flat" |
| 23 | OUTPUT_MODE_NESTED = "nested" |
| 24 | |
| 25 | |
| 26 | def _from_timestamp(timestamp: float) -> datetime: |
| 27 | return datetime.fromtimestamp(timestamp, tz=Localization.get().get_tzinfo()) |
| 28 | |
| 29 | |
| 30 | def file_tree( |
| 31 | relative_path: str, |
| 32 | *, |
| 33 | max_depth: int = 0, |
| 34 | max_lines: int = 0, |
| 35 | folders_first: bool = True, |
| 36 | max_folders: int = 0, |
| 37 | max_files: int = 0, |
| 38 | sort: tuple[Literal["name", "created", "modified"], Literal["asc", "desc"]] = ("modified", "desc"), |
| 39 | ignore: str | None = None, |
| 40 | output_mode: Literal["string", "flat", "nested"] = OUTPUT_MODE_STRING, |
| 41 | ) -> str | list[dict]: |
| 42 | """Render a directory tree relative to the repository base path. |
| 43 | |
| 44 | Parameters: |
| 45 | relative_path: Base directory (relative to project root) to scan with :func:`get_abs_path`. |
| 46 | max_depth: Maximum depth of traversal (0 = unlimited). Depth starts at 1 for root entries. |
| 47 | max_lines: Global limit for rendered lines (0 = unlimited). When exceeded, the current depth |
| 48 | finishes rendering before deeper levels are skipped. |
| 49 | folders_first: When True, folders render before files within each directory. |
| 50 | max_folders: Optional per-directory cap (0 = unlimited) on rendered folder entries before adding a |
| 51 | ``# N more folders`` comment. When only a single folder exceeds the limit and ``max_folders`` is greater than zero, that folder is rendered |
| 52 | directly instead of emitting a summary comment. |
| 53 | max_files: Optional per-directory cap (0 = unlimited) on rendered file entries before adding a ``# N more files`` comment. |
| 54 | As with folders, a single excess file is rendered when ``max_files`` is greater than zero. |
| 55 | sort: Tuple of ``(key, direction)`` where key is one of :data:`SORT_BY_NAME`, |
| 56 | :data:`SORT_BY_CREATED`, or :data:`SORT_BY_MODIFIED`; direction is :data:`SORT_ASC` |
| 57 | or :data:`SORT_DESC`. |
| 58 | ignore: Inline ``.gitignore`` content or ``file:`` reference. Examples:: |
| 59 | |
| 60 | ignore=\"\"\"\\n*.pyc\\n__pycache__/\\n!important.py\\n\"\"\" |
| 61 | ignore=\"file:.gitignore\" # relative to scan root |
| 62 | ignore=\"file://.gitignore\" # URI-style relative path |
| 63 | ignore=\"file:/abs/path/.gitignore\" |
| 64 | ignore=\"file:///abs/path/.gitignore\" |
| 65 | |
| 66 | output_mode: One of :data:`OUTPUT_MODE_STRING`, :data:`OUTPUT_MODE_FLAT`, or |
| 67 | :data:`OUTPUT_MODE_NESTED`. |
| 68 | |
| 69 | Returns: |
| 70 | ``OUTPUT_MODE_STRING`` → ``str``: multi-line ASCII tree. The first line is the root banner and |
| 71 | uses a dockerized absolute path for display. |
| 72 | ``OUTPUT_MODE_FLAT`` → ``list[dict]``: flattened sequence of TreeItem dictionaries, with a |
| 73 | synthetic root folder item prepended at index 0 (using a dockerized absolute path for display). |
| 74 | ``OUTPUT_MODE_NESTED`` → ``list[dict]``: a single synthetic root folder item (using a dockerized |
| 75 | absolute path for display) whose ``items`` contains the nested TreeItem dictionaries. |
| 76 | |
| 77 | Notes: |
| 78 | * The utility is synchronous; avoid calling from latency-sensitive async loops. |
| 79 | * The ASCII renderer walks the established tree depth-first so connectors reflect parent/child structure, |
| 80 | while traversal and limit calculations remain breadth-first by depth. When ``max_lines`` is set, the number |
| 81 | of non-comment entries (excluding the root banner) never exceeds that limit; informational summary comments |
| 82 | are emitted in addition when necessary. |
| 83 | * ``created`` and ``modified`` values in structured outputs are timezone-aware user-local |
| 84 | :class:`datetime.datetime` objects:: |
| 85 | |
| 86 | item = flat_items[0] |
| 87 | iso = item[\"created\"].isoformat() |
| 88 | epoch = item[\"created\"].timestamp() |
| 89 | |
| 90 | """ |
| 91 | abs_root = files_helper.get_abs_path(relative_path) |
| 92 | output_root = files_helper.get_abs_path_dockerized(relative_path) |
| 93 | |
| 94 | if not os.path.exists(abs_root): |
| 95 | raise FileNotFoundError(f"Path does not exist: {relative_path!r}") |
| 96 | if not os.path.isdir(abs_root): |
| 97 | raise NotADirectoryError(f"Expected a directory, received: {relative_path!r}") |
| 98 | |
| 99 | sort_key, sort_direction = sort |
| 100 | if sort_key not in {SORT_BY_NAME, SORT_BY_CREATED, SORT_BY_MODIFIED}: |
| 101 | raise ValueError(f"Unsupported sort key: {sort_key!r}") |
| 102 | if sort_direction not in {SORT_ASC, SORT_DESC}: |
| 103 | raise ValueError(f"Unsupported sort direction: {sort_direction!r}") |
| 104 | if output_mode not in {OUTPUT_MODE_STRING, OUTPUT_MODE_FLAT, OUTPUT_MODE_NESTED}: |
| 105 | raise ValueError(f"Unsupported output mode: {output_mode!r}") |
| 106 | if max_depth < 0: |
| 107 | raise ValueError("max_depth must be >= 0") |
| 108 | if max_lines < 0: |
| 109 | raise ValueError("max_lines must be >= 0") |
| 110 | |
| 111 | ignore_spec = _resolve_ignore_patterns(ignore, abs_root) |
| 112 | |
| 113 | root_stat = os.stat(abs_root, follow_symlinks=False) |
| 114 | root_name = os.path.basename(os.path.normpath(abs_root)) or os.path.basename(abs_root) |
| 115 | root_node = _TreeEntry( |
| 116 | name=root_name, |
| 117 | level=0, |
| 118 | item_type="folder", |
| 119 | created=_from_timestamp(root_stat.st_ctime), |
| 120 | modified=_from_timestamp(root_stat.st_mtime), |
| 121 | parent=None, |
| 122 | items=[], |
| 123 | rel_path="", |
| 124 | ) |
| 125 | |
| 126 | queue: deque[tuple[_TreeEntry, str, int]] = deque([(root_node, abs_root, 1)]) |
| 127 | nodes_in_order: list[_TreeEntry] = [] |
| 128 | rendered_count = 0 |
| 129 | limit_reached = False |
| 130 | visibility_cache: dict[str, bool] = {} |
| 131 | |
| 132 | def make_entry(entry: os.DirEntry, parent: _TreeEntry, level: int, item_type: Literal["file", "folder"]) -> _TreeEntry: |
| 133 | stat = entry.stat(follow_symlinks=False) |
| 134 | rel_path = os.path.relpath(entry.path, abs_root) |
| 135 | rel_posix = _normalize_relative_path(rel_path) |
| 136 | return _TreeEntry( |
| 137 | name=entry.name, |
| 138 | level=level, |
| 139 | item_type=item_type, |
| 140 | created=_from_timestamp(stat.st_ctime), |
| 141 | modified=_from_timestamp(stat.st_mtime), |
| 142 | parent=parent, |
| 143 | items=[] if item_type == "folder" else None, |
| 144 | rel_path=rel_posix, |
| 145 | ) |
| 146 | |
| 147 | while queue and not limit_reached: |
| 148 | parent_node, current_dir, level = queue.popleft() |
| 149 | |
| 150 | if max_depth and level > max_depth: |
| 151 | continue |
| 152 | |
| 153 | remaining_depth = max_depth - level if max_depth else -1 |
| 154 | folders, files = _list_directory_children( |
| 155 | current_dir, |
| 156 | abs_root, |
| 157 | ignore_spec, |
| 158 | max_depth_remaining=remaining_depth, |
| 159 | cache=visibility_cache, |
| 160 | ) |
| 161 | |
| 162 | folder_entries = [make_entry(folder, parent_node, level, "folder") for folder in folders] |
| 163 | file_entries = [make_entry(file_entry, parent_node, level, "file") for file_entry in files] |
| 164 | |
| 165 | children = _apply_sorting_and_limits( |
| 166 | folder_entries, |
| 167 | file_entries, |
| 168 | folders_first=folders_first, |
| 169 | sort=sort, |
| 170 | max_folders=max_folders, |
| 171 | max_files=max_files, |
| 172 | directory_node=parent_node, |
| 173 | ) |
| 174 | |
| 175 | trimmed_children: list[_TreeEntry] = [] |
| 176 | hidden_children_local: list[_TreeEntry] = [] |
| 177 | if max_lines and rendered_count >= max_lines: |
| 178 | limit_reached = True |
| 179 | hidden_children_local = children |
| 180 | else: |
| 181 | for index, child in enumerate(children): |
| 182 | if max_lines and rendered_count >= max_lines: |
| 183 | limit_reached = True |
| 184 | hidden_children_local = children[index:] |
| 185 | break |
| 186 | trimmed_children.append(child) |
| 187 | nodes_in_order.append(child) |
| 188 | is_global_summary = ( |
| 189 | child.item_type == "comment" |
| 190 | and child.rel_path.endswith("#summary:limit") |
| 191 | ) |
| 192 | if not is_global_summary: |
| 193 | rendered_count += 1 |
| 194 | if limit_reached and hidden_children_local: |
| 195 | summary = _create_global_limit_comment( |
| 196 | parent_node, |
| 197 | hidden_children_local, |
| 198 | ) |
| 199 | trimmed_children.append(summary) |
| 200 | nodes_in_order.append(summary) |
| 201 | |
| 202 | parent_node.items = trimmed_children or None |
| 203 | |
| 204 | if limit_reached: |
| 205 | break |
| 206 | |
| 207 | for child in trimmed_children: |
| 208 | if child.item_type != "folder": |
| 209 | continue |
| 210 | if max_depth and level >= max_depth: |
| 211 | continue |
| 212 | child_abs = os.path.join(current_dir, child.name) |
| 213 | queue.append((child, child_abs, level + 1)) |
| 214 | |
| 215 | remaining_queue = list(queue) if limit_reached else [] |
| 216 | queue.clear() |
| 217 | |
| 218 | if limit_reached and remaining_queue: |
| 219 | for folder_node, folder_path, _ in remaining_queue: |
| 220 | summary = _create_folder_unprocessed_comment( |
| 221 | folder_node, |
| 222 | folder_path, |
| 223 | abs_root, |
| 224 | ignore_spec, |
| 225 | ) |
| 226 | if summary is None: |
| 227 | continue |
| 228 | folder_node.items = (folder_node.items or []) + [summary] |
| 229 | nodes_in_order.append(summary) |
| 230 | |
| 231 | visible_nodes = nodes_in_order |
| 232 | |
| 233 | visible_ids = {id(node) for node in visible_nodes} |
| 234 | if visible_ids: |
| 235 | _prune_to_visible(root_node, visible_ids) |
| 236 | |
| 237 | _mark_last_flags(root_node) |
| 238 | _refresh_render_metadata(root_node) |
| 239 | |
| 240 | def iter_visible() -> Iterable[_TreeEntry]: |
| 241 | for node in _iter_depth_first(root_node.items or []): |
| 242 | if not visible_ids or id(node) in visible_ids: |
| 243 | yield node |
| 244 | |
| 245 | def make_root_item(items: list[dict] | None) -> dict: |
| 246 | root_item = root_node.as_dict() |
| 247 | root_item["name"] = output_root |
| 248 | root_item["text"] = f"{output_root.rstrip(os.sep)}/" |
| 249 | root_item["items"] = items |
| 250 | return root_item |
| 251 | |
| 252 | if output_mode == OUTPUT_MODE_STRING: |
| 253 | display_name = output_root #relative_path.strip() or root_name |
| 254 | root_line = f"{display_name.rstrip(os.sep)}/" |
| 255 | lines = [root_line] |
| 256 | for node in iter_visible(): |
| 257 | lines.append(node.text) |
| 258 | return "\n".join(lines) |
| 259 | |
| 260 | if output_mode == OUTPUT_MODE_FLAT: |
| 261 | return [make_root_item(None)] + _build_tree_items_flat(list(iter_visible())) |
| 262 | |
| 263 | return [make_root_item(_to_nested_structure(root_node.items or []))] |
| 264 | |
| 265 | |
| 266 | @dataclass(slots=True) |
| 267 | class _TreeEntry: |
| 268 | name: str |
| 269 | level: int |
| 270 | item_type: Literal["file", "folder", "comment"] |
| 271 | created: datetime |
| 272 | modified: datetime |
| 273 | parent: Optional["_TreeEntry"] = None |
| 274 | items: Optional[list["_TreeEntry"]] = None |
| 275 | is_last: bool = False |
| 276 | rel_path: str = "" |
| 277 | text: str = "" |
| 278 | |
| 279 | def as_dict(self) -> dict[str, Any]: |
| 280 | return { |
| 281 | "name": self.name, |
| 282 | "level": self.level, |
| 283 | "type": self.item_type, |
| 284 | "created": self.created, |
| 285 | "modified": self.modified, |
| 286 | "text": self.text, |
| 287 | "items": [child.as_dict() for child in self.items] if self.items is not None else None, |
| 288 | } |
| 289 | |
| 290 | |
| 291 | def _normalize_relative_path(path: str) -> str: |
| 292 | normalized = path.replace(os.sep, "/") |
| 293 | if normalized in {".", ""}: |
| 294 | return "" |
| 295 | while normalized.startswith("./"): |
| 296 | normalized = normalized[2:] |
| 297 | return normalized |
| 298 | |
| 299 | |
| 300 | def _directory_has_visible_entries( |
| 301 | directory: str, |
| 302 | root_abs_path: str, |
| 303 | ignore_spec: PathSpec, |
| 304 | cache: dict[str, bool], |
| 305 | max_depth_remaining: int, |
| 306 | ) -> bool: |
| 307 | if max_depth_remaining == 0: |
| 308 | return False |
| 309 | |
| 310 | cached = cache.get(directory) |
| 311 | if cached is not None: |
| 312 | return cached |
| 313 | |
| 314 | try: |
| 315 | with os.scandir(directory) as iterator: |
| 316 | for entry in iterator: |
| 317 | rel_path = os.path.relpath(entry.path, root_abs_path) |
| 318 | rel_posix = _normalize_relative_path(rel_path) |
| 319 | is_dir = entry.is_dir(follow_symlinks=False) |
| 320 | |
| 321 | if is_dir: |
| 322 | ignored = ignore_spec.match_file(rel_posix) or ignore_spec.match_file(f"{rel_posix}/") |
| 323 | if ignored: |
| 324 | next_depth = max_depth_remaining - 1 if max_depth_remaining > 0 else -1 |
| 325 | if next_depth == 0: |
| 326 | continue |
| 327 | if _directory_has_visible_entries( |
| 328 | entry.path, |
| 329 | root_abs_path, |
| 330 | ignore_spec, |
| 331 | cache, |
| 332 | next_depth, |
| 333 | ): |
| 334 | cache[directory] = True |
| 335 | return True |
| 336 | continue |
| 337 | else: |
| 338 | if ignore_spec.match_file(rel_posix): |
| 339 | continue |
| 340 | |
| 341 | cache[directory] = True |
| 342 | return True |
| 343 | except FileNotFoundError: |
| 344 | cache[directory] = False |
| 345 | return False |
| 346 | |
| 347 | cache[directory] = False |
| 348 | return False |
| 349 | |
| 350 | |
| 351 | def _create_summary_comment(parent: _TreeEntry, noun: str, count: int) -> _TreeEntry: |
| 352 | label = noun |
| 353 | if count == 1 and noun.endswith("s"): |
| 354 | label = noun[:-1] |
| 355 | elif count > 1 and not noun.endswith("s"): |
| 356 | label = f"{noun}s" |
| 357 | return _TreeEntry( |
| 358 | name=f"{count} more {label}", |
| 359 | level=parent.level + 1, |
| 360 | item_type="comment", |
| 361 | created=parent.created, |
| 362 | modified=parent.modified, |
| 363 | parent=parent, |
| 364 | items=None, |
| 365 | rel_path=f"{parent.rel_path}#summary:{noun}:{count}", |
| 366 | ) |
| 367 | |
| 368 | |
| 369 | def _create_global_limit_comment(parent: _TreeEntry, hidden_children: Sequence[_TreeEntry]) -> _TreeEntry: |
| 370 | folders = sum(1 for child in hidden_children if child.item_type == "folder") |
| 371 | files = sum(1 for child in hidden_children if child.item_type == "file") |
| 372 | parts: list[str] = [] |
| 373 | if folders: |
| 374 | label = "folder" if folders == 1 else "folders" |
| 375 | parts.append(f"{folders} {label}") |
| 376 | if files: |
| 377 | label = "file" if files == 1 else "files" |
| 378 | parts.append(f"{files} {label}") |
| 379 | if not parts: |
| 380 | remaining = len(hidden_children) |
| 381 | label = "item" if remaining == 1 else "items" |
| 382 | parts.append(f"{remaining} {label}") |
| 383 | label_text = ", ".join(parts) |
| 384 | return _TreeEntry( |
| 385 | name=f"limit reached – hidden: {label_text}", |
| 386 | level=parent.level + 1, |
| 387 | item_type="comment", |
| 388 | created=parent.created, |
| 389 | modified=parent.modified, |
| 390 | parent=parent, |
| 391 | items=None, |
| 392 | rel_path=f"{parent.rel_path}#summary:limit", |
| 393 | ) |
| 394 | |
| 395 | |
| 396 | def _create_folder_unprocessed_comment( |
| 397 | folder_node: _TreeEntry, |
| 398 | folder_path: str, |
| 399 | abs_root: str, |
| 400 | ignore_spec: Optional[PathSpec], |
| 401 | ) -> Optional[_TreeEntry]: |
| 402 | try: |
| 403 | folders, files = _list_directory_children( |
| 404 | folder_path, |
| 405 | abs_root, |
| 406 | ignore_spec, |
| 407 | max_depth_remaining=-1, |
| 408 | cache={}, |
| 409 | ) |
| 410 | except FileNotFoundError: |
| 411 | return None |
| 412 | |
| 413 | hidden_entries: list[_TreeEntry] = [] |
| 414 | for entry in folders: |
| 415 | stat = entry.stat(follow_symlinks=False) |
| 416 | hidden_entries.append( |
| 417 | _TreeEntry( |
| 418 | name=entry.name, |
| 419 | level=folder_node.level + 1, |
| 420 | item_type="folder", |
| 421 | created=_from_timestamp(stat.st_ctime), |
| 422 | modified=_from_timestamp(stat.st_mtime), |
| 423 | parent=folder_node, |
| 424 | items=None, |
| 425 | rel_path=os.path.join(folder_node.rel_path, entry.name), |
| 426 | ) |
| 427 | ) |
| 428 | for entry in files: |
| 429 | stat = entry.stat(follow_symlinks=False) |
| 430 | hidden_entries.append( |
| 431 | _TreeEntry( |
| 432 | name=entry.name, |
| 433 | level=folder_node.level + 1, |
| 434 | item_type="file", |
| 435 | created=_from_timestamp(stat.st_ctime), |
| 436 | modified=_from_timestamp(stat.st_mtime), |
| 437 | parent=folder_node, |
| 438 | items=None, |
| 439 | rel_path=os.path.join(folder_node.rel_path, entry.name), |
| 440 | ) |
| 441 | ) |
| 442 | |
| 443 | if not hidden_entries: |
| 444 | return None |
| 445 | |
| 446 | return _create_global_limit_comment(folder_node, hidden_entries) |
| 447 | |
| 448 | |
| 449 | def _prune_to_visible(node: _TreeEntry, visible_ids: set[int]) -> None: |
| 450 | if node.items is None: |
| 451 | return |
| 452 | filtered: list[_TreeEntry] = [] |
| 453 | for child in node.items: |
| 454 | if not visible_ids or id(child) in visible_ids: |
| 455 | _prune_to_visible(child, visible_ids) |
| 456 | filtered.append(child) |
| 457 | node.items = filtered or None |
| 458 | |
| 459 | |
| 460 | def _mark_last_flags(node: _TreeEntry) -> None: |
| 461 | if node.items is None: |
| 462 | return |
| 463 | total = len(node.items) |
| 464 | for index, child in enumerate(node.items): |
| 465 | child.is_last = index == total - 1 |
| 466 | _mark_last_flags(child) |
| 467 | |
| 468 | |
| 469 | def _refresh_render_metadata(node: _TreeEntry) -> None: |
| 470 | if node.items is None: |
| 471 | return |
| 472 | for child in node.items: |
| 473 | child.text = _format_line(child) |
| 474 | _refresh_render_metadata(child) |
| 475 | |
| 476 | |
| 477 | def _resolve_ignore_patterns(ignore: str | None, root_abs_path: str) -> Optional[PathSpec]: |
| 478 | if ignore is None: |
| 479 | return None |
| 480 | |
| 481 | content: str |
| 482 | if ignore.startswith("file:"): |
| 483 | reference = ignore[5:] |
| 484 | if reference.startswith("///"): |
| 485 | reference_path = reference[2:] |
| 486 | elif reference.startswith("//"): |
| 487 | reference_path = os.path.join(root_abs_path, reference[2:]) |
| 488 | elif reference.startswith("/"): |
| 489 | reference_path = reference |
| 490 | else: |
| 491 | reference_path = os.path.join(root_abs_path, reference) |
| 492 | |
| 493 | try: |
| 494 | with open(reference_path, "r", encoding="utf-8") as handle: |
| 495 | content = handle.read() |
| 496 | except FileNotFoundError as exc: |
| 497 | raise FileNotFoundError(f"Ignore file not found: {reference_path}") from exc |
| 498 | else: |
| 499 | content = ignore |
| 500 | |
| 501 | lines = [ |
| 502 | line.strip() |
| 503 | for line in content.splitlines() |
| 504 | if line.strip() and not line.strip().startswith("#") |
| 505 | ] |
| 506 | |
| 507 | if not lines: |
| 508 | return None |
| 509 | |
| 510 | return PathSpec.from_lines("gitignore", lines) |
| 511 | |
| 512 | |
| 513 | def _list_directory_children( |
| 514 | directory: str, |
| 515 | root_abs_path: str, |
| 516 | ignore_spec: Optional[PathSpec], |
| 517 | *, |
| 518 | max_depth_remaining: int, |
| 519 | cache: dict[str, bool], |
| 520 | ) -> tuple[list[os.DirEntry], list[os.DirEntry]]: |
| 521 | folders: list[os.DirEntry] = [] |
| 522 | files: list[os.DirEntry] = [] |
| 523 | |
| 524 | try: |
| 525 | with os.scandir(directory) as iterator: |
| 526 | for entry in iterator: |
| 527 | if entry.name in (".", ".."): |
| 528 | continue |
| 529 | rel_path = os.path.relpath(entry.path, root_abs_path) |
| 530 | rel_posix = _normalize_relative_path(rel_path) |
| 531 | is_directory = entry.is_dir(follow_symlinks=False) |
| 532 | |
| 533 | if ignore_spec: |
| 534 | if is_directory: |
| 535 | ignored = ignore_spec.match_file(rel_posix) or ignore_spec.match_file(f"{rel_posix}/") |
| 536 | if ignored: |
| 537 | if _directory_has_visible_entries( |
| 538 | entry.path, |
| 539 | root_abs_path, |
| 540 | ignore_spec, |
| 541 | cache, |
| 542 | max_depth_remaining - 1, |
| 543 | ): |
| 544 | folders.append(entry) |
| 545 | continue |
| 546 | else: |
| 547 | if ignore_spec.match_file(rel_posix): |
| 548 | continue |
| 549 | |
| 550 | if is_directory: |
| 551 | folders.append(entry) |
| 552 | else: |
| 553 | files.append(entry) |
| 554 | except FileNotFoundError: |
| 555 | return ([], []) |
| 556 | |
| 557 | return (folders, files) |
| 558 | |
| 559 | |
| 560 | def _apply_sorting_and_limits( |
| 561 | folders: list[_TreeEntry], |
| 562 | files: list[_TreeEntry], |
| 563 | *, |
| 564 | folders_first: bool, |
| 565 | sort: tuple[str, str], |
| 566 | max_folders: int | None, |
| 567 | max_files: int | None, |
| 568 | directory_node: _TreeEntry, |
| 569 | ) -> list[_TreeEntry]: |
| 570 | sort_key, sort_dir = sort |
| 571 | reverse = sort_dir == SORT_DESC |
| 572 | |
| 573 | def key_fn(node: _TreeEntry): |
| 574 | if sort_key == SORT_BY_NAME: |
| 575 | return node.name.casefold() |
| 576 | if sort_key == SORT_BY_CREATED: |
| 577 | return node.created |
| 578 | return node.modified |
| 579 | |
| 580 | folders_sorted = sorted(folders, key=key_fn, reverse=reverse) |
| 581 | files_sorted = sorted(files, key=key_fn, reverse=reverse) |
| 582 | combined: list[_TreeEntry] = [] |
| 583 | |
| 584 | def append_group(group: list[_TreeEntry], limit: int | None, noun: str) -> None: |
| 585 | if limit == 0: |
| 586 | limit = None |
| 587 | if not group: |
| 588 | return |
| 589 | if limit is None: |
| 590 | combined.extend(group) |
| 591 | return |
| 592 | |
| 593 | limit = max(limit, 0) |
| 594 | visible = group[:limit] |
| 595 | combined.extend(visible) |
| 596 | |
| 597 | overflow = group[limit:] |
| 598 | if not overflow: |
| 599 | return |
| 600 | |
| 601 | combined.append( |
| 602 | _create_summary_comment( |
| 603 | directory_node, |
| 604 | noun, |
| 605 | len(overflow), |
| 606 | ) |
| 607 | ) |
| 608 | |
| 609 | if folders_first: |
| 610 | append_group(folders_sorted, max_folders, "folder") |
| 611 | append_group(files_sorted, max_files, "file") |
| 612 | else: |
| 613 | append_group(files_sorted, max_files, "file") |
| 614 | append_group(folders_sorted, max_folders, "folder") |
| 615 | |
| 616 | return combined |
| 617 | |
| 618 | |
| 619 | def _format_line(node: _TreeEntry) -> str: |
| 620 | segments: list[str] = [] |
| 621 | ancestor = node.parent |
| 622 | while ancestor and ancestor.parent is not None: |
| 623 | segments.append(" " if ancestor.is_last else "│ ") |
| 624 | ancestor = ancestor.parent |
| 625 | segments.reverse() |
| 626 | |
| 627 | connector = "└── " if node.is_last else "├── " |
| 628 | if node.item_type == "folder": |
| 629 | label = f"{node.name}/" |
| 630 | elif node.item_type == "comment": |
| 631 | label = f"# {node.name}" |
| 632 | else: |
| 633 | label = node.name |
| 634 | |
| 635 | return "".join(segments) + connector + label |
| 636 | |
| 637 | |
| 638 | def _build_tree_items_flat(items: Sequence[_TreeEntry]) -> list[dict]: |
| 639 | return [ |
| 640 | { |
| 641 | "name": node.name, |
| 642 | "level": node.level, |
| 643 | "type": node.item_type, |
| 644 | "created": node.created, |
| 645 | "modified": node.modified, |
| 646 | "text": node.text, |
| 647 | "items": None, |
| 648 | } |
| 649 | for node in items |
| 650 | ] |
| 651 | |
| 652 | |
| 653 | def _to_nested_structure(items: Sequence[_TreeEntry]) -> list[dict]: |
| 654 | def convert(node: _TreeEntry) -> dict: |
| 655 | children = None |
| 656 | if node.items is not None: |
| 657 | children = [convert(child) for child in node.items] |
| 658 | return { |
| 659 | "name": node.name, |
| 660 | "level": node.level, |
| 661 | "type": node.item_type, |
| 662 | "created": node.created, |
| 663 | "modified": node.modified, |
| 664 | "text": node.text, |
| 665 | "items": children, |
| 666 | } |
| 667 | |
| 668 | return [convert(item) for item in items] |
| 669 | |
| 670 | |
| 671 | def _iter_depth_first(items: Sequence[_TreeEntry]) -> Iterable[_TreeEntry]: |
| 672 | for node in items: |
| 673 | yield node |
| 674 | if node.items: |
| 675 | yield from _iter_depth_first(node.items) |