| """Read Python syntax only; Rust owns selection, scheduling and execution. |
| |
| Protocol is tab-separated UTF-8. Never import repository modules. Dynamic file |
| and subprocess dependencies must be declared in checks.tsv. |
| """ |
| import ast |
| from pathlib import Path |
| import sys |
|
|
|
|
| def scan(root, extra=()): |
| files = sorted((root / 'scripts').glob('*.py')) |
| modules = {p.stem: 'scripts/' + p.name for p in files} |
| for p in extra: |
| if p.startswith('scripts/') and p.endswith('.py'): |
| modules[Path(p).stem] = p |
| rows = set() |
| for path in files: |
| name = 'scripts/' + path.name |
| rows.add(('node', name)) |
| tree = ast.parse(path.read_text(), filename=name) |
| outputs = set() |
| for node in ast.walk(tree): |
| if isinstance(node, ast.Import): |
| imports = [a.name.split('.')[0] for a in node.names] |
| elif isinstance(node, ast.ImportFrom): |
| imports = [(node.module or '').split('.')[0]] |
| else: |
| imports = [] |
| for module in imports: |
| if module in modules: |
| rows.add(('edge', modules[module], name)) |
| if isinstance(node, ast.Call) and isinstance(node.func, ast.Name) and node.func.id in ('write_json', 'write_rows', 'write_gzip') and node.args: |
| dest = node.args[0] |
| if isinstance(dest, ast.Call) and isinstance(dest.func, ast.Name) and dest.func.id == 'Path' and dest.args and isinstance(dest.args[0], ast.Constant) and isinstance(dest.args[0].value, str): |
| outputs.add(dest.args[0].value) |
| rows.add(('output', name, dest.args[0].value)) |
| else: |
| rows.add(('dynamic', name)) |
| if isinstance(node, ast.Call) and ((isinstance(node.func, ast.Name) and node.func.id == '__import__') or (isinstance(node.func, ast.Attribute) and node.func.attr == 'import_module')): |
| rows.add(('unsafe', name)) |
| for node in ast.walk(tree): |
| if isinstance(node, ast.Constant) and isinstance(node.value, str): |
| value = node.value |
| if value.startswith(('docs/', 'assets/', '.cache/')) and '\n' not in value and '\t' not in value and value not in outputs: |
| |
| if Path(value).suffix: |
| rows.add(('edge', value, name)) |
| return sorted(rows) |
|
|
|
|
| if __name__ == '__main__': |
| for row in scan(Path.cwd(), sys.argv[1:]): |
| print('\t'.join(row)) |
|
|
