]> git.madduck.net Git - etc/vim.git/blob - black.py

madduck's git repository

Every one of the projects in this repository is available at the canonical URL git://git.madduck.net/madduck/pub/<projectpath> — see each project's metadata for the exact URL.

All patches and comments are welcome. Please squash your changes to logical commits before using git-format-patch and git-send-email to patches@git.madduck.net. If you'd read over the Git project's submission guidelines and adhered to them, I'd be especially grateful.

SSH access, as well as push access can be individually arranged.

If you use my repositories frequently, consider adding the following snippet to ~/.gitconfig and using the third clone URL listed for each project:

[url "git://git.madduck.net/madduck/"]
  insteadOf = madduck:

f0e83f86df1c380c4be6c26364e96d1970f4d6ac
[etc/vim.git] / black.py
1 import ast
2 import asyncio
3 from concurrent.futures import Executor, ProcessPoolExecutor
4 from contextlib import contextmanager
5 from datetime import datetime
6 from enum import Enum
7 from functools import lru_cache, partial, wraps
8 import io
9 import itertools
10 import logging
11 from multiprocessing import Manager, freeze_support
12 import os
13 from pathlib import Path
14 import pickle
15 import regex as re
16 import signal
17 import sys
18 import tempfile
19 import tokenize
20 import traceback
21 from typing import (
22     Any,
23     Callable,
24     Collection,
25     Dict,
26     Generator,
27     Generic,
28     Iterable,
29     Iterator,
30     List,
31     Optional,
32     Pattern,
33     Sequence,
34     Set,
35     Tuple,
36     TypeVar,
37     Union,
38     cast,
39 )
40
41 from appdirs import user_cache_dir
42 from attr import dataclass, evolve, Factory
43 import click
44 import toml
45 from typed_ast import ast3, ast27
46 from pathspec import PathSpec
47
48 # lib2to3 fork
49 from blib2to3.pytree import Node, Leaf, type_repr
50 from blib2to3 import pygram, pytree
51 from blib2to3.pgen2 import driver, token
52 from blib2to3.pgen2.grammar import Grammar
53 from blib2to3.pgen2.parse import ParseError
54
55 from _black_version import version as __version__
56
57 DEFAULT_LINE_LENGTH = 88
58 DEFAULT_EXCLUDES = r"/(\.eggs|\.git|\.hg|\.mypy_cache|\.nox|\.tox|\.venv|\.svn|_build|buck-out|build|dist)/"  # noqa: B950
59 DEFAULT_INCLUDES = r"\.pyi?$"
60 CACHE_DIR = Path(user_cache_dir("black", version=__version__))
61
62
63 # types
64 FileContent = str
65 Encoding = str
66 NewLine = str
67 Depth = int
68 NodeType = int
69 LeafID = int
70 Priority = int
71 Index = int
72 LN = Union[Leaf, Node]
73 SplitFunc = Callable[["Line", Collection["Feature"]], Iterator["Line"]]
74 Timestamp = float
75 FileSize = int
76 CacheInfo = Tuple[Timestamp, FileSize]
77 Cache = Dict[Path, CacheInfo]
78 out = partial(click.secho, bold=True, err=True)
79 err = partial(click.secho, fg="red", err=True)
80
81 pygram.initialize(CACHE_DIR)
82 syms = pygram.python_symbols
83
84
85 class NothingChanged(UserWarning):
86     """Raised when reformatted code is the same as source."""
87
88
89 class CannotSplit(Exception):
90     """A readable split that fits the allotted line length is impossible."""
91
92
93 class InvalidInput(ValueError):
94     """Raised when input source code fails all parse attempts."""
95
96
97 class WriteBack(Enum):
98     NO = 0
99     YES = 1
100     DIFF = 2
101     CHECK = 3
102
103     @classmethod
104     def from_configuration(cls, *, check: bool, diff: bool) -> "WriteBack":
105         if check and not diff:
106             return cls.CHECK
107
108         return cls.DIFF if diff else cls.YES
109
110
111 class Changed(Enum):
112     NO = 0
113     CACHED = 1
114     YES = 2
115
116
117 class TargetVersion(Enum):
118     PY27 = 2
119     PY33 = 3
120     PY34 = 4
121     PY35 = 5
122     PY36 = 6
123     PY37 = 7
124     PY38 = 8
125
126     def is_python2(self) -> bool:
127         return self is TargetVersion.PY27
128
129
130 PY36_VERSIONS = {TargetVersion.PY36, TargetVersion.PY37, TargetVersion.PY38}
131
132
133 class Feature(Enum):
134     # All string literals are unicode
135     UNICODE_LITERALS = 1
136     F_STRINGS = 2
137     NUMERIC_UNDERSCORES = 3
138     TRAILING_COMMA_IN_CALL = 4
139     TRAILING_COMMA_IN_DEF = 5
140     # The following two feature-flags are mutually exclusive, and exactly one should be
141     # set for every version of python.
142     ASYNC_IDENTIFIERS = 6
143     ASYNC_KEYWORDS = 7
144     ASSIGNMENT_EXPRESSIONS = 8
145     POS_ONLY_ARGUMENTS = 9
146
147
148 VERSION_TO_FEATURES: Dict[TargetVersion, Set[Feature]] = {
149     TargetVersion.PY27: {Feature.ASYNC_IDENTIFIERS},
150     TargetVersion.PY33: {Feature.UNICODE_LITERALS, Feature.ASYNC_IDENTIFIERS},
151     TargetVersion.PY34: {Feature.UNICODE_LITERALS, Feature.ASYNC_IDENTIFIERS},
152     TargetVersion.PY35: {
153         Feature.UNICODE_LITERALS,
154         Feature.TRAILING_COMMA_IN_CALL,
155         Feature.ASYNC_IDENTIFIERS,
156     },
157     TargetVersion.PY36: {
158         Feature.UNICODE_LITERALS,
159         Feature.F_STRINGS,
160         Feature.NUMERIC_UNDERSCORES,
161         Feature.TRAILING_COMMA_IN_CALL,
162         Feature.TRAILING_COMMA_IN_DEF,
163         Feature.ASYNC_IDENTIFIERS,
164     },
165     TargetVersion.PY37: {
166         Feature.UNICODE_LITERALS,
167         Feature.F_STRINGS,
168         Feature.NUMERIC_UNDERSCORES,
169         Feature.TRAILING_COMMA_IN_CALL,
170         Feature.TRAILING_COMMA_IN_DEF,
171         Feature.ASYNC_KEYWORDS,
172     },
173     TargetVersion.PY38: {
174         Feature.UNICODE_LITERALS,
175         Feature.F_STRINGS,
176         Feature.NUMERIC_UNDERSCORES,
177         Feature.TRAILING_COMMA_IN_CALL,
178         Feature.TRAILING_COMMA_IN_DEF,
179         Feature.ASYNC_KEYWORDS,
180         Feature.ASSIGNMENT_EXPRESSIONS,
181         Feature.POS_ONLY_ARGUMENTS,
182     },
183 }
184
185
186 @dataclass
187 class FileMode:
188     target_versions: Set[TargetVersion] = Factory(set)
189     line_length: int = DEFAULT_LINE_LENGTH
190     string_normalization: bool = True
191     is_pyi: bool = False
192
193     def get_cache_key(self) -> str:
194         if self.target_versions:
195             version_str = ",".join(
196                 str(version.value)
197                 for version in sorted(self.target_versions, key=lambda v: v.value)
198             )
199         else:
200             version_str = "-"
201         parts = [
202             version_str,
203             str(self.line_length),
204             str(int(self.string_normalization)),
205             str(int(self.is_pyi)),
206         ]
207         return ".".join(parts)
208
209
210 def supports_feature(target_versions: Set[TargetVersion], feature: Feature) -> bool:
211     return all(feature in VERSION_TO_FEATURES[version] for version in target_versions)
212
213
214 def read_pyproject_toml(
215     ctx: click.Context, param: click.Parameter, value: Union[str, int, bool, None]
216 ) -> Optional[str]:
217     """Inject Black configuration from "pyproject.toml" into defaults in `ctx`.
218
219     Returns the path to a successfully found and read configuration file, None
220     otherwise.
221     """
222     assert not isinstance(value, (int, bool)), "Invalid parameter type passed"
223     if not value:
224         root = find_project_root(ctx.params.get("src", ()))
225         path = root / "pyproject.toml"
226         if path.is_file():
227             value = str(path)
228         else:
229             return None
230
231     try:
232         pyproject_toml = toml.load(value)
233         config = pyproject_toml.get("tool", {}).get("black", {})
234     except (toml.TomlDecodeError, OSError) as e:
235         raise click.FileError(
236             filename=value, hint=f"Error reading configuration file: {e}"
237         )
238
239     if not config:
240         return None
241
242     if ctx.default_map is None:
243         ctx.default_map = {}
244     ctx.default_map.update(  # type: ignore  # bad types in .pyi
245         {k.replace("--", "").replace("-", "_"): v for k, v in config.items()}
246     )
247     return value
248
249
250 @click.command(context_settings=dict(help_option_names=["-h", "--help"]))
251 @click.option("-c", "--code", type=str, help="Format the code passed in as a string.")
252 @click.option(
253     "-l",
254     "--line-length",
255     type=int,
256     default=DEFAULT_LINE_LENGTH,
257     help="How many characters per line to allow.",
258     show_default=True,
259 )
260 @click.option(
261     "-t",
262     "--target-version",
263     type=click.Choice([v.name.lower() for v in TargetVersion]),
264     callback=lambda c, p, v: [TargetVersion[val.upper()] for val in v],
265     multiple=True,
266     help=(
267         "Python versions that should be supported by Black's output. [default: "
268         "per-file auto-detection]"
269     ),
270 )
271 @click.option(
272     "--py36",
273     is_flag=True,
274     help=(
275         "Allow using Python 3.6-only syntax on all input files.  This will put "
276         "trailing commas in function signatures and calls also after *args and "
277         "**kwargs. Deprecated; use --target-version instead. "
278         "[default: per-file auto-detection]"
279     ),
280 )
281 @click.option(
282     "--pyi",
283     is_flag=True,
284     help=(
285         "Format all input files like typing stubs regardless of file extension "
286         "(useful when piping source on standard input)."
287     ),
288 )
289 @click.option(
290     "-S",
291     "--skip-string-normalization",
292     is_flag=True,
293     help="Don't normalize string quotes or prefixes.",
294 )
295 @click.option(
296     "--check",
297     is_flag=True,
298     help=(
299         "Don't write the files back, just return the status.  Return code 0 "
300         "means nothing would change.  Return code 1 means some files would be "
301         "reformatted.  Return code 123 means there was an internal error."
302     ),
303 )
304 @click.option(
305     "--diff",
306     is_flag=True,
307     help="Don't write the files back, just output a diff for each file on stdout.",
308 )
309 @click.option(
310     "--fast/--safe",
311     is_flag=True,
312     help="If --fast given, skip temporary sanity checks. [default: --safe]",
313 )
314 @click.option(
315     "--include",
316     type=str,
317     default=DEFAULT_INCLUDES,
318     help=(
319         "A regular expression that matches files and directories that should be "
320         "included on recursive searches.  An empty value means all files are "
321         "included regardless of the name.  Use forward slashes for directories on "
322         "all platforms (Windows, too).  Exclusions are calculated first, inclusions "
323         "later."
324     ),
325     show_default=True,
326 )
327 @click.option(
328     "--exclude",
329     type=str,
330     default=DEFAULT_EXCLUDES,
331     help=(
332         "A regular expression that matches files and directories that should be "
333         "excluded on recursive searches.  An empty value means no paths are excluded. "
334         "Use forward slashes for directories on all platforms (Windows, too).  "
335         "Exclusions are calculated first, inclusions later."
336     ),
337     show_default=True,
338 )
339 @click.option(
340     "-q",
341     "--quiet",
342     is_flag=True,
343     help=(
344         "Don't emit non-error messages to stderr. Errors are still emitted; "
345         "silence those with 2>/dev/null."
346     ),
347 )
348 @click.option(
349     "-v",
350     "--verbose",
351     is_flag=True,
352     help=(
353         "Also emit messages to stderr about files that were not changed or were "
354         "ignored due to --exclude=."
355     ),
356 )
357 @click.version_option(version=__version__)
358 @click.argument(
359     "src",
360     nargs=-1,
361     type=click.Path(
362         exists=True, file_okay=True, dir_okay=True, readable=True, allow_dash=True
363     ),
364     is_eager=True,
365 )
366 @click.option(
367     "--config",
368     type=click.Path(
369         exists=False, file_okay=True, dir_okay=False, readable=True, allow_dash=False
370     ),
371     is_eager=True,
372     callback=read_pyproject_toml,
373     help="Read configuration from PATH.",
374 )
375 @click.pass_context
376 def main(
377     ctx: click.Context,
378     code: Optional[str],
379     line_length: int,
380     target_version: List[TargetVersion],
381     check: bool,
382     diff: bool,
383     fast: bool,
384     pyi: bool,
385     py36: bool,
386     skip_string_normalization: bool,
387     quiet: bool,
388     verbose: bool,
389     include: str,
390     exclude: str,
391     src: Tuple[str],
392     config: Optional[str],
393 ) -> None:
394     """The uncompromising code formatter."""
395     write_back = WriteBack.from_configuration(check=check, diff=diff)
396     if target_version:
397         if py36:
398             err(f"Cannot use both --target-version and --py36")
399             ctx.exit(2)
400         else:
401             versions = set(target_version)
402     elif py36:
403         err(
404             "--py36 is deprecated and will be removed in a future version. "
405             "Use --target-version py36 instead."
406         )
407         versions = PY36_VERSIONS
408     else:
409         # We'll autodetect later.
410         versions = set()
411     mode = FileMode(
412         target_versions=versions,
413         line_length=line_length,
414         is_pyi=pyi,
415         string_normalization=not skip_string_normalization,
416     )
417     if config and verbose:
418         out(f"Using configuration from {config}.", bold=False, fg="blue")
419     if code is not None:
420         print(format_str(code, mode=mode))
421         ctx.exit(0)
422     try:
423         include_regex = re_compile_maybe_verbose(include)
424     except re.error:
425         err(f"Invalid regular expression for include given: {include!r}")
426         ctx.exit(2)
427     try:
428         exclude_regex = re_compile_maybe_verbose(exclude)
429     except re.error:
430         err(f"Invalid regular expression for exclude given: {exclude!r}")
431         ctx.exit(2)
432     report = Report(check=check, quiet=quiet, verbose=verbose)
433     root = find_project_root(src)
434     sources: Set[Path] = set()
435     path_empty(src, quiet, verbose, ctx)
436     for s in src:
437         p = Path(s)
438         if p.is_dir():
439             sources.update(
440                 gen_python_files_in_dir(
441                     p, root, include_regex, exclude_regex, report, get_gitignore(root)
442                 )
443             )
444         elif p.is_file() or s == "-":
445             # if a file was explicitly given, we don't care about its extension
446             sources.add(p)
447         else:
448             err(f"invalid path: {s}")
449     if len(sources) == 0:
450         if verbose or not quiet:
451             out("No Python files are present to be formatted. Nothing to do 😴")
452         ctx.exit(0)
453
454     if len(sources) == 1:
455         reformat_one(
456             src=sources.pop(),
457             fast=fast,
458             write_back=write_back,
459             mode=mode,
460             report=report,
461         )
462     else:
463         reformat_many(
464             sources=sources, fast=fast, write_back=write_back, mode=mode, report=report
465         )
466
467     if verbose or not quiet:
468         out("Oh no! 💥 💔 💥" if report.return_code else "All done! ✨ 🍰 ✨")
469         click.secho(str(report), err=True)
470     ctx.exit(report.return_code)
471
472
473 def path_empty(src: Tuple[str], quiet: bool, verbose: bool, ctx: click.Context) -> None:
474     """
475     Exit if there is no `src` provided for formatting
476     """
477     if not src:
478         if verbose or not quiet:
479             out("No Path provided. Nothing to do 😴")
480             ctx.exit(0)
481
482
483 def reformat_one(
484     src: Path, fast: bool, write_back: WriteBack, mode: FileMode, report: "Report"
485 ) -> None:
486     """Reformat a single file under `src` without spawning child processes.
487
488     `fast`, `write_back`, and `mode` options are passed to
489     :func:`format_file_in_place` or :func:`format_stdin_to_stdout`.
490     """
491     try:
492         changed = Changed.NO
493         if not src.is_file() and str(src) == "-":
494             if format_stdin_to_stdout(fast=fast, write_back=write_back, mode=mode):
495                 changed = Changed.YES
496         else:
497             cache: Cache = {}
498             if write_back != WriteBack.DIFF:
499                 cache = read_cache(mode)
500                 res_src = src.resolve()
501                 if res_src in cache and cache[res_src] == get_cache_info(res_src):
502                     changed = Changed.CACHED
503             if changed is not Changed.CACHED and format_file_in_place(
504                 src, fast=fast, write_back=write_back, mode=mode
505             ):
506                 changed = Changed.YES
507             if (write_back is WriteBack.YES and changed is not Changed.CACHED) or (
508                 write_back is WriteBack.CHECK and changed is Changed.NO
509             ):
510                 write_cache(cache, [src], mode)
511         report.done(src, changed)
512     except Exception as exc:
513         report.failed(src, str(exc))
514
515
516 def reformat_many(
517     sources: Set[Path],
518     fast: bool,
519     write_back: WriteBack,
520     mode: FileMode,
521     report: "Report",
522 ) -> None:
523     """Reformat multiple files using a ProcessPoolExecutor."""
524     loop = asyncio.get_event_loop()
525     worker_count = os.cpu_count()
526     if sys.platform == "win32":
527         # Work around https://bugs.python.org/issue26903
528         worker_count = min(worker_count, 61)
529     executor = ProcessPoolExecutor(max_workers=worker_count)
530     try:
531         loop.run_until_complete(
532             schedule_formatting(
533                 sources=sources,
534                 fast=fast,
535                 write_back=write_back,
536                 mode=mode,
537                 report=report,
538                 loop=loop,
539                 executor=executor,
540             )
541         )
542     finally:
543         shutdown(loop)
544         executor.shutdown()
545
546
547 async def schedule_formatting(
548     sources: Set[Path],
549     fast: bool,
550     write_back: WriteBack,
551     mode: FileMode,
552     report: "Report",
553     loop: asyncio.AbstractEventLoop,
554     executor: Executor,
555 ) -> None:
556     """Run formatting of `sources` in parallel using the provided `executor`.
557
558     (Use ProcessPoolExecutors for actual parallelism.)
559
560     `write_back`, `fast`, and `mode` options are passed to
561     :func:`format_file_in_place`.
562     """
563     cache: Cache = {}
564     if write_back != WriteBack.DIFF:
565         cache = read_cache(mode)
566         sources, cached = filter_cached(cache, sources)
567         for src in sorted(cached):
568             report.done(src, Changed.CACHED)
569     if not sources:
570         return
571
572     cancelled = []
573     sources_to_cache = []
574     lock = None
575     if write_back == WriteBack.DIFF:
576         # For diff output, we need locks to ensure we don't interleave output
577         # from different processes.
578         manager = Manager()
579         lock = manager.Lock()
580     tasks = {
581         asyncio.ensure_future(
582             loop.run_in_executor(
583                 executor, format_file_in_place, src, fast, mode, write_back, lock
584             )
585         ): src
586         for src in sorted(sources)
587     }
588     pending: Iterable[asyncio.Future] = tasks.keys()
589     try:
590         loop.add_signal_handler(signal.SIGINT, cancel, pending)
591         loop.add_signal_handler(signal.SIGTERM, cancel, pending)
592     except NotImplementedError:
593         # There are no good alternatives for these on Windows.
594         pass
595     while pending:
596         done, _ = await asyncio.wait(pending, return_when=asyncio.FIRST_COMPLETED)
597         for task in done:
598             src = tasks.pop(task)
599             if task.cancelled():
600                 cancelled.append(task)
601             elif task.exception():
602                 report.failed(src, str(task.exception()))
603             else:
604                 changed = Changed.YES if task.result() else Changed.NO
605                 # If the file was written back or was successfully checked as
606                 # well-formatted, store this information in the cache.
607                 if write_back is WriteBack.YES or (
608                     write_back is WriteBack.CHECK and changed is Changed.NO
609                 ):
610                     sources_to_cache.append(src)
611                 report.done(src, changed)
612     if cancelled:
613         await asyncio.gather(*cancelled, loop=loop, return_exceptions=True)
614     if sources_to_cache:
615         write_cache(cache, sources_to_cache, mode)
616
617
618 def format_file_in_place(
619     src: Path,
620     fast: bool,
621     mode: FileMode,
622     write_back: WriteBack = WriteBack.NO,
623     lock: Any = None,  # multiprocessing.Manager().Lock() is some crazy proxy
624 ) -> bool:
625     """Format file under `src` path. Return True if changed.
626
627     If `write_back` is DIFF, write a diff to stdout. If it is YES, write reformatted
628     code to the file.
629     `mode` and `fast` options are passed to :func:`format_file_contents`.
630     """
631     if src.suffix == ".pyi":
632         mode = evolve(mode, is_pyi=True)
633
634     then = datetime.utcfromtimestamp(src.stat().st_mtime)
635     with open(src, "rb") as buf:
636         src_contents, encoding, newline = decode_bytes(buf.read())
637     try:
638         dst_contents = format_file_contents(src_contents, fast=fast, mode=mode)
639     except NothingChanged:
640         return False
641
642     if write_back == write_back.YES:
643         with open(src, "w", encoding=encoding, newline=newline) as f:
644             f.write(dst_contents)
645     elif write_back == write_back.DIFF:
646         now = datetime.utcnow()
647         src_name = f"{src}\t{then} +0000"
648         dst_name = f"{src}\t{now} +0000"
649         diff_contents = diff(src_contents, dst_contents, src_name, dst_name)
650
651         with lock or nullcontext():
652             f = io.TextIOWrapper(
653                 sys.stdout.buffer,
654                 encoding=encoding,
655                 newline=newline,
656                 write_through=True,
657             )
658             f.write(diff_contents)
659             f.detach()
660
661     return True
662
663
664 def format_stdin_to_stdout(
665     fast: bool, *, write_back: WriteBack = WriteBack.NO, mode: FileMode
666 ) -> bool:
667     """Format file on stdin. Return True if changed.
668
669     If `write_back` is YES, write reformatted code back to stdout. If it is DIFF,
670     write a diff to stdout. The `mode` argument is passed to
671     :func:`format_file_contents`.
672     """
673     then = datetime.utcnow()
674     src, encoding, newline = decode_bytes(sys.stdin.buffer.read())
675     dst = src
676     try:
677         dst = format_file_contents(src, fast=fast, mode=mode)
678         return True
679
680     except NothingChanged:
681         return False
682
683     finally:
684         f = io.TextIOWrapper(
685             sys.stdout.buffer, encoding=encoding, newline=newline, write_through=True
686         )
687         if write_back == WriteBack.YES:
688             f.write(dst)
689         elif write_back == WriteBack.DIFF:
690             now = datetime.utcnow()
691             src_name = f"STDIN\t{then} +0000"
692             dst_name = f"STDOUT\t{now} +0000"
693             f.write(diff(src, dst, src_name, dst_name))
694         f.detach()
695
696
697 def format_file_contents(
698     src_contents: str, *, fast: bool, mode: FileMode
699 ) -> FileContent:
700     """Reformat contents a file and return new contents.
701
702     If `fast` is False, additionally confirm that the reformatted code is
703     valid by calling :func:`assert_equivalent` and :func:`assert_stable` on it.
704     `mode` is passed to :func:`format_str`.
705     """
706     if src_contents.strip() == "":
707         raise NothingChanged
708
709     dst_contents = format_str(src_contents, mode=mode)
710     if src_contents == dst_contents:
711         raise NothingChanged
712
713     if not fast:
714         assert_equivalent(src_contents, dst_contents)
715         assert_stable(src_contents, dst_contents, mode=mode)
716     return dst_contents
717
718
719 def format_str(src_contents: str, *, mode: FileMode) -> FileContent:
720     """Reformat a string and return new contents.
721
722     `mode` determines formatting options, such as how many characters per line are
723     allowed.
724     """
725     src_node = lib2to3_parse(src_contents.lstrip(), mode.target_versions)
726     dst_contents = []
727     future_imports = get_future_imports(src_node)
728     if mode.target_versions:
729         versions = mode.target_versions
730     else:
731         versions = detect_target_versions(src_node)
732     normalize_fmt_off(src_node)
733     lines = LineGenerator(
734         remove_u_prefix="unicode_literals" in future_imports
735         or supports_feature(versions, Feature.UNICODE_LITERALS),
736         is_pyi=mode.is_pyi,
737         normalize_strings=mode.string_normalization,
738     )
739     elt = EmptyLineTracker(is_pyi=mode.is_pyi)
740     empty_line = Line()
741     after = 0
742     split_line_features = {
743         feature
744         for feature in {Feature.TRAILING_COMMA_IN_CALL, Feature.TRAILING_COMMA_IN_DEF}
745         if supports_feature(versions, feature)
746     }
747     for current_line in lines.visit(src_node):
748         for _ in range(after):
749             dst_contents.append(str(empty_line))
750         before, after = elt.maybe_empty_lines(current_line)
751         for _ in range(before):
752             dst_contents.append(str(empty_line))
753         for line in split_line(
754             current_line, line_length=mode.line_length, features=split_line_features
755         ):
756             dst_contents.append(str(line))
757     return "".join(dst_contents)
758
759
760 def decode_bytes(src: bytes) -> Tuple[FileContent, Encoding, NewLine]:
761     """Return a tuple of (decoded_contents, encoding, newline).
762
763     `newline` is either CRLF or LF but `decoded_contents` is decoded with
764     universal newlines (i.e. only contains LF).
765     """
766     srcbuf = io.BytesIO(src)
767     encoding, lines = tokenize.detect_encoding(srcbuf.readline)
768     if not lines:
769         return "", encoding, "\n"
770
771     newline = "\r\n" if b"\r\n" == lines[0][-2:] else "\n"
772     srcbuf.seek(0)
773     with io.TextIOWrapper(srcbuf, encoding) as tiow:
774         return tiow.read(), encoding, newline
775
776
777 def get_grammars(target_versions: Set[TargetVersion]) -> List[Grammar]:
778     if not target_versions:
779         # No target_version specified, so try all grammars.
780         return [
781             # Python 3.7+
782             pygram.python_grammar_no_print_statement_no_exec_statement_async_keywords,
783             # Python 3.0-3.6
784             pygram.python_grammar_no_print_statement_no_exec_statement,
785             # Python 2.7 with future print_function import
786             pygram.python_grammar_no_print_statement,
787             # Python 2.7
788             pygram.python_grammar,
789         ]
790
791     if all(version.is_python2() for version in target_versions):
792         # Python 2-only code, so try Python 2 grammars.
793         return [
794             # Python 2.7 with future print_function import
795             pygram.python_grammar_no_print_statement,
796             # Python 2.7
797             pygram.python_grammar,
798         ]
799
800     # Python 3-compatible code, so only try Python 3 grammar.
801     grammars = []
802     # If we have to parse both, try to parse async as a keyword first
803     if not supports_feature(target_versions, Feature.ASYNC_IDENTIFIERS):
804         # Python 3.7+
805         grammars.append(
806             pygram.python_grammar_no_print_statement_no_exec_statement_async_keywords
807         )
808     if not supports_feature(target_versions, Feature.ASYNC_KEYWORDS):
809         # Python 3.0-3.6
810         grammars.append(pygram.python_grammar_no_print_statement_no_exec_statement)
811     # At least one of the above branches must have been taken, because every Python
812     # version has exactly one of the two 'ASYNC_*' flags
813     return grammars
814
815
816 def lib2to3_parse(src_txt: str, target_versions: Iterable[TargetVersion] = ()) -> Node:
817     """Given a string with source, return the lib2to3 Node."""
818     if src_txt[-1:] != "\n":
819         src_txt += "\n"
820
821     for grammar in get_grammars(set(target_versions)):
822         drv = driver.Driver(grammar, pytree.convert)
823         try:
824             result = drv.parse_string(src_txt, True)
825             break
826
827         except ParseError as pe:
828             lineno, column = pe.context[1]
829             lines = src_txt.splitlines()
830             try:
831                 faulty_line = lines[lineno - 1]
832             except IndexError:
833                 faulty_line = "<line number missing in source>"
834             exc = InvalidInput(f"Cannot parse: {lineno}:{column}: {faulty_line}")
835     else:
836         raise exc from None
837
838     if isinstance(result, Leaf):
839         result = Node(syms.file_input, [result])
840     return result
841
842
843 def lib2to3_unparse(node: Node) -> str:
844     """Given a lib2to3 node, return its string representation."""
845     code = str(node)
846     return code
847
848
849 T = TypeVar("T")
850
851
852 class Visitor(Generic[T]):
853     """Basic lib2to3 visitor that yields things of type `T` on `visit()`."""
854
855     def visit(self, node: LN) -> Iterator[T]:
856         """Main method to visit `node` and its children.
857
858         It tries to find a `visit_*()` method for the given `node.type`, like
859         `visit_simple_stmt` for Node objects or `visit_INDENT` for Leaf objects.
860         If no dedicated `visit_*()` method is found, chooses `visit_default()`
861         instead.
862
863         Then yields objects of type `T` from the selected visitor.
864         """
865         if node.type < 256:
866             name = token.tok_name[node.type]
867         else:
868             name = type_repr(node.type)
869         yield from getattr(self, f"visit_{name}", self.visit_default)(node)
870
871     def visit_default(self, node: LN) -> Iterator[T]:
872         """Default `visit_*()` implementation. Recurses to children of `node`."""
873         if isinstance(node, Node):
874             for child in node.children:
875                 yield from self.visit(child)
876
877
878 @dataclass
879 class DebugVisitor(Visitor[T]):
880     tree_depth: int = 0
881
882     def visit_default(self, node: LN) -> Iterator[T]:
883         indent = " " * (2 * self.tree_depth)
884         if isinstance(node, Node):
885             _type = type_repr(node.type)
886             out(f"{indent}{_type}", fg="yellow")
887             self.tree_depth += 1
888             for child in node.children:
889                 yield from self.visit(child)
890
891             self.tree_depth -= 1
892             out(f"{indent}/{_type}", fg="yellow", bold=False)
893         else:
894             _type = token.tok_name.get(node.type, str(node.type))
895             out(f"{indent}{_type}", fg="blue", nl=False)
896             if node.prefix:
897                 # We don't have to handle prefixes for `Node` objects since
898                 # that delegates to the first child anyway.
899                 out(f" {node.prefix!r}", fg="green", bold=False, nl=False)
900             out(f" {node.value!r}", fg="blue", bold=False)
901
902     @classmethod
903     def show(cls, code: Union[str, Leaf, Node]) -> None:
904         """Pretty-print the lib2to3 AST of a given string of `code`.
905
906         Convenience method for debugging.
907         """
908         v: DebugVisitor[None] = DebugVisitor()
909         if isinstance(code, str):
910             code = lib2to3_parse(code)
911         list(v.visit(code))
912
913
914 WHITESPACE = {token.DEDENT, token.INDENT, token.NEWLINE}
915 STATEMENT = {
916     syms.if_stmt,
917     syms.while_stmt,
918     syms.for_stmt,
919     syms.try_stmt,
920     syms.except_clause,
921     syms.with_stmt,
922     syms.funcdef,
923     syms.classdef,
924 }
925 STANDALONE_COMMENT = 153
926 token.tok_name[STANDALONE_COMMENT] = "STANDALONE_COMMENT"
927 LOGIC_OPERATORS = {"and", "or"}
928 COMPARATORS = {
929     token.LESS,
930     token.GREATER,
931     token.EQEQUAL,
932     token.NOTEQUAL,
933     token.LESSEQUAL,
934     token.GREATEREQUAL,
935 }
936 MATH_OPERATORS = {
937     token.VBAR,
938     token.CIRCUMFLEX,
939     token.AMPER,
940     token.LEFTSHIFT,
941     token.RIGHTSHIFT,
942     token.PLUS,
943     token.MINUS,
944     token.STAR,
945     token.SLASH,
946     token.DOUBLESLASH,
947     token.PERCENT,
948     token.AT,
949     token.TILDE,
950     token.DOUBLESTAR,
951 }
952 STARS = {token.STAR, token.DOUBLESTAR}
953 VARARGS_SPECIALS = STARS | {token.SLASH}
954 VARARGS_PARENTS = {
955     syms.arglist,
956     syms.argument,  # double star in arglist
957     syms.trailer,  # single argument to call
958     syms.typedargslist,
959     syms.varargslist,  # lambdas
960 }
961 UNPACKING_PARENTS = {
962     syms.atom,  # single element of a list or set literal
963     syms.dictsetmaker,
964     syms.listmaker,
965     syms.testlist_gexp,
966     syms.testlist_star_expr,
967 }
968 TEST_DESCENDANTS = {
969     syms.test,
970     syms.lambdef,
971     syms.or_test,
972     syms.and_test,
973     syms.not_test,
974     syms.comparison,
975     syms.star_expr,
976     syms.expr,
977     syms.xor_expr,
978     syms.and_expr,
979     syms.shift_expr,
980     syms.arith_expr,
981     syms.trailer,
982     syms.term,
983     syms.power,
984 }
985 ASSIGNMENTS = {
986     "=",
987     "+=",
988     "-=",
989     "*=",
990     "@=",
991     "/=",
992     "%=",
993     "&=",
994     "|=",
995     "^=",
996     "<<=",
997     ">>=",
998     "**=",
999     "//=",
1000 }
1001 COMPREHENSION_PRIORITY = 20
1002 COMMA_PRIORITY = 18
1003 TERNARY_PRIORITY = 16
1004 LOGIC_PRIORITY = 14
1005 STRING_PRIORITY = 12
1006 COMPARATOR_PRIORITY = 10
1007 MATH_PRIORITIES = {
1008     token.VBAR: 9,
1009     token.CIRCUMFLEX: 8,
1010     token.AMPER: 7,
1011     token.LEFTSHIFT: 6,
1012     token.RIGHTSHIFT: 6,
1013     token.PLUS: 5,
1014     token.MINUS: 5,
1015     token.STAR: 4,
1016     token.SLASH: 4,
1017     token.DOUBLESLASH: 4,
1018     token.PERCENT: 4,
1019     token.AT: 4,
1020     token.TILDE: 3,
1021     token.DOUBLESTAR: 2,
1022 }
1023 DOT_PRIORITY = 1
1024
1025
1026 @dataclass
1027 class BracketTracker:
1028     """Keeps track of brackets on a line."""
1029
1030     depth: int = 0
1031     bracket_match: Dict[Tuple[Depth, NodeType], Leaf] = Factory(dict)
1032     delimiters: Dict[LeafID, Priority] = Factory(dict)
1033     previous: Optional[Leaf] = None
1034     _for_loop_depths: List[int] = Factory(list)
1035     _lambda_argument_depths: List[int] = Factory(list)
1036
1037     def mark(self, leaf: Leaf) -> None:
1038         """Mark `leaf` with bracket-related metadata. Keep track of delimiters.
1039
1040         All leaves receive an int `bracket_depth` field that stores how deep
1041         within brackets a given leaf is. 0 means there are no enclosing brackets
1042         that started on this line.
1043
1044         If a leaf is itself a closing bracket, it receives an `opening_bracket`
1045         field that it forms a pair with. This is a one-directional link to
1046         avoid reference cycles.
1047
1048         If a leaf is a delimiter (a token on which Black can split the line if
1049         needed) and it's on depth 0, its `id()` is stored in the tracker's
1050         `delimiters` field.
1051         """
1052         if leaf.type == token.COMMENT:
1053             return
1054
1055         self.maybe_decrement_after_for_loop_variable(leaf)
1056         self.maybe_decrement_after_lambda_arguments(leaf)
1057         if leaf.type in CLOSING_BRACKETS:
1058             self.depth -= 1
1059             opening_bracket = self.bracket_match.pop((self.depth, leaf.type))
1060             leaf.opening_bracket = opening_bracket
1061         leaf.bracket_depth = self.depth
1062         if self.depth == 0:
1063             delim = is_split_before_delimiter(leaf, self.previous)
1064             if delim and self.previous is not None:
1065                 self.delimiters[id(self.previous)] = delim
1066             else:
1067                 delim = is_split_after_delimiter(leaf, self.previous)
1068                 if delim:
1069                     self.delimiters[id(leaf)] = delim
1070         if leaf.type in OPENING_BRACKETS:
1071             self.bracket_match[self.depth, BRACKET[leaf.type]] = leaf
1072             self.depth += 1
1073         self.previous = leaf
1074         self.maybe_increment_lambda_arguments(leaf)
1075         self.maybe_increment_for_loop_variable(leaf)
1076
1077     def any_open_brackets(self) -> bool:
1078         """Return True if there is an yet unmatched open bracket on the line."""
1079         return bool(self.bracket_match)
1080
1081     def max_delimiter_priority(self, exclude: Iterable[LeafID] = ()) -> Priority:
1082         """Return the highest priority of a delimiter found on the line.
1083
1084         Values are consistent with what `is_split_*_delimiter()` return.
1085         Raises ValueError on no delimiters.
1086         """
1087         return max(v for k, v in self.delimiters.items() if k not in exclude)
1088
1089     def delimiter_count_with_priority(self, priority: Priority = 0) -> int:
1090         """Return the number of delimiters with the given `priority`.
1091
1092         If no `priority` is passed, defaults to max priority on the line.
1093         """
1094         if not self.delimiters:
1095             return 0
1096
1097         priority = priority or self.max_delimiter_priority()
1098         return sum(1 for p in self.delimiters.values() if p == priority)
1099
1100     def maybe_increment_for_loop_variable(self, leaf: Leaf) -> bool:
1101         """In a for loop, or comprehension, the variables are often unpacks.
1102
1103         To avoid splitting on the comma in this situation, increase the depth of
1104         tokens between `for` and `in`.
1105         """
1106         if leaf.type == token.NAME and leaf.value == "for":
1107             self.depth += 1
1108             self._for_loop_depths.append(self.depth)
1109             return True
1110
1111         return False
1112
1113     def maybe_decrement_after_for_loop_variable(self, leaf: Leaf) -> bool:
1114         """See `maybe_increment_for_loop_variable` above for explanation."""
1115         if (
1116             self._for_loop_depths
1117             and self._for_loop_depths[-1] == self.depth
1118             and leaf.type == token.NAME
1119             and leaf.value == "in"
1120         ):
1121             self.depth -= 1
1122             self._for_loop_depths.pop()
1123             return True
1124
1125         return False
1126
1127     def maybe_increment_lambda_arguments(self, leaf: Leaf) -> bool:
1128         """In a lambda expression, there might be more than one argument.
1129
1130         To avoid splitting on the comma in this situation, increase the depth of
1131         tokens between `lambda` and `:`.
1132         """
1133         if leaf.type == token.NAME and leaf.value == "lambda":
1134             self.depth += 1
1135             self._lambda_argument_depths.append(self.depth)
1136             return True
1137
1138         return False
1139
1140     def maybe_decrement_after_lambda_arguments(self, leaf: Leaf) -> bool:
1141         """See `maybe_increment_lambda_arguments` above for explanation."""
1142         if (
1143             self._lambda_argument_depths
1144             and self._lambda_argument_depths[-1] == self.depth
1145             and leaf.type == token.COLON
1146         ):
1147             self.depth -= 1
1148             self._lambda_argument_depths.pop()
1149             return True
1150
1151         return False
1152
1153     def get_open_lsqb(self) -> Optional[Leaf]:
1154         """Return the most recent opening square bracket (if any)."""
1155         return self.bracket_match.get((self.depth - 1, token.RSQB))
1156
1157
1158 @dataclass
1159 class Line:
1160     """Holds leaves and comments. Can be printed with `str(line)`."""
1161
1162     depth: int = 0
1163     leaves: List[Leaf] = Factory(list)
1164     comments: Dict[LeafID, List[Leaf]] = Factory(dict)  # keys ordered like `leaves`
1165     bracket_tracker: BracketTracker = Factory(BracketTracker)
1166     inside_brackets: bool = False
1167     should_explode: bool = False
1168
1169     def append(self, leaf: Leaf, preformatted: bool = False) -> None:
1170         """Add a new `leaf` to the end of the line.
1171
1172         Unless `preformatted` is True, the `leaf` will receive a new consistent
1173         whitespace prefix and metadata applied by :class:`BracketTracker`.
1174         Trailing commas are maybe removed, unpacked for loop variables are
1175         demoted from being delimiters.
1176
1177         Inline comments are put aside.
1178         """
1179         has_value = leaf.type in BRACKETS or bool(leaf.value.strip())
1180         if not has_value:
1181             return
1182
1183         if token.COLON == leaf.type and self.is_class_paren_empty:
1184             del self.leaves[-2:]
1185         if self.leaves and not preformatted:
1186             # Note: at this point leaf.prefix should be empty except for
1187             # imports, for which we only preserve newlines.
1188             leaf.prefix += whitespace(
1189                 leaf, complex_subscript=self.is_complex_subscript(leaf)
1190             )
1191         if self.inside_brackets or not preformatted:
1192             self.bracket_tracker.mark(leaf)
1193             self.maybe_remove_trailing_comma(leaf)
1194         if not self.append_comment(leaf):
1195             self.leaves.append(leaf)
1196
1197     def append_safe(self, leaf: Leaf, preformatted: bool = False) -> None:
1198         """Like :func:`append()` but disallow invalid standalone comment structure.
1199
1200         Raises ValueError when any `leaf` is appended after a standalone comment
1201         or when a standalone comment is not the first leaf on the line.
1202         """
1203         if self.bracket_tracker.depth == 0:
1204             if self.is_comment:
1205                 raise ValueError("cannot append to standalone comments")
1206
1207             if self.leaves and leaf.type == STANDALONE_COMMENT:
1208                 raise ValueError(
1209                     "cannot append standalone comments to a populated line"
1210                 )
1211
1212         self.append(leaf, preformatted=preformatted)
1213
1214     @property
1215     def is_comment(self) -> bool:
1216         """Is this line a standalone comment?"""
1217         return len(self.leaves) == 1 and self.leaves[0].type == STANDALONE_COMMENT
1218
1219     @property
1220     def is_decorator(self) -> bool:
1221         """Is this line a decorator?"""
1222         return bool(self) and self.leaves[0].type == token.AT
1223
1224     @property
1225     def is_import(self) -> bool:
1226         """Is this an import line?"""
1227         return bool(self) and is_import(self.leaves[0])
1228
1229     @property
1230     def is_class(self) -> bool:
1231         """Is this line a class definition?"""
1232         return (
1233             bool(self)
1234             and self.leaves[0].type == token.NAME
1235             and self.leaves[0].value == "class"
1236         )
1237
1238     @property
1239     def is_stub_class(self) -> bool:
1240         """Is this line a class definition with a body consisting only of "..."?"""
1241         return self.is_class and self.leaves[-3:] == [
1242             Leaf(token.DOT, ".") for _ in range(3)
1243         ]
1244
1245     @property
1246     def is_collection_with_optional_trailing_comma(self) -> bool:
1247         """Is this line a collection literal with a trailing comma that's optional?
1248
1249         Note that the trailing comma in a 1-tuple is not optional.
1250         """
1251         if not self.leaves or len(self.leaves) < 4:
1252             return False
1253
1254         # Look for and address a trailing colon.
1255         if self.leaves[-1].type == token.COLON:
1256             closer = self.leaves[-2]
1257             close_index = -2
1258         else:
1259             closer = self.leaves[-1]
1260             close_index = -1
1261         if closer.type not in CLOSING_BRACKETS or self.inside_brackets:
1262             return False
1263
1264         if closer.type == token.RPAR:
1265             # Tuples require an extra check, because if there's only
1266             # one element in the tuple removing the comma unmakes the
1267             # tuple.
1268             #
1269             # We also check for parens before looking for the trailing
1270             # comma because in some cases (eg assigning a dict
1271             # literal) the literal gets wrapped in temporary parens
1272             # during parsing. This case is covered by the
1273             # collections.py test data.
1274             opener = closer.opening_bracket
1275             for _open_index, leaf in enumerate(self.leaves):
1276                 if leaf is opener:
1277                     break
1278
1279             else:
1280                 # Couldn't find the matching opening paren, play it safe.
1281                 return False
1282
1283             commas = 0
1284             comma_depth = self.leaves[close_index - 1].bracket_depth
1285             for leaf in self.leaves[_open_index + 1 : close_index]:
1286                 if leaf.bracket_depth == comma_depth and leaf.type == token.COMMA:
1287                     commas += 1
1288             if commas > 1:
1289                 # We haven't looked yet for the trailing comma because
1290                 # we might also have caught noop parens.
1291                 return self.leaves[close_index - 1].type == token.COMMA
1292
1293             elif commas == 1:
1294                 return False  # it's either a one-tuple or didn't have a trailing comma
1295
1296             if self.leaves[close_index - 1].type in CLOSING_BRACKETS:
1297                 close_index -= 1
1298                 closer = self.leaves[close_index]
1299                 if closer.type == token.RPAR:
1300                     # TODO: this is a gut feeling. Will we ever see this?
1301                     return False
1302
1303         if self.leaves[close_index - 1].type != token.COMMA:
1304             return False
1305
1306         return True
1307
1308     @property
1309     def is_def(self) -> bool:
1310         """Is this a function definition? (Also returns True for async defs.)"""
1311         try:
1312             first_leaf = self.leaves[0]
1313         except IndexError:
1314             return False
1315
1316         try:
1317             second_leaf: Optional[Leaf] = self.leaves[1]
1318         except IndexError:
1319             second_leaf = None
1320         return (first_leaf.type == token.NAME and first_leaf.value == "def") or (
1321             first_leaf.type == token.ASYNC
1322             and second_leaf is not None
1323             and second_leaf.type == token.NAME
1324             and second_leaf.value == "def"
1325         )
1326
1327     @property
1328     def is_class_paren_empty(self) -> bool:
1329         """Is this a class with no base classes but using parentheses?
1330
1331         Those are unnecessary and should be removed.
1332         """
1333         return (
1334             bool(self)
1335             and len(self.leaves) == 4
1336             and self.is_class
1337             and self.leaves[2].type == token.LPAR
1338             and self.leaves[2].value == "("
1339             and self.leaves[3].type == token.RPAR
1340             and self.leaves[3].value == ")"
1341         )
1342
1343     @property
1344     def is_triple_quoted_string(self) -> bool:
1345         """Is the line a triple quoted string?"""
1346         return (
1347             bool(self)
1348             and self.leaves[0].type == token.STRING
1349             and self.leaves[0].value.startswith(('"""', "'''"))
1350         )
1351
1352     def contains_standalone_comments(self, depth_limit: int = sys.maxsize) -> bool:
1353         """If so, needs to be split before emitting."""
1354         for leaf in self.leaves:
1355             if leaf.type == STANDALONE_COMMENT and leaf.bracket_depth <= depth_limit:
1356                 return True
1357
1358         return False
1359
1360     def contains_uncollapsable_type_comments(self) -> bool:
1361         ignored_ids = set()
1362         try:
1363             last_leaf = self.leaves[-1]
1364             ignored_ids.add(id(last_leaf))
1365             if last_leaf.type == token.COMMA or (
1366                 last_leaf.type == token.RPAR and not last_leaf.value
1367             ):
1368                 # When trailing commas or optional parens are inserted by Black for
1369                 # consistency, comments after the previous last element are not moved
1370                 # (they don't have to, rendering will still be correct).  So we ignore
1371                 # trailing commas and invisible.
1372                 last_leaf = self.leaves[-2]
1373                 ignored_ids.add(id(last_leaf))
1374         except IndexError:
1375             return False
1376
1377         # A type comment is uncollapsable if it is attached to a leaf
1378         # that isn't at the end of the line (since that could cause it
1379         # to get associated to a different argument) or if there are
1380         # comments before it (since that could cause it to get hidden
1381         # behind a comment.
1382         comment_seen = False
1383         for leaf_id, comments in self.comments.items():
1384             for comment in comments:
1385                 if is_type_comment(comment):
1386                     if leaf_id not in ignored_ids or comment_seen:
1387                         return True
1388
1389                 comment_seen = True
1390
1391         return False
1392
1393     def contains_unsplittable_type_ignore(self) -> bool:
1394         if not self.leaves:
1395             return False
1396
1397         # If a 'type: ignore' is attached to the end of a line, we
1398         # can't split the line, because we can't know which of the
1399         # subexpressions the ignore was meant to apply to.
1400         #
1401         # We only want this to apply to actual physical lines from the
1402         # original source, though: we don't want the presence of a
1403         # 'type: ignore' at the end of a multiline expression to
1404         # justify pushing it all onto one line. Thus we
1405         # (unfortunately) need to check the actual source lines and
1406         # only report an unsplittable 'type: ignore' if this line was
1407         # one line in the original code.
1408
1409         # Grab the first and last line numbers, skipping generated leaves
1410         first_line = next((l.lineno for l in self.leaves if l.lineno != 0), 0)
1411         last_line = next((l.lineno for l in reversed(self.leaves) if l.lineno != 0), 0)
1412
1413         if first_line == last_line:
1414             # We look at the last two leaves since a comma or an
1415             # invisible paren could have been added at the end of the
1416             # line.
1417             for node in self.leaves[-2:]:
1418                 for comment in self.comments.get(id(node), []):
1419                     if is_type_comment(comment, " ignore"):
1420                         return True
1421
1422         return False
1423
1424     def contains_multiline_strings(self) -> bool:
1425         for leaf in self.leaves:
1426             if is_multiline_string(leaf):
1427                 return True
1428
1429         return False
1430
1431     def maybe_remove_trailing_comma(self, closing: Leaf) -> bool:
1432         """Remove trailing comma if there is one and it's safe."""
1433         if not (self.leaves and self.leaves[-1].type == token.COMMA):
1434             return False
1435
1436         # We remove trailing commas only in the case of importing a
1437         # single name from a module.
1438         if not (
1439             self.leaves
1440             and self.is_import
1441             and len(self.leaves) > 4
1442             and self.leaves[-1].type == token.COMMA
1443             and closing.type in CLOSING_BRACKETS
1444             and self.leaves[-4].type == token.NAME
1445             and (
1446                 # regular `from foo import bar,`
1447                 self.leaves[-4].value == "import"
1448                 # `from foo import (bar as baz,)
1449                 or (
1450                     len(self.leaves) > 6
1451                     and self.leaves[-6].value == "import"
1452                     and self.leaves[-3].value == "as"
1453                 )
1454                 # `from foo import bar as baz,`
1455                 or (
1456                     len(self.leaves) > 5
1457                     and self.leaves[-5].value == "import"
1458                     and self.leaves[-3].value == "as"
1459                 )
1460             )
1461             and closing.type == token.RPAR
1462         ):
1463             return False
1464
1465         self.remove_trailing_comma()
1466         return True
1467
1468     def append_comment(self, comment: Leaf) -> bool:
1469         """Add an inline or standalone comment to the line."""
1470         if (
1471             comment.type == STANDALONE_COMMENT
1472             and self.bracket_tracker.any_open_brackets()
1473         ):
1474             comment.prefix = ""
1475             return False
1476
1477         if comment.type != token.COMMENT:
1478             return False
1479
1480         if not self.leaves:
1481             comment.type = STANDALONE_COMMENT
1482             comment.prefix = ""
1483             return False
1484
1485         last_leaf = self.leaves[-1]
1486         if (
1487             last_leaf.type == token.RPAR
1488             and not last_leaf.value
1489             and last_leaf.parent
1490             and len(list(last_leaf.parent.leaves())) <= 3
1491             and not is_type_comment(comment)
1492         ):
1493             # Comments on an optional parens wrapping a single leaf should belong to
1494             # the wrapped node except if it's a type comment. Pinning the comment like
1495             # this avoids unstable formatting caused by comment migration.
1496             if len(self.leaves) < 2:
1497                 comment.type = STANDALONE_COMMENT
1498                 comment.prefix = ""
1499                 return False
1500
1501             last_leaf = self.leaves[-2]
1502         self.comments.setdefault(id(last_leaf), []).append(comment)
1503         return True
1504
1505     def comments_after(self, leaf: Leaf) -> List[Leaf]:
1506         """Generate comments that should appear directly after `leaf`."""
1507         return self.comments.get(id(leaf), [])
1508
1509     def remove_trailing_comma(self) -> None:
1510         """Remove the trailing comma and moves the comments attached to it."""
1511         trailing_comma = self.leaves.pop()
1512         trailing_comma_comments = self.comments.pop(id(trailing_comma), [])
1513         self.comments.setdefault(id(self.leaves[-1]), []).extend(
1514             trailing_comma_comments
1515         )
1516
1517     def is_complex_subscript(self, leaf: Leaf) -> bool:
1518         """Return True iff `leaf` is part of a slice with non-trivial exprs."""
1519         open_lsqb = self.bracket_tracker.get_open_lsqb()
1520         if open_lsqb is None:
1521             return False
1522
1523         subscript_start = open_lsqb.next_sibling
1524
1525         if isinstance(subscript_start, Node):
1526             if subscript_start.type == syms.listmaker:
1527                 return False
1528
1529             if subscript_start.type == syms.subscriptlist:
1530                 subscript_start = child_towards(subscript_start, leaf)
1531         return subscript_start is not None and any(
1532             n.type in TEST_DESCENDANTS for n in subscript_start.pre_order()
1533         )
1534
1535     def __str__(self) -> str:
1536         """Render the line."""
1537         if not self:
1538             return "\n"
1539
1540         indent = "    " * self.depth
1541         leaves = iter(self.leaves)
1542         first = next(leaves)
1543         res = f"{first.prefix}{indent}{first.value}"
1544         for leaf in leaves:
1545             res += str(leaf)
1546         for comment in itertools.chain.from_iterable(self.comments.values()):
1547             res += str(comment)
1548         return res + "\n"
1549
1550     def __bool__(self) -> bool:
1551         """Return True if the line has leaves or comments."""
1552         return bool(self.leaves or self.comments)
1553
1554
1555 @dataclass
1556 class EmptyLineTracker:
1557     """Provides a stateful method that returns the number of potential extra
1558     empty lines needed before and after the currently processed line.
1559
1560     Note: this tracker works on lines that haven't been split yet.  It assumes
1561     the prefix of the first leaf consists of optional newlines.  Those newlines
1562     are consumed by `maybe_empty_lines()` and included in the computation.
1563     """
1564
1565     is_pyi: bool = False
1566     previous_line: Optional[Line] = None
1567     previous_after: int = 0
1568     previous_defs: List[int] = Factory(list)
1569
1570     def maybe_empty_lines(self, current_line: Line) -> Tuple[int, int]:
1571         """Return the number of extra empty lines before and after the `current_line`.
1572
1573         This is for separating `def`, `async def` and `class` with extra empty
1574         lines (two on module-level).
1575         """
1576         before, after = self._maybe_empty_lines(current_line)
1577         before = (
1578             # Black should not insert empty lines at the beginning
1579             # of the file
1580             0
1581             if self.previous_line is None
1582             else before - self.previous_after
1583         )
1584         self.previous_after = after
1585         self.previous_line = current_line
1586         return before, after
1587
1588     def _maybe_empty_lines(self, current_line: Line) -> Tuple[int, int]:
1589         max_allowed = 1
1590         if current_line.depth == 0:
1591             max_allowed = 1 if self.is_pyi else 2
1592         if current_line.leaves:
1593             # Consume the first leaf's extra newlines.
1594             first_leaf = current_line.leaves[0]
1595             before = first_leaf.prefix.count("\n")
1596             before = min(before, max_allowed)
1597             first_leaf.prefix = ""
1598         else:
1599             before = 0
1600         depth = current_line.depth
1601         while self.previous_defs and self.previous_defs[-1] >= depth:
1602             self.previous_defs.pop()
1603             if self.is_pyi:
1604                 before = 0 if depth else 1
1605             else:
1606                 before = 1 if depth else 2
1607         if current_line.is_decorator or current_line.is_def or current_line.is_class:
1608             return self._maybe_empty_lines_for_class_or_def(current_line, before)
1609
1610         if (
1611             self.previous_line
1612             and self.previous_line.is_import
1613             and not current_line.is_import
1614             and depth == self.previous_line.depth
1615         ):
1616             return (before or 1), 0
1617
1618         if (
1619             self.previous_line
1620             and self.previous_line.is_class
1621             and current_line.is_triple_quoted_string
1622         ):
1623             return before, 1
1624
1625         return before, 0
1626
1627     def _maybe_empty_lines_for_class_or_def(
1628         self, current_line: Line, before: int
1629     ) -> Tuple[int, int]:
1630         if not current_line.is_decorator:
1631             self.previous_defs.append(current_line.depth)
1632         if self.previous_line is None:
1633             # Don't insert empty lines before the first line in the file.
1634             return 0, 0
1635
1636         if self.previous_line.is_decorator:
1637             return 0, 0
1638
1639         if self.previous_line.depth < current_line.depth and (
1640             self.previous_line.is_class or self.previous_line.is_def
1641         ):
1642             return 0, 0
1643
1644         if (
1645             self.previous_line.is_comment
1646             and self.previous_line.depth == current_line.depth
1647             and before == 0
1648         ):
1649             return 0, 0
1650
1651         if self.is_pyi:
1652             if self.previous_line.depth > current_line.depth:
1653                 newlines = 1
1654             elif current_line.is_class or self.previous_line.is_class:
1655                 if current_line.is_stub_class and self.previous_line.is_stub_class:
1656                     # No blank line between classes with an empty body
1657                     newlines = 0
1658                 else:
1659                     newlines = 1
1660             elif current_line.is_def and not self.previous_line.is_def:
1661                 # Blank line between a block of functions and a block of non-functions
1662                 newlines = 1
1663             else:
1664                 newlines = 0
1665         else:
1666             newlines = 2
1667         if current_line.depth and newlines:
1668             newlines -= 1
1669         return newlines, 0
1670
1671
1672 @dataclass
1673 class LineGenerator(Visitor[Line]):
1674     """Generates reformatted Line objects.  Empty lines are not emitted.
1675
1676     Note: destroys the tree it's visiting by mutating prefixes of its leaves
1677     in ways that will no longer stringify to valid Python code on the tree.
1678     """
1679
1680     is_pyi: bool = False
1681     normalize_strings: bool = True
1682     current_line: Line = Factory(Line)
1683     remove_u_prefix: bool = False
1684
1685     def line(self, indent: int = 0) -> Iterator[Line]:
1686         """Generate a line.
1687
1688         If the line is empty, only emit if it makes sense.
1689         If the line is too long, split it first and then generate.
1690
1691         If any lines were generated, set up a new current_line.
1692         """
1693         if not self.current_line:
1694             self.current_line.depth += indent
1695             return  # Line is empty, don't emit. Creating a new one unnecessary.
1696
1697         complete_line = self.current_line
1698         self.current_line = Line(depth=complete_line.depth + indent)
1699         yield complete_line
1700
1701     def visit_default(self, node: LN) -> Iterator[Line]:
1702         """Default `visit_*()` implementation. Recurses to children of `node`."""
1703         if isinstance(node, Leaf):
1704             any_open_brackets = self.current_line.bracket_tracker.any_open_brackets()
1705             for comment in generate_comments(node):
1706                 if any_open_brackets:
1707                     # any comment within brackets is subject to splitting
1708                     self.current_line.append(comment)
1709                 elif comment.type == token.COMMENT:
1710                     # regular trailing comment
1711                     self.current_line.append(comment)
1712                     yield from self.line()
1713
1714                 else:
1715                     # regular standalone comment
1716                     yield from self.line()
1717
1718                     self.current_line.append(comment)
1719                     yield from self.line()
1720
1721             normalize_prefix(node, inside_brackets=any_open_brackets)
1722             if self.normalize_strings and node.type == token.STRING:
1723                 normalize_string_prefix(node, remove_u_prefix=self.remove_u_prefix)
1724                 normalize_string_quotes(node)
1725             if node.type == token.NUMBER:
1726                 normalize_numeric_literal(node)
1727             if node.type not in WHITESPACE:
1728                 self.current_line.append(node)
1729         yield from super().visit_default(node)
1730
1731     def visit_INDENT(self, node: Node) -> Iterator[Line]:
1732         """Increase indentation level, maybe yield a line."""
1733         # In blib2to3 INDENT never holds comments.
1734         yield from self.line(+1)
1735         yield from self.visit_default(node)
1736
1737     def visit_DEDENT(self, node: Node) -> Iterator[Line]:
1738         """Decrease indentation level, maybe yield a line."""
1739         # The current line might still wait for trailing comments.  At DEDENT time
1740         # there won't be any (they would be prefixes on the preceding NEWLINE).
1741         # Emit the line then.
1742         yield from self.line()
1743
1744         # While DEDENT has no value, its prefix may contain standalone comments
1745         # that belong to the current indentation level.  Get 'em.
1746         yield from self.visit_default(node)
1747
1748         # Finally, emit the dedent.
1749         yield from self.line(-1)
1750
1751     def visit_stmt(
1752         self, node: Node, keywords: Set[str], parens: Set[str]
1753     ) -> Iterator[Line]:
1754         """Visit a statement.
1755
1756         This implementation is shared for `if`, `while`, `for`, `try`, `except`,
1757         `def`, `with`, `class`, `assert` and assignments.
1758
1759         The relevant Python language `keywords` for a given statement will be
1760         NAME leaves within it. This methods puts those on a separate line.
1761
1762         `parens` holds a set of string leaf values immediately after which
1763         invisible parens should be put.
1764         """
1765         normalize_invisible_parens(node, parens_after=parens)
1766         for child in node.children:
1767             if child.type == token.NAME and child.value in keywords:  # type: ignore
1768                 yield from self.line()
1769
1770             yield from self.visit(child)
1771
1772     def visit_suite(self, node: Node) -> Iterator[Line]:
1773         """Visit a suite."""
1774         if self.is_pyi and is_stub_suite(node):
1775             yield from self.visit(node.children[2])
1776         else:
1777             yield from self.visit_default(node)
1778
1779     def visit_simple_stmt(self, node: Node) -> Iterator[Line]:
1780         """Visit a statement without nested statements."""
1781         is_suite_like = node.parent and node.parent.type in STATEMENT
1782         if is_suite_like:
1783             if self.is_pyi and is_stub_body(node):
1784                 yield from self.visit_default(node)
1785             else:
1786                 yield from self.line(+1)
1787                 yield from self.visit_default(node)
1788                 yield from self.line(-1)
1789
1790         else:
1791             if not self.is_pyi or not node.parent or not is_stub_suite(node.parent):
1792                 yield from self.line()
1793             yield from self.visit_default(node)
1794
1795     def visit_async_stmt(self, node: Node) -> Iterator[Line]:
1796         """Visit `async def`, `async for`, `async with`."""
1797         yield from self.line()
1798
1799         children = iter(node.children)
1800         for child in children:
1801             yield from self.visit(child)
1802
1803             if child.type == token.ASYNC:
1804                 break
1805
1806         internal_stmt = next(children)
1807         for child in internal_stmt.children:
1808             yield from self.visit(child)
1809
1810     def visit_decorators(self, node: Node) -> Iterator[Line]:
1811         """Visit decorators."""
1812         for child in node.children:
1813             yield from self.line()
1814             yield from self.visit(child)
1815
1816     def visit_SEMI(self, leaf: Leaf) -> Iterator[Line]:
1817         """Remove a semicolon and put the other statement on a separate line."""
1818         yield from self.line()
1819
1820     def visit_ENDMARKER(self, leaf: Leaf) -> Iterator[Line]:
1821         """End of file. Process outstanding comments and end with a newline."""
1822         yield from self.visit_default(leaf)
1823         yield from self.line()
1824
1825     def visit_STANDALONE_COMMENT(self, leaf: Leaf) -> Iterator[Line]:
1826         if not self.current_line.bracket_tracker.any_open_brackets():
1827             yield from self.line()
1828         yield from self.visit_default(leaf)
1829
1830     def visit_factor(self, node: Node) -> Iterator[Line]:
1831         """Force parentheses between a unary op and a binary power:
1832
1833         -2 ** 8 -> -(2 ** 8)
1834         """
1835         _operator, operand = node.children
1836         if (
1837             operand.type == syms.power
1838             and len(operand.children) == 3
1839             and operand.children[1].type == token.DOUBLESTAR
1840         ):
1841             lpar = Leaf(token.LPAR, "(")
1842             rpar = Leaf(token.RPAR, ")")
1843             index = operand.remove() or 0
1844             node.insert_child(index, Node(syms.atom, [lpar, operand, rpar]))
1845         yield from self.visit_default(node)
1846
1847     def __attrs_post_init__(self) -> None:
1848         """You are in a twisty little maze of passages."""
1849         v = self.visit_stmt
1850         Ø: Set[str] = set()
1851         self.visit_assert_stmt = partial(v, keywords={"assert"}, parens={"assert", ","})
1852         self.visit_if_stmt = partial(
1853             v, keywords={"if", "else", "elif"}, parens={"if", "elif"}
1854         )
1855         self.visit_while_stmt = partial(v, keywords={"while", "else"}, parens={"while"})
1856         self.visit_for_stmt = partial(v, keywords={"for", "else"}, parens={"for", "in"})
1857         self.visit_try_stmt = partial(
1858             v, keywords={"try", "except", "else", "finally"}, parens=Ø
1859         )
1860         self.visit_except_clause = partial(v, keywords={"except"}, parens=Ø)
1861         self.visit_with_stmt = partial(v, keywords={"with"}, parens=Ø)
1862         self.visit_funcdef = partial(v, keywords={"def"}, parens=Ø)
1863         self.visit_classdef = partial(v, keywords={"class"}, parens=Ø)
1864         self.visit_expr_stmt = partial(v, keywords=Ø, parens=ASSIGNMENTS)
1865         self.visit_return_stmt = partial(v, keywords={"return"}, parens={"return"})
1866         self.visit_import_from = partial(v, keywords=Ø, parens={"import"})
1867         self.visit_del_stmt = partial(v, keywords=Ø, parens={"del"})
1868         self.visit_async_funcdef = self.visit_async_stmt
1869         self.visit_decorated = self.visit_decorators
1870
1871
1872 IMPLICIT_TUPLE = {syms.testlist, syms.testlist_star_expr, syms.exprlist}
1873 BRACKET = {token.LPAR: token.RPAR, token.LSQB: token.RSQB, token.LBRACE: token.RBRACE}
1874 OPENING_BRACKETS = set(BRACKET.keys())
1875 CLOSING_BRACKETS = set(BRACKET.values())
1876 BRACKETS = OPENING_BRACKETS | CLOSING_BRACKETS
1877 ALWAYS_NO_SPACE = CLOSING_BRACKETS | {token.COMMA, STANDALONE_COMMENT}
1878
1879
1880 def whitespace(leaf: Leaf, *, complex_subscript: bool) -> str:  # noqa: C901
1881     """Return whitespace prefix if needed for the given `leaf`.
1882
1883     `complex_subscript` signals whether the given leaf is part of a subscription
1884     which has non-trivial arguments, like arithmetic expressions or function calls.
1885     """
1886     NO = ""
1887     SPACE = " "
1888     DOUBLESPACE = "  "
1889     t = leaf.type
1890     p = leaf.parent
1891     v = leaf.value
1892     if t in ALWAYS_NO_SPACE:
1893         return NO
1894
1895     if t == token.COMMENT:
1896         return DOUBLESPACE
1897
1898     assert p is not None, f"INTERNAL ERROR: hand-made leaf without parent: {leaf!r}"
1899     if t == token.COLON and p.type not in {
1900         syms.subscript,
1901         syms.subscriptlist,
1902         syms.sliceop,
1903     }:
1904         return NO
1905
1906     prev = leaf.prev_sibling
1907     if not prev:
1908         prevp = preceding_leaf(p)
1909         if not prevp or prevp.type in OPENING_BRACKETS:
1910             return NO
1911
1912         if t == token.COLON:
1913             if prevp.type == token.COLON:
1914                 return NO
1915
1916             elif prevp.type != token.COMMA and not complex_subscript:
1917                 return NO
1918
1919             return SPACE
1920
1921         if prevp.type == token.EQUAL:
1922             if prevp.parent:
1923                 if prevp.parent.type in {
1924                     syms.arglist,
1925                     syms.argument,
1926                     syms.parameters,
1927                     syms.varargslist,
1928                 }:
1929                     return NO
1930
1931                 elif prevp.parent.type == syms.typedargslist:
1932                     # A bit hacky: if the equal sign has whitespace, it means we
1933                     # previously found it's a typed argument.  So, we're using
1934                     # that, too.
1935                     return prevp.prefix
1936
1937         elif prevp.type in VARARGS_SPECIALS:
1938             if is_vararg(prevp, within=VARARGS_PARENTS | UNPACKING_PARENTS):
1939                 return NO
1940
1941         elif prevp.type == token.COLON:
1942             if prevp.parent and prevp.parent.type in {syms.subscript, syms.sliceop}:
1943                 return SPACE if complex_subscript else NO
1944
1945         elif (
1946             prevp.parent
1947             and prevp.parent.type == syms.factor
1948             and prevp.type in MATH_OPERATORS
1949         ):
1950             return NO
1951
1952         elif (
1953             prevp.type == token.RIGHTSHIFT
1954             and prevp.parent
1955             and prevp.parent.type == syms.shift_expr
1956             and prevp.prev_sibling
1957             and prevp.prev_sibling.type == token.NAME
1958             and prevp.prev_sibling.value == "print"  # type: ignore
1959         ):
1960             # Python 2 print chevron
1961             return NO
1962
1963     elif prev.type in OPENING_BRACKETS:
1964         return NO
1965
1966     if p.type in {syms.parameters, syms.arglist}:
1967         # untyped function signatures or calls
1968         if not prev or prev.type != token.COMMA:
1969             return NO
1970
1971     elif p.type == syms.varargslist:
1972         # lambdas
1973         if prev and prev.type != token.COMMA:
1974             return NO
1975
1976     elif p.type == syms.typedargslist:
1977         # typed function signatures
1978         if not prev:
1979             return NO
1980
1981         if t == token.EQUAL:
1982             if prev.type != syms.tname:
1983                 return NO
1984
1985         elif prev.type == token.EQUAL:
1986             # A bit hacky: if the equal sign has whitespace, it means we
1987             # previously found it's a typed argument.  So, we're using that, too.
1988             return prev.prefix
1989
1990         elif prev.type != token.COMMA:
1991             return NO
1992
1993     elif p.type == syms.tname:
1994         # type names
1995         if not prev:
1996             prevp = preceding_leaf(p)
1997             if not prevp or prevp.type != token.COMMA:
1998                 return NO
1999
2000     elif p.type == syms.trailer:
2001         # attributes and calls
2002         if t == token.LPAR or t == token.RPAR:
2003             return NO
2004
2005         if not prev:
2006             if t == token.DOT:
2007                 prevp = preceding_leaf(p)
2008                 if not prevp or prevp.type != token.NUMBER:
2009                     return NO
2010
2011             elif t == token.LSQB:
2012                 return NO
2013
2014         elif prev.type != token.COMMA:
2015             return NO
2016
2017     elif p.type == syms.argument:
2018         # single argument
2019         if t == token.EQUAL:
2020             return NO
2021
2022         if not prev:
2023             prevp = preceding_leaf(p)
2024             if not prevp or prevp.type == token.LPAR:
2025                 return NO
2026
2027         elif prev.type in {token.EQUAL} | VARARGS_SPECIALS:
2028             return NO
2029
2030     elif p.type == syms.decorator:
2031         # decorators
2032         return NO
2033
2034     elif p.type == syms.dotted_name:
2035         if prev:
2036             return NO
2037
2038         prevp = preceding_leaf(p)
2039         if not prevp or prevp.type == token.AT or prevp.type == token.DOT:
2040             return NO
2041
2042     elif p.type == syms.classdef:
2043         if t == token.LPAR:
2044             return NO
2045
2046         if prev and prev.type == token.LPAR:
2047             return NO
2048
2049     elif p.type in {syms.subscript, syms.sliceop}:
2050         # indexing
2051         if not prev:
2052             assert p.parent is not None, "subscripts are always parented"
2053             if p.parent.type == syms.subscriptlist:
2054                 return SPACE
2055
2056             return NO
2057
2058         elif not complex_subscript:
2059             return NO
2060
2061     elif p.type == syms.atom:
2062         if prev and t == token.DOT:
2063             # dots, but not the first one.
2064             return NO
2065
2066     elif p.type == syms.dictsetmaker:
2067         # dict unpacking
2068         if prev and prev.type == token.DOUBLESTAR:
2069             return NO
2070
2071     elif p.type in {syms.factor, syms.star_expr}:
2072         # unary ops
2073         if not prev:
2074             prevp = preceding_leaf(p)
2075             if not prevp or prevp.type in OPENING_BRACKETS:
2076                 return NO
2077
2078             prevp_parent = prevp.parent
2079             assert prevp_parent is not None
2080             if prevp.type == token.COLON and prevp_parent.type in {
2081                 syms.subscript,
2082                 syms.sliceop,
2083             }:
2084                 return NO
2085
2086             elif prevp.type == token.EQUAL and prevp_parent.type == syms.argument:
2087                 return NO
2088
2089         elif t in {token.NAME, token.NUMBER, token.STRING}:
2090             return NO
2091
2092     elif p.type == syms.import_from:
2093         if t == token.DOT:
2094             if prev and prev.type == token.DOT:
2095                 return NO
2096
2097         elif t == token.NAME:
2098             if v == "import":
2099                 return SPACE
2100
2101             if prev and prev.type == token.DOT:
2102                 return NO
2103
2104     elif p.type == syms.sliceop:
2105         return NO
2106
2107     return SPACE
2108
2109
2110 def preceding_leaf(node: Optional[LN]) -> Optional[Leaf]:
2111     """Return the first leaf that precedes `node`, if any."""
2112     while node:
2113         res = node.prev_sibling
2114         if res:
2115             if isinstance(res, Leaf):
2116                 return res
2117
2118             try:
2119                 return list(res.leaves())[-1]
2120
2121             except IndexError:
2122                 return None
2123
2124         node = node.parent
2125     return None
2126
2127
2128 def child_towards(ancestor: Node, descendant: LN) -> Optional[LN]:
2129     """Return the child of `ancestor` that contains `descendant`."""
2130     node: Optional[LN] = descendant
2131     while node and node.parent != ancestor:
2132         node = node.parent
2133     return node
2134
2135
2136 def container_of(leaf: Leaf) -> LN:
2137     """Return `leaf` or one of its ancestors that is the topmost container of it.
2138
2139     By "container" we mean a node where `leaf` is the very first child.
2140     """
2141     same_prefix = leaf.prefix
2142     container: LN = leaf
2143     while container:
2144         parent = container.parent
2145         if parent is None:
2146             break
2147
2148         if parent.children[0].prefix != same_prefix:
2149             break
2150
2151         if parent.type == syms.file_input:
2152             break
2153
2154         if parent.prev_sibling is not None and parent.prev_sibling.type in BRACKETS:
2155             break
2156
2157         container = parent
2158     return container
2159
2160
2161 def is_split_after_delimiter(leaf: Leaf, previous: Optional[Leaf] = None) -> Priority:
2162     """Return the priority of the `leaf` delimiter, given a line break after it.
2163
2164     The delimiter priorities returned here are from those delimiters that would
2165     cause a line break after themselves.
2166
2167     Higher numbers are higher priority.
2168     """
2169     if leaf.type == token.COMMA:
2170         return COMMA_PRIORITY
2171
2172     return 0
2173
2174
2175 def is_split_before_delimiter(leaf: Leaf, previous: Optional[Leaf] = None) -> Priority:
2176     """Return the priority of the `leaf` delimiter, given a line break before it.
2177
2178     The delimiter priorities returned here are from those delimiters that would
2179     cause a line break before themselves.
2180
2181     Higher numbers are higher priority.
2182     """
2183     if is_vararg(leaf, within=VARARGS_PARENTS | UNPACKING_PARENTS):
2184         # * and ** might also be MATH_OPERATORS but in this case they are not.
2185         # Don't treat them as a delimiter.
2186         return 0
2187
2188     if (
2189         leaf.type == token.DOT
2190         and leaf.parent
2191         and leaf.parent.type not in {syms.import_from, syms.dotted_name}
2192         and (previous is None or previous.type in CLOSING_BRACKETS)
2193     ):
2194         return DOT_PRIORITY
2195
2196     if (
2197         leaf.type in MATH_OPERATORS
2198         and leaf.parent
2199         and leaf.parent.type not in {syms.factor, syms.star_expr}
2200     ):
2201         return MATH_PRIORITIES[leaf.type]
2202
2203     if leaf.type in COMPARATORS:
2204         return COMPARATOR_PRIORITY
2205
2206     if (
2207         leaf.type == token.STRING
2208         and previous is not None
2209         and previous.type == token.STRING
2210     ):
2211         return STRING_PRIORITY
2212
2213     if leaf.type not in {token.NAME, token.ASYNC}:
2214         return 0
2215
2216     if (
2217         leaf.value == "for"
2218         and leaf.parent
2219         and leaf.parent.type in {syms.comp_for, syms.old_comp_for}
2220         or leaf.type == token.ASYNC
2221     ):
2222         if (
2223             not isinstance(leaf.prev_sibling, Leaf)
2224             or leaf.prev_sibling.value != "async"
2225         ):
2226             return COMPREHENSION_PRIORITY
2227
2228     if (
2229         leaf.value == "if"
2230         and leaf.parent
2231         and leaf.parent.type in {syms.comp_if, syms.old_comp_if}
2232     ):
2233         return COMPREHENSION_PRIORITY
2234
2235     if leaf.value in {"if", "else"} and leaf.parent and leaf.parent.type == syms.test:
2236         return TERNARY_PRIORITY
2237
2238     if leaf.value == "is":
2239         return COMPARATOR_PRIORITY
2240
2241     if (
2242         leaf.value == "in"
2243         and leaf.parent
2244         and leaf.parent.type in {syms.comp_op, syms.comparison}
2245         and not (
2246             previous is not None
2247             and previous.type == token.NAME
2248             and previous.value == "not"
2249         )
2250     ):
2251         return COMPARATOR_PRIORITY
2252
2253     if (
2254         leaf.value == "not"
2255         and leaf.parent
2256         and leaf.parent.type == syms.comp_op
2257         and not (
2258             previous is not None
2259             and previous.type == token.NAME
2260             and previous.value == "is"
2261         )
2262     ):
2263         return COMPARATOR_PRIORITY
2264
2265     if leaf.value in LOGIC_OPERATORS and leaf.parent:
2266         return LOGIC_PRIORITY
2267
2268     return 0
2269
2270
2271 FMT_OFF = {"# fmt: off", "# fmt:off", "# yapf: disable"}
2272 FMT_ON = {"# fmt: on", "# fmt:on", "# yapf: enable"}
2273
2274
2275 def generate_comments(leaf: LN) -> Iterator[Leaf]:
2276     """Clean the prefix of the `leaf` and generate comments from it, if any.
2277
2278     Comments in lib2to3 are shoved into the whitespace prefix.  This happens
2279     in `pgen2/driver.py:Driver.parse_tokens()`.  This was a brilliant implementation
2280     move because it does away with modifying the grammar to include all the
2281     possible places in which comments can be placed.
2282
2283     The sad consequence for us though is that comments don't "belong" anywhere.
2284     This is why this function generates simple parentless Leaf objects for
2285     comments.  We simply don't know what the correct parent should be.
2286
2287     No matter though, we can live without this.  We really only need to
2288     differentiate between inline and standalone comments.  The latter don't
2289     share the line with any code.
2290
2291     Inline comments are emitted as regular token.COMMENT leaves.  Standalone
2292     are emitted with a fake STANDALONE_COMMENT token identifier.
2293     """
2294     for pc in list_comments(leaf.prefix, is_endmarker=leaf.type == token.ENDMARKER):
2295         yield Leaf(pc.type, pc.value, prefix="\n" * pc.newlines)
2296
2297
2298 @dataclass
2299 class ProtoComment:
2300     """Describes a piece of syntax that is a comment.
2301
2302     It's not a :class:`blib2to3.pytree.Leaf` so that:
2303
2304     * it can be cached (`Leaf` objects should not be reused more than once as
2305       they store their lineno, column, prefix, and parent information);
2306     * `newlines` and `consumed` fields are kept separate from the `value`. This
2307       simplifies handling of special marker comments like ``# fmt: off/on``.
2308     """
2309
2310     type: int  # token.COMMENT or STANDALONE_COMMENT
2311     value: str  # content of the comment
2312     newlines: int  # how many newlines before the comment
2313     consumed: int  # how many characters of the original leaf's prefix did we consume
2314
2315
2316 @lru_cache(maxsize=4096)
2317 def list_comments(prefix: str, *, is_endmarker: bool) -> List[ProtoComment]:
2318     """Return a list of :class:`ProtoComment` objects parsed from the given `prefix`."""
2319     result: List[ProtoComment] = []
2320     if not prefix or "#" not in prefix:
2321         return result
2322
2323     consumed = 0
2324     nlines = 0
2325     ignored_lines = 0
2326     for index, line in enumerate(prefix.split("\n")):
2327         consumed += len(line) + 1  # adding the length of the split '\n'
2328         line = line.lstrip()
2329         if not line:
2330             nlines += 1
2331         if not line.startswith("#"):
2332             # Escaped newlines outside of a comment are not really newlines at
2333             # all. We treat a single-line comment following an escaped newline
2334             # as a simple trailing comment.
2335             if line.endswith("\\"):
2336                 ignored_lines += 1
2337             continue
2338
2339         if index == ignored_lines and not is_endmarker:
2340             comment_type = token.COMMENT  # simple trailing comment
2341         else:
2342             comment_type = STANDALONE_COMMENT
2343         comment = make_comment(line)
2344         result.append(
2345             ProtoComment(
2346                 type=comment_type, value=comment, newlines=nlines, consumed=consumed
2347             )
2348         )
2349         nlines = 0
2350     return result
2351
2352
2353 def make_comment(content: str) -> str:
2354     """Return a consistently formatted comment from the given `content` string.
2355
2356     All comments (except for "##", "#!", "#:", '#'", "#%%") should have a single
2357     space between the hash sign and the content.
2358
2359     If `content` didn't start with a hash sign, one is provided.
2360     """
2361     content = content.rstrip()
2362     if not content:
2363         return "#"
2364
2365     if content[0] == "#":
2366         content = content[1:]
2367     if content and content[0] not in " !:#'%":
2368         content = " " + content
2369     return "#" + content
2370
2371
2372 def split_line(
2373     line: Line,
2374     line_length: int,
2375     inner: bool = False,
2376     features: Collection[Feature] = (),
2377 ) -> Iterator[Line]:
2378     """Split a `line` into potentially many lines.
2379
2380     They should fit in the allotted `line_length` but might not be able to.
2381     `inner` signifies that there were a pair of brackets somewhere around the
2382     current `line`, possibly transitively. This means we can fallback to splitting
2383     by delimiters if the LHS/RHS don't yield any results.
2384
2385     `features` are syntactical features that may be used in the output.
2386     """
2387     if line.is_comment:
2388         yield line
2389         return
2390
2391     line_str = str(line).strip("\n")
2392
2393     if (
2394         not line.contains_uncollapsable_type_comments()
2395         and not line.should_explode
2396         and not line.is_collection_with_optional_trailing_comma
2397         and (
2398             is_line_short_enough(line, line_length=line_length, line_str=line_str)
2399             or line.contains_unsplittable_type_ignore()
2400         )
2401     ):
2402         yield line
2403         return
2404
2405     split_funcs: List[SplitFunc]
2406     if line.is_def:
2407         split_funcs = [left_hand_split]
2408     else:
2409
2410         def rhs(line: Line, features: Collection[Feature]) -> Iterator[Line]:
2411             for omit in generate_trailers_to_omit(line, line_length):
2412                 lines = list(right_hand_split(line, line_length, features, omit=omit))
2413                 if is_line_short_enough(lines[0], line_length=line_length):
2414                     yield from lines
2415                     return
2416
2417             # All splits failed, best effort split with no omits.
2418             # This mostly happens to multiline strings that are by definition
2419             # reported as not fitting a single line.
2420             # line_length=1 here was historically a bug that somehow became a feature.
2421             # See #762 and #781 for the full story.
2422             yield from right_hand_split(line, line_length=1, features=features)
2423
2424         if line.inside_brackets:
2425             split_funcs = [delimiter_split, standalone_comment_split, rhs]
2426         else:
2427             split_funcs = [rhs]
2428     for split_func in split_funcs:
2429         # We are accumulating lines in `result` because we might want to abort
2430         # mission and return the original line in the end, or attempt a different
2431         # split altogether.
2432         result: List[Line] = []
2433         try:
2434             for l in split_func(line, features):
2435                 if str(l).strip("\n") == line_str:
2436                     raise CannotSplit("Split function returned an unchanged result")
2437
2438                 result.extend(
2439                     split_line(
2440                         l, line_length=line_length, inner=True, features=features
2441                     )
2442                 )
2443         except CannotSplit:
2444             continue
2445
2446         else:
2447             yield from result
2448             break
2449
2450     else:
2451         yield line
2452
2453
2454 def left_hand_split(line: Line, features: Collection[Feature] = ()) -> Iterator[Line]:
2455     """Split line into many lines, starting with the first matching bracket pair.
2456
2457     Note: this usually looks weird, only use this for function definitions.
2458     Prefer RHS otherwise.  This is why this function is not symmetrical with
2459     :func:`right_hand_split` which also handles optional parentheses.
2460     """
2461     tail_leaves: List[Leaf] = []
2462     body_leaves: List[Leaf] = []
2463     head_leaves: List[Leaf] = []
2464     current_leaves = head_leaves
2465     matching_bracket = None
2466     for leaf in line.leaves:
2467         if (
2468             current_leaves is body_leaves
2469             and leaf.type in CLOSING_BRACKETS
2470             and leaf.opening_bracket is matching_bracket
2471         ):
2472             current_leaves = tail_leaves if body_leaves else head_leaves
2473         current_leaves.append(leaf)
2474         if current_leaves is head_leaves:
2475             if leaf.type in OPENING_BRACKETS:
2476                 matching_bracket = leaf
2477                 current_leaves = body_leaves
2478     if not matching_bracket:
2479         raise CannotSplit("No brackets found")
2480
2481     head = bracket_split_build_line(head_leaves, line, matching_bracket)
2482     body = bracket_split_build_line(body_leaves, line, matching_bracket, is_body=True)
2483     tail = bracket_split_build_line(tail_leaves, line, matching_bracket)
2484     bracket_split_succeeded_or_raise(head, body, tail)
2485     for result in (head, body, tail):
2486         if result:
2487             yield result
2488
2489
2490 def right_hand_split(
2491     line: Line,
2492     line_length: int,
2493     features: Collection[Feature] = (),
2494     omit: Collection[LeafID] = (),
2495 ) -> Iterator[Line]:
2496     """Split line into many lines, starting with the last matching bracket pair.
2497
2498     If the split was by optional parentheses, attempt splitting without them, too.
2499     `omit` is a collection of closing bracket IDs that shouldn't be considered for
2500     this split.
2501
2502     Note: running this function modifies `bracket_depth` on the leaves of `line`.
2503     """
2504     tail_leaves: List[Leaf] = []
2505     body_leaves: List[Leaf] = []
2506     head_leaves: List[Leaf] = []
2507     current_leaves = tail_leaves
2508     opening_bracket = None
2509     closing_bracket = None
2510     for leaf in reversed(line.leaves):
2511         if current_leaves is body_leaves:
2512             if leaf is opening_bracket:
2513                 current_leaves = head_leaves if body_leaves else tail_leaves
2514         current_leaves.append(leaf)
2515         if current_leaves is tail_leaves:
2516             if leaf.type in CLOSING_BRACKETS and id(leaf) not in omit:
2517                 opening_bracket = leaf.opening_bracket
2518                 closing_bracket = leaf
2519                 current_leaves = body_leaves
2520     if not (opening_bracket and closing_bracket and head_leaves):
2521         # If there is no opening or closing_bracket that means the split failed and
2522         # all content is in the tail.  Otherwise, if `head_leaves` are empty, it means
2523         # the matching `opening_bracket` wasn't available on `line` anymore.
2524         raise CannotSplit("No brackets found")
2525
2526     tail_leaves.reverse()
2527     body_leaves.reverse()
2528     head_leaves.reverse()
2529     head = bracket_split_build_line(head_leaves, line, opening_bracket)
2530     body = bracket_split_build_line(body_leaves, line, opening_bracket, is_body=True)
2531     tail = bracket_split_build_line(tail_leaves, line, opening_bracket)
2532     bracket_split_succeeded_or_raise(head, body, tail)
2533     if (
2534         # the body shouldn't be exploded
2535         not body.should_explode
2536         # the opening bracket is an optional paren
2537         and opening_bracket.type == token.LPAR
2538         and not opening_bracket.value
2539         # the closing bracket is an optional paren
2540         and closing_bracket.type == token.RPAR
2541         and not closing_bracket.value
2542         # it's not an import (optional parens are the only thing we can split on
2543         # in this case; attempting a split without them is a waste of time)
2544         and not line.is_import
2545         # there are no standalone comments in the body
2546         and not body.contains_standalone_comments(0)
2547         # and we can actually remove the parens
2548         and can_omit_invisible_parens(body, line_length)
2549     ):
2550         omit = {id(closing_bracket), *omit}
2551         try:
2552             yield from right_hand_split(line, line_length, features=features, omit=omit)
2553             return
2554
2555         except CannotSplit:
2556             if not (
2557                 can_be_split(body)
2558                 or is_line_short_enough(body, line_length=line_length)
2559             ):
2560                 raise CannotSplit(
2561                     "Splitting failed, body is still too long and can't be split."
2562                 )
2563
2564             elif head.contains_multiline_strings() or tail.contains_multiline_strings():
2565                 raise CannotSplit(
2566                     "The current optional pair of parentheses is bound to fail to "
2567                     "satisfy the splitting algorithm because the head or the tail "
2568                     "contains multiline strings which by definition never fit one "
2569                     "line."
2570                 )
2571
2572     ensure_visible(opening_bracket)
2573     ensure_visible(closing_bracket)
2574     for result in (head, body, tail):
2575         if result:
2576             yield result
2577
2578
2579 def bracket_split_succeeded_or_raise(head: Line, body: Line, tail: Line) -> None:
2580     """Raise :exc:`CannotSplit` if the last left- or right-hand split failed.
2581
2582     Do nothing otherwise.
2583
2584     A left- or right-hand split is based on a pair of brackets. Content before
2585     (and including) the opening bracket is left on one line, content inside the
2586     brackets is put on a separate line, and finally content starting with and
2587     following the closing bracket is put on a separate line.
2588
2589     Those are called `head`, `body`, and `tail`, respectively. If the split
2590     produced the same line (all content in `head`) or ended up with an empty `body`
2591     and the `tail` is just the closing bracket, then it's considered failed.
2592     """
2593     tail_len = len(str(tail).strip())
2594     if not body:
2595         if tail_len == 0:
2596             raise CannotSplit("Splitting brackets produced the same line")
2597
2598         elif tail_len < 3:
2599             raise CannotSplit(
2600                 f"Splitting brackets on an empty body to save "
2601                 f"{tail_len} characters is not worth it"
2602             )
2603
2604
2605 def bracket_split_build_line(
2606     leaves: List[Leaf], original: Line, opening_bracket: Leaf, *, is_body: bool = False
2607 ) -> Line:
2608     """Return a new line with given `leaves` and respective comments from `original`.
2609
2610     If `is_body` is True, the result line is one-indented inside brackets and as such
2611     has its first leaf's prefix normalized and a trailing comma added when expected.
2612     """
2613     result = Line(depth=original.depth)
2614     if is_body:
2615         result.inside_brackets = True
2616         result.depth += 1
2617         if leaves:
2618             # Since body is a new indent level, remove spurious leading whitespace.
2619             normalize_prefix(leaves[0], inside_brackets=True)
2620             # Ensure a trailing comma for imports and standalone function arguments, but
2621             # be careful not to add one after any comments or within type annotations.
2622             no_commas = (
2623                 original.is_def
2624                 and opening_bracket.value == "("
2625                 and not any(l.type == token.COMMA for l in leaves)
2626             )
2627
2628             if original.is_import or no_commas:
2629                 for i in range(len(leaves) - 1, -1, -1):
2630                     if leaves[i].type == STANDALONE_COMMENT:
2631                         continue
2632
2633                     if leaves[i].type != token.COMMA:
2634                         leaves.insert(i + 1, Leaf(token.COMMA, ","))
2635                     break
2636
2637     # Populate the line
2638     for leaf in leaves:
2639         result.append(leaf, preformatted=True)
2640         for comment_after in original.comments_after(leaf):
2641             result.append(comment_after, preformatted=True)
2642     if is_body:
2643         result.should_explode = should_explode(result, opening_bracket)
2644     return result
2645
2646
2647 def dont_increase_indentation(split_func: SplitFunc) -> SplitFunc:
2648     """Normalize prefix of the first leaf in every line returned by `split_func`.
2649
2650     This is a decorator over relevant split functions.
2651     """
2652
2653     @wraps(split_func)
2654     def split_wrapper(line: Line, features: Collection[Feature] = ()) -> Iterator[Line]:
2655         for l in split_func(line, features):
2656             normalize_prefix(l.leaves[0], inside_brackets=True)
2657             yield l
2658
2659     return split_wrapper
2660
2661
2662 @dont_increase_indentation
2663 def delimiter_split(line: Line, features: Collection[Feature] = ()) -> Iterator[Line]:
2664     """Split according to delimiters of the highest priority.
2665
2666     If the appropriate Features are given, the split will add trailing commas
2667     also in function signatures and calls that contain `*` and `**`.
2668     """
2669     try:
2670         last_leaf = line.leaves[-1]
2671     except IndexError:
2672         raise CannotSplit("Line empty")
2673
2674     bt = line.bracket_tracker
2675     try:
2676         delimiter_priority = bt.max_delimiter_priority(exclude={id(last_leaf)})
2677     except ValueError:
2678         raise CannotSplit("No delimiters found")
2679
2680     if delimiter_priority == DOT_PRIORITY:
2681         if bt.delimiter_count_with_priority(delimiter_priority) == 1:
2682             raise CannotSplit("Splitting a single attribute from its owner looks wrong")
2683
2684     current_line = Line(depth=line.depth, inside_brackets=line.inside_brackets)
2685     lowest_depth = sys.maxsize
2686     trailing_comma_safe = True
2687
2688     def append_to_line(leaf: Leaf) -> Iterator[Line]:
2689         """Append `leaf` to current line or to new line if appending impossible."""
2690         nonlocal current_line
2691         try:
2692             current_line.append_safe(leaf, preformatted=True)
2693         except ValueError:
2694             yield current_line
2695
2696             current_line = Line(depth=line.depth, inside_brackets=line.inside_brackets)
2697             current_line.append(leaf)
2698
2699     for leaf in line.leaves:
2700         yield from append_to_line(leaf)
2701
2702         for comment_after in line.comments_after(leaf):
2703             yield from append_to_line(comment_after)
2704
2705         lowest_depth = min(lowest_depth, leaf.bracket_depth)
2706         if leaf.bracket_depth == lowest_depth:
2707             if is_vararg(leaf, within={syms.typedargslist}):
2708                 trailing_comma_safe = (
2709                     trailing_comma_safe and Feature.TRAILING_COMMA_IN_DEF in features
2710                 )
2711             elif is_vararg(leaf, within={syms.arglist, syms.argument}):
2712                 trailing_comma_safe = (
2713                     trailing_comma_safe and Feature.TRAILING_COMMA_IN_CALL in features
2714                 )
2715
2716         leaf_priority = bt.delimiters.get(id(leaf))
2717         if leaf_priority == delimiter_priority:
2718             yield current_line
2719
2720             current_line = Line(depth=line.depth, inside_brackets=line.inside_brackets)
2721     if current_line:
2722         if (
2723             trailing_comma_safe
2724             and delimiter_priority == COMMA_PRIORITY
2725             and current_line.leaves[-1].type != token.COMMA
2726             and current_line.leaves[-1].type != STANDALONE_COMMENT
2727         ):
2728             current_line.append(Leaf(token.COMMA, ","))
2729         yield current_line
2730
2731
2732 @dont_increase_indentation
2733 def standalone_comment_split(
2734     line: Line, features: Collection[Feature] = ()
2735 ) -> Iterator[Line]:
2736     """Split standalone comments from the rest of the line."""
2737     if not line.contains_standalone_comments(0):
2738         raise CannotSplit("Line does not have any standalone comments")
2739
2740     current_line = Line(depth=line.depth, inside_brackets=line.inside_brackets)
2741
2742     def append_to_line(leaf: Leaf) -> Iterator[Line]:
2743         """Append `leaf` to current line or to new line if appending impossible."""
2744         nonlocal current_line
2745         try:
2746             current_line.append_safe(leaf, preformatted=True)
2747         except ValueError:
2748             yield current_line
2749
2750             current_line = Line(depth=line.depth, inside_brackets=line.inside_brackets)
2751             current_line.append(leaf)
2752
2753     for leaf in line.leaves:
2754         yield from append_to_line(leaf)
2755
2756         for comment_after in line.comments_after(leaf):
2757             yield from append_to_line(comment_after)
2758
2759     if current_line:
2760         yield current_line
2761
2762
2763 def is_import(leaf: Leaf) -> bool:
2764     """Return True if the given leaf starts an import statement."""
2765     p = leaf.parent
2766     t = leaf.type
2767     v = leaf.value
2768     return bool(
2769         t == token.NAME
2770         and (
2771             (v == "import" and p and p.type == syms.import_name)
2772             or (v == "from" and p and p.type == syms.import_from)
2773         )
2774     )
2775
2776
2777 def is_type_comment(leaf: Leaf, suffix: str = "") -> bool:
2778     """Return True if the given leaf is a special comment.
2779     Only returns true for type comments for now."""
2780     t = leaf.type
2781     v = leaf.value
2782     return t in {token.COMMENT, STANDALONE_COMMENT} and v.startswith("# type:" + suffix)
2783
2784
2785 def normalize_prefix(leaf: Leaf, *, inside_brackets: bool) -> None:
2786     """Leave existing extra newlines if not `inside_brackets`. Remove everything
2787     else.
2788
2789     Note: don't use backslashes for formatting or you'll lose your voting rights.
2790     """
2791     if not inside_brackets:
2792         spl = leaf.prefix.split("#")
2793         if "\\" not in spl[0]:
2794             nl_count = spl[-1].count("\n")
2795             if len(spl) > 1:
2796                 nl_count -= 1
2797             leaf.prefix = "\n" * nl_count
2798             return
2799
2800     leaf.prefix = ""
2801
2802
2803 def normalize_string_prefix(leaf: Leaf, remove_u_prefix: bool = False) -> None:
2804     """Make all string prefixes lowercase.
2805
2806     If remove_u_prefix is given, also removes any u prefix from the string.
2807
2808     Note: Mutates its argument.
2809     """
2810     match = re.match(r"^([furbFURB]*)(.*)$", leaf.value, re.DOTALL)
2811     assert match is not None, f"failed to match string {leaf.value!r}"
2812     orig_prefix = match.group(1)
2813     new_prefix = orig_prefix.lower()
2814     if remove_u_prefix:
2815         new_prefix = new_prefix.replace("u", "")
2816     leaf.value = f"{new_prefix}{match.group(2)}"
2817
2818
2819 def normalize_string_quotes(leaf: Leaf) -> None:
2820     """Prefer double quotes but only if it doesn't cause more escaping.
2821
2822     Adds or removes backslashes as appropriate. Doesn't parse and fix
2823     strings nested in f-strings (yet).
2824
2825     Note: Mutates its argument.
2826     """
2827     value = leaf.value.lstrip("furbFURB")
2828     if value[:3] == '"""':
2829         return
2830
2831     elif value[:3] == "'''":
2832         orig_quote = "'''"
2833         new_quote = '"""'
2834     elif value[0] == '"':
2835         orig_quote = '"'
2836         new_quote = "'"
2837     else:
2838         orig_quote = "'"
2839         new_quote = '"'
2840     first_quote_pos = leaf.value.find(orig_quote)
2841     if first_quote_pos == -1:
2842         return  # There's an internal error
2843
2844     prefix = leaf.value[:first_quote_pos]
2845     unescaped_new_quote = re.compile(rf"(([^\\]|^)(\\\\)*){new_quote}")
2846     escaped_new_quote = re.compile(rf"([^\\]|^)\\((?:\\\\)*){new_quote}")
2847     escaped_orig_quote = re.compile(rf"([^\\]|^)\\((?:\\\\)*){orig_quote}")
2848     body = leaf.value[first_quote_pos + len(orig_quote) : -len(orig_quote)]
2849     if "r" in prefix.casefold():
2850         if unescaped_new_quote.search(body):
2851             # There's at least one unescaped new_quote in this raw string
2852             # so converting is impossible
2853             return
2854
2855         # Do not introduce or remove backslashes in raw strings
2856         new_body = body
2857     else:
2858         # remove unnecessary escapes
2859         new_body = sub_twice(escaped_new_quote, rf"\1\2{new_quote}", body)
2860         if body != new_body:
2861             # Consider the string without unnecessary escapes as the original
2862             body = new_body
2863             leaf.value = f"{prefix}{orig_quote}{body}{orig_quote}"
2864         new_body = sub_twice(escaped_orig_quote, rf"\1\2{orig_quote}", new_body)
2865         new_body = sub_twice(unescaped_new_quote, rf"\1\\{new_quote}", new_body)
2866     if "f" in prefix.casefold():
2867         matches = re.findall(
2868             r"""
2869             (?:[^{]|^)\{  # start of the string or a non-{ followed by a single {
2870                 ([^{].*?)  # contents of the brackets except if begins with {{
2871             \}(?:[^}]|$)  # A } followed by end of the string or a non-}
2872             """,
2873             new_body,
2874             re.VERBOSE,
2875         )
2876         for m in matches:
2877             if "\\" in str(m):
2878                 # Do not introduce backslashes in interpolated expressions
2879                 return
2880
2881     if new_quote == '"""' and new_body[-1:] == '"':
2882         # edge case:
2883         new_body = new_body[:-1] + '\\"'
2884     orig_escape_count = body.count("\\")
2885     new_escape_count = new_body.count("\\")
2886     if new_escape_count > orig_escape_count:
2887         return  # Do not introduce more escaping
2888
2889     if new_escape_count == orig_escape_count and orig_quote == '"':
2890         return  # Prefer double quotes
2891
2892     leaf.value = f"{prefix}{new_quote}{new_body}{new_quote}"
2893
2894
2895 def normalize_numeric_literal(leaf: Leaf) -> None:
2896     """Normalizes numeric (float, int, and complex) literals.
2897
2898     All letters used in the representation are normalized to lowercase (except
2899     in Python 2 long literals).
2900     """
2901     text = leaf.value.lower()
2902     if text.startswith(("0o", "0b")):
2903         # Leave octal and binary literals alone.
2904         pass
2905     elif text.startswith("0x"):
2906         # Change hex literals to upper case.
2907         before, after = text[:2], text[2:]
2908         text = f"{before}{after.upper()}"
2909     elif "e" in text:
2910         before, after = text.split("e")
2911         sign = ""
2912         if after.startswith("-"):
2913             after = after[1:]
2914             sign = "-"
2915         elif after.startswith("+"):
2916             after = after[1:]
2917         before = format_float_or_int_string(before)
2918         text = f"{before}e{sign}{after}"
2919     elif text.endswith(("j", "l")):
2920         number = text[:-1]
2921         suffix = text[-1]
2922         # Capitalize in "2L" because "l" looks too similar to "1".
2923         if suffix == "l":
2924             suffix = "L"
2925         text = f"{format_float_or_int_string(number)}{suffix}"
2926     else:
2927         text = format_float_or_int_string(text)
2928     leaf.value = text
2929
2930
2931 def format_float_or_int_string(text: str) -> str:
2932     """Formats a float string like "1.0"."""
2933     if "." not in text:
2934         return text
2935
2936     before, after = text.split(".")
2937     return f"{before or 0}.{after or 0}"
2938
2939
2940 def normalize_invisible_parens(node: Node, parens_after: Set[str]) -> None:
2941     """Make existing optional parentheses invisible or create new ones.
2942
2943     `parens_after` is a set of string leaf values immediately after which parens
2944     should be put.
2945
2946     Standardizes on visible parentheses for single-element tuples, and keeps
2947     existing visible parentheses for other tuples and generator expressions.
2948     """
2949     for pc in list_comments(node.prefix, is_endmarker=False):
2950         if pc.value in FMT_OFF:
2951             # This `node` has a prefix with `# fmt: off`, don't mess with parens.
2952             return
2953
2954     check_lpar = False
2955     for index, child in enumerate(list(node.children)):
2956         # Add parentheses around long tuple unpacking in assignments.
2957         if (
2958             index == 0
2959             and isinstance(child, Node)
2960             and child.type == syms.testlist_star_expr
2961         ):
2962             check_lpar = True
2963
2964         if check_lpar:
2965             if is_walrus_assignment(child):
2966                 continue
2967
2968             if child.type == syms.atom:
2969                 # Determines if the underlying atom should be surrounded with
2970                 # invisible params - also makes parens invisible recursively
2971                 # within the atom and removes repeated invisible parens within
2972                 # the atom
2973                 should_surround_with_parens = maybe_make_parens_invisible_in_atom(
2974                     child, parent=node
2975                 )
2976
2977                 if should_surround_with_parens:
2978                     lpar = Leaf(token.LPAR, "")
2979                     rpar = Leaf(token.RPAR, "")
2980                     index = child.remove() or 0
2981                     node.insert_child(index, Node(syms.atom, [lpar, child, rpar]))
2982             elif is_one_tuple(child):
2983                 # wrap child in visible parentheses
2984                 lpar = Leaf(token.LPAR, "(")
2985                 rpar = Leaf(token.RPAR, ")")
2986                 child.remove()
2987                 node.insert_child(index, Node(syms.atom, [lpar, child, rpar]))
2988             elif node.type == syms.import_from:
2989                 # "import from" nodes store parentheses directly as part of
2990                 # the statement
2991                 if child.type == token.LPAR:
2992                     # make parentheses invisible
2993                     child.value = ""  # type: ignore
2994                     node.children[-1].value = ""  # type: ignore
2995                 elif child.type != token.STAR:
2996                     # insert invisible parentheses
2997                     node.insert_child(index, Leaf(token.LPAR, ""))
2998                     node.append_child(Leaf(token.RPAR, ""))
2999                 break
3000
3001             elif not (isinstance(child, Leaf) and is_multiline_string(child)):
3002                 # wrap child in invisible parentheses
3003                 lpar = Leaf(token.LPAR, "")
3004                 rpar = Leaf(token.RPAR, "")
3005                 index = child.remove() or 0
3006                 prefix = child.prefix
3007                 child.prefix = ""
3008                 new_child = Node(syms.atom, [lpar, child, rpar])
3009                 new_child.prefix = prefix
3010                 node.insert_child(index, new_child)
3011
3012         check_lpar = isinstance(child, Leaf) and child.value in parens_after
3013
3014
3015 def normalize_fmt_off(node: Node) -> None:
3016     """Convert content between `# fmt: off`/`# fmt: on` into standalone comments."""
3017     try_again = True
3018     while try_again:
3019         try_again = convert_one_fmt_off_pair(node)
3020
3021
3022 def convert_one_fmt_off_pair(node: Node) -> bool:
3023     """Convert content of a single `# fmt: off`/`# fmt: on` into a standalone comment.
3024
3025     Returns True if a pair was converted.
3026     """
3027     for leaf in node.leaves():
3028         previous_consumed = 0
3029         for comment in list_comments(leaf.prefix, is_endmarker=False):
3030             if comment.value in FMT_OFF:
3031                 # We only want standalone comments. If there's no previous leaf or
3032                 # the previous leaf is indentation, it's a standalone comment in
3033                 # disguise.
3034                 if comment.type != STANDALONE_COMMENT:
3035                     prev = preceding_leaf(leaf)
3036                     if prev and prev.type not in WHITESPACE:
3037                         continue
3038
3039                 ignored_nodes = list(generate_ignored_nodes(leaf))
3040                 if not ignored_nodes:
3041                     continue
3042
3043                 first = ignored_nodes[0]  # Can be a container node with the `leaf`.
3044                 parent = first.parent
3045                 prefix = first.prefix
3046                 first.prefix = prefix[comment.consumed :]
3047                 hidden_value = (
3048                     comment.value + "\n" + "".join(str(n) for n in ignored_nodes)
3049                 )
3050                 if hidden_value.endswith("\n"):
3051                     # That happens when one of the `ignored_nodes` ended with a NEWLINE
3052                     # leaf (possibly followed by a DEDENT).
3053                     hidden_value = hidden_value[:-1]
3054                 first_idx = None
3055                 for ignored in ignored_nodes:
3056                     index = ignored.remove()
3057                     if first_idx is None:
3058                         first_idx = index
3059                 assert parent is not None, "INTERNAL ERROR: fmt: on/off handling (1)"
3060                 assert first_idx is not None, "INTERNAL ERROR: fmt: on/off handling (2)"
3061                 parent.insert_child(
3062                     first_idx,
3063                     Leaf(
3064                         STANDALONE_COMMENT,
3065                         hidden_value,
3066                         prefix=prefix[:previous_consumed] + "\n" * comment.newlines,
3067                     ),
3068                 )
3069                 return True
3070
3071             previous_consumed = comment.consumed
3072
3073     return False
3074
3075
3076 def generate_ignored_nodes(leaf: Leaf) -> Iterator[LN]:
3077     """Starting from the container of `leaf`, generate all leaves until `# fmt: on`.
3078
3079     Stops at the end of the block.
3080     """
3081     container: Optional[LN] = container_of(leaf)
3082     while container is not None and container.type != token.ENDMARKER:
3083         for comment in list_comments(container.prefix, is_endmarker=False):
3084             if comment.value in FMT_ON:
3085                 return
3086
3087         yield container
3088
3089         container = container.next_sibling
3090
3091
3092 def maybe_make_parens_invisible_in_atom(node: LN, parent: LN) -> bool:
3093     """If it's safe, make the parens in the atom `node` invisible, recursively.
3094     Additionally, remove repeated, adjacent invisible parens from the atom `node`
3095     as they are redundant.
3096
3097     Returns whether the node should itself be wrapped in invisible parentheses.
3098
3099     """
3100     if (
3101         node.type != syms.atom
3102         or is_empty_tuple(node)
3103         or is_one_tuple(node)
3104         or (is_yield(node) and parent.type != syms.expr_stmt)
3105         or max_delimiter_priority_in_atom(node) >= COMMA_PRIORITY
3106     ):
3107         return False
3108
3109     first = node.children[0]
3110     last = node.children[-1]
3111     if first.type == token.LPAR and last.type == token.RPAR:
3112         middle = node.children[1]
3113         # make parentheses invisible
3114         first.value = ""  # type: ignore
3115         last.value = ""  # type: ignore
3116         maybe_make_parens_invisible_in_atom(middle, parent=parent)
3117
3118         if is_atom_with_invisible_parens(middle):
3119             # Strip the invisible parens from `middle` by replacing
3120             # it with the child in-between the invisible parens
3121             middle.replace(middle.children[1])
3122
3123         return False
3124
3125     return True
3126
3127
3128 def is_atom_with_invisible_parens(node: LN) -> bool:
3129     """Given a `LN`, determines whether it's an atom `node` with invisible
3130     parens. Useful in dedupe-ing and normalizing parens.
3131     """
3132     if isinstance(node, Leaf) or node.type != syms.atom:
3133         return False
3134
3135     first, last = node.children[0], node.children[-1]
3136     return (
3137         isinstance(first, Leaf)
3138         and first.type == token.LPAR
3139         and first.value == ""
3140         and isinstance(last, Leaf)
3141         and last.type == token.RPAR
3142         and last.value == ""
3143     )
3144
3145
3146 def is_empty_tuple(node: LN) -> bool:
3147     """Return True if `node` holds an empty tuple."""
3148     return (
3149         node.type == syms.atom
3150         and len(node.children) == 2
3151         and node.children[0].type == token.LPAR
3152         and node.children[1].type == token.RPAR
3153     )
3154
3155
3156 def unwrap_singleton_parenthesis(node: LN) -> Optional[LN]:
3157     """Returns `wrapped` if `node` is of the shape ( wrapped ).
3158
3159     Parenthesis can be optional. Returns None otherwise"""
3160     if len(node.children) != 3:
3161         return None
3162
3163     lpar, wrapped, rpar = node.children
3164     if not (lpar.type == token.LPAR and rpar.type == token.RPAR):
3165         return None
3166
3167     return wrapped
3168
3169
3170 def is_one_tuple(node: LN) -> bool:
3171     """Return True if `node` holds a tuple with one element, with or without parens."""
3172     if node.type == syms.atom:
3173         gexp = unwrap_singleton_parenthesis(node)
3174         if gexp is None or gexp.type != syms.testlist_gexp:
3175             return False
3176
3177         return len(gexp.children) == 2 and gexp.children[1].type == token.COMMA
3178
3179     return (
3180         node.type in IMPLICIT_TUPLE
3181         and len(node.children) == 2
3182         and node.children[1].type == token.COMMA
3183     )
3184
3185
3186 def is_walrus_assignment(node: LN) -> bool:
3187     """Return True iff `node` is of the shape ( test := test )"""
3188     inner = unwrap_singleton_parenthesis(node)
3189     return inner is not None and inner.type == syms.namedexpr_test
3190
3191
3192 def is_yield(node: LN) -> bool:
3193     """Return True if `node` holds a `yield` or `yield from` expression."""
3194     if node.type == syms.yield_expr:
3195         return True
3196
3197     if node.type == token.NAME and node.value == "yield":  # type: ignore
3198         return True
3199
3200     if node.type != syms.atom:
3201         return False
3202
3203     if len(node.children) != 3:
3204         return False
3205
3206     lpar, expr, rpar = node.children
3207     if lpar.type == token.LPAR and rpar.type == token.RPAR:
3208         return is_yield(expr)
3209
3210     return False
3211
3212
3213 def is_vararg(leaf: Leaf, within: Set[NodeType]) -> bool:
3214     """Return True if `leaf` is a star or double star in a vararg or kwarg.
3215
3216     If `within` includes VARARGS_PARENTS, this applies to function signatures.
3217     If `within` includes UNPACKING_PARENTS, it applies to right hand-side
3218     extended iterable unpacking (PEP 3132) and additional unpacking
3219     generalizations (PEP 448).
3220     """
3221     if leaf.type not in VARARGS_SPECIALS or not leaf.parent:
3222         return False
3223
3224     p = leaf.parent
3225     if p.type == syms.star_expr:
3226         # Star expressions are also used as assignment targets in extended
3227         # iterable unpacking (PEP 3132).  See what its parent is instead.
3228         if not p.parent:
3229             return False
3230
3231         p = p.parent
3232
3233     return p.type in within
3234
3235
3236 def is_multiline_string(leaf: Leaf) -> bool:
3237     """Return True if `leaf` is a multiline string that actually spans many lines."""
3238     value = leaf.value.lstrip("furbFURB")
3239     return value[:3] in {'"""', "'''"} and "\n" in value
3240
3241
3242 def is_stub_suite(node: Node) -> bool:
3243     """Return True if `node` is a suite with a stub body."""
3244     if (
3245         len(node.children) != 4
3246         or node.children[0].type != token.NEWLINE
3247         or node.children[1].type != token.INDENT
3248         or node.children[3].type != token.DEDENT
3249     ):
3250         return False
3251
3252     return is_stub_body(node.children[2])
3253
3254
3255 def is_stub_body(node: LN) -> bool:
3256     """Return True if `node` is a simple statement containing an ellipsis."""
3257     if not isinstance(node, Node) or node.type != syms.simple_stmt:
3258         return False
3259
3260     if len(node.children) != 2:
3261         return False
3262
3263     child = node.children[0]
3264     return (
3265         child.type == syms.atom
3266         and len(child.children) == 3
3267         and all(leaf == Leaf(token.DOT, ".") for leaf in child.children)
3268     )
3269
3270
3271 def max_delimiter_priority_in_atom(node: LN) -> Priority:
3272     """Return maximum delimiter priority inside `node`.
3273
3274     This is specific to atoms with contents contained in a pair of parentheses.
3275     If `node` isn't an atom or there are no enclosing parentheses, returns 0.
3276     """
3277     if node.type != syms.atom:
3278         return 0
3279
3280     first = node.children[0]
3281     last = node.children[-1]
3282     if not (first.type == token.LPAR and last.type == token.RPAR):
3283         return 0
3284
3285     bt = BracketTracker()
3286     for c in node.children[1:-1]:
3287         if isinstance(c, Leaf):
3288             bt.mark(c)
3289         else:
3290             for leaf in c.leaves():
3291                 bt.mark(leaf)
3292     try:
3293         return bt.max_delimiter_priority()
3294
3295     except ValueError:
3296         return 0
3297
3298
3299 def ensure_visible(leaf: Leaf) -> None:
3300     """Make sure parentheses are visible.
3301
3302     They could be invisible as part of some statements (see
3303     :func:`normalize_invisible_parens` and :func:`visit_import_from`).
3304     """
3305     if leaf.type == token.LPAR:
3306         leaf.value = "("
3307     elif leaf.type == token.RPAR:
3308         leaf.value = ")"
3309
3310
3311 def should_explode(line: Line, opening_bracket: Leaf) -> bool:
3312     """Should `line` immediately be split with `delimiter_split()` after RHS?"""
3313
3314     if not (
3315         opening_bracket.parent
3316         and opening_bracket.parent.type in {syms.atom, syms.import_from}
3317         and opening_bracket.value in "[{("
3318     ):
3319         return False
3320
3321     try:
3322         last_leaf = line.leaves[-1]
3323         exclude = {id(last_leaf)} if last_leaf.type == token.COMMA else set()
3324         max_priority = line.bracket_tracker.max_delimiter_priority(exclude=exclude)
3325     except (IndexError, ValueError):
3326         return False
3327
3328     return max_priority == COMMA_PRIORITY
3329
3330
3331 def get_features_used(node: Node) -> Set[Feature]:
3332     """Return a set of (relatively) new Python features used in this file.
3333
3334     Currently looking for:
3335     - f-strings;
3336     - underscores in numeric literals;
3337     - trailing commas after * or ** in function signatures and calls;
3338     - positional only arguments in function signatures and lambdas;
3339     """
3340     features: Set[Feature] = set()
3341     for n in node.pre_order():
3342         if n.type == token.STRING:
3343             value_head = n.value[:2]  # type: ignore
3344             if value_head in {'f"', 'F"', "f'", "F'", "rf", "fr", "RF", "FR"}:
3345                 features.add(Feature.F_STRINGS)
3346
3347         elif n.type == token.NUMBER:
3348             if "_" in n.value:  # type: ignore
3349                 features.add(Feature.NUMERIC_UNDERSCORES)
3350
3351         elif n.type == token.SLASH:
3352             if n.parent and n.parent.type in {syms.typedargslist, syms.arglist}:
3353                 features.add(Feature.POS_ONLY_ARGUMENTS)
3354
3355         elif n.type == token.COLONEQUAL:
3356             features.add(Feature.ASSIGNMENT_EXPRESSIONS)
3357
3358         elif (
3359             n.type in {syms.typedargslist, syms.arglist}
3360             and n.children
3361             and n.children[-1].type == token.COMMA
3362         ):
3363             if n.type == syms.typedargslist:
3364                 feature = Feature.TRAILING_COMMA_IN_DEF
3365             else:
3366                 feature = Feature.TRAILING_COMMA_IN_CALL
3367
3368             for ch in n.children:
3369                 if ch.type in STARS:
3370                     features.add(feature)
3371
3372                 if ch.type == syms.argument:
3373                     for argch in ch.children:
3374                         if argch.type in STARS:
3375                             features.add(feature)
3376
3377     return features
3378
3379
3380 def detect_target_versions(node: Node) -> Set[TargetVersion]:
3381     """Detect the version to target based on the nodes used."""
3382     features = get_features_used(node)
3383     return {
3384         version for version in TargetVersion if features <= VERSION_TO_FEATURES[version]
3385     }
3386
3387
3388 def generate_trailers_to_omit(line: Line, line_length: int) -> Iterator[Set[LeafID]]:
3389     """Generate sets of closing bracket IDs that should be omitted in a RHS.
3390
3391     Brackets can be omitted if the entire trailer up to and including
3392     a preceding closing bracket fits in one line.
3393
3394     Yielded sets are cumulative (contain results of previous yields, too).  First
3395     set is empty.
3396     """
3397
3398     omit: Set[LeafID] = set()
3399     yield omit
3400
3401     length = 4 * line.depth
3402     opening_bracket = None
3403     closing_bracket = None
3404     inner_brackets: Set[LeafID] = set()
3405     for index, leaf, leaf_length in enumerate_with_length(line, reversed=True):
3406         length += leaf_length
3407         if length > line_length:
3408             break
3409
3410         has_inline_comment = leaf_length > len(leaf.value) + len(leaf.prefix)
3411         if leaf.type == STANDALONE_COMMENT or has_inline_comment:
3412             break
3413
3414         if opening_bracket:
3415             if leaf is opening_bracket:
3416                 opening_bracket = None
3417             elif leaf.type in CLOSING_BRACKETS:
3418                 inner_brackets.add(id(leaf))
3419         elif leaf.type in CLOSING_BRACKETS:
3420             if index > 0 and line.leaves[index - 1].type in OPENING_BRACKETS:
3421                 # Empty brackets would fail a split so treat them as "inner"
3422                 # brackets (e.g. only add them to the `omit` set if another
3423                 # pair of brackets was good enough.
3424                 inner_brackets.add(id(leaf))
3425                 continue
3426
3427             if closing_bracket:
3428                 omit.add(id(closing_bracket))
3429                 omit.update(inner_brackets)
3430                 inner_brackets.clear()
3431                 yield omit
3432
3433             if leaf.value:
3434                 opening_bracket = leaf.opening_bracket
3435                 closing_bracket = leaf
3436
3437
3438 def get_future_imports(node: Node) -> Set[str]:
3439     """Return a set of __future__ imports in the file."""
3440     imports: Set[str] = set()
3441
3442     def get_imports_from_children(children: List[LN]) -> Generator[str, None, None]:
3443         for child in children:
3444             if isinstance(child, Leaf):
3445                 if child.type == token.NAME:
3446                     yield child.value
3447
3448             elif child.type == syms.import_as_name:
3449                 orig_name = child.children[0]
3450                 assert isinstance(orig_name, Leaf), "Invalid syntax parsing imports"
3451                 assert orig_name.type == token.NAME, "Invalid syntax parsing imports"
3452                 yield orig_name.value
3453
3454             elif child.type == syms.import_as_names:
3455                 yield from get_imports_from_children(child.children)
3456
3457             else:
3458                 raise AssertionError("Invalid syntax parsing imports")
3459
3460     for child in node.children:
3461         if child.type != syms.simple_stmt:
3462             break
3463
3464         first_child = child.children[0]
3465         if isinstance(first_child, Leaf):
3466             # Continue looking if we see a docstring; otherwise stop.
3467             if (
3468                 len(child.children) == 2
3469                 and first_child.type == token.STRING
3470                 and child.children[1].type == token.NEWLINE
3471             ):
3472                 continue
3473
3474             break
3475
3476         elif first_child.type == syms.import_from:
3477             module_name = first_child.children[1]
3478             if not isinstance(module_name, Leaf) or module_name.value != "__future__":
3479                 break
3480
3481             imports |= set(get_imports_from_children(first_child.children[3:]))
3482         else:
3483             break
3484
3485     return imports
3486
3487
3488 @lru_cache()
3489 def get_gitignore(root: Path) -> PathSpec:
3490     """ Return a PathSpec matching gitignore content if present."""
3491     gitignore = root / ".gitignore"
3492     lines: List[str] = []
3493     if gitignore.is_file():
3494         with gitignore.open() as gf:
3495             lines = gf.readlines()
3496     return PathSpec.from_lines("gitwildmatch", lines)
3497
3498
3499 def gen_python_files_in_dir(
3500     path: Path,
3501     root: Path,
3502     include: Pattern[str],
3503     exclude: Pattern[str],
3504     report: "Report",
3505     gitignore: PathSpec,
3506 ) -> Iterator[Path]:
3507     """Generate all files under `path` whose paths are not excluded by the
3508     `exclude` regex, but are included by the `include` regex.
3509
3510     Symbolic links pointing outside of the `root` directory are ignored.
3511
3512     `report` is where output about exclusions goes.
3513     """
3514     assert root.is_absolute(), f"INTERNAL ERROR: `root` must be absolute but is {root}"
3515     for child in path.iterdir():
3516         # First ignore files matching .gitignore
3517         if gitignore.match_file(child.as_posix()):
3518             report.path_ignored(child, f"matches the .gitignore file content")
3519             continue
3520
3521         # Then ignore with `exclude` option.
3522         try:
3523             normalized_path = "/" + child.resolve().relative_to(root).as_posix()
3524         except OSError as e:
3525             report.path_ignored(child, f"cannot be read because {e}")
3526             continue
3527
3528         except ValueError:
3529             if child.is_symlink():
3530                 report.path_ignored(
3531                     child, f"is a symbolic link that points outside {root}"
3532                 )
3533                 continue
3534
3535             raise
3536
3537         if child.is_dir():
3538             normalized_path += "/"
3539
3540         exclude_match = exclude.search(normalized_path)
3541         if exclude_match and exclude_match.group(0):
3542             report.path_ignored(child, f"matches the --exclude regular expression")
3543             continue
3544
3545         if child.is_dir():
3546             yield from gen_python_files_in_dir(
3547                 child, root, include, exclude, report, gitignore
3548             )
3549
3550         elif child.is_file():
3551             include_match = include.search(normalized_path)
3552             if include_match:
3553                 yield child
3554
3555
3556 @lru_cache()
3557 def find_project_root(srcs: Iterable[str]) -> Path:
3558     """Return a directory containing .git, .hg, or pyproject.toml.
3559
3560     That directory can be one of the directories passed in `srcs` or their
3561     common parent.
3562
3563     If no directory in the tree contains a marker that would specify it's the
3564     project root, the root of the file system is returned.
3565     """
3566     if not srcs:
3567         return Path("/").resolve()
3568
3569     common_base = min(Path(src).resolve() for src in srcs)
3570     if common_base.is_dir():
3571         # Append a fake file so `parents` below returns `common_base_dir`, too.
3572         common_base /= "fake-file"
3573     for directory in common_base.parents:
3574         if (directory / ".git").is_dir():
3575             return directory
3576
3577         if (directory / ".hg").is_dir():
3578             return directory
3579
3580         if (directory / "pyproject.toml").is_file():
3581             return directory
3582
3583     return directory
3584
3585
3586 @dataclass
3587 class Report:
3588     """Provides a reformatting counter. Can be rendered with `str(report)`."""
3589
3590     check: bool = False
3591     quiet: bool = False
3592     verbose: bool = False
3593     change_count: int = 0
3594     same_count: int = 0
3595     failure_count: int = 0
3596
3597     def done(self, src: Path, changed: Changed) -> None:
3598         """Increment the counter for successful reformatting. Write out a message."""
3599         if changed is Changed.YES:
3600             reformatted = "would reformat" if self.check else "reformatted"
3601             if self.verbose or not self.quiet:
3602                 out(f"{reformatted} {src}")
3603             self.change_count += 1
3604         else:
3605             if self.verbose:
3606                 if changed is Changed.NO:
3607                     msg = f"{src} already well formatted, good job."
3608                 else:
3609                     msg = f"{src} wasn't modified on disk since last run."
3610                 out(msg, bold=False)
3611             self.same_count += 1
3612
3613     def failed(self, src: Path, message: str) -> None:
3614         """Increment the counter for failed reformatting. Write out a message."""
3615         err(f"error: cannot format {src}: {message}")
3616         self.failure_count += 1
3617
3618     def path_ignored(self, path: Path, message: str) -> None:
3619         if self.verbose:
3620             out(f"{path} ignored: {message}", bold=False)
3621
3622     @property
3623     def return_code(self) -> int:
3624         """Return the exit code that the app should use.
3625
3626         This considers the current state of changed files and failures:
3627         - if there were any failures, return 123;
3628         - if any files were changed and --check is being used, return 1;
3629         - otherwise return 0.
3630         """
3631         # According to http://tldp.org/LDP/abs/html/exitcodes.html starting with
3632         # 126 we have special return codes reserved by the shell.
3633         if self.failure_count:
3634             return 123
3635
3636         elif self.change_count and self.check:
3637             return 1
3638
3639         return 0
3640
3641     def __str__(self) -> str:
3642         """Render a color report of the current state.
3643
3644         Use `click.unstyle` to remove colors.
3645         """
3646         if self.check:
3647             reformatted = "would be reformatted"
3648             unchanged = "would be left unchanged"
3649             failed = "would fail to reformat"
3650         else:
3651             reformatted = "reformatted"
3652             unchanged = "left unchanged"
3653             failed = "failed to reformat"
3654         report = []
3655         if self.change_count:
3656             s = "s" if self.change_count > 1 else ""
3657             report.append(
3658                 click.style(f"{self.change_count} file{s} {reformatted}", bold=True)
3659             )
3660         if self.same_count:
3661             s = "s" if self.same_count > 1 else ""
3662             report.append(f"{self.same_count} file{s} {unchanged}")
3663         if self.failure_count:
3664             s = "s" if self.failure_count > 1 else ""
3665             report.append(
3666                 click.style(f"{self.failure_count} file{s} {failed}", fg="red")
3667             )
3668         return ", ".join(report) + "."
3669
3670
3671 def parse_ast(src: str) -> Union[ast.AST, ast3.AST, ast27.AST]:
3672     filename = "<unknown>"
3673     if sys.version_info >= (3, 8):
3674         # TODO: support Python 4+ ;)
3675         for minor_version in range(sys.version_info[1], 4, -1):
3676             try:
3677                 return ast.parse(src, filename, feature_version=(3, minor_version))
3678             except SyntaxError:
3679                 continue
3680     else:
3681         for feature_version in (7, 6):
3682             try:
3683                 return ast3.parse(src, filename, feature_version=feature_version)
3684             except SyntaxError:
3685                 continue
3686
3687     return ast27.parse(src)
3688
3689
3690 def _fixup_ast_constants(
3691     node: Union[ast.AST, ast3.AST, ast27.AST]
3692 ) -> Union[ast.AST, ast3.AST, ast27.AST]:
3693     """Map ast nodes deprecated in 3.8 to Constant."""
3694     if isinstance(node, (ast.Str, ast3.Str, ast27.Str, ast.Bytes, ast3.Bytes)):
3695         return ast.Constant(value=node.s)
3696
3697     if isinstance(node, (ast.Num, ast3.Num, ast27.Num)):
3698         return ast.Constant(value=node.n)
3699
3700     if isinstance(node, (ast.NameConstant, ast3.NameConstant)):
3701         return ast.Constant(value=node.value)
3702
3703     return node
3704
3705
3706 def assert_equivalent(src: str, dst: str) -> None:
3707     """Raise AssertionError if `src` and `dst` aren't equivalent."""
3708
3709     def _v(node: Union[ast.AST, ast3.AST, ast27.AST], depth: int = 0) -> Iterator[str]:
3710         """Simple visitor generating strings to compare ASTs by content."""
3711
3712         node = _fixup_ast_constants(node)
3713
3714         yield f"{'  ' * depth}{node.__class__.__name__}("
3715
3716         for field in sorted(node._fields):
3717             # TypeIgnore has only one field 'lineno' which breaks this comparison
3718             type_ignore_classes = (ast3.TypeIgnore, ast27.TypeIgnore)
3719             if sys.version_info >= (3, 8):
3720                 type_ignore_classes += (ast.TypeIgnore,)
3721             if isinstance(node, type_ignore_classes):
3722                 break
3723
3724             try:
3725                 value = getattr(node, field)
3726             except AttributeError:
3727                 continue
3728
3729             yield f"{'  ' * (depth+1)}{field}="
3730
3731             if isinstance(value, list):
3732                 for item in value:
3733                     # Ignore nested tuples within del statements, because we may insert
3734                     # parentheses and they change the AST.
3735                     if (
3736                         field == "targets"
3737                         and isinstance(node, (ast.Delete, ast3.Delete, ast27.Delete))
3738                         and isinstance(item, (ast.Tuple, ast3.Tuple, ast27.Tuple))
3739                     ):
3740                         for item in item.elts:
3741                             yield from _v(item, depth + 2)
3742
3743                     elif isinstance(item, (ast.AST, ast3.AST, ast27.AST)):
3744                         yield from _v(item, depth + 2)
3745
3746             elif isinstance(value, (ast.AST, ast3.AST, ast27.AST)):
3747                 yield from _v(value, depth + 2)
3748
3749             else:
3750                 yield f"{'  ' * (depth+2)}{value!r},  # {value.__class__.__name__}"
3751
3752         yield f"{'  ' * depth})  # /{node.__class__.__name__}"
3753
3754     try:
3755         src_ast = parse_ast(src)
3756     except Exception as exc:
3757         raise AssertionError(
3758             f"cannot use --safe with this file; failed to parse source file.  "
3759             f"AST error message: {exc}"
3760         )
3761
3762     try:
3763         dst_ast = parse_ast(dst)
3764     except Exception as exc:
3765         log = dump_to_file("".join(traceback.format_tb(exc.__traceback__)), dst)
3766         raise AssertionError(
3767             f"INTERNAL ERROR: Black produced invalid code: {exc}. "
3768             f"Please report a bug on https://github.com/psf/black/issues.  "
3769             f"This invalid output might be helpful: {log}"
3770         ) from None
3771
3772     src_ast_str = "\n".join(_v(src_ast))
3773     dst_ast_str = "\n".join(_v(dst_ast))
3774     if src_ast_str != dst_ast_str:
3775         log = dump_to_file(diff(src_ast_str, dst_ast_str, "src", "dst"))
3776         raise AssertionError(
3777             f"INTERNAL ERROR: Black produced code that is not equivalent to "
3778             f"the source.  "
3779             f"Please report a bug on https://github.com/psf/black/issues.  "
3780             f"This diff might be helpful: {log}"
3781         ) from None
3782
3783
3784 def assert_stable(src: str, dst: str, mode: FileMode) -> None:
3785     """Raise AssertionError if `dst` reformats differently the second time."""
3786     newdst = format_str(dst, mode=mode)
3787     if dst != newdst:
3788         log = dump_to_file(
3789             diff(src, dst, "source", "first pass"),
3790             diff(dst, newdst, "first pass", "second pass"),
3791         )
3792         raise AssertionError(
3793             f"INTERNAL ERROR: Black produced different code on the second pass "
3794             f"of the formatter.  "
3795             f"Please report a bug on https://github.com/psf/black/issues.  "
3796             f"This diff might be helpful: {log}"
3797         ) from None
3798
3799
3800 def dump_to_file(*output: str) -> str:
3801     """Dump `output` to a temporary file. Return path to the file."""
3802     with tempfile.NamedTemporaryFile(
3803         mode="w", prefix="blk_", suffix=".log", delete=False, encoding="utf8"
3804     ) as f:
3805         for lines in output:
3806             f.write(lines)
3807             if lines and lines[-1] != "\n":
3808                 f.write("\n")
3809     return f.name
3810
3811
3812 @contextmanager
3813 def nullcontext() -> Iterator[None]:
3814     """Return an empty context manager.
3815
3816     To be used like `nullcontext` in Python 3.7.
3817     """
3818     yield
3819
3820
3821 def diff(a: str, b: str, a_name: str, b_name: str) -> str:
3822     """Return a unified diff string between strings `a` and `b`."""
3823     import difflib
3824
3825     a_lines = [line + "\n" for line in a.split("\n")]
3826     b_lines = [line + "\n" for line in b.split("\n")]
3827     return "".join(
3828         difflib.unified_diff(a_lines, b_lines, fromfile=a_name, tofile=b_name, n=5)
3829     )
3830
3831
3832 def cancel(tasks: Iterable[asyncio.Task]) -> None:
3833     """asyncio signal handler that cancels all `tasks` and reports to stderr."""
3834     err("Aborted!")
3835     for task in tasks:
3836         task.cancel()
3837
3838
3839 def shutdown(loop: asyncio.AbstractEventLoop) -> None:
3840     """Cancel all pending tasks on `loop`, wait for them, and close the loop."""
3841     try:
3842         if sys.version_info[:2] >= (3, 7):
3843             all_tasks = asyncio.all_tasks
3844         else:
3845             all_tasks = asyncio.Task.all_tasks
3846         # This part is borrowed from asyncio/runners.py in Python 3.7b2.
3847         to_cancel = [task for task in all_tasks(loop) if not task.done()]
3848         if not to_cancel:
3849             return
3850
3851         for task in to_cancel:
3852             task.cancel()
3853         loop.run_until_complete(
3854             asyncio.gather(*to_cancel, loop=loop, return_exceptions=True)
3855         )
3856     finally:
3857         # `concurrent.futures.Future` objects cannot be cancelled once they
3858         # are already running. There might be some when the `shutdown()` happened.
3859         # Silence their logger's spew about the event loop being closed.
3860         cf_logger = logging.getLogger("concurrent.futures")
3861         cf_logger.setLevel(logging.CRITICAL)
3862         loop.close()
3863
3864
3865 def sub_twice(regex: Pattern[str], replacement: str, original: str) -> str:
3866     """Replace `regex` with `replacement` twice on `original`.
3867
3868     This is used by string normalization to perform replaces on
3869     overlapping matches.
3870     """
3871     return regex.sub(replacement, regex.sub(replacement, original))
3872
3873
3874 def re_compile_maybe_verbose(regex: str) -> Pattern[str]:
3875     """Compile a regular expression string in `regex`.
3876
3877     If it contains newlines, use verbose mode.
3878     """
3879     if "\n" in regex:
3880         regex = "(?x)" + regex
3881     compiled: Pattern[str] = re.compile(regex)
3882     return compiled
3883
3884
3885 def enumerate_reversed(sequence: Sequence[T]) -> Iterator[Tuple[Index, T]]:
3886     """Like `reversed(enumerate(sequence))` if that were possible."""
3887     index = len(sequence) - 1
3888     for element in reversed(sequence):
3889         yield (index, element)
3890         index -= 1
3891
3892
3893 def enumerate_with_length(
3894     line: Line, reversed: bool = False
3895 ) -> Iterator[Tuple[Index, Leaf, int]]:
3896     """Return an enumeration of leaves with their length.
3897
3898     Stops prematurely on multiline strings and standalone comments.
3899     """
3900     op = cast(
3901         Callable[[Sequence[Leaf]], Iterator[Tuple[Index, Leaf]]],
3902         enumerate_reversed if reversed else enumerate,
3903     )
3904     for index, leaf in op(line.leaves):
3905         length = len(leaf.prefix) + len(leaf.value)
3906         if "\n" in leaf.value:
3907             return  # Multiline strings, we can't continue.
3908
3909         for comment in line.comments_after(leaf):
3910             length += len(comment.value)
3911
3912         yield index, leaf, length
3913
3914
3915 def is_line_short_enough(line: Line, *, line_length: int, line_str: str = "") -> bool:
3916     """Return True if `line` is no longer than `line_length`.
3917
3918     Uses the provided `line_str` rendering, if any, otherwise computes a new one.
3919     """
3920     if not line_str:
3921         line_str = str(line).strip("\n")
3922     return (
3923         len(line_str) <= line_length
3924         and "\n" not in line_str  # multiline strings
3925         and not line.contains_standalone_comments()
3926     )
3927
3928
3929 def can_be_split(line: Line) -> bool:
3930     """Return False if the line cannot be split *for sure*.
3931
3932     This is not an exhaustive search but a cheap heuristic that we can use to
3933     avoid some unfortunate formattings (mostly around wrapping unsplittable code
3934     in unnecessary parentheses).
3935     """
3936     leaves = line.leaves
3937     if len(leaves) < 2:
3938         return False
3939
3940     if leaves[0].type == token.STRING and leaves[1].type == token.DOT:
3941         call_count = 0
3942         dot_count = 0
3943         next = leaves[-1]
3944         for leaf in leaves[-2::-1]:
3945             if leaf.type in OPENING_BRACKETS:
3946                 if next.type not in CLOSING_BRACKETS:
3947                     return False
3948
3949                 call_count += 1
3950             elif leaf.type == token.DOT:
3951                 dot_count += 1
3952             elif leaf.type == token.NAME:
3953                 if not (next.type == token.DOT or next.type in OPENING_BRACKETS):
3954                     return False
3955
3956             elif leaf.type not in CLOSING_BRACKETS:
3957                 return False
3958
3959             if dot_count > 1 and call_count > 1:
3960                 return False
3961
3962     return True
3963
3964
3965 def can_omit_invisible_parens(line: Line, line_length: int) -> bool:
3966     """Does `line` have a shape safe to reformat without optional parens around it?
3967
3968     Returns True for only a subset of potentially nice looking formattings but
3969     the point is to not return false positives that end up producing lines that
3970     are too long.
3971     """
3972     bt = line.bracket_tracker
3973     if not bt.delimiters:
3974         # Without delimiters the optional parentheses are useless.
3975         return True
3976
3977     max_priority = bt.max_delimiter_priority()
3978     if bt.delimiter_count_with_priority(max_priority) > 1:
3979         # With more than one delimiter of a kind the optional parentheses read better.
3980         return False
3981
3982     if max_priority == DOT_PRIORITY:
3983         # A single stranded method call doesn't require optional parentheses.
3984         return True
3985
3986     assert len(line.leaves) >= 2, "Stranded delimiter"
3987
3988     first = line.leaves[0]
3989     second = line.leaves[1]
3990     penultimate = line.leaves[-2]
3991     last = line.leaves[-1]
3992
3993     # With a single delimiter, omit if the expression starts or ends with
3994     # a bracket.
3995     if first.type in OPENING_BRACKETS and second.type not in CLOSING_BRACKETS:
3996         remainder = False
3997         length = 4 * line.depth
3998         for _index, leaf, leaf_length in enumerate_with_length(line):
3999             if leaf.type in CLOSING_BRACKETS and leaf.opening_bracket is first:
4000                 remainder = True
4001             if remainder:
4002                 length += leaf_length
4003                 if length > line_length:
4004                     break
4005
4006                 if leaf.type in OPENING_BRACKETS:
4007                     # There are brackets we can further split on.
4008                     remainder = False
4009
4010         else:
4011             # checked the entire string and line length wasn't exceeded
4012             if len(line.leaves) == _index + 1:
4013                 return True
4014
4015         # Note: we are not returning False here because a line might have *both*
4016         # a leading opening bracket and a trailing closing bracket.  If the
4017         # opening bracket doesn't match our rule, maybe the closing will.
4018
4019     if (
4020         last.type == token.RPAR
4021         or last.type == token.RBRACE
4022         or (
4023             # don't use indexing for omitting optional parentheses;
4024             # it looks weird
4025             last.type == token.RSQB
4026             and last.parent
4027             and last.parent.type != syms.trailer
4028         )
4029     ):
4030         if penultimate.type in OPENING_BRACKETS:
4031             # Empty brackets don't help.
4032             return False
4033
4034         if is_multiline_string(first):
4035             # Additional wrapping of a multiline string in this situation is
4036             # unnecessary.
4037             return True
4038
4039         length = 4 * line.depth
4040         seen_other_brackets = False
4041         for _index, leaf, leaf_length in enumerate_with_length(line):
4042             length += leaf_length
4043             if leaf is last.opening_bracket:
4044                 if seen_other_brackets or length <= line_length:
4045                     return True
4046
4047             elif leaf.type in OPENING_BRACKETS:
4048                 # There are brackets we can further split on.
4049                 seen_other_brackets = True
4050
4051     return False
4052
4053
4054 def get_cache_file(mode: FileMode) -> Path:
4055     return CACHE_DIR / f"cache.{mode.get_cache_key()}.pickle"
4056
4057
4058 def read_cache(mode: FileMode) -> Cache:
4059     """Read the cache if it exists and is well formed.
4060
4061     If it is not well formed, the call to write_cache later should resolve the issue.
4062     """
4063     cache_file = get_cache_file(mode)
4064     if not cache_file.exists():
4065         return {}
4066
4067     with cache_file.open("rb") as fobj:
4068         try:
4069             cache: Cache = pickle.load(fobj)
4070         except (pickle.UnpicklingError, ValueError):
4071             return {}
4072
4073     return cache
4074
4075
4076 def get_cache_info(path: Path) -> CacheInfo:
4077     """Return the information used to check if a file is already formatted or not."""
4078     stat = path.stat()
4079     return stat.st_mtime, stat.st_size
4080
4081
4082 def filter_cached(cache: Cache, sources: Iterable[Path]) -> Tuple[Set[Path], Set[Path]]:
4083     """Split an iterable of paths in `sources` into two sets.
4084
4085     The first contains paths of files that modified on disk or are not in the
4086     cache. The other contains paths to non-modified files.
4087     """
4088     todo, done = set(), set()
4089     for src in sources:
4090         src = src.resolve()
4091         if cache.get(src) != get_cache_info(src):
4092             todo.add(src)
4093         else:
4094             done.add(src)
4095     return todo, done
4096
4097
4098 def write_cache(cache: Cache, sources: Iterable[Path], mode: FileMode) -> None:
4099     """Update the cache file."""
4100     cache_file = get_cache_file(mode)
4101     try:
4102         CACHE_DIR.mkdir(parents=True, exist_ok=True)
4103         new_cache = {**cache, **{src.resolve(): get_cache_info(src) for src in sources}}
4104         with tempfile.NamedTemporaryFile(dir=str(cache_file.parent), delete=False) as f:
4105             pickle.dump(new_cache, f, protocol=4)
4106         os.replace(f.name, cache_file)
4107     except OSError:
4108         pass
4109
4110
4111 def patch_click() -> None:
4112     """Make Click not crash.
4113
4114     On certain misconfigured environments, Python 3 selects the ASCII encoding as the
4115     default which restricts paths that it can access during the lifetime of the
4116     application.  Click refuses to work in this scenario by raising a RuntimeError.
4117
4118     In case of Black the likelihood that non-ASCII characters are going to be used in
4119     file paths is minimal since it's Python source code.  Moreover, this crash was
4120     spurious on Python 3.7 thanks to PEP 538 and PEP 540.
4121     """
4122     try:
4123         from click import core
4124         from click import _unicodefun  # type: ignore
4125     except ModuleNotFoundError:
4126         return
4127
4128     for module in (core, _unicodefun):
4129         if hasattr(module, "_verify_python3_env"):
4130             module._verify_python3_env = lambda: None
4131
4132
4133 def patched_main() -> None:
4134     freeze_support()
4135     patch_click()
4136     main()
4137
4138
4139 if __name__ == "__main__":
4140     patched_main()