|
| 1 | +""" |
| 2 | +``api.actions:`` splice helpers used by :mod:`writing`. |
| 3 | +
|
| 4 | +The ``api:`` block hosts a list of named callables under ``actions:`` |
| 5 | +— structurally near-identical to ``script:``, but nested two levels |
| 6 | +deep rather than at the top level. The lookup, item-locator, and |
| 7 | +re-indent helpers live here so :mod:`writing` stays close to the |
| 8 | +800-line file-size cap; the dispatch surface (the |
| 9 | +``_upsert_api_action`` / ``_delete_api_action`` branches of |
| 10 | +:func:`writing.render_upsert` / :func:`writing.render_delete`) |
| 11 | +stays in :mod:`writing` alongside every other location type for |
| 12 | +grep-ability. |
| 13 | +""" |
| 14 | + |
| 15 | +from __future__ import annotations |
| 16 | + |
| 17 | +import re |
| 18 | + |
| 19 | +from ...models.automations import YamlDiff |
| 20 | + |
| 21 | + |
| 22 | +def has_inline_actions_value( |
| 23 | + lines: list[str], |
| 24 | + api_span: tuple[int, int, str], |
| 25 | +) -> bool: |
| 26 | + """Return True iff ``actions:`` under *api_span* carries an inline value. |
| 27 | +
|
| 28 | + Flow-style values (``actions: []``, ``actions: null``, |
| 29 | + ``actions: !secret foo``, …) can't be spliced into the way the |
| 30 | + line-based writer wants. Callers should refuse the upsert / |
| 31 | + delete and surface a clear error rather than emit a second |
| 32 | + ``actions:`` key alongside the inline one. |
| 33 | + """ |
| 34 | + api_start, api_end, child_indent = api_span |
| 35 | + header = f"{child_indent}actions:" |
| 36 | + for idx in range(api_start + 1, api_end): |
| 37 | + text = lines[idx].rstrip("\n\r") |
| 38 | + if text == header: |
| 39 | + return False |
| 40 | + if text.startswith(header + " "): |
| 41 | + rest = text[len(header) :].strip() |
| 42 | + return bool(rest) and not rest.startswith("#") |
| 43 | + return False |
| 44 | + |
| 45 | + |
| 46 | +def locate_actions_list( |
| 47 | + lines: list[str], |
| 48 | + api_span: tuple[int, int, str], |
| 49 | +) -> tuple[int, int, str] | None: |
| 50 | + """Return ``(start, end, item_indent)`` for ``api.actions:`` or ``None``. |
| 51 | +
|
| 52 | + ``start`` is the line index of the ``actions:`` key; ``end`` is |
| 53 | + one past the last item line; ``item_indent`` is the leading |
| 54 | + whitespace shared by each ``- ...`` dash line. |
| 55 | + """ |
| 56 | + api_start, api_end, child_indent = api_span |
| 57 | + header = f"{child_indent}actions:" |
| 58 | + actions_start: int | None = None |
| 59 | + for idx in range(api_start + 1, api_end): |
| 60 | + text = lines[idx].rstrip("\n\r") |
| 61 | + if text == header or text.startswith(header + " "): |
| 62 | + actions_start = idx |
| 63 | + break |
| 64 | + if actions_start is None: |
| 65 | + return None |
| 66 | + actions_end = api_end |
| 67 | + for idx in range(actions_start + 1, api_end): |
| 68 | + content = lines[idx].rstrip("\n\r") |
| 69 | + if not content: |
| 70 | + continue |
| 71 | + leading = len(content) - len(content.lstrip(" ")) |
| 72 | + if leading <= len(child_indent): |
| 73 | + actions_end = idx |
| 74 | + break |
| 75 | + item_indent: str | None = None |
| 76 | + for idx in range(actions_start + 1, actions_end): |
| 77 | + raw = lines[idx].rstrip("\n\r") |
| 78 | + stripped = raw.lstrip(" ") |
| 79 | + if stripped.startswith("- "): |
| 80 | + item_indent = raw[: len(raw) - len(stripped)] |
| 81 | + break |
| 82 | + if item_indent is None: |
| 83 | + # Empty list — assume the canonical two-space nesting under |
| 84 | + # ``actions:`` so the first item still indents predictably. |
| 85 | + item_indent = child_indent + " " |
| 86 | + return actions_start, actions_end, item_indent |
| 87 | + |
| 88 | + |
| 89 | +def find_item( |
| 90 | + lines: list[str], |
| 91 | + actions_start: int, |
| 92 | + actions_end: int, |
| 93 | + item_indent: str, |
| 94 | + action_name: str, |
| 95 | +) -> tuple[int, int] | None: |
| 96 | + """Locate the line range of the list item whose discriminator matches.""" |
| 97 | + item_starts: list[int] = [] |
| 98 | + for idx in range(actions_start + 1, actions_end): |
| 99 | + raw = lines[idx].rstrip("\n\r") |
| 100 | + if not raw.startswith(item_indent + "- "): |
| 101 | + continue |
| 102 | + item_starts.append(idx) |
| 103 | + for run, start in enumerate(item_starts): |
| 104 | + end = item_starts[run + 1] if run + 1 < len(item_starts) else actions_end |
| 105 | + if _discriminator(lines, start, end, item_indent) == action_name: |
| 106 | + return start, end |
| 107 | + return None |
| 108 | + |
| 109 | + |
| 110 | +def count_siblings( |
| 111 | + lines: list[str], |
| 112 | + actions_start: int, |
| 113 | + actions_end: int, |
| 114 | + item_indent: str, |
| 115 | + matched: tuple[int, int], |
| 116 | +) -> int: |
| 117 | + """Count list items at *item_indent* that aren't the matched span.""" |
| 118 | + item_start, item_end = matched |
| 119 | + siblings = 0 |
| 120 | + for idx in range(actions_start + 1, actions_end): |
| 121 | + raw = lines[idx].rstrip("\n\r") |
| 122 | + if not raw.startswith(item_indent + "- "): |
| 123 | + continue |
| 124 | + if item_start <= idx < item_end: |
| 125 | + continue |
| 126 | + siblings += 1 |
| 127 | + return siblings |
| 128 | + |
| 129 | + |
| 130 | +def indent_for_list(rendered_item: str, item_indent: str) -> str: |
| 131 | + """Re-indent a ``dump([item])`` block to nest under ``api.actions:``. |
| 132 | +
|
| 133 | + The shared ruamel emitter writes top-level list items at column |
| 134 | + two (`` - key: value``) because ``offset=2`` indents each dash |
| 135 | + two columns inside its parent sequence. ``api.actions:`` items |
| 136 | + live two levels deeper than that, so add the *item_indent* - 2 |
| 137 | + delta to every non-empty line. Blank lines inside ``|`` block |
| 138 | + scalars (e.g. a multi-paragraph ``lambda:`` body) are left |
| 139 | + untouched — padding them with spaces would change the scalar's |
| 140 | + content, since YAML treats whitespace-only lines and fully |
| 141 | + empty lines differently inside a literal block. |
| 142 | + """ |
| 143 | + pad = " " * (len(item_indent) - 2) |
| 144 | + out_lines: list[str] = [] |
| 145 | + for line in rendered_item.splitlines(): |
| 146 | + if not line: |
| 147 | + out_lines.append("") |
| 148 | + continue |
| 149 | + out_lines.append(pad + line) |
| 150 | + if not out_lines or out_lines[-1] != "": |
| 151 | + out_lines.append("") |
| 152 | + return "\n".join(out_lines) |
| 153 | + |
| 154 | + |
| 155 | +def render_replacement( |
| 156 | + lines: list[str], |
| 157 | + item_start: int, |
| 158 | + item_end: int, |
| 159 | + rendered_text: str, |
| 160 | +) -> tuple[str, YamlDiff]: |
| 161 | + """Splice *rendered_text* over the lines spanning an existing item.""" |
| 162 | + new_lines = [*lines[:item_start], rendered_text, *lines[item_end:]] |
| 163 | + return "".join(new_lines), YamlDiff( |
| 164 | + fromLine=item_start + 1, |
| 165 | + toLine=item_end, |
| 166 | + replacement=rendered_text, |
| 167 | + ) |
| 168 | + |
| 169 | + |
| 170 | +def render_create_block(yaml_text: str, rendered: str) -> tuple[str, str]: |
| 171 | + """Return ``(new_yaml, block_text)`` for a fresh ``api:`` block at EOF.""" |
| 172 | + item_indent = " " |
| 173 | + item_text = indent_for_list(rendered, item_indent) |
| 174 | + # ``item_text`` always ends with ``\n``, so the block does too and |
| 175 | + # the final concatenation lands a trailing newline on EOF. |
| 176 | + block = "api:\n actions:\n" + item_text |
| 177 | + base = yaml_text.rstrip() |
| 178 | + separator = "\n\n" if base else "" |
| 179 | + return f"{base}{separator}{block}", block |
| 180 | + |
| 181 | + |
| 182 | +def render_insert_actions_key( |
| 183 | + lines: list[str], |
| 184 | + api_span: tuple[int, int, str], |
| 185 | + rendered: str, |
| 186 | +) -> tuple[str, YamlDiff]: |
| 187 | + """Insert an ``actions:`` key with one item under an existing ``api:`` block.""" |
| 188 | + api_start, api_end, child_indent = api_span |
| 189 | + item_indent = child_indent + " " |
| 190 | + item_text = indent_for_list(rendered, item_indent) |
| 191 | + block = f"{child_indent}actions:\n{item_text}" |
| 192 | + insert_at = api_end |
| 193 | + while insert_at > api_start + 1 and not lines[insert_at - 1].strip(): |
| 194 | + insert_at -= 1 |
| 195 | + new_lines = [*lines[:insert_at], block, *lines[insert_at:]] |
| 196 | + new_text = "".join(new_lines) |
| 197 | + return new_text, YamlDiff( |
| 198 | + fromLine=insert_at + 1, |
| 199 | + toLine=insert_at, |
| 200 | + replacement=block, |
| 201 | + ) |
| 202 | + |
| 203 | + |
| 204 | +def render_append( |
| 205 | + lines: list[str], |
| 206 | + actions_end: int, |
| 207 | + item_indent: str, |
| 208 | + rendered: str, |
| 209 | +) -> tuple[str, YamlDiff]: |
| 210 | + """Append a new list item at the end of an existing ``api.actions:``.""" |
| 211 | + item_text = indent_for_list(rendered, item_indent) |
| 212 | + insert_at = actions_end |
| 213 | + while insert_at > 0 and not lines[insert_at - 1].strip(): |
| 214 | + insert_at -= 1 |
| 215 | + new_lines = [*lines[:insert_at], item_text, *lines[insert_at:]] |
| 216 | + new_text = "".join(new_lines) |
| 217 | + return new_text, YamlDiff( |
| 218 | + fromLine=insert_at + 1, |
| 219 | + toLine=insert_at, |
| 220 | + replacement=item_text, |
| 221 | + ) |
| 222 | + |
| 223 | + |
| 224 | +def render_delete_item( |
| 225 | + lines: list[str], |
| 226 | + item_start: int, |
| 227 | + item_end: int, |
| 228 | +) -> tuple[str, YamlDiff]: |
| 229 | + """Remove the line range covering a single api-action list item.""" |
| 230 | + new_lines = [*lines[:item_start], *lines[item_end:]] |
| 231 | + return "".join(new_lines), YamlDiff( |
| 232 | + fromLine=item_start + 1, |
| 233 | + toLine=item_end, |
| 234 | + replacement="", |
| 235 | + ) |
| 236 | + |
| 237 | + |
| 238 | +def render_delete_actions_key( |
| 239 | + lines: list[str], |
| 240 | + actions_start: int, |
| 241 | + actions_end: int, |
| 242 | +) -> tuple[str, YamlDiff]: |
| 243 | + """Remove the entire ``actions:`` key when its last item is being dropped.""" |
| 244 | + new_lines = [*lines[:actions_start], *lines[actions_end:]] |
| 245 | + return "".join(new_lines), YamlDiff( |
| 246 | + fromLine=actions_start + 1, |
| 247 | + toLine=actions_end, |
| 248 | + replacement="", |
| 249 | + ) |
| 250 | + |
| 251 | + |
| 252 | +# --------------------------------------------------------------------------- |
| 253 | +# Internals |
| 254 | +# --------------------------------------------------------------------------- |
| 255 | + |
| 256 | + |
| 257 | +def _discriminator( |
| 258 | + lines: list[str], |
| 259 | + item_start: int, |
| 260 | + item_end: int, |
| 261 | + item_indent: str, |
| 262 | +) -> str | None: |
| 263 | + """Read the ``action:`` (or legacy ``service:``) key for a list item.""" |
| 264 | + child_indent = item_indent + " " |
| 265 | + inline = re.match( |
| 266 | + rf"^{re.escape(item_indent)}-\s*(?P<key>action|service):\s*(?P<val>\S+)", |
| 267 | + lines[item_start].rstrip("\n\r"), |
| 268 | + ) |
| 269 | + if inline: |
| 270 | + return inline.group("val").strip("'\"") |
| 271 | + child_re = re.compile( |
| 272 | + rf"^{re.escape(child_indent)}(?:action|service):\s*(?P<val>\S+)", |
| 273 | + ) |
| 274 | + for idx in range(item_start, item_end): |
| 275 | + m = child_re.match(lines[idx].rstrip("\n\r")) |
| 276 | + if m: |
| 277 | + return m.group("val").strip("'\"") |
| 278 | + return None |
0 commit comments