Add dataset-header validation & auto-fixing tool

.gitignore

@@ -8,3 +8,6 @@
 /.idea
 *.iml
 validation_report/
+
+__pycache__/
+*.pyc

buildScripts/dataset_header/README.md

@@ -0,0 +1,61 @@
+# Dataset-header validation & fixing
+
+A reusable, branch/version-agnostic tool that checks (and fixes) the DCAT
+dataset headers of ReliCapGrid instance files against the rules from Svein's
+PR #322 review. The **validator** and the **fixer** share one rule set
+(`rules.py`) so they can't drift apart.
+
+## Two tiers
+
+- **Tier A — structural / fixed-value.** Deterministic and auto-fixable:
+  required properties, fixed values (`accessRights`, `type`, `license`,
+  `rights`, `rightsHolder`), `xml:lang` tags, forbidden properties,
+  disallowed namespaces (`eumd`/`euvoc`), UUID and UTC-`Z` datetime formats,
+  and `TSO-`/`RCC-` publisher naming.
+- **Tier B — reference-data membership.** `conformsTo` / `isVersionOf` /
+  `spatial` / `wasGeneratedBy` / publisher-party validated against the
+  schemes in `Instance/referenceData`. **Report-only** — fixing is blocked
+  until those schemes are extended (the parked PR #322 items). New scheme
+  entries become valid automatically, no code change.
+
+## Usage
+
+```bash
+# Validate (report-only; exit 0). Add --strict to fail on Tier-A.
+uv run python buildScripts/validate_dataset_header.py --scope networkcode \
+    --details --json validation_report/dataset_header.json
+
+# Auto-fix the deterministic Tier-A subset (dry-run without --apply).
+uv run python buildScripts/fix_dataset_header.py --scope networkcode
+uv run python buildScripts/fix_dataset_header.py --scope networkcode --apply
+```
+
+`--scope` is `networkcode` (default) or `all`. The fixer is idempotent and
+formatting-preserving (text edits, not graph re-serialization), so diffs stay
+minimal. It never invents Tier-B values, real dates, titles, or spatial refs.
+
+## CI / pytest
+
+`tests/test_dataset_header.py` runs the validator over the in-scope files.
+It is **report-only** (`xfail`) until a branch's headers are clean; once you
+have run the fixer and resolved the manual gaps, flip the fixable-Tier-A check
+to a hard failure so regressions (e.g. re-introducing `eumd` or a forbidden
+property) fail the build. Fast unit tests cover the rule and fixer logic.
+
+## Layout
+
+| File | Role |
+|---|---|
+| `rules.py` | single source of truth: rule catalog + rdflib checks |
+| `schemes.py` | loads referenceData schemes into valid-value sets (Tier B) |
+| `discovery.py` | enumerates in-scope header files |
+| `fixer.py` | formatting-preserving Tier-A auto-fixes |
+| `report.py` | text / markdown / JSON reporting |
+| `../validate_dataset_header.py` | validator CLI |
+| `../fix_dataset_header.py` | fixer CLI |
+
+## Roadmap
+
+- Extend scope beyond NetworkCode cimxml (Grid, GridSituation, boundary).
+- Emit DCAT-AP-CIM SHACL shapes and run `pyshacl` as an independent
+  cross-check of the Python rules (`pyshacl` is already a dependency).

buildScripts/dataset_header/__init__.py

@@ -0,0 +1,15 @@
+"""Dataset-header validation and fixing for ReliCapGrid instance files.
+
+Single source of truth for the DCAT dataset-header rules (Svein's review of
+PR #322), consumed by two modes:
+
+- ``validate_dataset_header.py`` — semantic validation via rdflib (report).
+- ``fix_dataset_header.py``      — formatting-preserving auto-fix of the
+  deterministic (Tier-A) subset.
+
+Rules are split into two tiers:
+
+- **Tier A** — structural / fixed-value: deterministic, auto-fixable now.
+- **Tier B** — reference-data membership: checkable now, but fixing is blocked
+  until the reference schemes are extended (see the parked PR #322 items).
+"""

buildScripts/dataset_header/discovery.py

@@ -0,0 +1,29 @@
+"""Discover instance files that carry a dcat:Dataset header.
+
+Phase-1 scope is the NetworkCode cimxml surface (the PR #322 files). The
+existing ``create_cgm_zip.collect_cgm_files`` helper is too narrow (it skips
+the ``Dataset_version_dependency/`` subfolder and the Jotunheim NetworkCode
+IAM/SAR files), so we glob directly here. Scope can be widened later.
+"""
+from __future__ import annotations
+
+from pathlib import Path
+
+# Directories that are not instance datasets.
+_EXCLUDE_PARTS = {"referenceData"}
+
+
+def discover_headers(instance_dir: Path, scope: str = "networkcode") -> list[Path]:
+    """Return sorted instance XML files in scope that may contain a header."""
+    if scope == "networkcode":
+        pattern = "*/NetworkCode/**/*.xml"
+    elif scope == "all":
+        pattern = "**/*.xml"
+    else:
+        raise ValueError(f"unknown scope: {scope!r}")
+
+    files = [
+        p for p in instance_dir.glob(pattern)
+        if not (_EXCLUDE_PARTS & set(p.parts))
+    ]
+    return sorted(set(files))

buildScripts/dataset_header/fixer.py

@@ -0,0 +1,187 @@
+"""Formatting-preserving auto-fixer for the Tier-A header rules.
+
+Operates on the raw file text (not a re-serialized graph) so diffs stay
+minimal and reviewable. Only the deterministic Tier-A subset is fixed;
+Tier-B (reference-data) values are never invented here.
+
+Each ``fix_text`` call is idempotent: running it again yields no change.
+"""
+from __future__ import annotations
+
+import re
+
+# Namespace prefixes that must be dropped entirely.
+_DISALLOWED_PREFIXES = ("eumd", "euvoc")
+
+# Canonical publisher party names (bare -> prefixed). Extend as needed.
+KNOWN_PARTY = {
+    "Belgovia": "TSO-Belgovia",
+    "Espheim": "TSO-Espheim",
+    "Galia": "TSO-Galia",
+    "Svedala": "TSO-Svedala",
+    "Jotunheim": "RCC-Jotunheim",
+    "JOTUNHEIM": "RCC-Jotunheim",
+}
+
+# Fixed-value properties: label -> full element line body (value-forcing).
+_FIXED_URI = {
+    "dcterms:accessRights": "https://energy.referencedata.eu/Confidentiality/Public",
+    "dcterms:type": "https://energy.referencedata.eu/type/CIM-PowerSystemModel",
+    "dcterms:license": "https://creativecommons.org/licenses/by/4.0/",
+}
+_FIXED_LITERAL = {
+    "dcterms:rights": "Copyright",
+    "dcterms:rightsHolder": "ENTSO-E",
+}
+
+_FORBIDDEN = (
+    "dcatcim:alternativeVersionOf", "dcterms:accrualPeriodicity",
+    "dcterms:hasPart", "dcat:temporalResolution",
+    "dcatcim:preferredVersionOf", "dcat:inSeries",
+)
+
+_DATASET_OPEN = re.compile(r"<dcat:Dataset\b")
+
+
+def _indent_of(block: str) -> str:
+    m = re.search(r"\n([ \t]+)<", block)
+    return m.group(1) if m else "    "
+
+
+def fix_text(text: str) -> tuple[str, list[str]]:
+    """Return (new_text, list_of_applied_fix_ids)."""
+    applied: list[str] = []
+
+    # repair a doubled closing quote in an attribute value (well-formedness),
+    # e.g. rdf:resource=".../2.4""/> -> .../2.4"/>
+    dq = re.sub(r'(="[^"]*)""', r'\1"', text)
+    if dq != text:
+        text = dq
+        applied.append("fix-doubled-quote")
+
+    open_m = _DATASET_OPEN.search(text)
+    close_m = re.search(r"\n[ \t]*</dcat:Dataset>", text)
+    if not open_m or not close_m:
+        return text, applied  # not a header file
+
+    prolog, ds_start = text[:open_m.start()], open_m.start()
+    block = text[ds_start:close_m.start()]  # up to the newline before </dcat:Dataset>
+    tail = text[close_m.start():]           # newline + indent + close tag + rest
+
+    # --- drop disallowed prefixes (declaration AND any usage) --------------
+    for pfx in _DISALLOWED_PREFIXES:
+        p = pfx_esc(pfx)
+        new = re.sub(rf'\n[ \t]*xmlns:{p}="[^"]*"', "", prolog)
+        new = re.sub(rf'[ \t]*xmlns:{p}="[^"]*"', "", new)  # inline
+        if new != prolog:
+            prolog = new
+            applied.append(f"drop-ns:{pfx}")
+        # remove usage elements of the dropped prefix so it stays well-formed
+        for scope_name in ("block", "tail"):
+            src = block if scope_name == "block" else tail
+            cleaned = re.sub(rf'\n[ \t]*<{p}:[^>]*>.*?</{p}:[^>]*>', "", src)
+            cleaned = re.sub(rf'\n[ \t]*<{p}:[^>]*/>', "", cleaned)
+            if cleaned != src:
+                if scope_name == "block":
+                    block = cleaned
+                else:
+                    tail = cleaned
+                applied.append(f"remove-usage:{pfx}")
+
+    # --- prolog: fix dcterms '#' -------------------------------------------
+    if 'xmlns:dcterms="http://purl.org/dc/terms/#"' in prolog:
+        prolog = prolog.replace(
+            'xmlns:dcterms="http://purl.org/dc/terms/#"',
+            'xmlns:dcterms="http://purl.org/dc/terms/"')
+        applied.append("fix-dcterms-ns")
+
+    # --- block: remove forbidden properties --------------------------------
+    for label in _FORBIDDEN:
+        new = re.sub(rf'\n[ \t]*<{re.escape(label)}\b[^>]*/>', "", block)
+        new = re.sub(rf'\n[ \t]*<{re.escape(label)}\b[^>]*>.*?</{re.escape(label)}>',
+                     "", new)
+        if new != block:
+            block = new
+            applied.append(f"remove:{label}")
+
+    # --- block: force fixed values -----------------------------------------
+    for label, uri in _FIXED_URI.items():
+        pat = re.compile(rf'(<{re.escape(label)} rdf:resource=")[^"]*(")')
+        if pat.search(block) and f'"{uri}"' not in block:
+            block = pat.sub(rf'\g<1>{uri}\g<2>', block)
+            applied.append(f"force-value:{label}")
+    for label, lit in _FIXED_LITERAL.items():
+        pat = re.compile(rf'(<{re.escape(label)}>)[^<]*(</{re.escape(label)}>)')
+        m = pat.search(block)
+        if m and m.group(0) != f'<{label}>{lit}</{label}>':
+            block = pat.sub(rf'\g<1>{lit}\g<2>', block)
+            applied.append(f"force-value:{label}")
+
+    # --- block: add xml:lang to description / versionNotes -----------------
+    for label in ("dcterms:description", "adms:versionNotes"):
+        pat = re.compile(rf'<{re.escape(label)}>')
+        if pat.search(block):
+            block = pat.sub(f'<{label} xml:lang="en">', block)
+            applied.append(f"add-lang:{label}")
+
+    # --- block: normalize publisher party naming ---------------------------
+    def _pub(m):
+        name = m.group(2)
+        canon = KNOWN_PARTY.get(name)
+        return f'{m.group(1)}test/party/{canon}"' if canon else m.group(0)
+
+    new = re.sub(
+        r'(rdf:resource="https://energy\.referencedata\.eu/)[Tt]est/[Pp]arty/([^"/]+)"',
+        _pub, block)
+    if new != block:
+        block = new
+        applied.append("normalize-publisher")
+
+    # --- block: insert missing fixed-value props + versionNotes ------------
+    indent = _indent_of(block)
+    inserts: list[str] = []
+    for label, uri in _FIXED_URI.items():
+        if f"<{label}" not in block:
+            inserts.append(f'{indent}<{label} rdf:resource="{uri}"/>')
+    for label, lit in _FIXED_LITERAL.items():
+        if f"<{label}" not in block:
+            inserts.append(f'{indent}<{label}>{lit}</{label}>')
+    if "<adms:versionNotes" not in block:
+        inserts.append(f'{indent}<adms:versionNotes xml:lang="en">Initial version.</adms:versionNotes>')
+        applied.append("insert:adms:versionNotes(placeholder)")
+        if "xmlns:adms" not in prolog:
+            prolog = _add_adms_ns(prolog)
+            applied.append("add-ns:adms")
+    if inserts:
+        block = block.rstrip("\n") + "\n" + "\n".join(inserts)
+        for ins in inserts:
+            label = re.search(r"<([\w:]+)", ins).group(1)
+            if not label == "adms:versionNotes":
+                applied.append(f"insert:{label}")
+
+    text = prolog + block + tail
+
+    # --- whole file: drop dcatcim ns if now unused -------------------------
+    if "dcatcim:" not in re.sub(r'xmlns:dcatcim="[^"]*"', "", text):
+        new = re.sub(r'\n[ \t]*xmlns:dcatcim="[^"]*"', "", text)
+        new = re.sub(r'[ \t]*xmlns:dcatcim="[^"]*"', "", new)
+        if new != text:
+            text = new
+            applied.append("drop-ns:dcatcim")
+
+    return text, applied
+
+
+def _add_adms_ns(prolog: str) -> str:
+    """Insert an xmlns:adms declaration into the rdf:RDF open tag."""
+    decl = 'xmlns:adms="http://www.w3.org/ns/adms#"'
+    m = re.search(r"(<rdf:RDF\b)([ \t]*\n)", prolog)
+    if m:  # multi-line form: add as first prefix line
+        indent_m = re.search(r"\n([ \t]+)xmlns:", prolog)
+        indent = indent_m.group(1) if indent_m else "    "
+        return prolog.replace(m.group(0), f"{m.group(1)}\n{indent}{decl}\n", 1)
+    return prolog.replace("<rdf:RDF", f"<rdf:RDF {decl}", 1)
+
+
+def pfx_esc(pfx: str) -> str:
+    return re.escape(pfx)

buildScripts/dataset_header/report.py

@@ -0,0 +1,58 @@
+"""Reporting helpers shared by the validator CLI and the pytest integration."""
+from __future__ import annotations
+
+from collections import Counter
+from pathlib import Path
+
+from .rules import Violation
+
+
+def to_dicts(violations: list[Violation]) -> list[dict]:
+    return [
+        {
+            "file": v.file,
+            "rule_id": v.rule_id,
+            "tier": v.tier,
+            "review_ref": v.review_ref,
+            "message": v.message,
+            "fixable": v.fixable,
+            "detail": v.detail,
+        }
+        for v in violations
+    ]
+
+
+def summarize(violations: list[Violation], files_scanned: int, headers: int) -> str:
+    tier_a = [v for v in violations if v.tier == "A"]
+    tier_b = [v for v in violations if v.tier == "B"]
+    fixable = [v for v in tier_a if v.fixable]
+    lines = [
+        f"Scanned {files_scanned} files ({headers} with a dcat:Dataset header).",
+        f"Tier-A violations: {len(tier_a)} ({len(fixable)} auto-fixable)",
+        f"Tier-B violations: {len(tier_b)} (blocked on reference-scheme extensions)",
+    ]
+    return "\n".join(lines)
+
+
+def by_rule_table(violations: list[Violation]) -> str:
+    """Markdown table of violation counts per rule (for CI/PR comments)."""
+    counts = Counter((v.tier, v.rule_id, v.review_ref) for v in violations)
+    if not counts:
+        return "_No violations._"
+    rows = ["| Tier | Rule | Review | Count |", "|---|---|---|---|"]
+    for (tier, rule, ref), n in sorted(counts.items(), key=lambda x: (-x[1], x[0])):
+        rows.append(f"| {tier} | {rule} | {ref} | {n} |")
+    return "\n".join(rows)
+
+
+def details_text(violations: list[Violation], root: Path | None = None) -> str:
+    lines = []
+    for v in sorted(violations, key=lambda x: (x.tier, x.file, x.rule_id)):
+        f = v.file
+        if root is not None:
+            try:
+                f = str(Path(v.file).relative_to(root))
+            except ValueError:
+                pass
+        lines.append(f"{f}: {v}")
+    return "\n".join(lines)