File size: 2,603 Bytes
1550ce0 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 | """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:
# Direct file literals only. Prefixes/formatted paths belong in the registry.
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))
|