Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/black/ranges.py: 14%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1"""Functions related to Black's formatting by line ranges feature."""
3import difflib
4from collections.abc import Collection, Iterator, Sequence
5from dataclasses import dataclass
7from black.comments import FMT_SKIP, contains_fmt_directive
8from black.nodes import (
9 LN,
10 STANDALONE_COMMENT,
11 Leaf,
12 Node,
13 Visitor,
14 first_leaf,
15 furthest_ancestor_with_last_leaf,
16 last_leaf,
17 syms,
18)
19from blib2to3.pgen2.token import ASYNC, NEWLINE
22def parse_line_ranges(line_ranges: Sequence[str]) -> list[tuple[int, int]]:
23 lines: list[tuple[int, int]] = []
24 for lines_str in line_ranges:
25 parts = lines_str.split("-")
26 if len(parts) != 2:
27 raise ValueError(
28 "Incorrect --line-ranges format, expect 'START-END', found"
29 f" {lines_str!r}"
30 )
31 try:
32 start = int(parts[0])
33 end = int(parts[1])
34 except ValueError:
35 raise ValueError(
36 "Incorrect --line-ranges value, expect integer ranges, found"
37 f" {lines_str!r}"
38 ) from None
39 else:
40 if start < 1 or end < 1:
41 raise ValueError(
42 "Incorrect --line-ranges value, "
43 f"expect positive integers, found {lines_str!r}"
44 )
45 if start > end:
46 raise ValueError(
47 "Incorrect --line-ranges value, expect START <= END, "
48 f"found {lines_str!r}"
49 )
50 lines.append((start, end))
51 return lines
54def is_valid_line_range(lines: tuple[int, int]) -> bool:
55 """Returns whether the line range is valid."""
56 return lines[0] <= lines[1]
59def sanitized_lines(
60 lines: Collection[tuple[int, int]], src_contents: str
61) -> Collection[tuple[int, int]]:
62 """Returns the valid line ranges for the given source.
64 This removes ranges that are entirely outside the valid lines.
66 Other ranges are normalized so that the start values are at least 1 and the
67 end values are at most the (1-based) index of the last source line.
68 """
69 if not src_contents:
70 return []
71 good_lines = []
72 src_line_count = src_contents.count("\n")
73 if not src_contents.endswith("\n"):
74 src_line_count += 1
75 for start, end in lines:
76 if start > src_line_count:
77 continue
78 # line-ranges are 1-based
79 start = max(start, 1)
80 if end < start:
81 continue
82 end = min(end, src_line_count)
83 good_lines.append((start, end))
84 return good_lines
87def adjusted_lines(
88 lines: Collection[tuple[int, int]],
89 original_source: str,
90 modified_source: str,
91) -> list[tuple[int, int]]:
92 """Returns the adjusted line ranges based on edits from the original code.
94 This computes the new line ranges by diffing original_source and
95 modified_source, and adjust each range based on how the range overlaps with
96 the diffs.
98 Note the diff can contain lines outside of the original line ranges. This can
99 happen when the formatting has to be done in adjacent to maintain consistent
100 local results. For example:
102 1. def my_func(arg1, arg2,
103 2. arg3,):
104 3. pass
106 If it restricts to line 2-2, it can't simply reformat line 2, it also has
107 to reformat line 1:
109 1. def my_func(
110 2. arg1,
111 3. arg2,
112 4. arg3,
113 5. ):
114 6. pass
116 In this case, we will expand the line ranges to also include the whole diff
117 block.
119 Args:
120 lines: a collection of line ranges.
121 original_source: the original source.
122 modified_source: the modified source.
123 """
124 if len(lines) > 1:
125 original_lines = original_source.splitlines(keepends=True)
126 modified_lines = modified_source.splitlines(keepends=True)
127 if len(original_lines) == len(modified_lines) and all(
128 1 <= start <= end <= len(original_lines) for start, end in lines
129 ):
130 selected_lines = {
131 line for start, end in lines for line in range(start - 1, end)
132 }
133 # Equal-length edits confined to the requested ranges cannot shift
134 # their line numbers. Diffing repeated text can move them anyway.
135 if all(
136 index in selected_lines or original == modified
137 for index, (original, modified) in enumerate(
138 zip(original_lines, modified_lines, strict=True)
139 )
140 ):
141 return sorted(lines)
143 lines_mappings = _calculate_lines_mappings(original_source, modified_source)
145 new_lines = []
146 # Keep an index of the current search. Since the lines and lines_mappings are
147 # sorted, this makes the search complexity linear.
148 current_mapping_index = 0
149 for start, end in sorted(lines):
150 start_mapping_index = _find_lines_mapping_index(
151 start,
152 lines_mappings,
153 current_mapping_index,
154 )
155 end_mapping_index = _find_lines_mapping_index(
156 end,
157 lines_mappings,
158 start_mapping_index,
159 )
160 current_mapping_index = start_mapping_index
161 if start_mapping_index >= len(lines_mappings) or end_mapping_index >= len(
162 lines_mappings
163 ):
164 # Protect against invalid inputs.
165 continue
166 start_mapping = lines_mappings[start_mapping_index]
167 end_mapping = lines_mappings[end_mapping_index]
168 if start_mapping.is_changed_block:
169 # When the line falls into a changed block, expands to the whole block.
170 new_start = start_mapping.modified_start
171 else:
172 new_start = (
173 start - start_mapping.original_start + start_mapping.modified_start
174 )
175 if end_mapping.is_changed_block:
176 # When the line falls into a changed block, expands to the whole block.
177 new_end = end_mapping.modified_end
178 else:
179 new_end = end - end_mapping.original_start + end_mapping.modified_start
180 new_range = (new_start, new_end)
181 if is_valid_line_range(new_range):
182 new_lines.append(new_range)
183 return new_lines
186def convert_unchanged_lines(src_node: Node, lines: Collection[tuple[int, int]]) -> None:
187 r"""Converts unchanged lines to STANDALONE_COMMENT.
189 The idea is similar to how `# fmt: on/off` is implemented. It also converts the
190 nodes between those markers as a single `STANDALONE_COMMENT` leaf node with
191 the unformatted code as its value. `STANDALONE_COMMENT` is a "fake" token
192 that will be formatted as-is with its prefix normalized.
194 Here we perform two passes:
196 1. Visit the top-level statements, and convert them to a single
197 `STANDALONE_COMMENT` when unchanged. This speeds up formatting when some
198 of the top-level statements aren't changed.
199 2. Convert unchanged "unwrapped lines" to `STANDALONE_COMMENT` nodes line by
200 line. "unwrapped lines" are divided by the `NEWLINE` token. e.g. a
201 multi-line statement is *one* "unwrapped line" that ends with `NEWLINE`,
202 even though this statement itself can span multiple lines, and the
203 tokenizer only sees the last '\n' as the `NEWLINE` token.
205 NOTE: During pass (2), comment prefixes and indentations are ALWAYS
206 normalized even when the lines aren't changed. This is fixable by moving
207 more formatting to pass (1). However, it's hard to get it correct when
208 incorrect indentations are used. So we defer this to future optimizations.
209 """
210 lines_set = frozenset(n for start, end in lines for n in range(start, end + 1))
211 replacements = _NodeReplacements()
212 visitor = _TopLevelStatementsVisitor(lines_set, replacements)
213 _ = list(visitor.visit(src_node)) # Consume all results.
214 replacements.apply()
215 _convert_unchanged_line_by_line(src_node, lines_set)
218class _NodeReplacements:
219 """Collects STANDALONE_COMMENT conversions and applies them per parent.
221 Converting a node splices a `STANDALONE_COMMENT` leaf in place of a run of
222 sibling nodes. Doing that one node at a time with `Base.remove` /
223 `insert_child` scans and shifts the parent's children list on every call, so
224 converting the many sibling blocks of one parent (a long if/elif chain, a
225 match with many cases, a module with many top-level statements) is O(n^2).
226 Recording the conversions and rewriting each parent's children in a single
227 pass makes it O(n).
228 """
230 def __init__(self) -> None:
231 # id(parent) -> (parent, {id(child): standalone to splice in}, {ids to drop})
232 # A run's nodes can live under more than one parent (an `async` statement
233 # keeps the `ASYNC` leaf on the grandparent), so the standalone replaces
234 # the run's first node under its own parent while the remaining nodes are
235 # just dropped from wherever they sit.
236 self._by_parent: dict[int, tuple[Node, dict[int, Leaf], set[int]]] = {}
237 # NEWLINE leaves already inside a recorded run, so the line-by-line pass
238 # doesn't record an overlapping conversion for them.
239 self.covered_newlines: set[int] = set()
240 # Nodes already scheduled for removal. When conversions immediately
241 # mutated the tree, a second conversion touching an already-removed node
242 # was a no-op (`Base.remove` returned None); recording defers the
243 # mutation, so we skip such nodes explicitly to keep that behaviour.
244 self._recorded: set[int] = set()
245 # A leaf's prefix (indentation, blank lines, comments) moves onto its
246 # STANDALONE_COMMENT. When an inner and an outer node that share the same
247 # first leaf are both converted (a decorator inside a decorated block),
248 # the earlier conversion clears the prefix off the leaf, so the later one
249 # would see an empty prefix. Remember the taken prefix keyed by the leaf
250 # so both conversions reuse the original.
251 self._taken_prefixes: dict[int, str] = {}
253 def _entry(self, parent: Node) -> tuple[Node, dict[int, Leaf], set[int]]:
254 return self._by_parent.setdefault(id(parent), (parent, {}, set()))
256 def take_prefix(self, first: Leaf) -> str:
257 cached = self._taken_prefixes.get(id(first))
258 if cached is not None:
259 return cached
260 prefix = first.prefix
261 first.prefix = ""
262 # Immediate conversion replaced the whole run with a freshly built
263 # STANDALONE_COMMENT leaf, whose synthesized position is line 0. Recording
264 # the conversion leaves the original leaf in place, so mirror that line 0
265 # here: when a later, enclosing node shares this first leaf (a decorated
266 # block over an already-recorded decorator), _get_line_range sees the run
267 # as starting at line 0, matching the immediate behaviour.
268 first.lineno = 0
269 self._taken_prefixes[id(first)] = prefix
270 return prefix
272 def is_recorded(self, node: LN) -> bool:
273 return id(node) in self._recorded
275 def record(self, run: list[LN], standalone: Leaf) -> None:
276 first = run[0]
277 first_parent = first.parent
278 if first_parent is None or id(first) in self._recorded:
279 return
280 for node in run:
281 self._recorded.add(id(node))
282 for leaf in node.leaves():
283 if leaf.type == NEWLINE:
284 self.covered_newlines.add(id(leaf))
285 parent = node.parent
286 if parent is not None:
287 self._entry(parent)[2].add(id(node))
288 self._entry(first_parent)[1][id(first)] = standalone
290 def apply(self) -> None:
291 for parent, replaced, removed in self._by_parent.values():
292 new_children: list[LN] = []
293 for child in parent.children:
294 standalone = replaced.get(id(child))
295 if standalone is not None:
296 standalone.parent = parent
297 new_children.append(standalone)
298 child.parent = None
299 elif id(child) in removed:
300 child.parent = None
301 else:
302 new_children.append(child)
303 parent.children = new_children
304 parent.invalidate_sibling_maps()
307def _contains_standalone_comment(node: LN) -> bool:
308 if isinstance(node, Leaf):
309 return node.type == STANDALONE_COMMENT
310 else:
311 for child in node.children:
312 if _contains_standalone_comment(child):
313 return True
314 return False
317class _TopLevelStatementsVisitor(Visitor[None]):
318 """
319 A node visitor that converts unchanged top-level statements to
320 STANDALONE_COMMENT.
322 This is used in addition to _convert_unchanged_line_by_line, to
323 speed up formatting when there are unchanged top-level
324 classes/functions/statements.
325 """
327 def __init__(self, lines_set: frozenset[int], replacements: "_NodeReplacements"):
328 self._lines_set = lines_set
329 self._replacements = replacements
331 def visit_simple_stmt(self, node: Node) -> Iterator[None]:
332 # This is only called for top-level statements, since `visit_suite`
333 # won't visit its children nodes.
334 yield from []
335 newline_leaf = last_leaf(node)
336 if not newline_leaf:
337 return
338 assert (
339 newline_leaf.type == NEWLINE
340 ), f"Unexpectedly found leaf.type={newline_leaf.type}"
341 # We need to find the furthest ancestor with the NEWLINE as the last
342 # leaf, since a `suite` can simply be a `simple_stmt` when it puts
343 # its body on the same line. Example: `if cond: pass`.
344 ancestor = furthest_ancestor_with_last_leaf(newline_leaf)
345 if not _get_line_range(ancestor).intersection(self._lines_set):
346 _convert_node_to_standalone_comment(
347 ancestor, self._replacements, self._lines_set
348 )
350 def visit_suite(self, node: Node) -> Iterator[None]:
351 yield from []
352 # If there is a STANDALONE_COMMENT node, it means parts of the node tree
353 # have fmt on/off/skip markers. Those STANDALONE_COMMENT nodes can't
354 # be simply converted by calling str(node). So we just don't convert
355 # here.
356 if _contains_standalone_comment(node):
357 return
358 # Find the semantic parent of this suite. For `async_stmt` and
359 # `async_funcdef`, the ASYNC token is defined on a separate level by the
360 # grammar.
361 semantic_parent = node.parent
362 if semantic_parent is not None:
363 if (
364 semantic_parent.prev_sibling is not None
365 and semantic_parent.prev_sibling.type == ASYNC
366 ):
367 semantic_parent = semantic_parent.parent
368 if semantic_parent is not None and not _get_line_range(
369 semantic_parent
370 ).intersection(self._lines_set):
371 _convert_node_to_standalone_comment(
372 semantic_parent, self._replacements, self._lines_set
373 )
376def _convert_unchanged_line_by_line(node: Node, lines_set: frozenset[int]) -> None:
377 """Converts unchanged to STANDALONE_COMMENT line by line."""
378 replacements = _NodeReplacements()
379 for leaf in node.leaves():
380 if leaf.type != NEWLINE:
381 # We only consider "unwrapped lines", which are divided by the NEWLINE
382 # token.
383 continue
384 if id(leaf) in replacements.covered_newlines:
385 # This NEWLINE is inside a run already scheduled for conversion (e.g.
386 # a second decorator on a stacked decorator block).
387 continue
388 if leaf.parent and leaf.parent.type == syms.match_stmt:
389 # The `suite` node is defined as:
390 # match_stmt: "match" subject_expr ':' NEWLINE INDENT case_block+ DEDENT
391 # Here we need to check `subject_expr`. The `case_block+` will be
392 # checked by their own NEWLINEs.
393 nodes_to_ignore: list[LN] = []
394 prev_sibling = leaf.prev_sibling
395 while prev_sibling:
396 nodes_to_ignore.insert(0, prev_sibling)
397 prev_sibling = prev_sibling.prev_sibling
398 if not _get_line_range(nodes_to_ignore).intersection(lines_set):
399 _convert_nodes_to_standalone_comment(
400 nodes_to_ignore,
401 newline=leaf,
402 replacements=replacements,
403 lines_set=lines_set,
404 )
405 elif leaf.parent and leaf.parent.type == syms.suite:
406 # The `suite` node is defined as:
407 # suite: simple_stmt | NEWLINE INDENT stmt+ DEDENT
408 # We will check `simple_stmt` and `stmt+` separately against the lines set
409 parent_sibling = leaf.parent.prev_sibling
410 nodes_to_ignore = []
411 while parent_sibling and parent_sibling.type != syms.suite:
412 # NOTE: Multiple suite nodes can exist as siblings in e.g. `if_stmt`.
413 nodes_to_ignore.insert(0, parent_sibling)
414 parent_sibling = parent_sibling.prev_sibling
415 # Special case for `async_stmt` and `async_funcdef` where the ASYNC
416 # token is on the grandparent node.
417 grandparent = leaf.parent.parent
418 if (
419 grandparent is not None
420 and grandparent.prev_sibling is not None
421 and grandparent.prev_sibling.type == ASYNC
422 ):
423 nodes_to_ignore.insert(0, grandparent.prev_sibling)
424 if not _get_line_range(nodes_to_ignore).intersection(lines_set):
425 _convert_nodes_to_standalone_comment(
426 nodes_to_ignore,
427 newline=leaf,
428 replacements=replacements,
429 lines_set=lines_set,
430 )
431 else:
432 ancestor = furthest_ancestor_with_last_leaf(leaf)
433 # Consider multiple decorators as a whole block, as their
434 # newlines have different behaviors than the rest of the grammar.
435 if (
436 ancestor.type == syms.decorator
437 and ancestor.parent
438 and ancestor.parent.type == syms.decorators
439 ):
440 ancestor = ancestor.parent
441 if not _get_line_range(ancestor).intersection(lines_set):
442 _convert_node_to_standalone_comment(ancestor, replacements, lines_set)
443 replacements.apply()
446def _str_with_standalone_comments(*nodes: LN) -> str:
447 """Like `"".join(str(node) for node in nodes)`, but a nested STANDALONE_COMMENT
448 that ends in a comment is followed by a newline.
450 A STANDALONE_COMMENT made by `# fmt: skip` handling, or by a `# fmt: off` /
451 `# fmt: on` block that only holds comments, can end with the comment and has no
452 trailing newline: the line generator puts one after it. Joining it straight to
453 the next leaf would append that leaf (e.g. a closing bracket) to the comment's
454 line and turn it into part of the comment.
455 """
456 leaves = [leaf for node in nodes for leaf in node.leaves()]
457 parts = []
458 for leaf, next_leaf in zip(leaves, [*leaves[1:], None], strict=True):
459 text = str(leaf)
460 if (
461 leaf.type == STANDALONE_COMMENT
462 and next_leaf is not None
463 and (
464 # A comment-only fmt:off/on block ends with the `# fmt: on` comment.
465 # A newline in the next leaf's prefix is a blank line after it.
466 leaf.fmt_pass_converted_first_leaf is None
467 or (not str(next_leaf).startswith("\n") and _ends_with_fmt_skip(text))
468 )
469 ):
470 text += "\n"
471 parts.append(text)
472 return "".join(parts)
475def _ends_with_fmt_skip(text: str) -> bool:
476 """Whether the last line of a STANDALONE_COMMENT value ends in a `# fmt: skip`.
478 Only that directive makes such a value end in a comment without a trailing
479 newline. Any other `#` is ignored: it may be inside a string.
480 """
481 last_line = text.rpartition("\n")[2]
482 comment_start = last_line.find("#")
483 return comment_start != -1 and contains_fmt_directive(
484 last_line[comment_start:], FMT_SKIP
485 )
488def _convert_node_to_standalone_comment(
489 node: LN, replacements: "_NodeReplacements", lines_set: frozenset[int]
490) -> None:
491 """Convert node to STANDALONE_COMMENT by modifying the tree inline."""
492 parent = node.parent
493 if not parent or replacements.is_recorded(node):
494 return
495 first = first_leaf(node)
496 last = last_leaf(node)
497 if not first or not last:
498 return
499 if first is last:
500 # This can happen on the following edge cases:
501 # 1. A block of `# fmt: off/on` code except the `# fmt: on` is placed
502 # on the end of the last line instead of on a new line.
503 # 2. A single backslash on its own line followed by a comment line.
504 # Ideally we don't want to format them when not requested, but fixing
505 # isn't easy. These cases are also badly formatted code, so it isn't
506 # too bad we reformat them.
507 return
508 # The prefix contains comments and indentation whitespaces. They are
509 # reformatted accordingly to the correct indentation level.
510 # This also means the indentation will be changed on the unchanged lines, and
511 # this is actually required to not break incremental reformatting.
512 first_lineno = first.lineno
513 prefix = replacements.take_prefix(first)
514 # For a single-line decorated item the decorator and the item need a newline
515 # between them. The conversions are recorded and applied in a single pass
516 # instead of mutating the tree per node, so the decorator here is still its
517 # own child node carrying its trailing NEWLINE, which already separates it
518 # from the item; there's nothing to add back. (A decorated item spanning
519 # multiple lines is handled by the earlier suite case in
520 # _convert_unchanged_line_by_line, which manages the newlines itself.)
521 # Remove the '\n', as STANDALONE_COMMENT will have '\n' appended when
522 # generating the formatted code.
523 value = _str_with_standalone_comments(node)[:-1]
524 replacements.record(
525 [node],
526 Leaf(
527 STANDALONE_COMMENT,
528 value,
529 prefix=prefix,
530 fmt_pass_converted_first_leaf=first,
531 line_ranges_first_lineno=first_lineno,
532 line_ranges_selected=lines_set,
533 ),
534 )
537def _convert_nodes_to_standalone_comment(
538 nodes: Sequence[LN],
539 *,
540 newline: Leaf,
541 replacements: "_NodeReplacements",
542 lines_set: frozenset[int],
543) -> None:
544 """Convert nodes to STANDALONE_COMMENT by modifying the tree inline."""
545 if not nodes:
546 return
547 parent = nodes[0].parent
548 first = first_leaf(nodes[0])
549 if not parent or not first or replacements.is_recorded(nodes[0]):
550 return
551 first_lineno = first.lineno
552 prefix = replacements.take_prefix(first)
553 value = _str_with_standalone_comments(*nodes)
554 # The prefix comment on the NEWLINE leaf is the trailing comment of the statement.
555 if newline.prefix:
556 value += newline.prefix
557 newline.prefix = ""
558 replacements.record(
559 list(nodes),
560 Leaf(
561 STANDALONE_COMMENT,
562 value,
563 prefix=prefix,
564 fmt_pass_converted_first_leaf=first,
565 line_ranges_first_lineno=first_lineno,
566 line_ranges_selected=lines_set,
567 ),
568 )
571def _leaf_line_end(leaf: Leaf) -> int:
572 """Returns the line number of the leaf node's last line."""
573 if leaf.type == NEWLINE:
574 return leaf.lineno
575 else:
576 # Leaf nodes like multiline strings can occupy multiple lines.
577 return leaf.lineno + str(leaf).count("\n")
580def _get_line_range(node_or_nodes: LN | list[LN]) -> set[int]:
581 """Returns the line range of this node or list of nodes."""
582 if isinstance(node_or_nodes, list):
583 nodes = node_or_nodes
584 if not nodes:
585 return set()
586 first = first_leaf(nodes[0])
587 last = last_leaf(nodes[-1])
588 if first and last:
589 line_start = first.lineno
590 line_end = _leaf_line_end(last)
591 return set(range(line_start, line_end + 1))
592 else:
593 return set()
594 else:
595 node = node_or_nodes
596 if isinstance(node, Leaf):
597 return set(range(node.lineno, _leaf_line_end(node) + 1))
598 else:
599 first = first_leaf(node)
600 last = last_leaf(node)
601 if first and last:
602 return set(range(first.lineno, _leaf_line_end(last) + 1))
603 else:
604 return set()
607@dataclass
608class _LinesMapping:
609 """1-based lines mapping from original source to modified source.
611 Lines [original_start, original_end] from original source
612 are mapped to [modified_start, modified_end].
614 The ranges are inclusive on both ends.
615 """
617 original_start: int
618 original_end: int
619 modified_start: int
620 modified_end: int
621 # Whether this range corresponds to a changed block, or an unchanged block.
622 is_changed_block: bool
625def _calculate_lines_mappings(
626 original_source: str,
627 modified_source: str,
628) -> Sequence[_LinesMapping]:
629 """Returns a sequence of _LinesMapping by diffing the sources.
631 For example, given the following diff:
632 import re
633 - def func(arg1,
634 - arg2, arg3):
635 + def func(arg1, arg2, arg3):
636 pass
637 It returns the following mappings:
638 original -> modified
639 (1, 1) -> (1, 1), is_changed_block=False (the "import re" line)
640 (2, 3) -> (2, 2), is_changed_block=True (the diff)
641 (4, 4) -> (3, 3), is_changed_block=False (the "pass" line)
643 You can think of this visually as if it brings up a side-by-side diff, and tries
644 to map the line ranges from the left side to the right side:
646 (1, 1)->(1, 1) 1. import re 1. import re
647 (2, 3)->(2, 2) 2. def func(arg1, 2. def func(arg1, arg2, arg3):
648 3. arg2, arg3):
649 (4, 4)->(3, 3) 4. pass 3. pass
651 Args:
652 original_source: the original source.
653 modified_source: the modified source.
654 """
655 original_lines = original_source.splitlines(keepends=True)
656 modified_lines = modified_source.splitlines(keepends=True)
658 # SequenceMatcher can align repeated unchanged lines at a different position,
659 # making a change in the middle look like an insertion at the start and a
660 # deletion at the end. Anchor the common edges before diffing the middle.
661 prefix = 0
662 while (
663 prefix < min(len(original_lines), len(modified_lines))
664 and original_lines[prefix] == modified_lines[prefix]
665 ):
666 prefix += 1
668 suffix = 0
669 while (
670 suffix < min(len(original_lines), len(modified_lines)) - prefix
671 and original_lines[-suffix - 1] == modified_lines[-suffix - 1]
672 ):
673 suffix += 1
675 original_middle_end = len(original_lines) - suffix
676 modified_middle_end = len(modified_lines) - suffix
677 # The first formatting pass re-indents the bodies of the statements it
678 # formats, so the same line can have a different indentation in the two
679 # sources. Comparing the lines without indentation keeps them matched.
680 matcher = difflib.SequenceMatcher(
681 None,
682 [line.strip() for line in original_lines[prefix:original_middle_end]],
683 [line.strip() for line in modified_lines[prefix:modified_middle_end]],
684 )
685 matching_blocks = []
686 if prefix:
687 matching_blocks.append(difflib.Match(0, 0, prefix))
688 matching_blocks.extend(
689 difflib.Match(block.a + prefix, block.b + prefix, block.size)
690 for block in matcher.get_matching_blocks()
691 if block.size
692 )
693 if suffix:
694 matching_blocks.append(
695 difflib.Match(original_middle_end, modified_middle_end, suffix)
696 )
697 matching_blocks.append(difflib.Match(len(original_lines), len(modified_lines), 0))
698 lines_mappings: list[_LinesMapping] = []
699 # matching_blocks is a sequence of "same block of code ranges", see
700 # https://docs.python.org/3/library/difflib.html#difflib.SequenceMatcher.get_matching_blocks
701 # Each block corresponds to a _LinesMapping with is_changed_block=False,
702 # and the ranges between two blocks corresponds to a _LinesMapping with
703 # is_changed_block=True,
704 # NOTE: matching_blocks is 0-based, but _LinesMapping is 1-based.
705 for i, block in enumerate(matching_blocks):
706 if i == 0:
707 if block.a != 0 or block.b != 0:
708 lines_mappings.append(
709 _LinesMapping(
710 original_start=1,
711 original_end=block.a,
712 modified_start=1,
713 modified_end=block.b,
714 # The lines before the first matching block are not part of
715 # any matching block, so this range is a changed block. A line
716 # that only exists on one side has no counterpart to map onto.
717 is_changed_block=True,
718 )
719 )
720 else:
721 previous_block = matching_blocks[i - 1]
722 lines_mappings.append(
723 _LinesMapping(
724 original_start=previous_block.a + previous_block.size + 1,
725 original_end=block.a,
726 modified_start=previous_block.b + previous_block.size + 1,
727 modified_end=block.b,
728 is_changed_block=True,
729 )
730 )
731 if i < len(matching_blocks) - 1:
732 lines_mappings.append(
733 _LinesMapping(
734 original_start=block.a + 1,
735 original_end=block.a + block.size,
736 modified_start=block.b + 1,
737 modified_end=block.b + block.size,
738 is_changed_block=False,
739 )
740 )
741 return lines_mappings
744def _find_lines_mapping_index(
745 original_line: int,
746 lines_mappings: Sequence[_LinesMapping],
747 start_index: int,
748) -> int:
749 """Returns the original index of the lines mappings for the original line."""
750 index = start_index
751 while index < len(lines_mappings):
752 mapping = lines_mappings[index]
753 if mapping.original_start <= original_line <= mapping.original_end:
754 return index
755 index += 1
756 return index