#!/usr/bin/env python3 """Normalize the independent coverage indexes into stable cross-index records.""" from __future__ import annotations import argparse import hashlib import json from collections import defaultdict from pathlib import Path, PurePosixPath from typing import Iterable INDEXES = ( ("source", Path("coverage/source-index.jsonl")), ("document", Path("coverage/document-capabilities.jsonl")), ("symbol", Path("coverage/source-symbols.jsonl")), ("resource", Path("coverage/config-resources.jsonl")), ("test", Path("coverage/test-entries.jsonl")), ) def sha256(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as handle: for block in iter(lambda: handle.read(1024 * 1024), b""): digest.update(block) return digest.hexdigest() def read_jsonl(path: Path) -> list[dict[str, object]]: rows: list[dict[str, object]] = [] with path.open("r", encoding="utf-8") as handle: for line_number, line in enumerate(handle, 1): if not line.strip(): continue value = json.loads(line) if not isinstance(value, dict): raise ValueError(f"{path}:{line_number}: expected object") rows.append(value) return rows def canonical_path(value: object) -> str | None: if not isinstance(value, str) or not value.strip(): return None raw = value.replace("\\", "/").strip() if raw.startswith("/"): return None parts: list[str] = [] for part in PurePosixPath(raw).parts: if part in ("", "."): continue if part == "..": if not parts: return None parts.pop() continue parts.append(part) return "/".join(parts) or None def as_text(value: object) -> str: return str(value) if value is not None else "" def source_key(index_name: str, row: dict[str, object], path: str | None) -> str: if index_name == "source": return f"src:file:{path or as_text(row.get('source_id'))}" if index_name == "document": return f"src:doc:{path or 'unknown'}#{as_text(row.get('anchor')) or 'root'}" if index_name == "symbol": return "src:symbol:{}:{}:{}:{}".format( path or "unknown", as_text(row.get("line")), as_text(row.get("symbol_kind")), as_text(row.get("symbol")), ) if index_name == "resource": entrypoint = canonical_path(row.get("entrypoint")) or "unknown" return f"src:resource:{entrypoint}:{as_text(row.get('line'))}:{as_text(row.get('edge_kind'))}:{path or as_text(row.get('raw_value'))}" return f"src:test:{path or as_text(row.get('test_source_id'))}" def row_path(index_name: str, row: dict[str, object]) -> str | None: if index_name == "document": return canonical_path(row.get("document")) if index_name == "resource": return canonical_path(row.get("path")) return canonical_path(row.get("path")) def hash_path(index_name: str, row: dict[str, object]) -> str | None: if index_name == "resource": return canonical_path(row.get("origin_path")) or canonical_path(row.get("entrypoint")) return row_path(index_name, row) def original_id(index_name: str, row: dict[str, object]) -> str: keys = { "source": "source_id", "document": "capability_source_id", "symbol": "symbol_source_id", "resource": "resource_source_id", "test": "test_source_id", } return as_text(row.get(keys[index_name])) def normalize_row(index_name: str, row: dict[str, object], root: Path, path_to_file_id: dict[str, str]) -> dict[str, object]: path = row_path(index_name, row) source_id = source_key(index_name, row, path) source_hash = row.get("source_sha", row.get("sha256")) if source_hash is not None: source_hash = as_text(source_hash) related: set[str] = set() if path in path_to_file_id: related.add(path_to_file_id[path]) if index_name == "resource": origin = canonical_path(row.get("origin_path")) target = canonical_path(row.get("path")) if origin in path_to_file_id: related.add(path_to_file_id[origin]) if target in path_to_file_id: related.add(path_to_file_id[target]) elif index_name == "test": for ref in row.get("config_refs", []) if isinstance(row.get("config_refs"), list) else []: ref_path = canonical_path(ref) if ref_path in path_to_file_id: related.add(path_to_file_id[ref_path]) record: dict[str, object] = { "source_id": source_id, "origin_index": index_name, "origin_source_id": original_id(index_name, row), "path": path, "hash_path": hash_path(index_name, row), "source_sha": source_hash, "domain": as_text(row.get("domain")) or ({"source": "source", "document": "documentation", "symbol": "code", "resource": "configuration", "test": "test"}[index_name]), "kind": as_text(row.get("kind")) or as_text(row.get("symbol_kind")) or as_text(row.get("resource_kind")) or as_text(row.get("test_kind")) or index_name, "line": row.get("line"), "parse_status": as_text(row.get("parse_status")) or as_text(row.get("mapping_status")) or "parsed", "related_source_ids": sorted(related), } if index_name == "resource": record.update({ "edge_kind": row.get("edge_kind"), "raw_value": row.get("raw_value"), "resource_kind": row.get("resource_kind"), "target_status": row.get("parse_status"), }) elif index_name == "test": record.update({ "command": row.get("command"), "inputs": row.get("inputs", []), "expected": row.get("expected", []), "config_refs": row.get("config_refs", []), "test_kind": row.get("test_kind"), }) return record def generate(root: Path, output: Path, report_path: Path) -> tuple[int, str]: loaded: dict[str, list[dict[str, object]]] = {} input_summary: dict[str, dict[str, object]] = {} for index_name, relative_path in INDEXES: path = root / relative_path if not path.is_file(): raise FileNotFoundError(f"required index is missing: {relative_path}") loaded[index_name] = read_jsonl(path) input_summary[index_name] = { "path": relative_path.as_posix(), "rows": len(loaded[index_name]), "sha256": sha256(path), } path_to_file_id: dict[str, str] = {} for row in loaded["source"]: path = canonical_path(row.get("path")) if path: path_to_file_id.setdefault(path, f"src:file:{path}") records: list[dict[str, object]] = [] source_ids: set[str] = set() source_id_counts: dict[str, int] = defaultdict(int) collisions: list[dict[str, object]] = [] path_hashes: dict[str, set[str]] = defaultdict(set) missing_paths: set[str] = set() for index_name, _ in INDEXES: for row in loaded[index_name]: record = normalize_row(index_name, row, root, path_to_file_id) base_source_id = str(record["source_id"]) source_id_counts[base_source_id] += 1 if base_source_id in source_ids: collisions.append({"source_id": base_source_id, "occurrence": source_id_counts[base_source_id], "origin_index": index_name, "origin_source_id": record["origin_source_id"]}) record["source_id"] = f"{base_source_id}~dup{source_id_counts[base_source_id]}" source_ids.add(str(record["source_id"])) hash_value = record.get("source_sha") hash_path_value = record.get("hash_path") if hash_value and hash_path_value: path_hashes[str(hash_path_value)].add(str(hash_value)) primary_path = record.get("path") if primary_path and not (root / str(primary_path)).exists(): if not (index_name == "resource" and as_text(row.get("parse_status")) in {"missing", "unresolved", "host_or_variable_path", "outside_repo"}): missing_paths.add(str(primary_path)) records.append(record) records.sort(key=lambda row: str(row["source_id"])) output.parent.mkdir(parents=True, exist_ok=True) with output.open("w", encoding="utf-8", newline="\n") as handle: for record in records: handle.write(json.dumps(record, ensure_ascii=False, sort_keys=True, separators=(",", ":")) + "\n") digest = sha256(output) report = { "schema": "normalized-sources-report.v1", "records": len(records), "unique_source_ids": len(source_ids), "input_summary": input_summary, "collisions": collisions, "hash_conflicts": {path: sorted(values) for path, values in sorted(path_hashes.items()) if len(values) > 1}, "missing_paths": sorted(missing_paths), "orphan_related_ids": sum(1 for row in records if not row["related_source_ids"] and row["origin_index"] != "source"), "output_sha256": digest, } report_path.parent.mkdir(parents=True, exist_ok=True) report_path.write_text(json.dumps(report, ensure_ascii=False, sort_keys=True, indent=2) + "\n", encoding="utf-8") return len(records), digest def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--repo-root", type=Path, default=Path(__file__).resolve().parents[1]) parser.add_argument("--output", type=Path, default=Path("coverage/normalized-sources.jsonl")) parser.add_argument("--report", type=Path, default=Path("coverage/normalized-sources.report.json")) args = parser.parse_args() root = args.repo_root.resolve() count, digest = generate(root, root / args.output, root / args.report) print(f"normalized_sources={count} sha256={digest}") return 0 if __name__ == "__main__": raise SystemExit(main())