|
| 1 | +#!/usr/bin/env python3 |
| 2 | +"""Collect .sightmap/ definitions and upload them to a Lidar MCP session. |
| 3 | +
|
| 4 | +Walks all .sightmap/**/*.yaml files under a given root, parses component definitions, flattens |
| 5 | +hierarchical children into compound CSS selectors suitable for the subtext MCP's NFA matcher, and |
| 6 | +uploads the result to the sightmap upload endpoint. |
| 7 | +
|
| 8 | +Usage: |
| 9 | + python3 collect_and_upload_sightmap.py --url <sightmap_upload_url> [--root DIR] |
| 10 | +
|
| 11 | +The upload URL is returned by open_session / open_connection and includes a |
| 12 | +single-use authentication token. |
| 13 | +""" |
| 14 | + |
| 15 | +# Python 3.9 compat |
| 16 | +from __future__ import annotations |
| 17 | + |
| 18 | +import argparse |
| 19 | +import json |
| 20 | +import os |
| 21 | +import ssl |
| 22 | +import sys |
| 23 | +import urllib.error |
| 24 | +import urllib.parse |
| 25 | +import urllib.request |
| 26 | +from typing import Optional |
| 27 | + |
| 28 | +try: |
| 29 | + import yaml # type: ignore[import-not-found] |
| 30 | +except ImportError: |
| 31 | + sys.exit("PyYAML is required: pip install pyyaml") |
| 32 | + |
| 33 | + |
| 34 | +# --------------------------------------------------------------------------- |
| 35 | +# Sightmap collection |
| 36 | +# --------------------------------------------------------------------------- |
| 37 | + |
| 38 | + |
| 39 | +def find_sightmap_files(root: str) -> list[str]: |
| 40 | + """Find all .yaml/.yml files under root/.sightmap/. |
| 41 | +
|
| 42 | + Checks only the direct .sightmap/ child of root to avoid walking |
| 43 | + potentially massive directory trees (node_modules, go, etc.). |
| 44 | + """ |
| 45 | + sdir = os.path.join(root, ".sightmap") |
| 46 | + if not os.path.isdir(sdir): |
| 47 | + return [] |
| 48 | + |
| 49 | + files = [] |
| 50 | + for dirpath, _, filenames in os.walk(sdir): |
| 51 | + for name in sorted(filenames): |
| 52 | + if name.endswith((".yaml", ".yml")): |
| 53 | + files.append(os.path.join(dirpath, name)) |
| 54 | + return files |
| 55 | + |
| 56 | + |
| 57 | +def flatten_components( |
| 58 | + components: list[dict], |
| 59 | + parent_selectors: Optional[list[str]] = None, |
| 60 | + parent_source: str = "", |
| 61 | +) -> list[dict]: |
| 62 | + """Flatten hierarchical component definitions into a flat list. |
| 63 | +
|
| 64 | + Children inherit the parent's selectors as prefixes (descendant combinator) |
| 65 | + and the parent's source if they don't specify their own. |
| 66 | +
|
| 67 | + The YAML ``selector`` field may be a string or a list of strings. The output |
| 68 | + always uses ``selectors`` (a JSON array) so the Go side never needs to split |
| 69 | + comma-separated values. |
| 70 | + """ |
| 71 | + if parent_selectors is None: |
| 72 | + parent_selectors = [] |
| 73 | + result = [] |
| 74 | + for comp in components: |
| 75 | + name = comp.get("name", "") |
| 76 | + raw = comp.get("selector", "") |
| 77 | + source = comp.get("source", "") or parent_source |
| 78 | + |
| 79 | + # Normalise to a list — YAML authors may write a string or a list. |
| 80 | + if isinstance(raw, list): |
| 81 | + selectors = [s for s in raw if s] |
| 82 | + elif raw: |
| 83 | + selectors = [raw] |
| 84 | + else: |
| 85 | + selectors = [] |
| 86 | + |
| 87 | + # Build full selector chains by combining with parent selectors. |
| 88 | + if parent_selectors and selectors: |
| 89 | + full_selectors = [f"{p} {s}" for p in parent_selectors for s in selectors] |
| 90 | + elif parent_selectors: |
| 91 | + full_selectors = list(parent_selectors) |
| 92 | + else: |
| 93 | + full_selectors = selectors |
| 94 | + |
| 95 | + if name and full_selectors: |
| 96 | + memory = comp.get("memory", []) |
| 97 | + if not isinstance(memory, list): |
| 98 | + memory = [memory] if memory else [] |
| 99 | + entry = { |
| 100 | + "name": name, |
| 101 | + "selectors": full_selectors, |
| 102 | + "source": source or "", |
| 103 | + "memory": memory, |
| 104 | + } |
| 105 | + result.append(entry) |
| 106 | + |
| 107 | + # Recurse into children |
| 108 | + children = comp.get("children", []) |
| 109 | + if children: |
| 110 | + result.extend(flatten_components(children, full_selectors, source)) |
| 111 | + |
| 112 | + return result |
| 113 | + |
| 114 | + |
| 115 | +def parse_file(path: str) -> list[dict]: |
| 116 | + """Parse a single sightmap YAML file and return flattened components.""" |
| 117 | + with open(path) as f: |
| 118 | + data = yaml.safe_load(f) |
| 119 | + |
| 120 | + if not isinstance(data, dict): |
| 121 | + return [] |
| 122 | + |
| 123 | + components = data.get("components", []) |
| 124 | + if not isinstance(components, list): |
| 125 | + components = [] |
| 126 | + |
| 127 | + result = flatten_components(components) |
| 128 | + |
| 129 | + # Also flatten view-scoped components |
| 130 | + views = data.get("views", []) |
| 131 | + if isinstance(views, list): |
| 132 | + for view in views: |
| 133 | + view_components = view.get("components", []) |
| 134 | + if isinstance(view_components, list): |
| 135 | + result.extend(flatten_components(view_components)) |
| 136 | + |
| 137 | + return result |
| 138 | + |
| 139 | + |
| 140 | +def collect(root: str) -> list[dict]: |
| 141 | + """Collect all sightmap definitions from a root directory.""" |
| 142 | + files = find_sightmap_files(root) |
| 143 | + result = [] |
| 144 | + for path in files: |
| 145 | + result.extend(parse_file(path)) |
| 146 | + return result |
| 147 | + |
| 148 | + |
| 149 | +def collect_memory(root: str) -> list[str]: |
| 150 | + """Collect top-level memory entries from .sightmap/ YAML files.""" |
| 151 | + files = find_sightmap_files(root) |
| 152 | + result: list[str] = [] |
| 153 | + for path in files: |
| 154 | + with open(path) as f: |
| 155 | + data = yaml.safe_load(f) |
| 156 | + if not isinstance(data, dict): |
| 157 | + continue |
| 158 | + memory = data.get("memory", []) |
| 159 | + if isinstance(memory, str): |
| 160 | + memory = [memory] |
| 161 | + if isinstance(memory, list): |
| 162 | + result.extend(str(m) for m in memory if m) |
| 163 | + return result |
| 164 | + |
| 165 | + |
| 166 | +# --------------------------------------------------------------------------- |
| 167 | +# Sightmap root discovery |
| 168 | +# --------------------------------------------------------------------------- |
| 169 | + |
| 170 | + |
| 171 | +def find_sightmap_root(cwd: str) -> Optional[str]: |
| 172 | + """Find a directory containing .sightmap/, checking cwd and ancestors.""" |
| 173 | + d = cwd |
| 174 | + while d != os.path.dirname(d): |
| 175 | + if os.path.isdir(os.path.join(d, ".sightmap")): |
| 176 | + return d |
| 177 | + d = os.path.dirname(d) |
| 178 | + return None |
| 179 | + |
| 180 | + |
| 181 | +# --------------------------------------------------------------------------- |
| 182 | +# Upload |
| 183 | +# --------------------------------------------------------------------------- |
| 184 | + |
| 185 | + |
| 186 | +def main(): |
| 187 | + parser = argparse.ArgumentParser( |
| 188 | + description="Collect and upload .sightmap/ definitions" |
| 189 | + ) |
| 190 | + parser.add_argument( |
| 191 | + "--url", |
| 192 | + required=True, |
| 193 | + help="Sightmap upload URL (from open_session/open_connection response)", |
| 194 | + ) |
| 195 | + parser.add_argument( |
| 196 | + "--root", |
| 197 | + default=None, |
| 198 | + help="Root directory containing .sightmap/ (auto-detected if omitted)", |
| 199 | + ) |
| 200 | + args = parser.parse_args() |
| 201 | + |
| 202 | + root = ( |
| 203 | + args.root or os.environ.get("SIGHTMAP_ROOT") or find_sightmap_root(os.getcwd()) |
| 204 | + ) |
| 205 | + if not root: |
| 206 | + print("No .sightmap/ directory found", file=sys.stderr) |
| 207 | + sys.exit(1) |
| 208 | + |
| 209 | + components = collect(root) |
| 210 | + memory = collect_memory(root) |
| 211 | + |
| 212 | + if not components and not memory: |
| 213 | + print("No sightmap definitions found") |
| 214 | + sys.exit(0) |
| 215 | + |
| 216 | + body = json.dumps( |
| 217 | + { |
| 218 | + "sightmap": components, |
| 219 | + "memory": memory, |
| 220 | + } |
| 221 | + ).encode("utf-8") |
| 222 | + |
| 223 | + req = urllib.request.Request( |
| 224 | + args.url, |
| 225 | + data=body, |
| 226 | + headers={"Content-Type": "application/json"}, |
| 227 | + method="POST", |
| 228 | + ) |
| 229 | + |
| 230 | + # Allow self-signed certs for local dev servers (.test, localhost). |
| 231 | + ssl_ctx = None |
| 232 | + parsed_url = urllib.parse.urlparse(args.url) |
| 233 | + if parsed_url.hostname and ( |
| 234 | + parsed_url.hostname.endswith(".test") |
| 235 | + or parsed_url.hostname in ("localhost", "127.0.0.1") |
| 236 | + ): |
| 237 | + ssl_ctx = ssl.create_default_context() |
| 238 | + ssl_ctx.check_hostname = False |
| 239 | + ssl_ctx.verify_mode = ssl.CERT_NONE |
| 240 | + |
| 241 | + try: |
| 242 | + with urllib.request.urlopen(req, timeout=30, context=ssl_ctx) as resp: |
| 243 | + result = json.loads(resp.read()) |
| 244 | + count = result.get("components", 0) |
| 245 | + print(f"Uploaded {count} sightmap component(s)") |
| 246 | + except urllib.error.HTTPError as e: |
| 247 | + body_text = e.read().decode("utf-8", errors="replace") |
| 248 | + print(f"Upload failed ({e.code}): {body_text}", file=sys.stderr) |
| 249 | + sys.exit(1) |
| 250 | + except urllib.error.URLError as e: |
| 251 | + print(f"Upload failed: {e.reason}", file=sys.stderr) |
| 252 | + sys.exit(1) |
| 253 | + |
| 254 | + |
| 255 | +if __name__ == "__main__": |
| 256 | + main() |
0 commit comments