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

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