modify documents

scripts/validate_reference_artifacts.py

@@ -5,6 +5,7 @@ from __future__ import annotations
 
 import hashlib
 import json
+import re
 import sys
 from pathlib import Path
 
@@ -12,6 +13,7 @@ from pathlib import Path
 SCHEMA_VERSION = "abaqus-user-subroutine-artifact-v1"
 VALID_STATUSES = {"draft", "needs-reference-artifacts", "ready-for-comparison", "blocked"}
 READY_STATUS = "ready-for-comparison"
+SHA256_RE = re.compile(r"^[0-9a-fA-F]{64}$")
 
 
 def sha256_file(path: Path) -> str:
@@ -64,6 +66,70 @@ def _require_ready_key(path: Path, payload: dict, *keys: str) -> list[str]:
     return []
 
 
+def _is_safe_relative_path(path_text: str) -> bool:
+    candidate = Path(path_text)
+    return not candidate.is_absolute() and ".." not in candidate.parts
+
+
+def _is_extracted_csv_path(path_text: str) -> bool:
+    candidate = Path(path_text)
+    return (
+        _is_safe_relative_path(path_text)
+        and len(candidate.parts) == 2
+        and candidate.parts[0] == "extracted"
+        and candidate.suffix.lower() == ".csv"
+    )
+
+
+def _validate_optional_sha256_file(path: Path, model_dir: Path, key: str, value: object) -> list[str]:
+    if value is None:
+        return []
+    if not isinstance(value, str) or not value:
+        return [f"{path}: invalid {key}"]
+    if not _is_safe_relative_path(value):
+        return [f"{path}: {key} must be a relative path inside the artifact bundle"]
+
+    sha_path = model_dir / value
+    if not sha_path.exists():
+        return [f"{path}: missing {key}: {value}"]
+
+    first_token = sha_path.read_text(encoding="utf-8").strip().split(maxsplit=1)[0]
+    if not SHA256_RE.match(first_token):
+        return [f"{path}: invalid {key}: {value}"]
+    return []
+
+
+def _validate_extraction(path: Path, model_dir: Path, payload: dict) -> list[str]:
+    errors: list[str] = []
+    extraction = payload.get("extraction")
+    if not isinstance(extraction, dict):
+        return [f"{path}: extraction provenance must be an object"]
+
+    for key in ("source_odb", "tool", "extracted_at", "csv_directory"):
+        if not extraction.get(key):
+            errors.append(f"{path}: missing extraction provenance key {key}")
+
+    csv_directory = extraction.get("csv_directory")
+    if isinstance(csv_directory, str) and csv_directory != "extracted":
+        errors.append(f"{path}: extraction.csv_directory must be extracted")
+
+    script = extraction.get("script")
+    if script is not None:
+        if not isinstance(script, str) or not script:
+            errors.append(f"{path}: invalid extraction script")
+        elif not _is_safe_relative_path(script):
+            errors.append(f"{path}: extraction script must be a relative path inside the artifact bundle")
+        elif not (model_dir / script).exists():
+            errors.append(f"{path}: missing extraction script: {script}")
+
+    odb_sha256 = extraction.get("odb_sha256")
+    if odb_sha256 is not None and (not isinstance(odb_sha256, str) or not SHA256_RE.match(odb_sha256)):
+        errors.append(f"{path}: invalid odb_sha256")
+
+    errors.extend(_validate_optional_sha256_file(path, model_dir, "odb_sha256_file", extraction.get("odb_sha256_file")))
+    return errors
+
+
 def _validate_ready_files(path: Path, root: Path, payload: dict) -> list[str]:
     errors: list[str] = []
     model_dir = path.parent
@@ -71,7 +137,6 @@ def _validate_ready_files(path: Path, root: Path, payload: dict) -> list[str]:
     for keys in (
         ("abaqus", "version"),
         ("abaqus", "precision"),
-        ("abaqus", "command"),
         ("compiler", "vendor"),
         ("compiler", "name"),
         ("compiler", "version"),
@@ -80,6 +145,7 @@ def _validate_ready_files(path: Path, root: Path, payload: dict) -> list[str]:
         ("input_file",),
         ("outputs", "tails"),
         ("outputs", "csv"),
+        ("extraction",),
     ):
         errors.extend(_require_ready_key(path, payload, *keys))
 
@@ -89,7 +155,7 @@ def _validate_ready_files(path: Path, root: Path, payload: dict) -> list[str]:
 
     tails = _nested(payload, "outputs", "tails")
     if isinstance(tails, dict):
-        for key in ("msg", "dat", "log"):
+        for key in ("msg", "dat", "log", "sta"):
             tail_path = tails.get(key)
             if not isinstance(tail_path, str) or not tail_path:
                 errors.append(f"{path}: missing output tail {key}")
@@ -103,9 +169,14 @@ def _validate_ready_files(path: Path, root: Path, payload: dict) -> list[str]:
         for key, csv_path in csv_outputs.items():
             if not isinstance(csv_path, str) or not csv_path:
                 errors.append(f"{path}: missing csv output {key}")
+            elif not _is_extracted_csv_path(csv_path):
+                errors.append(f"{path}: csv output {key} must match extracted/*.csv")
             elif not (model_dir / csv_path).exists():
                 errors.append(f"{path}: missing csv output {key}: {csv_path}")
 
+    if "extraction" in payload:
+        errors.extend(_validate_extraction(path, model_dir, payload))
+
     source_files = _nested(payload, "subroutine", "source_files")
     if isinstance(source_files, list):
         if not source_files: