]> 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:

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