#!/usr/bin/env python3 """ vocab_gen.py — Entity Lexicon + Acronym Vocabulary Generator (Tier 2) Primary source: vault docs (reliable, complete) Secondary source: live infra inventory (best-effort, SSH, read-only) Outputs: engine/vocab.json — structured vocabulary vault/glossary.md — browsable Obsidian note """ import json import os import re import subprocess import sys from datetime import datetime, timezone from pathlib import Path # --------------------------------------------------------------------------- # Config loader (stdlib-only YAML parser for the keys we need) # --------------------------------------------------------------------------- def load_config(config_path: str) -> dict: """Minimal YAML parser for flat / simple-list config.yaml.""" config = { "vault_dir": "/home/zvx/projects/.ref/vault", "engine_dir": "/home/zvx/projects/.ref/engine", "topic_categories": [ "mesh", "matrix", "recon", "media", "auth", "dns", "vpn", "storage", "proxmox", "ai", "mail", ], "inventory": { "proxmox_nodes": ["data", "utility", "cloud", "media", "toc"], "proxmox_vps": ["edge1", "edge2"], "docker_hosts": ["cortex", "utility", "media"], "headscale_host": "edge2", "ssh_user": "zvx", }, } try: with open(config_path) as f: lines = f.readlines() # Parse vault_dir / engine_dir for line in lines: m = re.match(r'^(vault_dir|engine_dir):\s*(.+)', line) if m: config[m.group(1)] = m.group(2).strip() # Parse topic_categories list in_topics = False topics = [] for line in lines: if re.match(r'^topic_categories:', line): in_topics = True continue if in_topics: m = re.match(r'^\s+-\s+(\S+)', line) if m: topics.append(m.group(1).strip()) elif line.strip() and not line.startswith(' '): in_topics = False if topics: config["topic_categories"] = topics # Parse inventory.proxmox_nodes (inline list) for line in lines: m = re.match(r'\s*proxmox_nodes:\s*\[(.+)\]', line) if m: config["inventory"]["proxmox_nodes"] = [ x.strip().strip("'\"") for x in m.group(1).split(',') ] m2 = re.match(r'\s*proxmox_vps:\s*\[(.+)\]', line) if m2: config["inventory"]["proxmox_vps"] = [ x.strip().strip("'\"") for x in m2.group(1).split(',') ] m3 = re.match(r'\s*docker_hosts:\s*\[(.+)\]', line) if m3: config["inventory"]["docker_hosts"] = [ x.strip().strip("'\"") for x in m3.group(1).split(',') ] m4 = re.match(r'\s*headscale_host:\s*(\S+)', line) if m4: config["inventory"]["headscale_host"] = m4.group(1).strip() m5 = re.match(r'\s*ssh_user:\s*(\S+)', line) if m5: config["inventory"]["ssh_user"] = m5.group(1).strip() except Exception as e: print(f"[warn] Could not fully parse {config_path}: {e}; using defaults", file=sys.stderr) return config # --------------------------------------------------------------------------- # Host IP map (for enrichment during doc parsing) # --------------------------------------------------------------------------- HOST_IPS = { "data": ("192.168.1.240", "100.64.0.6"), "utility": ("192.168.1.241", "100.64.0.5"), "cloud": ("192.168.1.242", "100.64.0.4"), "media": ("192.168.1.243", "100.64.0.3"), "toc": ("192.168.1.244", "100.64.0.13"), "cortex": ("192.168.1.150", "100.64.0.14"), "recon-vm": ("192.168.1.130", "100.64.0.24"), "arr": ("192.168.1.160", "100.64.0.18"), "aida-nebra": ("192.168.1.253", "100.64.0.9"), "mt-isr": ("192.168.1.141", "100.100.0.5"), "mt-burleybutte": ("192.168.1.185", None), "pi-nas": ("192.168.1.245", "100.64.0.21"), "matt-desktop": ("192.168.1.111", "100.64.0.10"), "contabo": ("5.189.158.149", "100.64.0.40"), "edge1": ("5.189.158.149", "100.64.0.40"), "edge2": ("184.174.35.153", "100.64.0.26"), } # --------------------------------------------------------------------------- # Fallback English-word stoplist (~300 most common words). # Used when /usr/share/dict/words is absent. # A candidate token whose .lower() is in this set is rejected as a plain # English word and NOT treated as a domain acronym. # --------------------------------------------------------------------------- _COMMON_ENGLISH_WORDS = { # articles / determiners "a", "an", "the", "this", "that", "these", "those", "its", "our", "your", "their", "my", "his", "her", "all", "both", "each", "every", "few", "more", "most", "other", "some", "such", "no", "nor", "not", "only", "same", "so", "than", "too", "very", # pronouns "i", "me", "we", "us", "you", "he", "she", "it", "they", "them", "who", "whom", "which", "what", "one", # conjunctions / prepositions "and", "but", "or", "yet", "for", "nor", "so", "at", "by", "in", "of", "on", "to", "up", "as", "be", "do", "if", "is", "it", "no", "was", "are", "had", "has", "have", "how", "may", "per", "did", "due", "out", "via", "any", "can", "could", "from", "into", "over", "then", "they", "with", "will", "been", "down", "also", "when", "where", "while", "after", "about", "above", "below", "under", "until", "before", "between", "through", "during", "without", "within", "against", "around", "despite", # verbs "run", "get", "set", "put", "use", "let", "see", "say", "add", "act", "ask", "cut", "end", "hit", "log", "map", "mix", "off", "own", "pass", "read", "send", "stop", "test", "try", "wait", "want", "work", "show", "load", "save", "move", "list", "boot", "call", "copy", "drop", "edit", "exit", "fail", "find", "give", "help", "hold", "keep", "kill", "link", "lock", "make", "mark", "must", "need", "note", "open", "pick", "ping", "push", "pull", "quit", "skip", "take", "tell", "used", "view", "warn", "check", "clean", "clear", "close", "count", "cover", "create", "defer", "delete", "deploy", "enable", "export", "fetch", "flush", "force", "grant", "import", "inject", "insert", "launch", "manage", "mount", "output", "parse", "patch", "print", "prove", "query", "raise", "reboot", "reload", "remove", "rename", "reset", "restart", "return", "rotate", "select", "should", "signal", "start", "stream", "submit", "switch", "sync", "update", "upload", "verify", "write", # adjectives / adverbs "new", "old", "big", "low", "high", "raw", "hot", "cold", "fast", "slow", "free", "full", "good", "best", "bad", "next", "last", "long", "main", "same", "true", "false", "empty", "first", "local", "valid", "basic", "clean", "close", "direct", "extra", "final", "fixed", "global", "human", "large", "light", "minor", "never", "plain", "quick", "ready", "right", "short", "small", "smart", "solid", "spare", "super", "total", "unique", "upper", "valid", "whole", "active", "actual", "custom", "failed", "hidden", "modern", "native", "normal", "public", "random", "recent", "remote", "simple", "single", "stable", "static", "strong", "target", # common nouns (non-technical) "way", "day", "man", "end", "top", "yes", "now", "ago", "area", "base", "case", "date", "door", "edge", "else", "face", "fact", "file", "form", "game", "home", "host", "idea", "info", "item", "kind", "lack", "lane", "life", "line", "link", "list", "mode", "name", "node", "note", "page", "part", "path", "plan", "plug", "plus", "port", "post", "rate", "rule", "side", "site", "size", "slot", "spec", "step", "tag", "task", "term", "text", "time", "type", "user", "word", "year", "admin", "agent", "alias", "array", "audit", "batch", "block", "board", "build", "cache", "chain", "chart", "chunk", "class", "cloud", "codec", "color", "count", "debug", "delta", "depth", "error", "event", "field", "flags", "floor", "frame", "front", "graph", "group", "guard", "guide", "image", "index", "input", "issue", "label", "layer", "lease", "level", "limit", "match", "media", "model", "mount", "mutex", "order", "owner", "panel", "phase", "place", "point", "power", "queue", "quota", "range", "relay", "reply", "retry", "route", "round", "scale", "scope", "score", "shard", "share", "shift", "skill", "slice", "space", "stack", "stage", "state", "stats", "store", "suite", "table", "theme", "token", "trace", "track", "trait", "trunk", "tuple", "union", "unit", "value", "vault", "watch", "wheel", "world", # words that look like acronyms when uppercased "ward", "warp", "wrap", "wire", "wars", "star", "mark", "mars", "arc", # common uppercase bash/config tokens that are not acronyms "eof", "ok", "yes", "true", "false", "none", "null", "enabled", "disabled", "missing", "blocked", "forward", "nopasswd", "before", "after", "begin", "end", "stop", "start", "pass", "fail", "skip", "done", "running", "stopped", "exited", "created", "password", "hostname", "logfile", "timestamp", "database", "metric", "metrics", "redacted", "domain", # standard computing abbreviations that are generic (not Echo6-specific) "http", "https", "dns", "ssh", "api", "url", "json", "yaml", "cpu", "gpu", "ram", "ip", "tcp", "udp", "ssl", "tls", "cors", "nvme", "usb", "lan", "vpn", "sso", "nas", "mqtt", "html", "css", "nfs", "mac", "tun", "uri", "wal", "pve", "rss", "sim", "vram", "vps", "lxc", "cli", "pdf", "smtp", "jwt", "jwks", "rsa", "bbs", "web", "llm", "rag", "pem", "aio", "crud", "uuid", "hls", "wsl", "tsv", "esp", "gps", "cuda", "epub", "zim", "npc", "hmac", "nat", "utc", "dhcp", "sql", "vm", "ct", "ai", "ddr", "rtx", "ssd", "nvidia", "amd", "arm", "gnu", "iso", "xml", "csv", "svg", "png", "mp3", "zip", "ini", "toml", "ttl", "irq", "nic", "imap", "smtp", "scp", "sftp", "ftp", "rgb", "utf", "utf8", "ascii", "hex", "eof", "bom", "iommu", "vfio", "acl", "cname", "mx", "txt", "ptr", "icmp", "arp", "bgp", "ospf", "vlan", "mpls", "gre", "tun", "tap", "psk", "rsa", "ecdsa", "aes", "hmac", "jwt", "oidc", # oidc is in entity_upper_tokens so rule 4 will re-admit if needed "rs256", "es512", # crypto alg names "orm", "spa", "dom", "ajax", "mvc", "api", "uefi", "bios", "grub", "raid", "lvm", "xfs", "ext4", "hevc", "h264", "nvenc", "cuda", "opencl", "sdr", "fm", "am", "uhf", "vhf", "hf", "pii", "gdpr", "opsec", "ml", "ai", "nlp", "ocr", # generic AI/ML terms "tbd", "tbr", "wip", "poc", "mvp", "rfc", "pr", "mr", "ui", "ux", "seo", "cms", # shell / scripting env vars and common uppercase tokens "path", "home", "user", "term", "lang", "shell", "editor", "pager", "display", "tmpdir", "config", "debug", "verbose", "output", "input", "timeout", # 2-char garbage tokens (generic abbreviations, not Echo6-specific) "db", "id", "mb", "gb", "kb", "tb", "hp", "ui", "ux", "cc", "ad", "pk", "vk", "fk", "js", "ts", "ws", "wl", "de", "en", "uk", "us", "ca", "eu", "au", "nz", "sf", "tx", "fl", "pr", "bx", "la", "cs", "ml", "dl", "rl", "ve", "ia", "os", "re", "io", "ok", "ps", "ls", "rm", "mv", "cp", "ln", "cd", "bg", "fg", "mm", "em", "px", "ms", "ns", "hz", "mhz", "ghz", "tb", "pb", # generic ALL-CAPS words that aren't acronyms "client", "server", "binary", "script", "memory", "cores", "router", "bridge", "backup", "latest", "stale", "broken", "health", "login", "works", "found", "exists", "nodes", "scopes", "phases", "height", "weight", "depth", "width", "accept", "reject", "allow", "deny", "block", "permit", "pragma", "pragma", "upsert", "revert", "regen", "comms", "apikey", "pubkey", "dbpass", "ssheof", "pyeof", "xxxx", "ctid", "vmid", "puid", "pgid", "uid", "gid", "pid", "oid", "cid", "rid", "sid", "tid", "mid", "bid", "vid", "fid", "dest", "src", "dst", "tmp", "var", "buf", "ptr", "ref", "max", "min", "avg", "sum", "cnt", "num", "idx", "pos", "key", "val", "str", "int", "obj", "arr", "len", "cap", "lts", "oom", "tui", "isp", "acl", "crl", "csr", "der", "pem", "ber", "asn", "oid", "dn", "cn", "ou", "dc", "sata", "nvme", "pcie", "usb", "hdmi", "vga", "dvi", "ddr4", "ecc", "efi", "gpt", "mbr", "xxx", "xxxx", "xxxxxx", "tz", "bbs", "dcrg", "embr", "lure", "gate", "shop", "lord", "dead", "bank", "rap", "spd", "def", "ff", "rpt", "ys", "bk", # generic tech + project acronyms that are NOT Echo6-specific "rest", "smb", "cifs", "nfs", "omv", "obs", "tui", "gui", "diy", "blm", "usfs", "arrl", "csec", "wspr", # ham/org abbreviations "tv", "pro", "yt", "mp4", "mkv", "dvd", "blu", "hdr", "dcrg", "embr", # already in but restate "cannot", "plenty", "works", "stale", "broken", "hard", "two", "three", "four", "five", "six", "seven", "eight", "nine", "yyyy", "mmdd", "hhmm", "utc", "gmt", "xx", "xxx", "xxxx", "glob", "role", "scope", "claim", "grant", "token", "login", "logout", "signup", "register", "atak", # Android Team Awareness Kit — covered as TAK "mpeg", "mpeg4", "hevc", "h264", "h265", "avc", "av1", "vp9", "wsl2", # Windows Subsystem — generic "derp", "stun", "turn", # Tailscale/DERP networking — generic "amqp", "mqtt", "stomp", # messaging protocols — generic "hnsw", # vector index algo — generic tech "jwk", "jwe", "jws", # JWT variants — generic "ec", "rsa", "dsa", "dh", # crypto primitives "cmp", "sub", "add", "xor", "and", "div", "mod", "nbr", "lp", "lp", "sut", # System Under Test "fm10", "g10", "g9", "g8", "g7", "g6", "g5", "g3", "g2", "g1", "g11", "g13", "g12", # camera model series "n2", "n3", "n4", "n5", # Raspberry Pi / Compute Module designators "cm3", "cm4", "cm5", # Compute Module designators "a4000", "a5000", "a6000", "rtx4090", "rx580", # GPU model numbers "mv51", "mv52", # hardware model numbers "sx1262", "sx1268", "sx1276", # LoRa chip designators "tsip", "tsop", # chip/protocol designators "nebra", # Nebra is a specific brand entity covered as aida-nebra "sigint", # signals intelligence — generic intel term "dnat", "snat", "masq", # iptables — generic networking "opsec", "pii", # generic security terms "gdal", "geos", "proj", # geospatial libs — generic "jsonl", "toml", "yaml", "msgpack", # data formats "grst", # unknown but generic-looking "mrn", "trvl", "j51b", # unclear short codes "f0d848", # hex color / hash "sata", "pcie", "nvme", "ddr4", "ddr5", # hardware bus standards "e6", "e7", "e8", # generic model/version numbers "claude", # Anthropic AI — not an Echo6 acronym "bible", # not an acronym "spd", "rpm", "rps", # metrics # common English words missed above "rest", "stop", "hold", "pass", "fail", "warn", "info", "send", "recv", "read", "write", "open", "close", "seek", "two", "one", "ten", "max", "min", "sum", "avg", "hard", "soft", "fast", "slow", "long", "short", "wide", "thin", "new", "old", "raw", "hot", "cold", "full", "empty", } # Curated acronym expansions — these OVERRIDE auto-derived expansions. # Add entries here for Echo6-specific acronyms that need a fixed, authoritative expansion. CURATED_EXPANSIONS: dict[str, str] = { "AIDA": "Autonomous Idaho Assistant", "MMUD": "Mesh Multi-User Dungeon", } # Generic/googleable acronyms — excluded; glossary is for Echo6-specific terms only. Add to this list as needed. GENERIC_ACRONYM_BLOCKLIST: set[str] = { # Seeded from identified generic/googleable terms "MAS", "TAK", "OTS", "OSINT", "DEM", "DM", "E2EE", "E2BE", # Common tech/web/protocol acronyms "OIDC", "SAML", "SSO", "JWT", "OAUTH", "NVENC", "DNAT", "SNAT", "ACL", "CIDR", "DHCP", "NTP", "SMTP", "IMAP", "LDAP", "REST", "GRPC", "CORS", "MQTT", "RAID", "ZFS", "LVM", "VLAN", "NAT", "WAF", "CDN", "UPS", "PSU", # Military / ATAK / comms "ATAK", "COT", # Generic networking "RTT", "TTL", "MTU", "RPC", # Hardware / SBC "SBC", # Web / app "PWA", } # --------------------------------------------------------------------------- # A. Parse vault docs # --------------------------------------------------------------------------- def parse_environment_md(path: Path) -> list[dict]: """Extract entities from environment.md tables.""" entities = [] seen_names = set() try: text = path.read_text() except Exception as e: print(f"[warn] Cannot read {path}: {e}", file=sys.stderr) return entities def add(name, display, etype, aliases, runs_on=None, extra=None): canonical = name.lower().replace(' ', '-').replace('_', '-') if canonical in seen_names: return seen_names.add(canonical) entry = { "name": canonical, "display": display, "type": etype, "aliases": [a for a in aliases if a], "runs_on": runs_on, "source": "docs", } if extra: entry.update(extra) entities.append(entry) # --- Proxmox cluster nodes table --- # | Node | Local IP | Tailscale | Hardware | RAM | Purpose | node_pat = re.compile( r'^\|\s*(data|utility|cloud|media|toc)\s*\|\s*(\d+\.\d+\.\d+\.\d+)\s*\|\s*(\S+)\s*\|', re.MULTILINE ) for m in node_pat.finditer(text): node, lip, ts = m.group(1), m.group(2), m.group(3) add(node, node.capitalize(), "host", [lip, ts]) # --- VMs table --- # | VM | Host | VMID | Local IP | Tailscale | Purpose | vm_pat = re.compile( r'^\|\s*(\S+)\s*\|\s*(\S+)\s*\|\s*(\d+)\s*\|\s*(\d+\.\d+\.\d+\.\d+)\s*\|\s*(\S+)\s*\|', re.MULTILINE ) for m in vm_pat.finditer(text): name, host, vmid, lip, ts = m.group(1), m.group(2), m.group(3), m.group(4), m.group(5) if name.startswith('--') or name == 'VM': continue aliases = [lip] if ts and ts not in ('—', '-', '|'): aliases.append(ts) aliases.append(f"VM {vmid}") add(name, name, "vm", aliases, runs_on=host) # --- Key Servers table --- # | Server | Local IP | Tailscale | Purpose | server_pat = re.compile( r'^\|\s*([a-zA-Z][a-zA-Z0-9\-\.]+)\s*\|\s*(\d+\.\d+\.\d+\.\d+)\s*\|\s*([^\|]+)\s*\|\s*([^\|]+)\s*\|', re.MULTILINE ) for m in server_pat.finditer(text): name, lip, ts_raw, purpose = ( m.group(1).strip(), m.group(2).strip(), m.group(3).strip(), m.group(4).strip() ) # Skip header rows and already-parsed nodes if name in ('Node', 'VM', 'Container', 'Server', 'Node Storage'): continue if name.startswith('-'): continue ts = ts_raw if re.match(r'100\.\d+', ts_raw) else None aliases = [lip] if ts: aliases.append(ts) # Special: aida-nebra has !27780c47 node_id_m = re.search(r'(![\da-f]+)', purpose) if node_id_m: aliases.append(node_id_m.group(1)) # Infer type if 'Contabo' in name or 'edge1' in name.lower() or 'edge2' in name.lower() or 'VPS' in name.upper(): etype = 'vps' elif re.search(r'pi|Pi|raspberry', purpose, re.I) or name.startswith('pi-') or name.startswith('mt-') or name == 'aida-nebra': etype = 'pi' elif 'desktop' in name.lower() or 'Desktop' in purpose: etype = 'desktop' else: etype = 'host' add(name, name, etype, aliases) # --- LXC Containers table --- # | Container | Host | Local IP | Tailscale | Purpose | ct_pat = re.compile( r'^\|\s*([a-zA-Z][a-zA-Z0-9\-]+)\s*\|\s*([^\|]+?)\(CT (\d+)\)\s*\|\s*(\d+\.\d+\.\d+\.\d+)\s*\|\s*([^\|]*)\s*\|\s*([^\|]+)\s*\|', re.MULTILINE ) for m in ct_pat.finditer(text): cname, host_raw, ctid, lip, ts_raw, purpose = ( m.group(1).strip(), m.group(2).strip(), m.group(3).strip(), m.group(4).strip(), m.group(5).strip(), m.group(6).strip() ) if cname.startswith('-') or cname == 'Container': continue host_name = host_raw.split()[0].lower() aliases = [lip, f"CT {ctid}"] if ts_raw and re.match(r'100\.\d+', ts_raw): aliases.append(ts_raw) add(cname, cname, "ct", aliases, runs_on=host_name) # --- Headscale node list table --- # | Node | Tailscale IP | Type | hs_pat = re.compile( r'^\|\s*([a-zA-Z][a-zA-Z0-9\-]+)\s*\|\s*(100\.\d+\.\d+\.\d+)\s*\|\s*([^\|]+)\s*\|', re.MULTILINE ) for m in hs_pat.finditer(text): node, ts_ip, ntype_raw = m.group(1).strip(), m.group(2).strip(), m.group(3).strip() if node in ('Node', '--'): continue # These should already be in entities; just enrich aliases if missing canonical = node.lower() found = next((e for e in entities if e['name'] == canonical), None) if found: if ts_ip not in found['aliases']: found['aliases'].append(ts_ip) else: # Map HS type to our type nt = ntype_raw.split()[0].lower() type_map = {'proxmox': 'host', 'lxc': 'ct', 'vm': 'vm', 'pi': 'pi', 'vps': 'vps', 'desktop': 'desktop', 'mobile': 'mobile', 'router': 'router'} etype = type_map.get(nt, 'host') if canonical not in seen_names: seen_names.add(canonical) entities.append({ "name": canonical, "display": node, "type": etype, "aliases": [ts_ip], "runs_on": None, "source": "docs", }) return entities def parse_services_md(path: Path, known_entities: list[dict]) -> list[dict]: """Extract service entities from services.md table.""" entities = [] seen_names = set(e['name'] for e in known_entities) try: text = path.read_text() except Exception as e: print(f"[warn] Cannot read {path}: {e}", file=sys.stderr) return entities # Main table: | Service | Location | IP:Port | Access | Notes | svc_pat = re.compile( r'^\|\s*([^\|]+?)\s*\|\s*([^\|]+?)\s*\|\s*([^\|]+?)\s*\|\s*([^\|]+?)\s*\|', re.MULTILINE ) for m in svc_pat.finditer(text): svc_raw, loc_raw, ipport_raw, access_raw = ( m.group(1).strip(), m.group(2).strip(), m.group(3).strip(), m.group(4).strip() ) # Skip headers and separators if svc_raw.startswith('-') or svc_raw in ('Service', 'Container', 'Node', 'Server'): continue if '**Decommissioned' in svc_raw or 'Decommissioned' in loc_raw: continue # Canonical name: lowercase-kebab of service name, strip ~~ svc_clean = re.sub(r'~~', '', svc_raw) svc_clean = re.sub(r'\*+', '', svc_clean).strip() if not svc_clean or svc_clean.startswith('-'): continue canonical = re.sub(r'[^a-z0-9]+', '-', svc_clean.lower()).strip('-') if not canonical or canonical in seen_names: continue seen_names.add(canonical) # Extract subdomain alias from access URL aliases = [svc_clean] url_m = re.search(r'https?://([a-z0-9\-\.]+\.echo6\.co)', access_raw, re.I) if url_m: sub = url_m.group(1).split('.')[0] if sub not in aliases: aliases.append(sub) # runs_on: first word of location runs_on = None loc_clean = re.sub(r'\([^)]*\)', '', loc_raw).strip() if loc_clean: first_word = loc_clean.split()[0].lower() # normalize host names host_map = { 'utility': 'utility', 'cloud': 'cloud', 'media': 'media', 'data': 'data', 'toc': 'toc', 'cortex': 'cortex', 'contabo': 'contabo', 'edge1': 'edge1', 'edge2': 'edge2', 'aida-nebra': 'aida-nebra', 'pi-nas': 'pi-nas', } runs_on = host_map.get(first_word, first_word) entities.append({ "name": canonical, "display": svc_clean, "type": "service", "aliases": aliases, "runs_on": runs_on, "source": "docs", }) return entities def parse_projects(vault_dir: Path) -> list[dict]: """Each *.md under vault/projects/ is a project entity.""" entities = [] projects_dir = vault_dir / "projects" if not projects_dir.exists(): return entities for md in sorted(projects_dir.glob("*.md")): stem = md.stem canonical = re.sub(r'[^a-z0-9]+', '-', stem.lower()).strip('-') entities.append({ "name": canonical, "display": stem, "type": "project", "aliases": [stem], "runs_on": None, "source": "docs", }) return entities def _expansion_matches_acronym(full: str, acro: str) -> bool: """ Heuristic: does 'full' plausibly expand 'acro'? Two strategies: 1. Multi-word: initials of words map to acronym letters (standard expansion) 2. CamelCase compound word: split on uppercase transitions (OpenTAKServer → O,T,S → OTS) At minimum, must start with the same letter as the acronym. """ if not full or not acro: return False if full[0].upper() != acro[0]: return False # Strategy 1: multi-word words = [w for w in re.split(r'[\s\-]+', full) if w] if len(words) >= 2: initials = ''.join(w[0].upper() for w in words if len(w) > 1 or w[0].isupper()) acro_letters = re.sub(r'[0-9]', '', acro) matches = sum(1 for ch in acro_letters if ch in initials) if matches >= max(1, len(acro_letters) - 1): return True # Strategy 2: CamelCase single compound word (e.g. OpenTAKServer → OTS) if len(words) == 1: # Extract uppercase letters (treating runs of uppercase as separate tokens) caps = re.findall(r'[A-Z][a-z]*|[A-Z]+(?=[A-Z]|$)', full) initials = ''.join(c[0] for c in caps) acro_letters = re.sub(r'[0-9]', '', acro) if len(acro_letters) >= 2 and initials.upper().startswith(acro_letters[0]): matches = sum(1 for ch in acro_letters if ch in initials.upper()) if matches >= max(1, len(acro_letters) - 1): return True return False def _load_dict_words() -> tuple[set, bool]: """Load system dictionary; fall back to built-in stoplist. Returns (word_set, used_system_dict).""" dict_path = "/usr/share/dict/words" try: import os if os.path.isfile(dict_path) and os.path.getsize(dict_path) > 1000: words = set() with open(dict_path) as f: for line in f: w = line.strip().lower() if w: words.add(w) return words, True except Exception: pass return set(_COMMON_ENGLISH_WORDS), False def harvest_acronyms(vault_dir: Path, all_entities: list[dict]) -> list[dict]: """ Scan all vault/**/*.md for acronym tokens and expansion patterns. Returns list of {acronym, expansion, source}. Qualification rules — ALL must pass: 1. Shape: 2–6 chars, uppercase letters/digits only, at least one letter. 2. NOT a plain English word (checked against /usr/share/dict/words or built-in stoplist — case-insensitive). 3. NOT already an entity name or alias (those are covered as entities). 4. At least ONE of: a. Has an expansion found via adjacency patterns in these docs. b. Referenced by / maps to an entity (appears in entity display name or aliases — e.g. AIDA↔aida-nebra). c. Appears as a standalone all-caps token ≥3 times across all docs. """ # ---- shape regex: 2–6 chars, A-Z0-9, at least one letter ---- # We harvest everything matching [A-Z][A-Z0-9]{1,5} and filter below. acro_pat = re.compile(r'\b([A-Z][A-Z0-9]{1,5})\b') # Expansion patterns # "Full Name (ACRO)" or "Full Name — ACRO" or "Full Name: ACRO" expand_pat = re.compile( r'((?:[A-Z][A-Za-z0-9]+(?:[ \-][A-Za-z][A-Za-z0-9]*){1,6})|(?:[A-Z][A-Za-z0-9]{4,}))' r'(?:\s*[\(—:]\s*)([A-Z][A-Z0-9]{1,5})(?:[\)\s]|$)' ) # "ACRO (Full Name)" or "ACRO — Full Name" or "ACRO: Full Name" expand_pat2 = re.compile( r'\b([A-Z][A-Z0-9]{1,5})\s+(?:\(([A-Z][A-Za-z0-9 \-]{3,50})\)|(?:—\s*|:\s*)([A-Z][A-Za-z0-9 \-]{3,50}))' ) # ---- build entity lookup sets ---- entity_names: set[str] = set() entity_aliases_upper: set[str] = set() for ent in all_entities: entity_names.add(ent['name'].lower()) if ent.get('display'): entity_names.add(ent['display'].lower()) for al in ent.get('aliases', []): entity_names.add(str(al).lower()) # Build set of uppercase tokens that are entity names/aliases # (for rule 3: reject if token.lower() in entity_names) # We also build a set of uppercase strings that appear in entity display names # (for rule 4b: token is referenced by an entity) entity_upper_tokens: set[str] = set() for ent in all_entities: # Split display name and aliases into words, collect ≥2-char uppercase-looking words for text_val in [ent.get('display', ''), ent['name']] + list(ent.get('aliases', [])): for word in re.split(r'[\s\-_]+', str(text_val)): if re.match(r'[A-Z][A-Z0-9]{1,5}$', word): entity_upper_tokens.add(word) # ---- load dictionary ---- dict_words, used_system_dict = _load_dict_words() # ---- scan docs ---- acro_counts: dict[str, int] = {} expansions: dict[str, str] = {} for md in vault_dir.rglob("*.md"): try: doc_text = md.read_text(errors='replace') except Exception: continue # Harvest raw tokens for m in acro_pat.finditer(doc_text): acro = m.group(1) # Rule 1: must have at least one letter (not pure digits) if not re.search(r'[A-Z]', acro): continue acro_counts[acro] = acro_counts.get(acro, 0) + 1 # Harvest expansions "Full Name (ACRO)" and "Full Name — ACRO" for m in expand_pat.finditer(doc_text): full, acro = m.group(1).strip(), m.group(2) if not re.search(r'[A-Z]', acro): continue if not _expansion_matches_acronym(full, acro): continue if acro not in expansions: expansions[acro] = full acro_counts[acro] = acro_counts.get(acro, 0) + 1 # Harvest expansions "ACRO (Full Name)" and "ACRO — Full Name" for m in expand_pat2.finditer(doc_text): acro = m.group(1) full = (m.group(2) or m.group(3) or '').strip() if not full or not re.search(r'[A-Z]', acro): continue if acro not in expansions and _expansion_matches_acronym(full, acro): expansions[acro] = full acro_counts[acro] = acro_counts.get(acro, 0) + 1 # ---- apply filtering rules ---- result = [] for acro, count in sorted(acro_counts.items(), key=lambda x: (-x[1], x[0])): # Rule 1: shape already enforced by regex + letter check above # Extra length check just to be safe if not (2 <= len(acro) <= 6): continue if not re.match(r'[A-Z][A-Z0-9]{1,5}$', acro): continue if not re.search(r'[A-Z]', acro): continue # Rule 2: reject plain English words if acro.lower() in dict_words: continue # Rule 2b: reject generic/googleable acronyms (not Echo6-specific) if acro in GENERIC_ACRONYM_BLOCKLIST: continue # Rule 3: reject if it IS an entity name / alias if acro.lower() in entity_names: continue # Rule 4: must qualify via at least one signal has_expansion = acro in expansions maps_to_entity = acro in entity_upper_tokens high_freq = count >= 3 if not (has_expansion or maps_to_entity or high_freq): continue # Curated expansion takes priority over auto-derived if acro in CURATED_EXPANSIONS: exp = CURATED_EXPANSIONS[acro] else: exp = expansions.get(acro) if exp: exp = re.sub(r'\\s+', ' ', exp).strip() exp = re.sub(r'[,;:]+$', '', exp) if len(exp) > 60 or len(exp.split()) > 8: exp = None result.append({ "acronym": acro, "expansion": exp, "source": "docs", }) return result # --------------------------------------------------------------------------- # B. Live inventory (best-effort) # --------------------------------------------------------------------------- def _ssh_run(host: str, user: str, cmd: str, timeout: int = 5) -> str | None: """Run a remote command; return stdout or None on any failure.""" try: r = subprocess.run( ["ssh", "-o", "BatchMode=yes", "-o", f"ConnectTimeout={timeout}", "-o", "StrictHostKeyChecking=no", f"{user}@{host}", cmd], capture_output=True, text=True, timeout=timeout + 2 ) if r.returncode == 0: return r.stdout return None except Exception: return None def _proxmox_ip_for_name(name: str) -> str | None: """Return the local or Tailscale IP for a known Proxmox host name.""" ip_map = { "data": "192.168.1.240", "utility": "192.168.1.241", "cloud": "192.168.1.242", "media": "192.168.1.243", "toc": "192.168.1.244", "edge1": "5.189.158.149", "edge2": "184.174.35.153", } return ip_map.get(name) def _host_access(name: str) -> tuple[str, str]: """Return (ssh_user, sudo_prefix) for a host. edge2 is hardened: root login is disabled; use admin + passwordless sudo. All other Proxmox hosts (home nodes, edge1) use root.""" if name == "edge2": return ("admin", "sudo ") return ("root", "") def query_proxmox_node(host_name: str) -> list[dict]: """Try pct list + qm list on a Proxmox host; return entity dicts.""" ip = _proxmox_ip_for_name(host_name) if not ip: return [] entities = [] user, sudo = _host_access(host_name) out_pct = _ssh_run(ip, user, f"{sudo}pct list 2>/dev/null") out_qm = _ssh_run(ip, user, f"{sudo}qm list 2>/dev/null") if out_pct: for line in out_pct.splitlines(): m = re.match(r'^\s*(\d+)\s+(\S+)', line) if m and m.group(1) != 'VMID': ctid, name = m.group(1), m.group(2).lower() entities.append({ "name": name, "display": name, "type": "ct", "aliases": [f"CT {ctid}"], "runs_on": host_name, "source": "live", }) if out_qm: for line in out_qm.splitlines(): m = re.match(r'^\s*(\d+)\s+(\S+)', line) if m and m.group(1) != 'VMID': vmid, name = m.group(1), m.group(2).lower() entities.append({ "name": name, "display": name, "type": "vm", "aliases": [f"VM {vmid}"], "runs_on": host_name, "source": "live", }) return entities def query_docker_host(host_name: str, user: str) -> list[dict]: """Try docker ps on a host; return container name entities.""" ip_map = { "cortex": "192.168.1.150", "utility": "192.168.1.241", "media": "192.168.1.160", } ip = ip_map.get(host_name) if not ip: return [] out = _ssh_run(ip, user, "docker ps --format '{{.Names}}' 2>/dev/null") if not out: out = _ssh_run(ip, "root", "docker ps --format '{{.Names}}' 2>/dev/null") if not out: return [] entities = [] for name in out.splitlines(): name = name.strip() if not name: continue canonical = re.sub(r'[^a-z0-9]+', '-', name.lower()).strip('-') entities.append({ "name": canonical, "display": name, "type": "service", "aliases": [name], "runs_on": host_name, "source": "live", }) return entities def query_headscale(headscale_host: str) -> list[dict]: """Try headscale nodes list; Headscale runs dockerized inside edge2 CT 107.""" ip = _proxmox_ip_for_name(headscale_host) or headscale_host user, sudo = _host_access(headscale_host) if headscale_host == "edge2": cmd = f"{sudo}pct exec 107 -- docker exec headscale headscale nodes list 2>/dev/null" else: cmd = "headscale nodes list 2>/dev/null" out = _ssh_run(ip, user, cmd) if not out: return [] entities = [] for line in out.splitlines(): # headscale output: ID | Name | Prefix | IPs | Ephemeral | Last seen | ... parts = [p.strip() for p in line.split('|')] if len(parts) >= 3 and parts[1] and not parts[1].startswith('Name'): name = parts[1].strip().lower() if name: canonical = re.sub(r'[^a-z0-9]+', '-', name).strip('-') entities.append({ "name": canonical, "display": name, "type": "host", "aliases": [], "runs_on": None, "source": "live", }) return entities def run_live_inventory(config: dict) -> tuple[list[dict], list[str]]: """ Run all live inventory queries. Returns (entities, list_of_sources_that_succeeded). """ inv = config.get("inventory", {}) ssh_user = inv.get("ssh_user", "zvx") live_entities: list[dict] = [] succeeded: list[str] = [] # Proxmox nodes for node_name in inv.get("proxmox_nodes", []): try: ents = query_proxmox_node(node_name) if ents: live_entities.extend(ents) succeeded.append(f"proxmox:{node_name}") except Exception: pass for node_name in inv.get("proxmox_vps", []): try: ents = query_proxmox_node(node_name) if ents: live_entities.extend(ents) succeeded.append(f"proxmox:{node_name}") except Exception: pass # Docker hosts for host_name in inv.get("docker_hosts", []): try: ents = query_docker_host(host_name, ssh_user) if ents: live_entities.extend(ents) succeeded.append(f"docker:{host_name}") except Exception: pass # Headscale try: ents = query_headscale(inv.get("headscale_host", "edge2")) if ents: live_entities.extend(ents) succeeded.append(f"headscale:{inv.get('headscale_host', 'edge2')}") except Exception: pass return live_entities, succeeded # --------------------------------------------------------------------------- # C. Build vocab.json # --------------------------------------------------------------------------- def merge_entities(doc_entities: list[dict], live_entities: list[dict]) -> list[dict]: """Merge live entities that aren't already present from docs.""" existing_names = {e['name'] for e in doc_entities} result = list(doc_entities) for le in live_entities: if le['name'] not in existing_names: result.append(le) existing_names.add(le['name']) return result def build_vocab(config: dict) -> dict: vault_dir = Path(config["vault_dir"]) engine_dir = Path(config["engine_dir"]) print("[1/4] Parsing environment.md ...", file=sys.stderr) env_md = vault_dir / "docs" / "hardware" / "environment.md" host_entities = parse_environment_md(env_md) print(f" → {len(host_entities)} entities from environment.md", file=sys.stderr) print("[2/4] Parsing services.md ...", file=sys.stderr) svc_md = vault_dir / "docs" / "services" / "services.md" svc_entities = parse_services_md(svc_md, host_entities) print(f" → {len(svc_entities)} entities from services.md", file=sys.stderr) print("[3/4] Parsing vault/projects/ ...", file=sys.stderr) project_entities = parse_projects(vault_dir) print(f" → {len(project_entities)} project entities", file=sys.stderr) doc_entities = host_entities + svc_entities + project_entities print("[4/4] Harvesting acronyms from all vault docs ...", file=sys.stderr) acronyms = harvest_acronyms(vault_dir, doc_entities) print(f" → {len(acronyms)} acronyms ({sum(1 for a in acronyms if a['expansion'])} with expansions)", file=sys.stderr) print("[5/5] Running live inventory (best-effort) ...", file=sys.stderr) live_entities, live_sources = run_live_inventory(config) print(f" → {len(live_entities)} live entities, sources: {live_sources or ['none']}", file=sys.stderr) all_entities = merge_entities(doc_entities, live_entities) # Counts by type type_counts: dict[str, int] = {} for e in all_entities: t = e['type'] type_counts[t] = type_counts.get(t, 0) + 1 expansions_count = sum(1 for a in acronyms if a['expansion']) vocab = { "_meta": { "generated": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"), "generated_from": ["docs"] + live_sources, "counts": { "entities_total": len(all_entities), "entities_by_type": type_counts, "acronyms_total": len(acronyms), "acronyms_with_expansion": expansions_count, }, }, "topic_categories": config["topic_categories"], "entities": all_entities, "acronyms": acronyms, } return vocab # --------------------------------------------------------------------------- # D. Write outputs # --------------------------------------------------------------------------- def write_vocab_json(vocab: dict, engine_dir: Path) -> Path: out_path = engine_dir / "vocab.json" out_path.write_text(json.dumps(vocab, indent=2, ensure_ascii=False)) return out_path def write_glossary_md(vocab: dict, vault_dir: Path) -> Path: today = datetime.now(timezone.utc).strftime("%Y-%m-%d") lines = [ "---", "title: Glossary & Vocabulary", "type: reference", "tags: [meta]", f"updated: {today}", "---", "# Glossary & Vocabulary", "", "> Auto-generated by the vault engine (`engine/lib/vocab_gen.py`).", "> Acronym expansions marked _(unconfirmed)_ need a human pass.", "", "## Topic categories", "", " · ".join(vocab["topic_categories"]), "", "## Acronyms", "", "| Acronym | Expansion |", "|---|---|", ] for a in vocab["acronyms"]: exp = a["expansion"] if a["expansion"] else "_(unconfirmed)_" lines.append(f"| {a['acronym']} | {exp} |") lines += [ "", "## Entities", "", ] # Group by type type_order = ["host", "vm", "ct", "pi", "vps", "desktop", "mobile", "router", "service", "project"] type_labels = { "host": "Hosts / Proxmox nodes", "vm": "Virtual Machines", "ct": "LXC Containers", "pi": "Raspberry Pi / Edge nodes", "vps": "VPS / External servers", "desktop": "Desktops", "mobile": "Mobile devices", "router": "Routers / Network devices", "service": "Services", "project": "Projects", } by_type: dict[str, list[dict]] = {t: [] for t in type_order} by_type["other"] = [] for e in vocab["entities"]: t = e.get("type", "other") if t in by_type: by_type[t].append(e) else: by_type.setdefault("other", []).append(e) for etype in type_order: ents = by_type.get(etype, []) if not ents: continue label = type_labels.get(etype, etype.title()) lines.append(f"### {label}") lines.append("") for e in sorted(ents, key=lambda x: x['name']): display = e.get('display', e['name']) name_part = f"**{e['name']}**" if display != e['name']: name_part += f" ({display})" alias_str = "" if e.get('aliases'): alias_str = " — aliases: " + ", ".join(e['aliases']) runs_str = "" if e.get('runs_on'): runs_str = f" — on: {e['runs_on']}" src = " _(live)_" if e.get('source') == 'live' else "" lines.append(f"- {name_part}{alias_str}{runs_str}{src}") lines.append("") out_path = vault_dir / "glossary.md" out_path.write_text("\n".join(lines) + "\n") return out_path # --------------------------------------------------------------------------- # Main # --------------------------------------------------------------------------- def main(): config_path = Path(__file__).parent.parent / "config.yaml" alt_path = Path("/home/zvx/projects/.ref/engine/config.yaml") if not config_path.exists() and alt_path.exists(): config_path = alt_path print(f"Loading config from {config_path}", file=sys.stderr) config = load_config(str(config_path)) vault_dir = Path(config["vault_dir"]) engine_dir = Path(config["engine_dir"]) if not vault_dir.exists(): print(f"[error] vault_dir not found: {vault_dir}", file=sys.stderr) sys.exit(1) if not engine_dir.exists(): print(f"[error] engine_dir not found: {engine_dir}", file=sys.stderr) sys.exit(1) vocab = build_vocab(config) vocab_path = write_vocab_json(vocab, engine_dir) print(f"\nWrote {vocab_path}", file=sys.stderr) glossary_path = write_glossary_md(vocab, vault_dir) print(f"Wrote {glossary_path}", file=sys.stderr) # Summary meta = vocab["_meta"] print(f"\n=== vocab_gen complete ===", file=sys.stderr) print(f" Entities : {meta['counts']['entities_total']} total", file=sys.stderr) for t, c in sorted(meta['counts']['entities_by_type'].items()): print(f" {t:12s}: {c}", file=sys.stderr) print(f" Acronyms : {meta['counts']['acronyms_total']} total, " f"{meta['counts']['acronyms_with_expansion']} with expansions", file=sys.stderr) _, used_sys = _load_dict_words() dict_src = "/usr/share/dict/words" if used_sys else "built-in stoplist (~300 words)" print(f" Dict source : {dict_src}", file=sys.stderr) print(f" Live sources: {meta['generated_from']}", file=sys.stderr) if __name__ == "__main__": main()