Report Python SAV encoding fidelity loss

Author: TonisOrmisson

src/openstatspec/api.py

@@ -15,7 +15,9 @@ def capability_matrix() -> Mapping[str, Any]:
     return {
         "spss": {
             "values": "supported", "variable_labels": "supported",
-            "value_labels": "supported", "formats": "supported",
+            "value_labels": "supported",
+            "print_format": "supported", "write_format": "unobservable",
+            "source_encoding": "preserved only when UTF-8",
             "measurement_level": "supported", "user_missing_rules": "supported",
             "documents": "supported", "multiple_response_sets": "lossy",
             "variable_alignment": "lossy", "variable_sets": "unobservable",

src/openstatspec/spss/sav.py

@@ -13,6 +13,9 @@
 from ..sql.wide import create_wide_dataset, physical_name, read_fidelity_events, read_wide_dataset, record_export_operation
 
 
+_UTF8_ENCODINGS = {"UTF-8", "UTF8"}
+
+
 def _variables(meta: Any, names: list[str]) -> list[dict[str, Any]]:
     used = {"__case_ordinal"}
     labels = dict(getattr(meta, "column_names_to_labels", {}) or {})
@@ -167,21 +170,30 @@ def _sha256(path: Path) -> str:
     return digest.hexdigest()
 
 def _import_loss_report(meta: Any) -> tuple[dict[str, str], ...]:
-    events = [{"code": "unobservable-source-dictionary-features", "detail": "The reader does not expose SPSS variable sets or custom file/variable attributes, so their presence cannot be established or preserved."}]
+    events = [{"code": "unobservable-source-dictionary-features", "detail": "The reader does not expose SPSS variable sets, custom file/variable attributes, variable roles, or separate write formats, so their presence cannot be established or preserved."}]
+    if _is_non_utf8_encoding(getattr(meta, "file_encoding", None)):
+        events.append({"code": "source-encoding-not-preserved", "detail": "The SAV writer has no output-encoding option and writes UTF-8, so a non-UTF-8 source encoding cannot be restored."})
     if getattr(meta, "mr_sets", {}) or {}:
         events.append({"code": "multiple-response-sets-not-exportable", "detail": "Multiple-response sets are catalogued but pyreadstat cannot write them back to SAV."})
     return tuple(events)
 
 
 def _export_loss_report(dataset: dict[str, Any], variables: list[dict[str, Any]]) -> tuple[dict[str, str], ...]:
-    events = [{"code": "unobservable-source-dictionary-features", "detail": "Variable sets and custom file/variable attributes are not observable through the reader and therefore cannot be restored by export."}]
+    events = [{"code": "unobservable-source-dictionary-features", "detail": "Variable sets, custom file/variable attributes, variable roles, and separate SPSS write formats are not observable through the reader and therefore cannot be restored by export."}]
+    if _is_non_utf8_encoding(dataset.get("source_encoding")):
+        events.append({"code": "source-encoding-not-preserved", "detail": "The SAV writer has no output-encoding option and writes UTF-8, so a non-UTF-8 source encoding cannot be restored."})
     if dataset["multiple_response_sets"] != "{}":
         events.append({"code": "multiple-response-sets-not-exported", "detail": "The SAV writer has no multiple-response-set output capability."})
     if any(item.get("alignment") not in (None, "unknown") for item in variables):
         events.append({"code": "variable-alignment-not-exported", "detail": "The SAV writer has no variable-alignment output capability."})
     return tuple(events)
 
 
+def _is_non_utf8_encoding(encoding: Any) -> bool:
+    """Whether a known source encoding differs from the writer's UTF-8 output."""
+    return encoding is not None and str(encoding).strip().replace("_", "-").upper() not in _UTF8_ENCODINGS
+
+
 def _merge_loss_reports(*reports: tuple[dict[str, str], ...]) -> tuple[dict[str, str], ...]:
     """Keep one deterministic diagnostic per loss code across import and export."""
     events: dict[str, dict[str, str]] = {}

tests/test_loss_reports.py

@@ -108,3 +108,49 @@ def test_loss_allowed_export_persists_accepted_diagnostics(tmp_path) -> None:
     ).fetchone()
     assert (direction, severity, code) == ("export", "warning", "unobservable-source-dictionary-features")
     assert '"accepted_by_user": true' in details
+
+
+def test_non_utf8_source_encoding_is_explicit_export_loss(tmp_path) -> None:
+    """The writer has no output-encoding argument, so it must fail closed."""
+    import pyreadstat
+    import pytest
+
+    from openstatspec.api import capability_matrix
+    from openstatspec.core import UnsupportedOperationError
+
+    database_path = tmp_path / "legacy-encoding.sqlite"
+    database = f"sqlite:///{database_path}"
+    destination = tmp_path / "legacy-encoding.sav"
+    create_wide_dataset(
+        database_url=database, dataset_id="legacy-encoding", source_name="legacy.sav",
+        source_format="SAV", source_encoding="WINDOWS-1252",
+        rows=[{"name": "Muller"}], variables=[{
+            "ordinal": 1, "source_name": "name", "physical_name": "name",
+            "storage_kind": "string", "string_width": 8, "label": "",
+            "format": "A8", "measure": "nominal", "alignment": None,
+            "display_width": 8, "value_labels": "{}", "missing_ranges": "[]",
+        }],
+    )
+
+    with pytest.raises(UnsupportedOperationError, match="source-encoding-not-preserved"):
+        export_sav(
+            database_url=database, dataset_id="legacy-encoding", destination=destination,
+            allow_loss=["unobservable-source-dictionary-features"],
+        )
+    assert not destination.exists()
+
+    exported = export_sav(
+        database_url=database, dataset_id="legacy-encoding", destination=destination,
+        allow_loss=[
+            "unobservable-source-dictionary-features",
+            "source-encoding-not-preserved",
+        ],
+    )
+    assert {event["code"] for event in exported["loss_report"]} == {
+        "source-encoding-not-preserved",
+        "unobservable-source-dictionary-features",
+    }
+    _, metadata = pyreadstat.read_sav(destination, metadataonly=True)
+    assert metadata.file_encoding == "UTF-8"
+    assert capability_matrix()["spss"]["print_format"] == "supported"
+    assert capability_matrix()["spss"]["write_format"] == "unobservable"