(filename, is_builtin=False)
| 230 | |
| 231 | # scan |
| 232 | def scan_in_file(filename, is_builtin=False): |
| 233 | global builtin_nodes |
| 234 | |
| 235 | with open(filename, encoding='utf-8', errors='ignore') as file: |
| 236 | code = file.read() |
| 237 | |
| 238 | # Support type annotations (e.g., NODE_CLASS_MAPPINGS: Type = {...}) and line continuations (\) |
| 239 | pattern = r"_CLASS_MAPPINGS\s*(?::\s*\w+\s*)?=\s*(?:\\\s*)?{([^}]*)}" |
| 240 | regex = re.compile(pattern, re.MULTILINE | re.DOTALL) |
| 241 | |
| 242 | nodes = set() |
| 243 | class_dict = {} |
| 244 | |
| 245 | # V1 nodes detection |
| 246 | nodes |= extract_nodes(code) |
| 247 | |
| 248 | # V3 nodes detection |
| 249 | nodes |= extract_v3_nodes(code) |
| 250 | code = re.sub(r'^#.*?$', '', code, flags=re.MULTILINE) |
| 251 | |
| 252 | def extract_keys(pattern, code): |
| 253 | keys = re.findall(pattern, code) |
| 254 | return {key.strip() for key in keys} |
| 255 | |
| 256 | def update_nodes(nodes, new_keys): |
| 257 | nodes |= new_keys |
| 258 | |
| 259 | patterns = [ |
| 260 | r'^[^=]*_CLASS_MAPPINGS\["(.*?)"\]', |
| 261 | r'^[^=]*_CLASS_MAPPINGS\[\'(.*?)\'\]', |
| 262 | r'@register_node\("(.+)",\s*\".+"\)', |
| 263 | r'"(\w+)"\s*:\s*{"class":\s*\w+\s*' |
| 264 | ] |
| 265 | |
| 266 | with concurrent.futures.ThreadPoolExecutor() as executor: |
| 267 | futures = {executor.submit(extract_keys, pattern, code): pattern for pattern in patterns} |
| 268 | for future in concurrent.futures.as_completed(futures): |
| 269 | update_nodes(nodes, future.result()) |
| 270 | |
| 271 | matches = regex.findall(code) |
| 272 | for match in matches: |
| 273 | dict_text = match |
| 274 | |
| 275 | key_value_pairs = re.findall(r"\"([^\"]*)\"\s*:\s*([^,\n]*)", dict_text) |
| 276 | for key, value in key_value_pairs: |
| 277 | class_dict[key.strip()] = value.strip() |
| 278 | |
| 279 | key_value_pairs = re.findall(r"'([^']*)'\s*:\s*([^,\n]*)", dict_text) |
| 280 | for key, value in key_value_pairs: |
| 281 | class_dict[key.strip()] = value.strip() |
| 282 | |
| 283 | for key, value in class_dict.items(): |
| 284 | nodes.add(key.strip()) |
| 285 | |
| 286 | update_pattern = r"_CLASS_MAPPINGS.update\s*\({([^}]*)}\)" |
| 287 | update_match = re.search(update_pattern, code) |
| 288 | if update_match: |
| 289 | update_dict_text = update_match.group(1) |
no test coverage detected