144 lines
4.5 KiB
Python

#!/usr/bin/env python3
import os
import argparse
import yaml
import json
from collections import deque
from typing import List, Dict, Any
def find_role_meta(roles_dir: str, role: str) -> str:
path = os.path.join(roles_dir, role, 'meta', 'main.yml')
if not os.path.isfile(path):
raise FileNotFoundError(f"Metadata not found for role: {role}")
return path
def load_meta(path: str) -> Dict[str, Any]:
"""
Load meta/main.yml → return galaxy_info + run_after + dependencies
"""
with open(path, 'r') as f:
data = yaml.safe_load(f) or {}
galaxy_info = data.get('galaxy_info', {}) or {}
return {
'galaxy_info': galaxy_info,
'run_after': galaxy_info.get('run_after', []) or [],
'dependencies': data.get('dependencies', []) or []
}
def build_single_graph(start_role: str, recurse_on: str, roles_dir: str) -> Dict[str, Any]:
"""
Build one graph for exactly one dependency type:
- includes all links for context
- recurses only on `recurse_on`
"""
nodes: Dict[str, Dict[str, Any]] = {}
links: List[Dict[str, str]] = []
visited = set()
queue = deque([start_role])
while queue:
role = queue.popleft()
if role in visited:
continue
visited.add(role)
try:
meta = load_meta(find_role_meta(roles_dir, role))
except FileNotFoundError:
continue
# register node
node = {'id': role}
node.update(meta['galaxy_info'])
node['doc_url'] = f"https://docs.cymais.cloud/roles/{role}/README.html"
node['source_url'] = f"https://github.com/kevinveenbirkenbach/cymais/tree/master/roles/{role}"
nodes[role] = node
# emit all links
for lt in ('run_after', 'dependencies'):
for tgt in meta[lt]:
links.append({'source': role, 'target': tgt, 'type': lt})
# recurse only on chosen type
for tgt in meta[recurse_on]:
if tgt not in visited:
queue.append(tgt)
return {'nodes': list(nodes.values()), 'links': links}
def build_graphs(start_role: str, types: List[str], roles_dir: str) -> Dict[str, Any]:
"""
If multiple types: return { type1: graph1, type2: graph2, ... }
If single type: return that single graph dict
"""
if len(types) == 1:
return build_single_graph(start_role, types[0], roles_dir)
combined = {}
for t in types:
combined[t] = build_single_graph(start_role, t, roles_dir)
return combined
def output_graph(graph_data: Any, fmt: str, start: str, types: List[str]):
"""
Write to file or print to console. If multiple types, join them in filename.
"""
if len(types) == 1:
base = f"{start}_{types[0]}"
else:
base = f"{start}_{'_'.join(types)}"
if fmt == 'console':
print(yaml.safe_dump(graph_data, sort_keys=False))
elif fmt in ('yaml', 'json'):
path = f"{base}.{fmt}"
with open(path, 'w') as f:
if fmt == 'yaml':
yaml.safe_dump(graph_data, f, sort_keys=False)
else:
json.dump(graph_data, f, indent=2)
print(f"Wrote {fmt.upper()} to {path}")
else:
raise ValueError(f"Unknown format: {fmt}")
def main():
script_dir = os.path.dirname(os.path.abspath(__file__))
default_roles_dir = os.path.abspath(os.path.join(script_dir, '..', '..', 'roles'))
parser = argparse.ArgumentParser(
description="Generate dependency graphs from Ansible roles' meta/main.yml"
)
parser.add_argument(
'-r', '--role', required=True,
help="Starting role name (directory under roles/)"
)
parser.add_argument(
'-t', '--type',
choices=['run_after', 'dependencies'],
default=['run_after'],
nargs='+',
help="Dependency type(s) to recurse on; can specify multiple"
)
parser.add_argument(
'-o', '--output',
choices=['yaml', 'json', 'console'],
default='yaml',
help="Output as files (yaml/json) or print to console"
)
parser.add_argument(
'--roles-dir',
default=default_roles_dir,
help=f"Path to the roles directory (default: {default_roles_dir})"
)
args = parser.parse_args()
graph_data = build_graphs(
start_role=args.role,
types=args.type,
roles_dir=args.roles_dir
)
output_graph(graph_data, args.output, args.role, args.type)
if __name__ == '__main__':
main()