1"""
2This module provides the :module:`hyperscan` backend for :class:`~pathspec.gitignore.GitIgnoreSpec`.
3
4WARNING: The *pathspec._backends.hyperscan* package is not part of the public
5API. Its contents and structure are likely to change.
6"""
7from __future__ import annotations
8
9from collections.abc import (
10 Sequence)
11from typing import (
12 Any,
13 Callable, # Replaced by `collections.abc.Callable` in 3.9.2.
14 Optional, # Replaced by `X | None` in 3.10.
15 Union) # Replaced by `X | Y` in 3.10.
16
17try:
18 import hyperscan
19except ModuleNotFoundError:
20 hyperscan = None # type: ignore[assignment]
21
22from pathspec.pattern import (
23 RegexPattern)
24from pathspec.patterns.gitignore.spec import (
25 GitIgnoreSpecPattern,
26 _BYTES_ENCODING,
27 _DIR_MARK_CG,
28 _DIR_MARK_OPT)
29from pathspec._typing import (
30 override) # Added in 3.12.
31
32from .._utils import (
33 get_regex_flags,
34 translate_end_anchor,
35 translate_regex_flags)
36
37from ._base import (
38 HS_FLAGS,
39 HS_VERSION,
40 HS_VERSION_0_9_1,
41 HyperscanExprDat,
42 HyperscanExprDebug,
43 compat_convert_utf8)
44from .pathspec import (
45 HyperscanPsBackend)
46
47
48class HyperscanGiBackend(HyperscanPsBackend):
49 """
50 The :class:`HyperscanGiBackend` class is the :module:`hyperscan`
51 implementation used by :class:`~pathspec.gitignore.GitIgnoreSpec`. The
52 Hyperscan database uses block mode for matching files.
53 """
54
55 # Change type hint.
56 _out: tuple[Optional[bool], int, Optional[bool], int] # type: ignore[assignment]
57
58 def __init__(
59 self,
60 patterns: Sequence[RegexPattern],
61 *,
62 _debug_exprs: Optional[bool] = None,
63 _test_sort: Optional[Callable[[list], None]] = None,
64 ) -> None:
65 """
66 Initialize the :class:`HyperscanMatcher` instance.
67
68 *patterns* (:class:`Sequence` of :class:`.RegexPattern`) contains the
69 compiled patterns.
70 """
71 super().__init__(patterns, _debug_exprs=_debug_exprs, _test_sort=_test_sort)
72
73 self._out = (None, -1, None, -1)
74 """
75 *_out* (:class:`tuple`) stores the current match:
76
77 - *0* (:class:`bool` or :data:`None`) is the directory match include.
78
79 - *1* (:class:`int`) is the directory match index.
80
81 - *2* (:class:`bool` or :data:`None`) is the file match include.
82
83 - *3* (:class:`int`) is the file match index.
84 """
85
86 @override
87 @staticmethod
88 def _init_db(
89 db: hyperscan.Database, # type: ignore
90 debug: bool,
91 patterns: list[tuple[int, RegexPattern]],
92 sort_ids: Optional[Callable[[list[int]], None]],
93 ) -> list[HyperscanExprDat]:
94 """
95 Create the Hyperscan database from the given patterns.
96
97 *db* (:class:`hyperscan.Hyperscan`) is the Hyperscan database.
98
99 *debug* (:class:`bool`) is whether to include additional debugging
100 information for the expressions.
101
102 *patterns* (:class:`~collections.abc.Sequence` of :class:`.RegexPattern`)
103 contains the patterns.
104
105 *sort_ids* (:class:`callable` or :data:`None`) is a function used to sort
106 the compiled expression ids. This is used during testing to ensure the order
107 of expressions is not accidentally relied on.
108
109 Returns a :class:`list` indexed by expression id (:class:`int`) to its data
110 (:class:`HyperscanExprDat`).
111 """
112 # WARNING: Hyperscan raises a `hyperscan.error` exception when compiled with
113 # zero elements.
114 assert patterns, patterns
115
116 # Prepare patterns.
117 expr_data: list[HyperscanExprDat] = []
118 exprs: list[bytes] = []
119 for pattern_index, pattern in patterns:
120 pattern_include = pattern.include
121 assert pattern_include is not None, (pattern_index, pattern)
122 assert pattern.regex is not None, (pattern_index, pattern)
123
124 # Encode regex.
125 assert isinstance(pattern, RegexPattern), pattern
126 regex = pattern.regex.pattern
127
128 use_regexes: list[tuple[Union[str, bytes], bool]] = []
129 if isinstance(pattern, GitIgnoreSpecPattern):
130 # GitIgnoreSpecPattern uses capture groups for its directory marker but
131 # Hyperscan does not support capture groups. Handle this scenario.
132 regex_str: str
133 if isinstance(regex, str):
134 regex_str = regex
135 else:
136 assert isinstance(regex, bytes), regex
137 regex_str = regex.decode(_BYTES_ENCODING)
138
139 if _DIR_MARK_CG in regex_str:
140 # Found directory marker.
141 if regex_str.endswith(_DIR_MARK_OPT):
142 # Regex has optional directory marker. Split regex into directory
143 # and file variants. The directory variant must require a child
144 # segment: without one the query *is* that directory, which is a
145 # direct match and not an excluded ancestor.
146 base_regex = regex_str[:-len(_DIR_MARK_OPT)]
147 use_regexes.append((f'{base_regex}/(?s:.)', True))
148 use_regexes.append((rf'{base_regex}/?\z', False))
149 else:
150 # Remove capture group.
151 base_regex = regex_str.replace(_DIR_MARK_CG, '/')
152 use_regexes.append((f'{base_regex}(?s:.)', True))
153 use_regexes.append((rf'{base_regex}\z', False))
154
155 if not use_regexes:
156 # No special case for regex.
157 use_regexes.append((regex, False))
158
159 for regex, is_dir_pattern in use_regexes:
160 regex = translate_end_anchor(regex)
161 regex = translate_regex_flags(
162 regex,
163 get_regex_flags(pattern.regex),
164 )
165 if isinstance(regex, bytes):
166 regex_bytes = regex
167 else:
168 assert isinstance(regex, str), regex
169 if HS_VERSION >= HS_VERSION_0_9_1:
170 regex_bytes = regex.encode('utf8')
171 else:
172 # Convert unicode for older version of hyperscan.
173 regex_bytes = compat_convert_utf8(regex)
174
175 if debug:
176 expr_data.append(HyperscanExprDebug(
177 include=pattern_include,
178 index=pattern_index,
179 is_dir_pattern=is_dir_pattern,
180 regex=regex,
181 ))
182 else:
183 expr_data.append(HyperscanExprDat(
184 include=pattern_include,
185 index=pattern_index,
186 is_dir_pattern=is_dir_pattern,
187 ))
188
189 exprs.append(regex_bytes)
190
191 # Sort expressions.
192 ids = list(range(len(exprs)))
193 if sort_ids is not None:
194 sort_ids(ids)
195 exprs = [exprs[__id] for __id in ids]
196
197 # Compile patterns.
198 db.compile(
199 expressions=exprs,
200 ids=ids,
201 elements=len(exprs),
202 flags=HS_FLAGS,
203 )
204 return expr_data
205
206 @override
207 def match_file(self, file: str) -> tuple[Optional[bool], Optional[int]]:
208 """
209 Check the file against the patterns.
210
211 *file* (:class:`str`) is the normalized file path to check.
212
213 Returns a :class:`tuple` containing whether to include *file* (:class:`bool`
214 or :data:`None`), and the index of the last matched pattern (:class:`int` or
215 :data:`None`).
216 """
217 return self._match(file, check_ancestors=True)
218
219 def _match(
220 self, file: str, check_ancestors: bool,
221 ) -> tuple[Optional[bool], Optional[int]]:
222 """
223 Implements :meth:`match_file`. *check_ancestors* (:class:`bool`) is
224 whether to ask if an ancestor directory of *file* is excluded; see
225 :meth:`_match_ancestor` for why it can be skipped.
226 """
227 # NOTICE: According to benchmarking, a method callback is 13% faster than
228 # using a closure here.
229 db = self._db
230 if db is None:
231 # Database was not initialized because there were no patterns. Return no
232 # match.
233 return (None, None)
234
235 self._out = (None, -1, None, -1)
236 db.scan(file.encode('utf8'), match_event_handler=self.__on_match)
237
238 dir_include, dir_index, file_include, file_index = self._out
239 if check_ancestors and (dir_include is not None or file_include is not None):
240 # Neither file nor directory negations override an excluded ancestor.
241 # Some patterns (e.g. "a/**") match without a directory marker.
242 ancestor_index = self._match_ancestor(file)
243 if ancestor_index is not None:
244 return (True, ancestor_index)
245
246 if file_include is not None:
247 out_include, out_index = file_include, file_index
248 elif dir_include:
249 # An ancestor matched an exclude pattern, but the spec as a whole
250 # re-includes that ancestor, so the rule does not apply.
251 out_include, out_index = None, -1
252 else:
253 out_include, out_index = dir_include, dir_index
254
255 return (out_include, out_index if out_index != -1 else None)
256
257 def _match_ancestor(self, file: str) -> Optional[int]:
258 """
259 Return the excluding pattern index for a strict ancestor, or :data:`None`.
260 Git stops descending at the first excluded directory, so the ancestors are
261 asked outermost first, each as a directory query (trailing slash included).
262 By the time an ancestor is asked, every ancestor above it is known not to
263 be excluded, so it is matched without checking its own ancestors again;
264 otherwise each level re-asks all the levels above it and the work grows
265 exponentially with depth.
266 """
267 index = file.find('/')
268 while index != -1 and index + 1 < len(file):
269 ancestor_include, ancestor_index = self._match(
270 file[:index + 1], check_ancestors=False,
271 )
272 if ancestor_include:
273 return ancestor_index
274 index = file.find('/', index + 1)
275
276 return None
277
278
279 @override
280 def __on_match(
281 self,
282 expr_id: int,
283 _from: int,
284 _to: int,
285 _flags: int,
286 _context: Any,
287 ) -> Optional[bool]:
288 """
289 Called on each match.
290
291 *expr_id* (:class:`int`) is the expression id (index) of the matched
292 pattern.
293 """
294 expr_dat = self._expr_data[expr_id]
295
296 # WARNING: Hyperscan does not guarantee matches will be produced in order!
297 # Resolve the ancestor directory and the file separately: a file negation
298 # only applies while no ancestor directory is excluded.
299 include = expr_dat.include
300 index = expr_dat.index
301 dir_include, dir_index, file_include, file_index = self._out
302 if expr_dat.is_dir_pattern:
303 # Pattern matched by a directory pattern.
304 if index > dir_index:
305 self._out = (include, index, file_include, file_index) # type: ignore
306 elif index > file_index:
307 # Pattern matched by a file pattern.
308 self._out = (dir_include, dir_index, include, index) # type: ignore
309
310 return None