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

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