utils.py 11 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351
  1. """Utility methods for flake8."""
  2. import collections
  3. import fnmatch as _fnmatch
  4. import functools
  5. import io
  6. import logging
  7. import os
  8. import platform
  9. import re
  10. import sys
  11. import textwrap
  12. import tokenize
  13. from typing import Dict
  14. from typing import List
  15. from typing import NamedTuple
  16. from typing import Optional
  17. from typing import Pattern
  18. from typing import Sequence
  19. from typing import Set
  20. from typing import Tuple
  21. from typing import Union
  22. from flake8 import exceptions
  23. DIFF_HUNK_REGEXP = re.compile(r"^@@ -\d+(?:,\d+)? \+(\d+)(?:,(\d+))? @@.*$")
  24. COMMA_SEPARATED_LIST_RE = re.compile(r"[,\s]")
  25. LOCAL_PLUGIN_LIST_RE = re.compile(r"[,\t\n\r\f\v]")
  26. NORMALIZE_PACKAGE_NAME_RE = re.compile(r"[-_.]+")
  27. def parse_comma_separated_list(
  28. value: str, regexp: Pattern[str] = COMMA_SEPARATED_LIST_RE
  29. ) -> List[str]:
  30. """Parse a comma-separated list.
  31. :param value:
  32. String to be parsed and normalized.
  33. :param regexp:
  34. Compiled regular expression used to split the value when it is a
  35. string.
  36. :returns:
  37. List of values with whitespace stripped.
  38. """
  39. assert isinstance(value, str), value
  40. separated = regexp.split(value)
  41. item_gen = (item.strip() for item in separated)
  42. return [item for item in item_gen if item]
  43. class _Token(NamedTuple):
  44. tp: str
  45. src: str
  46. _CODE, _FILE, _COLON, _COMMA, _WS = "code", "file", "colon", "comma", "ws"
  47. _EOF = "eof"
  48. _FILE_LIST_TOKEN_TYPES = [
  49. (re.compile(r"[A-Z]+[0-9]*(?=$|\s|,)"), _CODE),
  50. (re.compile(r"[^\s:,]+"), _FILE),
  51. (re.compile(r"\s*:\s*"), _COLON),
  52. (re.compile(r"\s*,\s*"), _COMMA),
  53. (re.compile(r"\s+"), _WS),
  54. ]
  55. def _tokenize_files_to_codes_mapping(value: str) -> List[_Token]:
  56. tokens = []
  57. i = 0
  58. while i < len(value):
  59. for token_re, token_name in _FILE_LIST_TOKEN_TYPES:
  60. match = token_re.match(value, i)
  61. if match:
  62. tokens.append(_Token(token_name, match.group().strip()))
  63. i = match.end()
  64. break
  65. else:
  66. raise AssertionError("unreachable", value, i)
  67. tokens.append(_Token(_EOF, ""))
  68. return tokens
  69. def parse_files_to_codes_mapping( # noqa: C901
  70. value_: Union[Sequence[str], str]
  71. ) -> List[Tuple[str, List[str]]]:
  72. """Parse a files-to-codes mapping.
  73. A files-to-codes mapping a sequence of values specified as
  74. `filenames list:codes list ...`. Each of the lists may be separated by
  75. either comma or whitespace tokens.
  76. :param value: String to be parsed and normalized.
  77. """
  78. if not isinstance(value_, str):
  79. value = "\n".join(value_)
  80. else:
  81. value = value_
  82. ret: List[Tuple[str, List[str]]] = []
  83. if not value.strip():
  84. return ret
  85. class State:
  86. seen_sep = True
  87. seen_colon = False
  88. filenames: List[str] = []
  89. codes: List[str] = []
  90. def _reset() -> None:
  91. if State.codes:
  92. for filename in State.filenames:
  93. ret.append((filename, State.codes))
  94. State.seen_sep = True
  95. State.seen_colon = False
  96. State.filenames = []
  97. State.codes = []
  98. def _unexpected_token() -> exceptions.ExecutionError:
  99. return exceptions.ExecutionError(
  100. f"Expected `per-file-ignores` to be a mapping from file exclude "
  101. f"patterns to ignore codes.\n\n"
  102. f"Configured `per-file-ignores` setting:\n\n"
  103. f"{textwrap.indent(value.strip(), ' ')}"
  104. )
  105. for token in _tokenize_files_to_codes_mapping(value):
  106. # legal in any state: separator sets the sep bit
  107. if token.tp in {_COMMA, _WS}:
  108. State.seen_sep = True
  109. # looking for filenames
  110. elif not State.seen_colon:
  111. if token.tp == _COLON:
  112. State.seen_colon = True
  113. State.seen_sep = True
  114. elif State.seen_sep and token.tp == _FILE:
  115. State.filenames.append(token.src)
  116. State.seen_sep = False
  117. else:
  118. raise _unexpected_token()
  119. # looking for codes
  120. else:
  121. if token.tp == _EOF:
  122. _reset()
  123. elif State.seen_sep and token.tp == _CODE:
  124. State.codes.append(token.src)
  125. State.seen_sep = False
  126. elif State.seen_sep and token.tp == _FILE:
  127. _reset()
  128. State.filenames.append(token.src)
  129. State.seen_sep = False
  130. else:
  131. raise _unexpected_token()
  132. return ret
  133. def normalize_paths(
  134. paths: Sequence[str], parent: str = os.curdir
  135. ) -> List[str]:
  136. """Normalize a list of paths relative to a parent directory.
  137. :returns:
  138. The normalized paths.
  139. """
  140. assert isinstance(paths, list), paths
  141. return [normalize_path(p, parent) for p in paths]
  142. def normalize_path(path: str, parent: str = os.curdir) -> str:
  143. """Normalize a single-path.
  144. :returns:
  145. The normalized path.
  146. """
  147. # NOTE(sigmavirus24): Using os.path.sep and os.path.altsep allow for
  148. # Windows compatibility with both Windows-style paths (c:\foo\bar) and
  149. # Unix style paths (/foo/bar).
  150. separator = os.path.sep
  151. # NOTE(sigmavirus24): os.path.altsep may be None
  152. alternate_separator = os.path.altsep or ""
  153. if (
  154. path == "."
  155. or separator in path
  156. or (alternate_separator and alternate_separator in path)
  157. ):
  158. path = os.path.abspath(os.path.join(parent, path))
  159. return path.rstrip(separator + alternate_separator)
  160. @functools.lru_cache(maxsize=1)
  161. def stdin_get_value() -> str:
  162. """Get and cache it so plugins can use it."""
  163. stdin_value = sys.stdin.buffer.read()
  164. fd = io.BytesIO(stdin_value)
  165. try:
  166. coding, _ = tokenize.detect_encoding(fd.readline)
  167. fd.seek(0)
  168. return io.TextIOWrapper(fd, coding).read()
  169. except (LookupError, SyntaxError, UnicodeError):
  170. return stdin_value.decode("utf-8")
  171. def stdin_get_lines() -> List[str]:
  172. """Return lines of stdin split according to file splitting."""
  173. return list(io.StringIO(stdin_get_value()))
  174. def parse_unified_diff(diff: Optional[str] = None) -> Dict[str, Set[int]]:
  175. """Parse the unified diff passed on stdin.
  176. :returns:
  177. dictionary mapping file names to sets of line numbers
  178. """
  179. # Allow us to not have to patch out stdin_get_value
  180. if diff is None:
  181. diff = stdin_get_value()
  182. number_of_rows = None
  183. current_path = None
  184. parsed_paths: Dict[str, Set[int]] = collections.defaultdict(set)
  185. for line in diff.splitlines():
  186. if number_of_rows:
  187. if not line or line[0] != "-":
  188. number_of_rows -= 1
  189. # We're in the part of the diff that has lines starting with +, -,
  190. # and ' ' to show context and the changes made. We skip these
  191. # because the information we care about is the filename and the
  192. # range within it.
  193. # When number_of_rows reaches 0, we will once again start
  194. # searching for filenames and ranges.
  195. continue
  196. # NOTE(sigmavirus24): Diffs that we support look roughly like:
  197. # diff a/file.py b/file.py
  198. # ...
  199. # --- a/file.py
  200. # +++ b/file.py
  201. # Below we're looking for that last line. Every diff tool that
  202. # gives us this output may have additional information after
  203. # ``b/file.py`` which it will separate with a \t, e.g.,
  204. # +++ b/file.py\t100644
  205. # Which is an example that has the new file permissions/mode.
  206. # In this case we only care about the file name.
  207. if line[:3] == "+++":
  208. current_path = line[4:].split("\t", 1)[0]
  209. # NOTE(sigmavirus24): This check is for diff output from git.
  210. if current_path[:2] == "b/":
  211. current_path = current_path[2:]
  212. # We don't need to do anything else. We have set up our local
  213. # ``current_path`` variable. We can skip the rest of this loop.
  214. # The next line we will see will give us the hung information
  215. # which is in the next section of logic.
  216. continue
  217. hunk_match = DIFF_HUNK_REGEXP.match(line)
  218. # NOTE(sigmavirus24): pep8/pycodestyle check for:
  219. # line[:3] == '@@ '
  220. # But the DIFF_HUNK_REGEXP enforces that the line start with that
  221. # So we can more simply check for a match instead of slicing and
  222. # comparing.
  223. if hunk_match:
  224. (row, number_of_rows) = (
  225. 1 if not group else int(group) for group in hunk_match.groups()
  226. )
  227. assert current_path is not None
  228. parsed_paths[current_path].update(range(row, row + number_of_rows))
  229. # We have now parsed our diff into a dictionary that looks like:
  230. # {'file.py': set(range(10, 16), range(18, 20)), ...}
  231. return parsed_paths
  232. def is_using_stdin(paths: List[str]) -> bool:
  233. """Determine if we're going to read from stdin.
  234. :param paths:
  235. The paths that we're going to check.
  236. :returns:
  237. True if stdin (-) is in the path, otherwise False
  238. """
  239. return "-" in paths
  240. def fnmatch(filename: str, patterns: Sequence[str]) -> bool:
  241. """Wrap :func:`fnmatch.fnmatch` to add some functionality.
  242. :param filename:
  243. Name of the file we're trying to match.
  244. :param patterns:
  245. Patterns we're using to try to match the filename.
  246. :param default:
  247. The default value if patterns is empty
  248. :returns:
  249. True if a pattern matches the filename, False if it doesn't.
  250. ``True`` if patterns is empty.
  251. """
  252. if not patterns:
  253. return True
  254. return any(_fnmatch.fnmatch(filename, pattern) for pattern in patterns)
  255. def matches_filename(
  256. path: str,
  257. patterns: Sequence[str],
  258. log_message: str,
  259. logger: logging.Logger,
  260. ) -> bool:
  261. """Use fnmatch to discern if a path exists in patterns.
  262. :param path:
  263. The path to the file under question
  264. :param patterns:
  265. The patterns to match the path against.
  266. :param log_message:
  267. The message used for logging purposes.
  268. :returns:
  269. True if path matches patterns, False otherwise
  270. """
  271. if not patterns:
  272. return False
  273. basename = os.path.basename(path)
  274. if basename not in {".", ".."} and fnmatch(basename, patterns):
  275. logger.debug(log_message, {"path": basename, "whether": ""})
  276. return True
  277. absolute_path = os.path.abspath(path)
  278. match = fnmatch(absolute_path, patterns)
  279. logger.debug(
  280. log_message,
  281. {"path": absolute_path, "whether": "" if match else "not "},
  282. )
  283. return match
  284. def get_python_version() -> str:
  285. """Find and format the python implementation and version.
  286. :returns:
  287. Implementation name, version, and platform as a string.
  288. """
  289. return "{} {} on {}".format(
  290. platform.python_implementation(),
  291. platform.python_version(),
  292. platform.system(),
  293. )
  294. def normalize_pypi_name(s: str) -> str:
  295. """Normalize a distribution name according to PEP 503."""
  296. return NORMALIZE_PACKAGE_NAME_RE.sub("-", s).lower()