Navegador

navegador / navegador / ingestion / ansible.py
Source Blame History 674 lines
b45288f… lmata 1 """
b45288f… lmata 2 Ansible playbook/task parser — extracts plays, tasks, handlers, roles,
b45288f… lmata 3 and variables from Ansible YAML files into the navegador graph.
b45288f… lmata 4
b45288f… lmata 5 Unlike other parsers this does NOT use tree-sitter. Ansible semantics
b45288f… lmata 6 are encoded in YAML structure (dicts with well-known keys like ``hosts``,
b45288f… lmata 7 ``tasks``, ``handlers``), so we parse with ``yaml.safe_load()`` and walk
b45288f… lmata 8 the resulting Python data structures directly.
b45288f… lmata 9
b45288f… lmata 10 Invoked via a hook in RepoIngester rather than through LANGUAGE_MAP.
b45288f… lmata 11 """
b45288f… lmata 12
b45288f… lmata 13 import logging
b45288f… lmata 14 import re
b45288f… lmata 15 from pathlib import Path
b45288f… lmata 16
b45288f… lmata 17 from navegador.graph.schema import EdgeType, NodeLabel
b45288f… lmata 18 from navegador.graph.store import GraphStore
b45288f… lmata 19 from navegador.ingestion.parser import LanguageParser
b45288f… lmata 20
b45288f… lmata 21 logger = logging.getLogger(__name__)
b45288f… lmata 22
b45288f… lmata 23 # Well-known Ansible module names — used to identify task dicts that lack
b45288f… lmata 24 # an explicit ``name`` key and to extract the module used by a task.
b45288f… lmata 25 _ANSIBLE_MODULES = {
b45288f… lmata 26 "apt",
b45288f… lmata 27 "yum",
b45288f… lmata 28 "dnf",
b45288f… lmata 29 "pip",
b45288f… lmata 30 "gem",
b45288f… lmata 31 "npm",
b45288f… lmata 32 "copy",
b45288f… lmata 33 "template",
b45288f… lmata 34 "file",
b45288f… lmata 35 "lineinfile",
b45288f… lmata 36 "blockinfile",
b45288f… lmata 37 "service",
b45288f… lmata 38 "systemd",
b45288f… lmata 39 "command",
b45288f… lmata 40 "shell",
b45288f… lmata 41 "raw",
b45288f… lmata 42 "script",
b45288f… lmata 43 "git",
b45288f… lmata 44 "get_url",
b45288f… lmata 45 "uri",
b45288f… lmata 46 "unarchive",
b45288f… lmata 47 "user",
b45288f… lmata 48 "group",
b45288f… lmata 49 "cron",
b45288f… lmata 50 "mount",
b45288f… lmata 51 "docker_container",
b45288f… lmata 52 "docker_image",
b45288f… lmata 53 "k8s",
b45288f… lmata 54 "helm",
b45288f… lmata 55 "debug",
b45288f… lmata 56 "assert",
b45288f… lmata 57 "fail",
b45288f… lmata 58 "set_fact",
b45288f… lmata 59 "include_tasks",
b45288f… lmata 60 "import_tasks",
b45288f… lmata 61 "include_role",
b45288f… lmata 62 "import_role",
b45288f… lmata 63 "block",
b45288f… lmata 64 "rescue",
b45288f… lmata 65 "always",
b45288f… lmata 66 "wait_for",
b45288f… lmata 67 "pause",
b45288f… lmata 68 "stat",
b45288f… lmata 69 "find",
b45288f… lmata 70 "replace",
b45288f… lmata 71 "package",
b45288f… lmata 72 "hostname",
b45288f… lmata 73 "timezone",
b45288f… lmata 74 "sysctl",
b45288f… lmata 75 "authorized_key",
b45288f… lmata 76 "firewalld",
b45288f… lmata 77 "iptables",
b45288f… lmata 78 "aws_s3",
b45288f… lmata 79 "ec2",
b45288f… lmata 80 "ec2_instance",
b45288f… lmata 81 "s3_bucket",
b45288f… lmata 82 "ansible.builtin.copy",
b45288f… lmata 83 "ansible.builtin.template",
b45288f… lmata 84 "ansible.builtin.file",
b45288f… lmata 85 "ansible.builtin.command",
b45288f… lmata 86 "ansible.builtin.shell",
b45288f… lmata 87 "ansible.builtin.service",
b45288f… lmata 88 "ansible.builtin.debug",
b45288f… lmata 89 "ansible.builtin.set_fact",
b45288f… lmata 90 "ansible.builtin.include_tasks",
b45288f… lmata 91 "ansible.builtin.import_tasks",
b45288f… lmata 92 "ansible.builtin.include_role",
b45288f… lmata 93 "ansible.builtin.import_role",
b45288f… lmata 94 "ansible.builtin.apt",
b45288f… lmata 95 "ansible.builtin.yum",
b45288f… lmata 96 "ansible.builtin.pip",
b45288f… lmata 97 "ansible.builtin.git",
b45288f… lmata 98 "ansible.builtin.user",
b45288f… lmata 99 "ansible.builtin.group",
b45288f… lmata 100 "ansible.builtin.uri",
b45288f… lmata 101 "ansible.builtin.get_url",
b45288f… lmata 102 "ansible.builtin.lineinfile",
b45288f… lmata 103 "ansible.builtin.blockinfile",
b45288f… lmata 104 "ansible.builtin.systemd",
b45288f… lmata 105 "ansible.builtin.raw",
b45288f… lmata 106 "ansible.builtin.script",
b45288f… lmata 107 "ansible.builtin.unarchive",
b45288f… lmata 108 "ansible.builtin.assert",
b45288f… lmata 109 "ansible.builtin.fail",
b45288f… lmata 110 "ansible.builtin.wait_for",
b45288f… lmata 111 "ansible.builtin.pause",
b45288f… lmata 112 "ansible.builtin.stat",
b45288f… lmata 113 "ansible.builtin.find",
b45288f… lmata 114 "ansible.builtin.replace",
b45288f… lmata 115 "ansible.builtin.package",
b45288f… lmata 116 }
b45288f… lmata 117
b45288f… lmata 118 # Patterns in file paths that strongly suggest Ansible content
b45288f… lmata 119 _ROLE_TASKS_RE = re.compile(r"roles/[^/]+/tasks/")
b45288f… lmata 120 _ROLE_HANDLERS_RE = re.compile(r"roles/[^/]+/handlers/")
b45288f… lmata 121 _ROLE_DEFAULTS_RE = re.compile(r"roles/[^/]+/defaults/")
b45288f… lmata 122 _ROLE_VARS_RE = re.compile(r"roles/[^/]+/vars/")
b45288f… lmata 123 _PLAYBOOKS_DIR_RE = re.compile(r"(^|/)playbooks/")
b45288f… lmata 124 _COMMON_PLAYBOOK_RE = re.compile(
b45288f… lmata 125 r"(^|/)(playbook[^/]*|site|main|common|deploy|provision|setup|configure)\.(yml|yaml)$"
b45288f… lmata 126 )
b45288f… lmata 127 _GROUP_VARS_RE = re.compile(r"(^|/)group_vars/")
b45288f… lmata 128 _HOST_VARS_RE = re.compile(r"(^|/)host_vars/")
b45288f… lmata 129
b45288f… lmata 130
b45288f… lmata 131 class AnsibleParser(LanguageParser):
b45288f… lmata 132 """Parses Ansible YAML files into the navegador graph."""
b45288f… lmata 133
b45288f… lmata 134 def __init__(self) -> None:
b45288f… lmata 135 pass # no tree-sitter parser needed
b45288f… lmata 136
b45288f… lmata 137 @staticmethod
b45288f… lmata 138 def is_ansible_file(path: Path, repo_root: Path | None = None) -> bool:
b45288f… lmata 139 """Return True if *path* looks like an Ansible YAML file."""
b45288f… lmata 140 if path.suffix not in (".yml", ".yaml"):
b45288f… lmata 141 return False
b45288f… lmata 142
b45288f… lmata 143 rel = str(path)
b45288f… lmata 144 if repo_root is not None:
b45288f… lmata 145 try:
b45288f… lmata 146 rel = str(path.relative_to(repo_root))
b45288f… lmata 147 except ValueError:
b45288f… lmata 148 pass
b45288f… lmata 149
b45288f… lmata 150 # Structural heuristics based on path
b45288f… lmata 151 if _ROLE_TASKS_RE.search(rel):
b45288f… lmata 152 return True
b45288f… lmata 153 if _ROLE_HANDLERS_RE.search(rel):
b45288f… lmata 154 return True
b45288f… lmata 155 if _ROLE_DEFAULTS_RE.search(rel):
b45288f… lmata 156 return True
b45288f… lmata 157 if _ROLE_VARS_RE.search(rel):
b45288f… lmata 158 return True
b45288f… lmata 159 if _PLAYBOOKS_DIR_RE.search(rel):
b45288f… lmata 160 return True
b45288f… lmata 161 if _GROUP_VARS_RE.search(rel):
b45288f… lmata 162 return True
b45288f… lmata 163 if _HOST_VARS_RE.search(rel):
b45288f… lmata 164 return True
b45288f… lmata 165
b45288f… lmata 166 # ansible.cfg sibling in repo root
b45288f… lmata 167 if repo_root is not None and (repo_root / "ansible.cfg").exists():
b45288f… lmata 168 if _COMMON_PLAYBOOK_RE.search(rel):
b45288f… lmata 169 return True
b45288f… lmata 170
b45288f… lmata 171 # Content-based: top-level list whose items contain "hosts:" key
b45288f… lmata 172 try:
b45288f… lmata 173 text = path.read_text(encoding="utf-8", errors="replace")
b45288f… lmata 174 except OSError:
b45288f… lmata 175 return False
b45288f… lmata 176
b45288f… lmata 177 if not text.lstrip().startswith("---"):
b45288f… lmata 178 return False
b45288f… lmata 179
b45288f… lmata 180 try:
66c6384… lmata 181 import yaml
66c6384… lmata 182
b45288f… lmata 183 data = yaml.safe_load(text)
66c6384… lmata 184 except Exception:
b45288f… lmata 185 return False
b45288f… lmata 186
b45288f… lmata 187 if isinstance(data, list) and data:
b45288f… lmata 188 if any(isinstance(item, dict) and "hosts" in item for item in data):
b45288f… lmata 189 return True
b45288f… lmata 190
b45288f… lmata 191 return False
b45288f… lmata 192
b45288f… lmata 193 # ── Main entry point ─────────────────────────────────────────────────────
b45288f… lmata 194
b45288f… lmata 195 def parse_file(self, path: Path, repo_root: Path, store: GraphStore) -> dict[str, int]:
b45288f… lmata 196 rel_path = str(path.relative_to(repo_root))
b45288f… lmata 197 stats = {"functions": 0, "classes": 0, "edges": 0}
b45288f… lmata 198
b45288f… lmata 199 try:
66c6384… lmata 200 import yaml
66c6384… lmata 201
b45288f… lmata 202 text = path.read_text(encoding="utf-8", errors="replace")
b45288f… lmata 203 data = yaml.safe_load(text)
66c6384… lmata 204 except Exception as exc:
b45288f… lmata 205 logger.warning("Could not parse Ansible file %s: %s", rel_path, exc)
b45288f… lmata 206 return stats
b45288f… lmata 207
b45288f… lmata 208 if data is None:
b45288f… lmata 209 return stats
b45288f… lmata 210
b45288f… lmata 211 # File node
b45288f… lmata 212 store.create_node(
b45288f… lmata 213 NodeLabel.File,
b45288f… lmata 214 {
b45288f… lmata 215 "name": path.name,
b45288f… lmata 216 "path": rel_path,
b45288f… lmata 217 "language": "ansible",
b45288f… lmata 218 "line_count": text.count("\n"),
b45288f… lmata 219 },
b45288f… lmata 220 )
b45288f… lmata 221
b45288f… lmata 222 rel_str = rel_path.replace("\\", "/")
b45288f… lmata 223
b45288f… lmata 224 # Dispatch based on file type
b45288f… lmata 225 if _ROLE_DEFAULTS_RE.search(rel_str) or _ROLE_VARS_RE.search(rel_str):
b45288f… lmata 226 self._parse_variable_file(data, rel_path, store, stats)
b45288f… lmata 227 elif _GROUP_VARS_RE.search(rel_str) or _HOST_VARS_RE.search(rel_str):
b45288f… lmata 228 self._parse_variable_file(data, rel_path, store, stats)
b45288f… lmata 229 elif _ROLE_HANDLERS_RE.search(rel_str):
b45288f… lmata 230 self._parse_handler_file(data, rel_path, store, stats)
b45288f… lmata 231 elif _ROLE_TASKS_RE.search(rel_str):
b45288f… lmata 232 self._parse_task_file(data, rel_path, store, stats)
b45288f… lmata 233 elif (
b45288f… lmata 234 isinstance(data, list)
b45288f… lmata 235 and data
b45288f… lmata 236 and any(isinstance(item, dict) and "hosts" in item for item in data)
b45288f… lmata 237 ):
b45288f… lmata 238 self._parse_playbook(data, rel_path, store, stats)
b45288f… lmata 239 elif isinstance(data, list):
b45288f… lmata 240 # Might be a task list (e.g. included task file)
b45288f… lmata 241 self._parse_task_file(data, rel_path, store, stats)
b45288f… lmata 242 elif isinstance(data, dict):
b45288f… lmata 243 # Standalone variable file
b45288f… lmata 244 self._parse_variable_file(data, rel_path, store, stats)
b45288f… lmata 245
b45288f… lmata 246 return stats
b45288f… lmata 247
b45288f… lmata 248 # ── Playbook parsing ─────────────────────────────────────────────────────
b45288f… lmata 249
b45288f… lmata 250 def _parse_playbook(
b45288f… lmata 251 self,
b45288f… lmata 252 data: list,
b45288f… lmata 253 file_path: str,
b45288f… lmata 254 store: GraphStore,
b45288f… lmata 255 stats: dict,
b45288f… lmata 256 ) -> None:
b45288f… lmata 257 """Parse a full playbook (list of plays)."""
b45288f… lmata 258 playbook_name = Path(file_path).stem
b45288f… lmata 259
b45288f… lmata 260 # Module node for the playbook file
b45288f… lmata 261 store.create_node(
b45288f… lmata 262 NodeLabel.Module,
b45288f… lmata 263 {
b45288f… lmata 264 "name": playbook_name,
b45288f… lmata 265 "file_path": file_path,
b45288f… lmata 266 "docstring": "",
b45288f… lmata 267 "semantic_type": "ansible_playbook",
b45288f… lmata 268 },
b45288f… lmata 269 )
b45288f… lmata 270 store.create_edge(
b45288f… lmata 271 NodeLabel.File,
b45288f… lmata 272 {"path": file_path},
b45288f… lmata 273 EdgeType.CONTAINS,
b45288f… lmata 274 NodeLabel.Module,
b45288f… lmata 275 {"name": playbook_name, "file_path": file_path},
b45288f… lmata 276 )
b45288f… lmata 277 stats["edges"] += 1
b45288f… lmata 278
b45288f… lmata 279 for play in data:
b45288f… lmata 280 if not isinstance(play, dict):
b45288f… lmata 281 continue
b45288f… lmata 282 if "hosts" not in play:
b45288f… lmata 283 continue
b45288f… lmata 284 self._parse_play(play, file_path, playbook_name, store, stats)
b45288f… lmata 285
b45288f… lmata 286 def _parse_play(
b45288f… lmata 287 self,
b45288f… lmata 288 play: dict,
b45288f… lmata 289 file_path: str,
b45288f… lmata 290 playbook_name: str,
b45288f… lmata 291 store: GraphStore,
b45288f… lmata 292 stats: dict,
b45288f… lmata 293 ) -> None:
b45288f… lmata 294 """Parse a single play dict."""
b45288f… lmata 295 play_name = play.get("name", f"play:{play.get('hosts', 'unknown')}")
b45288f… lmata 296
b45288f… lmata 297 store.create_node(
b45288f… lmata 298 NodeLabel.Class,
b45288f… lmata 299 {
b45288f… lmata 300 "name": play_name,
b45288f… lmata 301 "file_path": file_path,
b45288f… lmata 302 "line_start": 0,
b45288f… lmata 303 "line_end": 0,
b45288f… lmata 304 "docstring": f"hosts: {play.get('hosts', '')}",
b45288f… lmata 305 "semantic_type": "ansible_play",
b45288f… lmata 306 },
b45288f… lmata 307 )
b45288f… lmata 308 store.create_edge(
b45288f… lmata 309 NodeLabel.Module,
b45288f… lmata 310 {"name": playbook_name, "file_path": file_path},
b45288f… lmata 311 EdgeType.CONTAINS,
b45288f… lmata 312 NodeLabel.Class,
b45288f… lmata 313 {"name": play_name, "file_path": file_path},
b45288f… lmata 314 )
b45288f… lmata 315 stats["classes"] += 1
b45288f… lmata 316 stats["edges"] += 1
b45288f… lmata 317
b45288f… lmata 318 # Tasks
b45288f… lmata 319 for task_dict in play.get("tasks", []) or []:
b45288f… lmata 320 if isinstance(task_dict, dict):
b45288f… lmata 321 self._parse_task(task_dict, file_path, play_name, store, stats)
b45288f… lmata 322
b45288f… lmata 323 # Pre-tasks
b45288f… lmata 324 for task_dict in play.get("pre_tasks", []) or []:
b45288f… lmata 325 if isinstance(task_dict, dict):
b45288f… lmata 326 self._parse_task(task_dict, file_path, play_name, store, stats)
b45288f… lmata 327
b45288f… lmata 328 # Post-tasks
b45288f… lmata 329 for task_dict in play.get("post_tasks", []) or []:
b45288f… lmata 330 if isinstance(task_dict, dict):
b45288f… lmata 331 self._parse_task(task_dict, file_path, play_name, store, stats)
b45288f… lmata 332
b45288f… lmata 333 # Handlers
b45288f… lmata 334 for handler_dict in play.get("handlers", []) or []:
b45288f… lmata 335 if isinstance(handler_dict, dict):
b45288f… lmata 336 self._parse_handler(handler_dict, file_path, play_name, store, stats)
b45288f… lmata 337
b45288f… lmata 338 # Roles
b45288f… lmata 339 for role in play.get("roles", []) or []:
b45288f… lmata 340 self._parse_role_reference(role, file_path, play_name, store, stats)
b45288f… lmata 341
b45288f… lmata 342 # Variables
b45288f… lmata 343 self._parse_vars_block(play.get("vars"), file_path, play_name, store, stats)
b45288f… lmata 344
b45288f… lmata 345 # ── Task parsing ─────────────────────────────────────────────────────────
b45288f… lmata 346
b45288f… lmata 347 def _task_name(self, task: dict) -> str:
b45288f… lmata 348 """Derive a task name from the dict."""
b45288f… lmata 349 if "name" in task and task["name"]:
b45288f… lmata 350 return str(task["name"])
b45288f… lmata 351 # Fall back to module name
b45288f… lmata 352 for key in task:
b45288f… lmata 353 if key in _ANSIBLE_MODULES:
b45288f… lmata 354 return key
b45288f… lmata 355 # Last resort: first non-meta key
b45288f… lmata 356 _meta_keys = {
b45288f… lmata 357 "name",
b45288f… lmata 358 "register",
b45288f… lmata 359 "when",
b45288f… lmata 360 "notify",
b45288f… lmata 361 "tags",
b45288f… lmata 362 "become",
b45288f… lmata 363 "become_user",
b45288f… lmata 364 "ignore_errors",
b45288f… lmata 365 "changed_when",
b45288f… lmata 366 "failed_when",
b45288f… lmata 367 "loop",
b45288f… lmata 368 "with_items",
b45288f… lmata 369 "with_dict",
b45288f… lmata 370 "with_fileglob",
b45288f… lmata 371 "until",
b45288f… lmata 372 "retries",
b45288f… lmata 373 "delay",
b45288f… lmata 374 "no_log",
b45288f… lmata 375 "environment",
b45288f… lmata 376 "vars",
b45288f… lmata 377 "listen",
b45288f… lmata 378 "delegate_to",
b45288f… lmata 379 "run_once",
b45288f… lmata 380 "timeout",
b45288f… lmata 381 }
b45288f… lmata 382 for key in task:
b45288f… lmata 383 if key not in _meta_keys:
b45288f… lmata 384 return key
b45288f… lmata 385 return "unnamed_task"
b45288f… lmata 386
b45288f… lmata 387 def _parse_task(
b45288f… lmata 388 self,
b45288f… lmata 389 task: dict,
b45288f… lmata 390 file_path: str,
b45288f… lmata 391 parent_name: str,
b45288f… lmata 392 store: GraphStore,
b45288f… lmata 393 stats: dict,
b45288f… lmata 394 ) -> None:
b45288f… lmata 395 """Parse a single task dict into a Function node."""
b45288f… lmata 396 task_name = self._task_name(task)
b45288f… lmata 397
b45288f… lmata 398 store.create_node(
b45288f… lmata 399 NodeLabel.Function,
b45288f… lmata 400 {
b45288f… lmata 401 "name": task_name,
b45288f… lmata 402 "file_path": file_path,
b45288f… lmata 403 "line_start": 0,
b45288f… lmata 404 "line_end": 0,
b45288f… lmata 405 "docstring": "",
b45288f… lmata 406 "semantic_type": "ansible_task",
b45288f… lmata 407 },
b45288f… lmata 408 )
b45288f… lmata 409 store.create_edge(
b45288f… lmata 410 NodeLabel.Class,
b45288f… lmata 411 {"name": parent_name, "file_path": file_path},
b45288f… lmata 412 EdgeType.CONTAINS,
b45288f… lmata 413 NodeLabel.Function,
b45288f… lmata 414 {"name": task_name, "file_path": file_path},
b45288f… lmata 415 )
b45288f… lmata 416 stats["functions"] += 1
b45288f… lmata 417 stats["edges"] += 1
b45288f… lmata 418
b45288f… lmata 419 # notify: -> CALLS edge to handler
b45288f… lmata 420 notify = task.get("notify")
b45288f… lmata 421 if notify:
b45288f… lmata 422 if isinstance(notify, str):
b45288f… lmata 423 notify = [notify]
b45288f… lmata 424 for handler_name in notify:
b45288f… lmata 425 store.create_edge(
b45288f… lmata 426 NodeLabel.Function,
b45288f… lmata 427 {"name": task_name, "file_path": file_path},
b45288f… lmata 428 EdgeType.CALLS,
b45288f… lmata 429 NodeLabel.Function,
b45288f… lmata 430 {"name": str(handler_name), "file_path": file_path},
b45288f… lmata 431 )
b45288f… lmata 432 stats["edges"] += 1
b45288f… lmata 433
b45288f… lmata 434 # Handle block/rescue/always
b45288f… lmata 435 for block_key in ("block", "rescue", "always"):
b45288f… lmata 436 block_tasks = task.get(block_key)
b45288f… lmata 437 if isinstance(block_tasks, list):
b45288f… lmata 438 for sub_task in block_tasks:
b45288f… lmata 439 if isinstance(sub_task, dict):
b45288f… lmata 440 self._parse_task(sub_task, file_path, parent_name, store, stats)
b45288f… lmata 441
b45288f… lmata 442 # ── Handler parsing ──────────────────────────────────────────────────────
b45288f… lmata 443
b45288f… lmata 444 def _parse_handler(
b45288f… lmata 445 self,
b45288f… lmata 446 handler: dict,
b45288f… lmata 447 file_path: str,
b45288f… lmata 448 parent_name: str,
b45288f… lmata 449 store: GraphStore,
b45288f… lmata 450 stats: dict,
b45288f… lmata 451 ) -> None:
b45288f… lmata 452 """Parse a handler dict into a Function node."""
b45288f… lmata 453 handler_name = handler.get("name", self._task_name(handler))
b45288f… lmata 454
b45288f… lmata 455 store.create_node(
b45288f… lmata 456 NodeLabel.Function,
b45288f… lmata 457 {
b45288f… lmata 458 "name": handler_name,
b45288f… lmata 459 "file_path": file_path,
b45288f… lmata 460 "line_start": 0,
b45288f… lmata 461 "line_end": 0,
b45288f… lmata 462 "docstring": "",
b45288f… lmata 463 "semantic_type": "ansible_handler",
b45288f… lmata 464 },
b45288f… lmata 465 )
b45288f… lmata 466 store.create_edge(
b45288f… lmata 467 NodeLabel.Class,
b45288f… lmata 468 {"name": parent_name, "file_path": file_path},
b45288f… lmata 469 EdgeType.CONTAINS,
b45288f… lmata 470 NodeLabel.Function,
b45288f… lmata 471 {"name": handler_name, "file_path": file_path},
b45288f… lmata 472 )
b45288f… lmata 473 stats["functions"] += 1
b45288f… lmata 474 stats["edges"] += 1
b45288f… lmata 475
b45288f… lmata 476 # ── Role reference parsing ───────────────────────────────────────────────
b45288f… lmata 477
b45288f… lmata 478 def _parse_role_reference(
b45288f… lmata 479 self,
b45288f… lmata 480 role,
b45288f… lmata 481 file_path: str,
b45288f… lmata 482 play_name: str,
b45288f… lmata 483 store: GraphStore,
b45288f… lmata 484 stats: dict,
b45288f… lmata 485 ) -> None:
b45288f… lmata 486 """Parse a role reference (string or dict with 'role' key)."""
b45288f… lmata 487 if isinstance(role, str):
b45288f… lmata 488 role_name = role
b45288f… lmata 489 elif isinstance(role, dict):
b45288f… lmata 490 role_name = role.get("role") or role.get("name", "")
b45288f… lmata 491 else:
b45288f… lmata 492 return
b45288f… lmata 493
b45288f… lmata 494 if not role_name:
b45288f… lmata 495 return
b45288f… lmata 496
b45288f… lmata 497 store.create_node(
b45288f… lmata 498 NodeLabel.Import,
b45288f… lmata 499 {
b45288f… lmata 500 "name": role_name,
b45288f… lmata 501 "file_path": file_path,
b45288f… lmata 502 "line_start": 0,
b45288f… lmata 503 "module": role_name,
b45288f… lmata 504 "semantic_type": "ansible_role",
b45288f… lmata 505 },
b45288f… lmata 506 )
b45288f… lmata 507 store.create_edge(
b45288f… lmata 508 NodeLabel.Class,
b45288f… lmata 509 {"name": play_name, "file_path": file_path},
b45288f… lmata 510 EdgeType.IMPORTS,
b45288f… lmata 511 NodeLabel.Import,
b45288f… lmata 512 {"name": role_name, "file_path": file_path},
b45288f… lmata 513 )
b45288f… lmata 514 stats["edges"] += 1
b45288f… lmata 515
b45288f… lmata 516 # ── Variable parsing ─────────────────────────────────────────────────────
b45288f… lmata 517
b45288f… lmata 518 def _parse_vars_block(
b45288f… lmata 519 self,
b45288f… lmata 520 vars_data,
b45288f… lmata 521 file_path: str,
b45288f… lmata 522 parent_name: str,
b45288f… lmata 523 store: GraphStore,
b45288f… lmata 524 stats: dict,
b45288f… lmata 525 ) -> None:
b45288f… lmata 526 """Parse a vars: block (dict) into Variable nodes."""
b45288f… lmata 527 if not isinstance(vars_data, dict):
b45288f… lmata 528 return
b45288f… lmata 529
b45288f… lmata 530 for var_name, var_value in vars_data.items():
b45288f… lmata 531 store.create_node(
b45288f… lmata 532 NodeLabel.Variable,
b45288f… lmata 533 {
b45288f… lmata 534 "name": str(var_name),
b45288f… lmata 535 "file_path": file_path,
b45288f… lmata 536 "line_start": 0,
b45288f… lmata 537 "semantic_type": "ansible_variable",
b45288f… lmata 538 },
b45288f… lmata 539 )
b45288f… lmata 540 store.create_edge(
b45288f… lmata 541 NodeLabel.Class,
b45288f… lmata 542 {"name": parent_name, "file_path": file_path},
b45288f… lmata 543 EdgeType.CONTAINS,
b45288f… lmata 544 NodeLabel.Variable,
b45288f… lmata 545 {"name": str(var_name), "file_path": file_path},
b45288f… lmata 546 )
b45288f… lmata 547 stats["edges"] += 1
b45288f… lmata 548
b45288f… lmata 549 # ── Standalone file parsers ──────────────────────────────────────────────
b45288f… lmata 550
b45288f… lmata 551 def _parse_task_file(
b45288f… lmata 552 self,
b45288f… lmata 553 data,
b45288f… lmata 554 file_path: str,
b45288f… lmata 555 store: GraphStore,
b45288f… lmata 556 stats: dict,
b45288f… lmata 557 ) -> None:
b45288f… lmata 558 """Parse a standalone task file (roles/*/tasks/main.yml or included file)."""
b45288f… lmata 559 if not isinstance(data, list):
b45288f… lmata 560 return
b45288f… lmata 561
b45288f… lmata 562 # Use file stem as a synthetic parent class
b45288f… lmata 563 parent_name = Path(file_path).stem
b45288f… lmata 564 store.create_node(
b45288f… lmata 565 NodeLabel.Class,
b45288f… lmata 566 {
b45288f… lmata 567 "name": parent_name,
b45288f… lmata 568 "file_path": file_path,
b45288f… lmata 569 "line_start": 0,
b45288f… lmata 570 "line_end": 0,
b45288f… lmata 571 "docstring": "",
b45288f… lmata 572 "semantic_type": "ansible_play",
b45288f… lmata 573 },
b45288f… lmata 574 )
b45288f… lmata 575 store.create_edge(
b45288f… lmata 576 NodeLabel.File,
b45288f… lmata 577 {"path": file_path},
b45288f… lmata 578 EdgeType.CONTAINS,
b45288f… lmata 579 NodeLabel.Class,
b45288f… lmata 580 {"name": parent_name, "file_path": file_path},
b45288f… lmata 581 )
b45288f… lmata 582 stats["classes"] += 1
b45288f… lmata 583 stats["edges"] += 1
b45288f… lmata 584
b45288f… lmata 585 for task_dict in data:
b45288f… lmata 586 if isinstance(task_dict, dict):
b45288f… lmata 587 self._parse_task(task_dict, file_path, parent_name, store, stats)
b45288f… lmata 588
b45288f… lmata 589 def _parse_handler_file(
b45288f… lmata 590 self,
b45288f… lmata 591 data,
b45288f… lmata 592 file_path: str,
b45288f… lmata 593 store: GraphStore,
b45288f… lmata 594 stats: dict,
b45288f… lmata 595 ) -> None:
b45288f… lmata 596 """Parse a standalone handler file (roles/*/handlers/main.yml)."""
b45288f… lmata 597 if not isinstance(data, list):
b45288f… lmata 598 return
b45288f… lmata 599
b45288f… lmata 600 parent_name = Path(file_path).stem
b45288f… lmata 601 store.create_node(
b45288f… lmata 602 NodeLabel.Class,
b45288f… lmata 603 {
b45288f… lmata 604 "name": parent_name,
b45288f… lmata 605 "file_path": file_path,
b45288f… lmata 606 "line_start": 0,
b45288f… lmata 607 "line_end": 0,
b45288f… lmata 608 "docstring": "",
b45288f… lmata 609 "semantic_type": "ansible_play",
b45288f… lmata 610 },
b45288f… lmata 611 )
b45288f… lmata 612 store.create_edge(
b45288f… lmata 613 NodeLabel.File,
b45288f… lmata 614 {"path": file_path},
b45288f… lmata 615 EdgeType.CONTAINS,
b45288f… lmata 616 NodeLabel.Class,
b45288f… lmata 617 {"name": parent_name, "file_path": file_path},
b45288f… lmata 618 )
b45288f… lmata 619 stats["classes"] += 1
b45288f… lmata 620 stats["edges"] += 1
b45288f… lmata 621
b45288f… lmata 622 for handler_dict in data:
b45288f… lmata 623 if isinstance(handler_dict, dict):
b45288f… lmata 624 self._parse_handler(handler_dict, file_path, parent_name, store, stats)
b45288f… lmata 625
b45288f… lmata 626 def _parse_variable_file(
b45288f… lmata 627 self,
b45288f… lmata 628 data,
b45288f… lmata 629 file_path: str,
b45288f… lmata 630 store: GraphStore,
b45288f… lmata 631 stats: dict,
b45288f… lmata 632 ) -> None:
b45288f… lmata 633 """Parse a variable file (defaults/main.yml, vars/main.yml, group_vars/, host_vars/)."""
b45288f… lmata 634 if not isinstance(data, dict):
b45288f… lmata 635 return
b45288f… lmata 636
b45288f… lmata 637 # Use file stem as a synthetic parent
b45288f… lmata 638 parent_name = Path(file_path).stem
b45288f… lmata 639 store.create_node(
b45288f… lmata 640 NodeLabel.Module,
b45288f… lmata 641 {
b45288f… lmata 642 "name": parent_name,
b45288f… lmata 643 "file_path": file_path,
b45288f… lmata 644 "docstring": "",
b45288f… lmata 645 "semantic_type": "ansible_playbook",
b45288f… lmata 646 },
b45288f… lmata 647 )
b45288f… lmata 648 store.create_edge(
b45288f… lmata 649 NodeLabel.File,
b45288f… lmata 650 {"path": file_path},
b45288f… lmata 651 EdgeType.CONTAINS,
b45288f… lmata 652 NodeLabel.Module,
b45288f… lmata 653 {"name": parent_name, "file_path": file_path},
b45288f… lmata 654 )
b45288f… lmata 655 stats["edges"] += 1
b45288f… lmata 656
b45288f… lmata 657 for var_name in data:
b45288f… lmata 658 store.create_node(
b45288f… lmata 659 NodeLabel.Variable,
b45288f… lmata 660 {
b45288f… lmata 661 "name": str(var_name),
b45288f… lmata 662 "file_path": file_path,
b45288f… lmata 663 "line_start": 0,
b45288f… lmata 664 "semantic_type": "ansible_variable",
b45288f… lmata 665 },
b45288f… lmata 666 )
b45288f… lmata 667 store.create_edge(
b45288f… lmata 668 NodeLabel.Module,
b45288f… lmata 669 {"name": parent_name, "file_path": file_path},
b45288f… lmata 670 EdgeType.CONTAINS,
b45288f… lmata 671 NodeLabel.Variable,
b45288f… lmata 672 {"name": str(var_name), "file_path": file_path},
b45288f… lmata 673 )
b45288f… lmata 674 stats["edges"] += 1

Keyboard Shortcuts

Open search /
Next entry (timeline) j
Previous entry (timeline) k
Open focused entry Enter
Show this help ?
Toggle theme Top nav button