| 7 | from concurrent.futures import ThreadPoolExecutor |
| 8 | |
| 9 | class AsyncRegexChecker: |
| 10 | def __init__(self, jsonl_file): |
| 11 | """Load regex patterns from a JSONL file.""" |
| 12 | self.patterns = self.load_patterns(jsonl_file) |
| 13 | self.executor = ThreadPoolExecutor() |
| 14 | |
| 15 | def compile_pattern(self, pattern): |
| 16 | with warnings.catch_warnings(): |
| 17 | warnings.simplefilter("ignore", category=DeprecationWarning) |
| 18 | try: |
| 19 | return re.compile(pattern) |
| 20 | except re.error as e: |
| 21 | print(f"Regex compilation error for pattern: {pattern[:50]}... -> {e}") |
| 22 | return None |
| 23 | |
| 24 | def load_patterns(self, jsonl_file): |
| 25 | """Load all regex patterns from a JSONL file.""" |
| 26 | patterns = [] |
| 27 | with open(jsonl_file, 'r', encoding='utf-8') as file: |
| 28 | for line in file: |
| 29 | rule = json.loads(line.strip()) |
| 30 | patterns.append({'name': rule['name'], 'id': rule['id'], 'pattern': self.compile_pattern(rule['pattern'])}) # Precompile regex |
| 31 | return patterns |
| 32 | |
| 33 | async def check_document(self, document): |
| 34 | """Check document against regex rules asynchronously.""" |
| 35 | loop = asyncio.get_running_loop() |
| 36 | tasks = [ |
| 37 | loop.run_in_executor(self.executor, self.match_pattern, rule, document) |
| 38 | for rule in self.patterns |
| 39 | ] |
| 40 | results = await asyncio.gather(*tasks) |
| 41 | return [match for match in results if match] |
| 42 | |
| 43 | def match_pattern(self, rule, document): |
| 44 | """Match a single regex pattern synchronously (runs in a thread pool).""" |
| 45 | matches = [ |
| 46 | {'rule_name': rule['name'], 'rule_id': rule['id'], 'match': m.group(), 'start': m.start(), 'end': m.end()} |
| 47 | for m in rule['pattern'].finditer(document) |
| 48 | ] |
| 49 | return matches if matches else None |
| 50 | |
| 51 | |
| 52 | if __name__ == "__main__": |
no outgoing calls
no test coverage detected