diff --git a/landing/search-authority.json b/landing/search-authority.json
new file mode 100644
index 000000000..edad7ed26
--- /dev/null
+++ b/landing/search-authority.json
@@ -0,0 +1,78 @@
+{
+ "schemaVersion": 1,
+ "clusters": {
+ "connection": {"label": {"en": "MMS connection and live model", "id": "Koneksi MMS dan live model"}},
+ "reporting": {"label": {"en": "Reporting, RCB and DataSet evidence", "id": "Evidence reporting, RCB dan DataSet"}},
+ "file-transfer": {"label": {"en": "MMS file service and COMTRADE", "id": "MMS file service dan COMTRADE"}},
+ "goose": {"label": {"en": "GOOSE sequence and network evidence", "id": "Sequence dan evidence network GOOSE"}},
+ "scl": {"label": {"en": "SCL intent and live configuration", "id": "Intent SCL dan konfigurasi live"}},
+ "control": {"label": {"en": "IEC 61850 control evidence", "id": "Evidence control IEC 61850"}},
+ "commissioning": {"label": {"en": "FAT, SAT and commissioning workflow", "id": "Workflow FAT, SAT dan commissioning"}}
+ },
+ "nodes": {
+ "mms-client.html": {"cluster": "connection", "role": "capability", "label": {"en": "IEC 61850 MMS Client", "id": "MMS Client IEC 61850"}, "summary": {"en": "Association, live model, DataSet, RCB and control-object evidence.", "id": "Evidence association, live model, DataSet, RCB dan control object."}, "primaryQuery": "IEC 61850 MMS client"},
+ "mms-client-iec61850.html": {"cluster": "connection", "role": "capability", "label": {"en": "IEC 61850 MMS Client in Indonesian", "id": "MMS Client IEC 61850"}, "summary": {"en": "Indonesian workflow for association and live-model evidence.", "id": "Workflow association dan evidence live model dalam Bahasa Indonesia."}, "primaryQuery": "MMS client IEC 61850"},
+ "smart-reporting.html": {"cluster": "reporting", "role": "capability", "label": {"en": "IEC 61850 Smart Reporting", "id": "Smart Reporting IEC 61850"}, "summary": {"en": "BRCB, URCB, exact DataSet coverage and bounded polling fallback.", "id": "BRCB, URCB, exact coverage DataSet dan polling fallback terkontrol."}, "primaryQuery": "IEC 61850 reporting BRCB URCB"},
+ "smart-reporting-iec61850.html": {"cluster": "reporting", "role": "capability", "label": {"en": "Smart Reporting in Indonesian", "id": "Smart Reporting IEC 61850"}, "summary": {"en": "Indonesian reporting workflow and fallback evidence.", "id": "Workflow reporting dan evidence fallback dalam Bahasa Indonesia."}, "primaryQuery": "reporting IEC 61850 BRCB URCB"},
+ "goose-analyzer.html": {"cluster": "goose", "role": "capability", "label": {"en": "IEC 61850 GOOSE Analyzer", "id": "Analyzer GOOSE IEC 61850"}, "summary": {"en": "APPID, VLAN, stNum, sqNum, TAL and payload continuity.", "id": "APPID, VLAN, stNum, sqNum, TAL dan continuity payload."}, "primaryQuery": "IEC 61850 GOOSE analyzer"},
+ "analyzer-goose-iec61850.html": {"cluster": "goose", "role": "capability", "label": {"en": "GOOSE Analyzer in Indonesian", "id": "Analyzer GOOSE IEC 61850"}, "summary": {"en": "Indonesian packet and sequence evidence workflow.", "id": "Workflow evidence packet dan sequence dalam Bahasa Indonesia."}, "primaryQuery": "analyzer GOOSE IEC 61850"},
+ "file-transfer.html": {"cluster": "file-transfer", "role": "capability", "label": {"en": "IEC 61850 File Transfer", "id": "Transfer File IEC 61850"}, "summary": {"en": "MMS file service, remote directories and COMTRADE record sets.", "id": "MMS file service, remote directory dan record set COMTRADE."}, "primaryQuery": "IEC 61850 file transfer COMTRADE"},
+ "transfer-file-comtrade-iec61850.html": {"cluster": "file-transfer", "role": "capability", "label": {"en": "COMTRADE File Transfer in Indonesian", "id": "Transfer File COMTRADE IEC 61850"}, "summary": {"en": "Indonesian file-service and record-set workflow.", "id": "Workflow file service dan record set dalam Bahasa Indonesia."}, "primaryQuery": "download COMTRADE IEC 61850"},
+ "scl-workspace.html": {"cluster": "scl", "role": "capability", "label": {"en": "IEC 61850 SCL Workspace", "id": "Workspace SCL IEC 61850"}, "summary": {"en": "Compare configured SCL intent with live IED evidence.", "id": "Bandingkan intent SCL dengan evidence live dari IED."}, "primaryQuery": "IEC 61850 SCL viewer workspace"},
+ "workspace-scl-iec61850.html": {"cluster": "scl", "role": "capability", "label": {"en": "SCL Workspace in Indonesian", "id": "Workspace SCL IEC 61850"}, "summary": {"en": "Indonesian configured-intent and live-evidence workflow.", "id": "Workflow configured intent dan live evidence dalam Bahasa Indonesia."}, "primaryQuery": "workspace SCL IEC 61850"},
+ "control.html": {"cluster": "control", "role": "capability", "label": {"en": "IEC 61850 Control Testing", "id": "Pengujian Control IEC 61850"}, "summary": {"en": "Direct, SBO, typed values, origin, checks and final termination.", "id": "Direct, SBO, typed value, origin, check dan final termination."}, "primaryQuery": "IEC 61850 direct SBO control testing"},
+ "multi-ied-monitoring.html": {"cluster": "reporting", "role": "capability", "label": {"en": "Multi-IED Monitoring", "id": "Monitoring Multi-IED"}, "summary": {"en": "Independent associations with attributable live-value evidence.", "id": "Association independen dengan evidence live value yang tetap attributable."}, "primaryQuery": "IEC 61850 multi IED monitoring"},
+ "fat-testing.html": {"cluster": "commissioning", "role": "solution", "label": {"en": "IEC 61850 FAT Testing", "id": "Pengujian FAT IEC 61850"}, "summary": {"en": "Repeatable factory evidence across MMS, reports, GOOSE and control.", "id": "Evidence factory yang repeatable untuk MMS, report, GOOSE dan control."}, "primaryQuery": "IEC 61850 FAT testing"},
+ "pengujian-fat-iec61850.html": {"cluster": "commissioning", "role": "solution", "label": {"en": "IEC 61850 FAT in Indonesian", "id": "Pengujian FAT IEC 61850"}, "summary": {"en": "Indonesian FAT evidence workflow.", "id": "Workflow evidence FAT dalam Bahasa Indonesia."}, "primaryQuery": "pengujian FAT IEC 61850"},
+ "sat-testing.html": {"cluster": "commissioning", "role": "solution", "label": {"en": "IEC 61850 SAT Testing", "id": "Pengujian SAT IEC 61850"}, "summary": {"en": "Site evidence for network, reports, GOOSE and control paths.", "id": "Evidence site untuk network, report, GOOSE dan control path."}, "primaryQuery": "IEC 61850 SAT testing"},
+ "pengujian-sat-iec61850.html": {"cluster": "commissioning", "role": "solution", "label": {"en": "IEC 61850 SAT in Indonesian", "id": "Pengujian SAT IEC 61850"}, "summary": {"en": "Indonesian SAT evidence workflow.", "id": "Workflow evidence SAT dalam Bahasa Indonesia."}, "primaryQuery": "pengujian SAT IEC 61850"},
+ "commissioning.html": {"cluster": "commissioning", "role": "solution", "label": {"en": "IEC 61850 Commissioning", "id": "Commissioning IEC 61850"}, "summary": {"en": "Traceable commissioning evidence from connection to final operation.", "id": "Evidence commissioning yang traceable dari koneksi sampai final operation."}, "primaryQuery": "IEC 61850 commissioning tool"},
+ "commissioning-iec61850.html": {"cluster": "commissioning", "role": "solution", "label": {"en": "IEC 61850 Commissioning in Indonesian", "id": "Commissioning IEC 61850"}, "summary": {"en": "Indonesian commissioning evidence workflow.", "id": "Workflow evidence commissioning dalam Bahasa Indonesia."}, "primaryQuery": "commissioning IEC 61850"},
+ "multi-vendor-integration.html": {"cluster": "scl", "role": "solution", "label": {"en": "Multi-vendor IEC 61850 Integration", "id": "Integrasi Multi-vendor IEC 61850"}, "summary": {"en": "Separate configured intent, live behavior and vendor-specific boundaries.", "id": "Pisahkan configured intent, behavior live dan batas vendor-specific."}, "primaryQuery": "multi vendor IEC 61850 integration"},
+ "integrasi-multi-vendor-iec61850.html": {"cluster": "scl", "role": "solution", "label": {"en": "Multi-vendor Integration in Indonesian", "id": "Integrasi Multi-vendor IEC 61850"}, "summary": {"en": "Indonesian multi-vendor evidence workflow.", "id": "Workflow evidence multi-vendor dalam Bahasa Indonesia."}, "primaryQuery": "integrasi multi vendor IEC 61850"},
+ "reporting-silent.html": {"cluster": "reporting", "role": "guide", "label": {"en": "Reporting is silent", "id": "Reporting tidak masuk"}, "summary": {"en": "Separate association, reads, DataSet coverage, RCB enablement and delivery evidence.", "id": "Pisahkan association, read, coverage DataSet, enablement RCB dan evidence delivery."}, "primaryQuery": "IEC 61850 reporting not working"},
+ "brcb-vs-urcb.html": {"cluster": "reporting", "role": "guide", "label": {"en": "BRCB versus URCB", "id": "BRCB dibanding URCB"}, "summary": {"en": "Choose and test buffered or unbuffered reporting without assuming continuity.", "id": "Pilih dan uji buffered atau unbuffered reporting tanpa mengasumsikan continuity."}, "primaryQuery": "BRCB vs URCB"},
+ "rcb-reserved.html": {"cluster": "reporting", "role": "guide", "label": {"en": "RCB reserved or occupied", "id": "RCB reserved atau occupied"}, "summary": {"en": "Inspect ownership and reservation without forcing takeover.", "id": "Periksa ownership dan reservation tanpa forced takeover."}, "primaryQuery": "IEC 61850 RCB reserved"},
+ "empty-dataset.html": {"cluster": "reporting", "role": "guide", "label": {"en": "Empty report DataSet", "id": "DataSet report kosong"}, "summary": {"en": "Confirm references, member lists, permissions and fallback evidence.", "id": "Konfirmasi reference, member list, permission dan evidence fallback."}, "primaryQuery": "IEC 61850 empty dataset"},
+ "port-102-connection-failed.html": {"cluster": "connection", "role": "guide", "label": {"en": "Port 102 connection failed", "id": "Koneksi port 102 gagal"}, "summary": {"en": "Separate routing, ping, TCP 102, firewall, association and application evidence.", "id": "Pisahkan routing, ping, TCP 102, firewall, association dan evidence aplikasi."}, "primaryQuery": "IEC 61850 port 102 connection failed"},
+ "comtrade-download.html": {"cluster": "file-transfer", "role": "guide", "label": {"en": "COMTRADE download troubleshooting", "id": "Troubleshooting download COMTRADE"}, "summary": {"en": "Distinguish file-service support, transfer success and complete record sets.", "id": "Bedakan dukungan file service, transfer berhasil dan record set lengkap."}, "primaryQuery": "IEC 61850 COMTRADE download"},
+ "goose-sequence.html": {"cluster": "goose", "role": "guide", "label": {"en": "GOOSE stNum and sqNum", "id": "GOOSE stNum dan sqNum"}, "summary": {"en": "Interpret state changes, retransmissions, TAL and capture continuity together.", "id": "Interpretasikan state change, retransmission, TAL dan capture continuity bersama-sama."}, "primaryQuery": "GOOSE stNum sqNum"},
+ "cid-rejected.html": {"cluster": "scl", "role": "guide", "label": {"en": "CID or SCL rejected", "id": "CID atau SCL ditolak"}, "summary": {"en": "Inspect XML, namespace, edition, references and vendor requirements.", "id": "Periksa XML, namespace, edition, reference dan kebutuhan vendor."}, "primaryQuery": "IEC 61850 CID rejected"},
+ "live-model-vs-scl.html": {"cluster": "scl", "role": "guide", "label": {"en": "Live model versus SCL", "id": "Live model dibanding SCL"}, "summary": {"en": "Use SCL as configured intent and the online model as current server evidence.", "id": "Gunakan SCL sebagai configured intent dan model online sebagai evidence server aktif."}, "primaryQuery": "IEC 61850 live model vs SCL"},
+ "direct-vs-sbo.html": {"cluster": "control", "role": "guide", "label": {"en": "Direct versus SBO control", "id": "Control Direct dibanding SBO"}, "summary": {"en": "Confirm the control model and service sequence before operating.", "id": "Konfirmasi control model dan urutan service sebelum operate."}, "primaryQuery": "IEC 61850 direct vs SBO"},
+ "commandtermination-addcause.html": {"cluster": "control", "role": "guide", "label": {"en": "CommandTermination and AddCause", "id": "CommandTermination dan AddCause"}, "summary": {"en": "Separate immediate acceptance from final process completion evidence.", "id": "Pisahkan immediate acceptance dari evidence final process completion."}, "primaryQuery": "IEC 61850 CommandTermination AddCause"}
+ },
+ "pages": {
+ "mms-client.html": ["port-102-connection-failed.html", "live-model-vs-scl.html", "reporting-silent.html"],
+ "mms-client-iec61850.html": ["port-102-connection-failed.html", "live-model-vs-scl.html", "smart-reporting-iec61850.html"],
+ "smart-reporting.html": ["reporting-silent.html", "brcb-vs-urcb.html", "rcb-reserved.html", "empty-dataset.html"],
+ "smart-reporting-iec61850.html": ["reporting-silent.html", "brcb-vs-urcb.html", "rcb-reserved.html", "empty-dataset.html"],
+ "goose-analyzer.html": ["goose-sequence.html", "fat-testing.html", "sat-testing.html"],
+ "analyzer-goose-iec61850.html": ["goose-sequence.html", "pengujian-fat-iec61850.html", "pengujian-sat-iec61850.html"],
+ "file-transfer.html": ["comtrade-download.html", "port-102-connection-failed.html", "commissioning.html"],
+ "transfer-file-comtrade-iec61850.html": ["comtrade-download.html", "port-102-connection-failed.html", "commissioning-iec61850.html"],
+ "scl-workspace.html": ["cid-rejected.html", "live-model-vs-scl.html", "multi-vendor-integration.html"],
+ "workspace-scl-iec61850.html": ["cid-rejected.html", "live-model-vs-scl.html", "integrasi-multi-vendor-iec61850.html"],
+ "control.html": ["direct-vs-sbo.html", "commandtermination-addcause.html", "commissioning.html"],
+ "multi-ied-monitoring.html": ["reporting-silent.html", "port-102-connection-failed.html", "smart-reporting.html"],
+ "fat-testing.html": ["goose-sequence.html", "live-model-vs-scl.html", "direct-vs-sbo.html"],
+ "pengujian-fat-iec61850.html": ["goose-sequence.html", "live-model-vs-scl.html", "direct-vs-sbo.html"],
+ "sat-testing.html": ["port-102-connection-failed.html", "goose-sequence.html", "commandtermination-addcause.html"],
+ "pengujian-sat-iec61850.html": ["port-102-connection-failed.html", "goose-sequence.html", "commandtermination-addcause.html"],
+ "commissioning.html": ["port-102-connection-failed.html", "comtrade-download.html", "reporting-silent.html", "commandtermination-addcause.html"],
+ "commissioning-iec61850.html": ["port-102-connection-failed.html", "comtrade-download.html", "reporting-silent.html", "commandtermination-addcause.html"],
+ "multi-vendor-integration.html": ["cid-rejected.html", "live-model-vs-scl.html", "brcb-vs-urcb.html"],
+ "integrasi-multi-vendor-iec61850.html": ["cid-rejected.html", "live-model-vs-scl.html", "brcb-vs-urcb.html"],
+ "reporting-silent.html": ["smart-reporting.html", "rcb-reserved.html", "empty-dataset.html"],
+ "brcb-vs-urcb.html": ["smart-reporting.html", "rcb-reserved.html", "reporting-silent.html"],
+ "rcb-reserved.html": ["smart-reporting.html", "brcb-vs-urcb.html", "reporting-silent.html"],
+ "empty-dataset.html": ["smart-reporting.html", "reporting-silent.html", "live-model-vs-scl.html"],
+ "port-102-connection-failed.html": ["mms-client.html", "file-transfer.html", "commissioning.html"],
+ "comtrade-download.html": ["file-transfer.html", "port-102-connection-failed.html", "commissioning.html"],
+ "goose-sequence.html": ["goose-analyzer.html", "fat-testing.html", "sat-testing.html"],
+ "cid-rejected.html": ["scl-workspace.html", "live-model-vs-scl.html", "multi-vendor-integration.html"],
+ "live-model-vs-scl.html": ["scl-workspace.html", "cid-rejected.html", "multi-vendor-integration.html"],
+ "direct-vs-sbo.html": ["control.html", "commandtermination-addcause.html", "fat-testing.html"],
+ "commandtermination-addcause.html": ["control.html", "direct-vs-sbo.html", "sat-testing.html"]
+ }
+}
diff --git a/scripts/apply-search-authority.py b/scripts/apply-search-authority.py
new file mode 100644
index 000000000..5b45ac6e3
--- /dev/null
+++ b/scripts/apply-search-authority.py
@@ -0,0 +1,247 @@
+#!/usr/bin/env python3
+"""Validate and apply the ARSAS contextual search-authority graph to a built site."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+from collections import defaultdict
+from html import escape
+from pathlib import Path
+from typing import Any
+
+ROOT = Path(__file__).resolve().parents[1]
+DEFAULT_GRAPH = ROOT / "landing" / "search-authority.json"
+GUIDE_ROLES = {"guide"}
+DISCOVERY_ROLES = {"capability", "solution"}
+
+
+def load_object(path: Path, label: str) -> dict[str, Any]:
+ try:
+ value = json.loads(path.read_text(encoding="utf-8"))
+ except (OSError, json.JSONDecodeError) as exc:
+ raise SystemExit(f"Cannot read {label}: {exc}") from exc
+ if not isinstance(value, dict):
+ raise SystemExit(f"{label} must contain a JSON object")
+ return value
+
+
+def registry(site: Path) -> dict[str, dict[str, Any]]:
+ config = load_object(site / "site.json", "built site registry")
+ pages = config.get("pages")
+ if not isinstance(pages, list):
+ raise SystemExit("Built site registry has no pages list")
+ result: dict[str, dict[str, Any]] = {}
+ for item in pages:
+ if not isinstance(item, dict) or not isinstance(item.get("path"), str):
+ raise SystemExit("Built site registry contains an invalid page entry")
+ path = item["path"] or "index.html"
+ result[path] = item
+ return result
+
+
+def text(value: Any, label: str) -> str:
+ if not isinstance(value, str) or not value.strip():
+ raise SystemExit(f"Search authority graph has invalid {label}")
+ return value.strip()
+
+
+def localized(node: dict[str, Any], key: str, language: str) -> str:
+ values = node.get(key)
+ if not isinstance(values, dict):
+ raise SystemExit(f"Search authority node is missing {key}")
+ return text(values.get(language) or values.get("en"), f"{key}.{language}")
+
+
+def validate_graph(graph: dict[str, Any], entries: dict[str, dict[str, Any]]) -> tuple[dict[str, dict[str, Any]], dict[str, list[str]], dict[str, int]]:
+ if graph.get("schemaVersion") != 1:
+ raise SystemExit("Search authority graph schemaVersion must be 1")
+ clusters = graph.get("clusters")
+ nodes = graph.get("nodes")
+ pages = graph.get("pages")
+ if not isinstance(clusters, dict) or not clusters:
+ raise SystemExit("Search authority graph must define clusters")
+ if not isinstance(nodes, dict) or not nodes:
+ raise SystemExit("Search authority graph must define nodes")
+ if not isinstance(pages, dict) or not pages:
+ raise SystemExit("Search authority graph must define page relationships")
+
+ indexed = {path for path, item in entries.items() if item.get("index", True) is not False}
+ for cluster, value in clusters.items():
+ text(cluster, "cluster key")
+ if not isinstance(value, dict):
+ raise SystemExit(f"Search authority cluster {cluster} must be an object")
+ labels = value.get("label")
+ if not isinstance(labels, dict):
+ raise SystemExit(f"Search authority cluster {cluster} is missing labels")
+ text(labels.get("en"), f"clusters.{cluster}.label.en")
+ text(labels.get("id"), f"clusters.{cluster}.label.id")
+
+ normalized_nodes: dict[str, dict[str, Any]] = {}
+ for path, node in nodes.items():
+ if path not in indexed:
+ raise SystemExit(f"Search authority node is not an indexable page: {path}")
+ if not isinstance(node, dict):
+ raise SystemExit(f"Search authority node {path} must be an object")
+ cluster = text(node.get("cluster"), f"nodes.{path}.cluster")
+ if cluster not in clusters:
+ raise SystemExit(f"Search authority node {path} uses unknown cluster {cluster}")
+ role = text(node.get("role"), f"nodes.{path}.role")
+ if role not in {"capability", "solution", "guide", "hub", "authority"}:
+ raise SystemExit(f"Search authority node {path} has invalid role {role}")
+ localized(node, "label", "en")
+ localized(node, "label", "id")
+ localized(node, "summary", "en")
+ localized(node, "summary", "id")
+ text(node.get("primaryQuery"), f"nodes.{path}.primaryQuery")
+ normalized_nodes[path] = node
+
+ normalized_pages: dict[str, list[str]] = {}
+ inbound: dict[str, int] = defaultdict(int)
+ discovery_inbound: dict[str, int] = defaultdict(int)
+ for source, raw_targets in pages.items():
+ if source not in normalized_nodes:
+ raise SystemExit(f"Search authority source has no node metadata: {source}")
+ if not isinstance(raw_targets, list) or not 2 <= len(raw_targets) <= 4:
+ raise SystemExit(f"Search authority source {source} must have two to four related pages")
+ targets = [text(item, f"pages.{source}") for item in raw_targets]
+ if len(targets) != len(set(targets)):
+ raise SystemExit(f"Search authority source {source} contains duplicate targets")
+ if source in targets:
+ raise SystemExit(f"Search authority source {source} links to itself")
+ for target in targets:
+ if target not in normalized_nodes:
+ raise SystemExit(f"Search authority target has no node metadata: {target}")
+ inbound[target] += 1
+ if normalized_nodes[source]["role"] in DISCOVERY_ROLES:
+ discovery_inbound[target] += 1
+ normalized_pages[source] = targets
+
+ guide_paths = {
+ path for path, item in entries.items()
+ if item.get("contentType") == "guide"
+ }
+ missing_nodes = sorted(guide_paths - set(normalized_nodes))
+ if missing_nodes:
+ raise SystemExit("Troubleshooting guides missing from authority graph: " + ", ".join(missing_nodes))
+ missing_discovery = sorted(path for path in guide_paths if discovery_inbound[path] < 1)
+ if missing_discovery:
+ raise SystemExit("Guides lack capability/solution inbound paths: " + ", ".join(missing_discovery))
+
+ stats = {
+ "clusterCount": len(clusters),
+ "nodeCount": len(normalized_nodes),
+ "mappedPageCount": len(normalized_pages),
+ "guideCount": len(guide_paths),
+ "guidesWithDiscoveryInbound": sum(1 for path in guide_paths if discovery_inbound[path] >= 1),
+ "relationshipCount": sum(len(items) for items in normalized_pages.values()),
+ }
+ return normalized_nodes, normalized_pages, stats
+
+
+def target_attributes(source_language: str, target_language: str) -> str:
+ if source_language == target_language:
+ return ""
+ return f' lang="{escape(target_language)}" hreflang="{escape(target_language)}"'
+
+
+def section(source: str, language: str, nodes: dict[str, dict[str, Any]], pages: dict[str, list[str]]) -> str:
+ source_node = nodes[source]
+ cluster = str(source_node["cluster"])
+ is_id = language == "id"
+ kicker = "Jalur evidence terkait" if is_id else "Related evidence paths"
+ heading = "Lanjutkan ke gejala atau workflow yang paling dekat." if is_id else "Continue with the closest failure mode or workflow."
+ intro = (
+ "Link berikut dipilih dari cluster engineering yang sama agar diagnosis berpindah dari gejala ke evidence yang relevan."
+ if is_id else
+ "These links stay inside the same engineering evidence cluster, moving from the current workflow to the most relevant failure mode."
+ )
+ cards: list[str] = []
+ for target in pages[source]:
+ target_node = nodes[target]
+ target_language = "id" if target.endswith(".html") and target in {
+ "mms-client-iec61850.html", "smart-reporting-iec61850.html", "analyzer-goose-iec61850.html",
+ "transfer-file-comtrade-iec61850.html", "workspace-scl-iec61850.html",
+ "pengujian-fat-iec61850.html", "pengujian-sat-iec61850.html",
+ "commissioning-iec61850.html", "integrasi-multi-vendor-iec61850.html",
+ } else "en"
+ label = localized(target_node, "label", language)
+ summary = localized(target_node, "summary", language)
+ role = str(target_node["role"])
+ role_label = {
+ "guide": "Panduan troubleshooting" if is_id else "Troubleshooting guide",
+ "capability": "Capability produk" if is_id else "Product capability",
+ "solution": "Workflow solusi" if is_id else "Solution workflow",
+ }.get(role, "Evidence path")
+ attrs = target_attributes(language, target_language)
+ cards.append(
+ f'
{escape(role_label)}'
+ f'{escape(label)}
{escape(summary)}
'
+ f'{"Buka jalur evidence" if is_id else "Open evidence path"} →'
+ )
+ return (
+ f'
\n\n'
+ )
+
+
+def apply(site: Path, graph_path: Path) -> dict[str, int]:
+ entries = registry(site)
+ graph = load_object(graph_path, "landing/search-authority.json")
+ nodes, pages, stats = validate_graph(graph, entries)
+ for source in sorted(pages):
+ page = site / source
+ if not page.is_file():
+ raise SystemExit(f"Built page required by authority graph is missing: {source}")
+ html = page.read_text(encoding="utf-8")
+ if "data-search-authority=" in html:
+ raise SystemExit(f"Built page already contains a search-authority section: {source}")
+ language = str(entries[source].get("language", "en"))
+ generated = section(source, language, nodes, pages)
+ marker = '
'
+ if marker in html:
+ html = html.replace(marker, generated + marker, 1)
+ elif "" in html:
+ html = html.replace("", generated + "", 1)
+ else:
+ raise SystemExit(f"Cannot place search-authority section in {source}")
+ page.write_text(html, encoding="utf-8")
+
+ build_info_path = site / "build-info.json"
+ build_info = load_object(build_info_path, "build-info.json")
+ build_info["searchAuthority"] = {
+ "schemaVersion": 1,
+ **stats,
+ "allGuidesHaveCapabilityOrSolutionInbound": stats["guidesWithDiscoveryInbound"] == stats["guideCount"],
+ }
+ build_info_path.write_text(json.dumps(build_info, indent=2, ensure_ascii=False) + "\n", encoding="utf-8")
+ deployed_graph = site / "search-authority.json"
+ if deployed_graph.exists():
+ deployed_graph.unlink()
+ return stats
+
+
+def main() -> int:
+ parser = argparse.ArgumentParser()
+ parser.add_argument("site", nargs="?", default="_site")
+ parser.add_argument("--graph", default=str(DEFAULT_GRAPH))
+ args = parser.parse_args()
+ stats = apply(Path(args.site).resolve(), Path(args.graph).resolve())
+ print(
+ "Applied ARSAS search authority graph: "
+ f"{stats['mappedPageCount']} pages, {stats['relationshipCount']} contextual links, "
+ f"{stats['guidesWithDiscoveryInbound']}/{stats['guideCount']} guides with capability/solution inbound paths."
+ )
+ return 0
+
+
+if __name__ == "__main__":
+ try:
+ raise SystemExit(main())
+ except Exception as exc:
+ print(f"ARSAS search authority application failed: {exc}", file=sys.stderr)
+ raise
diff --git a/scripts/build-search-growth-report.py b/scripts/build-search-growth-report.py
new file mode 100644
index 000000000..b4050a4a5
--- /dev/null
+++ b/scripts/build-search-growth-report.py
@@ -0,0 +1,285 @@
+#!/usr/bin/env python3
+"""Build a private ARSAS search-growth queue from structure and optional measurement evidence."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import math
+from collections import defaultdict
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any
+from urllib.parse import urlsplit
+
+ROOT = Path(__file__).resolve().parents[1]
+DEFAULT_GRAPH = ROOT / "landing" / "search-authority.json"
+CANONICAL_PREFIX = "/arsas/"
+
+
+def load(path: Path, required: bool = True) -> dict[str, Any]:
+ if not path.exists():
+ if required:
+ raise SystemExit(f"Missing JSON input: {path}")
+ return {}
+ try:
+ value = json.loads(path.read_text(encoding="utf-8"))
+ except json.JSONDecodeError as exc:
+ raise SystemExit(f"Invalid JSON input {path}: {exc}") from exc
+ if not isinstance(value, dict):
+ raise SystemExit(f"JSON input must contain an object: {path}")
+ return value
+
+
+def normalize_path(value: str) -> str:
+ parsed = urlsplit(value)
+ path = parsed.path
+ if path in {"/arsas", "/arsas/", ""}:
+ return "index.html"
+ if path.startswith(CANONICAL_PREFIX):
+ path = path[len(CANONICAL_PREFIX):]
+ return path.lstrip("/") or "index.html"
+
+
+def registry(site: Path) -> dict[str, dict[str, Any]]:
+ config = load(site / "site.json")
+ result: dict[str, dict[str, Any]] = {}
+ for item in config.get("pages", []):
+ if isinstance(item, dict) and isinstance(item.get("path"), str):
+ result[item["path"] or "index.html"] = item
+ return result
+
+
+def localized_counterparts(entries: dict[str, dict[str, Any]]) -> dict[str, str]:
+ pairs: dict[str, str] = {}
+ for path, item in entries.items():
+ if item.get("language", "en") != "en":
+ continue
+ alternates = item.get("alternates")
+ if isinstance(alternates, dict) and isinstance(alternates.get("id"), str):
+ pairs[path] = alternates["id"] or "id.html"
+ return pairs
+
+
+def add(queue: list[dict[str, Any]], *, priority: int, category: str, page: str = "", query: str = "", evidence: str, action: str, source: str) -> None:
+ queue.append({
+ "priority": max(1, min(100, int(priority))),
+ "category": category,
+ "page": page,
+ "query": query,
+ "evidence": evidence,
+ "action": action,
+ "source": source,
+ })
+
+
+def structural_queue(entries: dict[str, dict[str, Any]], graph: dict[str, Any]) -> tuple[list[dict[str, Any]], dict[str, Any]]:
+ nodes = graph.get("nodes", {})
+ pages = graph.get("pages", {})
+ inbound: dict[str, int] = defaultdict(int)
+ for targets in pages.values():
+ if isinstance(targets, list):
+ for target in targets:
+ if isinstance(target, str):
+ inbound[target] += 1
+
+ queue: list[dict[str, Any]] = []
+ pairs = localized_counterparts(entries)
+ localized_targets = set(pairs.values())
+ for path, node in nodes.items():
+ if not isinstance(node, dict):
+ continue
+ role = str(node.get("role", ""))
+ language = str(entries.get(path, {}).get("language", "en"))
+ if language == "en" and role in {"guide", "capability", "solution"} and path not in pairs:
+ base = {"guide": 68, "capability": 54, "solution": 50}.get(role, 45)
+ add(
+ queue,
+ priority=base,
+ category="localization-gap",
+ page=path,
+ query=str(node.get("primaryQuery", "")),
+ evidence=f"Indexable English {role} page has no declared Indonesian alternate.",
+ action="Translate only after search impressions or Indonesian demand justify the work; then add reciprocal hreflang and hub links.",
+ source="authority-graph",
+ )
+ if inbound[path] < 2:
+ add(
+ queue,
+ priority=42 if role == "guide" else 32,
+ category="authority-depth",
+ page=path,
+ query=str(node.get("primaryQuery", "")),
+ evidence=f"Contextual authority graph provides {inbound[path]} inbound relationship(s).",
+ action="Add another relevant contextual path from a capability, solution or sibling guide without creating generic sitewide links.",
+ source="authority-graph",
+ )
+
+ stats = {
+ "indexablePages": sum(1 for item in entries.values() if item.get("index", True) is not False),
+ "authorityNodes": len(nodes),
+ "mappedPages": len(pages),
+ "relationships": sum(len(value) for value in pages.values() if isinstance(value, list)),
+ "localizedPairs": len(pairs),
+ "localizedTargets": len(localized_targets),
+ "structuralQueueItems": len(queue),
+ }
+ return queue, stats
+
+
+def measurement_queue(report: dict[str, Any]) -> tuple[list[dict[str, Any]], dict[str, str]]:
+ queue: list[dict[str, Any]] = []
+ ga4 = report.get("ga4") if isinstance(report.get("ga4"), dict) else {}
+ search = report.get("searchConsole") if isinstance(report.get("searchConsole"), dict) else {}
+ speed = report.get("coreWebVitals") if isinstance(report.get("coreWebVitals"), dict) else {}
+
+ for item in search.get("lowCtrPages", [])[:50]:
+ if not isinstance(item, dict):
+ continue
+ impressions = float(item.get("impressions", 0) or 0)
+ ctr = float(item.get("ctr", 0) or 0)
+ position = float(item.get("position", 0) or 0)
+ score = 62 + min(25, math.log10(max(impressions, 1)) * 7) + (8 if position <= 10 else 0)
+ add(
+ queue,
+ priority=score,
+ category="low-ctr-page",
+ page=normalize_path(str(item.get("page", ""))),
+ evidence=f"{int(impressions)} impressions, {ctr * 100:.2f}% CTR, average position {position:.1f}.",
+ action="Review title, description, visible answer and query intent before changing page scope; compare the next reporting window.",
+ source="search-console",
+ )
+
+ for item in search.get("lowCtrQueries", [])[:50]:
+ if not isinstance(item, dict):
+ continue
+ impressions = float(item.get("impressions", 0) or 0)
+ ctr = float(item.get("ctr", 0) or 0)
+ position = float(item.get("position", 0) or 0)
+ score = 58 + min(22, math.log10(max(impressions, 1)) * 6) + (8 if position <= 10 else 0)
+ add(
+ queue,
+ priority=score,
+ category="low-ctr-query",
+ query=str(item.get("query", "")),
+ evidence=f"{int(impressions)} impressions, {ctr * 100:.2f}% CTR, average position {position:.1f}.",
+ action="Map the query to one existing evidence page first; improve intent alignment rather than creating a near-duplicate page.",
+ source="search-console",
+ )
+
+ for item in ga4.get("notFound", [])[:30]:
+ if not isinstance(item, dict):
+ continue
+ count = int(item.get("eventCount", 0) or 0)
+ add(
+ queue,
+ priority=92 if count >= 5 else 78,
+ category="observed-404",
+ page=normalize_path(str(item.get("pagePath", ""))),
+ evidence=f"{count} consented 404 event(s); referrer {item.get('pageReferrer', '') or 'unavailable'}.",
+ action="Repair the source link or add a precise valid route; do not redirect unrelated paths to the homepage.",
+ source="ga4",
+ )
+
+ for item in speed.get("pages", []):
+ if not isinstance(item, dict) or item.get("status") != "available":
+ continue
+ category = str(item.get("fieldCategory", "none")).lower()
+ lab_score = item.get("lab", {}).get("performanceScore") if isinstance(item.get("lab"), dict) else None
+ poor_lab = isinstance(lab_score, (int, float)) and lab_score < 0.75
+ if category in {"slow", "poor"} or poor_lab:
+ add(
+ queue,
+ priority=86 if category in {"slow", "poor"} else 70,
+ category="core-web-vitals",
+ page=normalize_path(str(item.get("url", ""))),
+ evidence=f"Field category {category}; Lighthouse performance score {lab_score if lab_score is not None else 'unavailable'}.",
+ action="Profile the exact page and remove the largest blocking or layout-shift source; confirm with the next field-data cycle.",
+ source="pagespeed-crux",
+ )
+
+ coverage = {
+ "ga4": str(ga4.get("status", "not-configured")),
+ "searchConsole": str(search.get("status", "not-configured")),
+ "coreWebVitals": str(speed.get("status", "not-run")),
+ }
+ return queue, coverage
+
+
+def deduplicate(queue: list[dict[str, Any]]) -> list[dict[str, Any]]:
+ best: dict[tuple[str, str, str], dict[str, Any]] = {}
+ for item in queue:
+ key = (str(item.get("category", "")), str(item.get("page", "")), str(item.get("query", "")))
+ if key not in best or int(item["priority"]) > int(best[key]["priority"]):
+ best[key] = item
+ return sorted(best.values(), key=lambda item: (-int(item["priority"]), str(item["category"]), str(item["page"]), str(item["query"])))
+
+
+def markdown(payload: dict[str, Any]) -> str:
+ coverage = payload["dataCoverage"]
+ lines = [
+ "# ARSAS search growth queue",
+ "",
+ f"Generated: `{payload['generatedAtUtc']}`",
+ "",
+ "## Evidence coverage",
+ "",
+ f"- GA4: **{coverage['ga4']}**",
+ f"- Search Console: **{coverage['searchConsole']}**",
+ f"- Core Web Vitals: **{coverage['coreWebVitals']}**",
+ f"- Authority graph: **available** ({payload['structure']['mappedPages']} mapped pages, {payload['structure']['relationships']} relationships)",
+ "",
+ "## Prioritized actions",
+ "",
+ "| Priority | Category | Page / query | Evidence | Action |",
+ "|---:|---|---|---|---|",
+ ]
+ for item in payload["queue"][:50]:
+ subject = item.get("page") or item.get("query") or "—"
+ values = [
+ str(item["priority"]), str(item["category"]), str(subject),
+ str(item["evidence"]), str(item["action"]),
+ ]
+ lines.append("| " + " | ".join(value.replace("|", "\\|").replace("\n", " ") for value in values) + " |")
+ if not payload["queue"]:
+ lines.append("| — | none | — | No threshold breach or structural gap detected. | Continue the weekly evidence cycle. |")
+ lines.extend([
+ "",
+ "This queue is private workflow evidence. It does not publish traffic, queries, credentials or customer/project data to the website.",
+ "",
+ ])
+ return "\n".join(lines)
+
+
+def main() -> int:
+ parser = argparse.ArgumentParser()
+ parser.add_argument("--site", default="_site")
+ parser.add_argument("--graph", default=str(DEFAULT_GRAPH))
+ parser.add_argument("--measurement-json", default="")
+ parser.add_argument("--output", default="_growth")
+ args = parser.parse_args()
+
+ site = Path(args.site).resolve()
+ graph = load(Path(args.graph).resolve())
+ entries = registry(site)
+ structural, structure_stats = structural_queue(entries, graph)
+ measurement = load(Path(args.measurement_json).resolve(), required=False) if args.measurement_json else {}
+ measured, coverage = measurement_queue(measurement)
+ queue = deduplicate([*measured, *structural])
+ payload = {
+ "schemaVersion": 1,
+ "generatedAtUtc": datetime.now(timezone.utc).isoformat(),
+ "dataCoverage": coverage,
+ "structure": structure_stats,
+ "queue": queue,
+ }
+ output = Path(args.output).resolve()
+ output.mkdir(parents=True, exist_ok=True)
+ (output / "search-growth.json").write_text(json.dumps(payload, indent=2, ensure_ascii=False) + "\n", encoding="utf-8")
+ (output / "search-growth.md").write_text(markdown(payload), encoding="utf-8")
+ print(f"ARSAS search growth queue generated: {len(queue)} prioritized item(s), output {output}.")
+ return 0
+
+
+if __name__ == "__main__":
+ raise SystemExit(main())
diff --git a/scripts/validate-search-authority.py b/scripts/validate-search-authority.py
new file mode 100644
index 000000000..40b6d89b6
--- /dev/null
+++ b/scripts/validate-search-authority.py
@@ -0,0 +1,138 @@
+#!/usr/bin/env python3
+"""Validate rendered ARSAS search-authority relationships and build metadata."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import re
+import sys
+from collections import defaultdict
+from pathlib import Path
+from typing import Any
+
+ROOT = Path(__file__).resolve().parents[1]
+DEFAULT_GRAPH = ROOT / "landing" / "search-authority.json"
+SECTION_PATTERN = re.compile(
+ r']*>(.*?)',
+ re.IGNORECASE | re.DOTALL,
+)
+HREF_PATTERN = re.compile(r'href="([^"]+)"', re.IGNORECASE)
+
+
+def load(path: Path, label: str) -> dict[str, Any]:
+ try:
+ value = json.loads(path.read_text(encoding="utf-8"))
+ except (OSError, json.JSONDecodeError) as exc:
+ raise SystemExit(f"Cannot read {label}: {exc}") from exc
+ if not isinstance(value, dict):
+ raise SystemExit(f"{label} must contain an object")
+ return value
+
+
+def main() -> int:
+ parser = argparse.ArgumentParser()
+ parser.add_argument("site", nargs="?", default="_site")
+ parser.add_argument("--graph", default=str(DEFAULT_GRAPH))
+ args = parser.parse_args()
+
+ site = Path(args.site).resolve()
+ graph = load(Path(args.graph).resolve(), "search authority graph")
+ config = load(site / "site.json", "built site registry")
+ build_info = load(site / "build-info.json", "build-info.json")
+ errors: list[str] = []
+
+ nodes = graph.get("nodes")
+ pages = graph.get("pages")
+ clusters = graph.get("clusters")
+ if not isinstance(nodes, dict) or not isinstance(pages, dict) or not isinstance(clusters, dict):
+ raise SystemExit("Search authority graph is incomplete")
+
+ registry: dict[str, dict[str, Any]] = {}
+ for item in config.get("pages", []):
+ if isinstance(item, dict) and isinstance(item.get("path"), str):
+ registry[item["path"] or "index.html"] = item
+
+ inbound: dict[str, int] = defaultdict(int)
+ relationship_count = 0
+ for source, raw_targets in pages.items():
+ targets = raw_targets if isinstance(raw_targets, list) else []
+ relationship_count += len(targets)
+ page = site / source
+ if not page.is_file():
+ errors.append(f"mapped page is missing: {source}")
+ continue
+ html = page.read_text(encoding="utf-8")
+ matches = SECTION_PATTERN.findall(html)
+ if len(matches) != 1:
+ errors.append(f"{source}: expected one rendered authority section, found {len(matches)}")
+ continue
+ cluster, body = matches[0]
+ node = nodes.get(source)
+ expected_cluster = node.get("cluster") if isinstance(node, dict) else None
+ if cluster != expected_cluster:
+ errors.append(f"{source}: authority cluster is {cluster!r}, expected {expected_cluster!r}")
+ hrefs = HREF_PATTERN.findall(body)
+ for target in targets:
+ if hrefs.count(target) != 1:
+ errors.append(f"{source}: expected exactly one contextual link to {target}")
+ inbound[target] += 1
+ unexpected = sorted(set(hrefs) - set(targets))
+ if unexpected:
+ errors.append(f"{source}: authority section has unexpected links {unexpected}")
+ if html.find('data-search-authority=') > html.find('data-section="download-cta"') >= 0:
+ errors.append(f"{source}: authority section must appear before the download CTA")
+ if 'aria-labelledby="related-evidence-heading"' not in html:
+ errors.append(f"{source}: authority section is missing accessible heading linkage")
+
+ for path in registry:
+ if path in pages:
+ continue
+ page = site / path
+ if page.is_file() and "data-search-authority=" in page.read_text(encoding="utf-8"):
+ errors.append(f"{path}: unexpected authority section on unmapped page")
+
+ guide_paths = {
+ path for path, item in registry.items()
+ if item.get("contentType") == "guide"
+ }
+ for guide in sorted(guide_paths):
+ if inbound[guide] < 1:
+ errors.append(f"{guide}: no rendered inbound authority relationship")
+
+ authority = build_info.get("searchAuthority")
+ expected_stats = {
+ "schemaVersion": 1,
+ "clusterCount": len(clusters),
+ "nodeCount": len(nodes),
+ "mappedPageCount": len(pages),
+ "guideCount": len(guide_paths),
+ "guidesWithDiscoveryInbound": len(guide_paths),
+ "relationshipCount": relationship_count,
+ "allGuidesHaveCapabilityOrSolutionInbound": True,
+ }
+ if not isinstance(authority, dict):
+ errors.append("build-info.json is missing searchAuthority metadata")
+ else:
+ for key, value in expected_stats.items():
+ if authority.get(key) != value:
+ errors.append(f"build-info.json searchAuthority.{key} must be {value!r}")
+
+ if (site / "search-authority.json").exists():
+ errors.append("source search-authority.json must not be deployed publicly")
+
+ errors = list(dict.fromkeys(errors))
+ if errors:
+ print("ARSAS search authority validation failed:", file=sys.stderr)
+ for error in errors:
+ print(f"- {error}", file=sys.stderr)
+ return 1
+ print(
+ f"ARSAS search authority validation passed: {len(pages)} mapped pages, "
+ f"{relationship_count} contextual links, {len(guide_paths)} guides with inbound paths."
+ )
+ return 0
+
+
+if __name__ == "__main__":
+ raise SystemExit(main())
diff --git a/scripts/verify-pages-deployment.py b/scripts/verify-pages-deployment.py
index 4b2adaef2..9b8dc397a 100644
--- a/scripts/verify-pages-deployment.py
+++ b/scripts/verify-pages-deployment.py
@@ -33,7 +33,7 @@ def fetch(url: str, timeout: int = 20) -> tuple[int, str, dict[str, str]]:
request = Request(
url,
headers={
- "User-Agent": "ARSAS-Pages-Attestation/1.0",
+ "User-Agent": "ARSAS-Pages-Attestation/1.1",
"Accept": "application/json,text/html;q=0.9,*/*;q=0.8",
"Cache-Control": "no-cache",
"Pragma": "no-cache",
@@ -116,6 +116,19 @@ def check_once(base_url: str, source_commit: str, stable_version: str, measureme
if privacy.get("analyticsClientLoadedOnPolicyPages") is not False:
errors.append("public privacy metadata must prohibit analytics client loading on policy pages")
+ authority = info.get("searchAuthority")
+ if not isinstance(authority, dict):
+ errors.append("public build is missing search-authority metadata")
+ else:
+ if authority.get("schemaVersion") != 1:
+ errors.append("public search-authority schema is invalid")
+ if authority.get("allGuidesHaveCapabilityOrSolutionInbound") is not True:
+ errors.append("public search-authority metadata does not prove guide discovery coverage")
+ if int(authority.get("mappedPageCount", 0) or 0) < 20:
+ errors.append("public search-authority graph maps too few pages")
+ if int(authority.get("relationshipCount", 0) or 0) < 60:
+ errors.append("public search-authority graph has too few contextual relationships")
+
for relative in ("privacy.html", "privasi.html"):
url = urljoin(base_url, relative) + "?" + urlencode({"attest": source_commit, "n": nonce})
page_status, page, _ = fetch(url)
@@ -149,12 +162,37 @@ def check_once(base_url: str, source_commit: str, stable_version: str, measureme
if 'src="analytics.js"' in home or "googletagmanager.com" in home:
errors.append("homepage must not load analytics before consent")
+ authority_url = urljoin(base_url, "smart-reporting.html") + "?" + urlencode({"attest": source_commit, "n": nonce})
+ authority_status, authority_page, _ = fetch(authority_url)
+ evidence["authorityProbeStatus"] = authority_status
+ if authority_status != 200:
+ errors.append(f"smart-reporting.html returned HTTP {authority_status}")
+ else:
+ for required in (
+ 'data-search-authority="reporting"',
+ 'href="reporting-silent.html"',
+ 'href="brcb-vs-urcb.html"',
+ 'href="rcb-reserved.html"',
+ 'href="empty-dataset.html"',
+ 'data-section="download-cta"',
+ ):
+ if required not in authority_page:
+ errors.append(f"smart-reporting.html is missing contextual authority evidence {required}")
+ if authority_page.find('data-search-authority="reporting"') > authority_page.find('data-section="download-cta"') >= 0:
+ errors.append("smart-reporting.html authority section appears after the download CTA")
+
+ graph_url = urljoin(base_url, "search-authority.json") + "?" + urlencode({"attest": source_commit, "n": nonce})
+ graph_status, _, _ = fetch(graph_url)
+ evidence["privateAuthoritySourceStatus"] = graph_status
+ if graph_status != 404:
+ errors.append(f"search-authority.json must not be publicly deployed; observed HTTP {graph_status}")
+
return errors, evidence
def write_report(path: Path, success: bool, attempts: int, evidence: dict[str, object], errors: list[str]) -> None:
payload = {
- "schemaVersion": 2,
+ "schemaVersion": 3,
"verifiedAtUtc": datetime.now(timezone.utc).isoformat(),
"success": success,
"attempts": attempts,