Canonicalize-SPSS-reference-rewrites

Author: TonisOrmisson

src/openstatspec/spss/raw_dictionary.py

@@ -176,6 +176,7 @@ def write_compatible_names(
         encoding=encoding,
         compatible_replacements=replacements,
         known_compatible_names=[name for _record, name in primary_variables],
+        known_long_names=[long_name for _short_name, long_name in pairs],
     )
     record_replacements.update(reference_replacements)
 
@@ -419,16 +420,49 @@ def _assert_dictionary_semantics(
         raise RawDictionaryError("Multiple-response Reader readback changed semantics.")
 
 
+def _rewrite_reference_member(
+    raw_member: bytes,
+    *,
+    subtype: int,
+    encoding: str,
+    known_compatible: dict[str, str],
+    known_long: dict[str, str],
+    encoded_replacements: dict[str, bytes],
+) -> bytes:
+    try:
+        compatible = raw_member.decode("ascii")
+        _validated_compatible_name(compatible)
+    except (UnicodeDecodeError, RawDictionaryError):
+        compatible = None
+    if compatible is not None and compatible.casefold() in known_compatible:
+        return encoded_replacements.get(compatible.casefold(), raw_member)
+    try:
+        long_name = raw_member.decode(encoding)
+    except UnicodeDecodeError as error:
+        raise RawDictionaryError(
+            f"Invalid subtype-{subtype} variable member token."
+        ) from error
+    if long_name.casefold() in known_long:
+        return raw_member
+    raise RawDictionaryError(
+        f"Subtype-{subtype} references unknown variable {long_name!r}."
+    )
+
+
 def _rewrite_reference_records(
     data: bytes, *, byte_order: str, records: list[_Record], encoding: str,
     compatible_replacements: dict[str, str], known_compatible_names: Iterable[str],
+    known_long_names: Iterable[str],
 ) -> tuple[dict[int, bytes], dict[int, tuple[bytes, ...]]]:
     known_names = list(known_compatible_names)
     _require_unique_names(known_names, "reference-compatible variable")
     known = {name.casefold(): name for name in known_names}
     encoded_replacements = {
         key: value.encode("ascii") for key, value in compatible_replacements.items()
     }
+    long_names = list(known_long_names)
+    _require_unique_names(long_names, "reference long variable")
+    known_long = {name.casefold(): name for name in long_names}
     replacements: dict[int, bytes] = {}
     expected: dict[int, list[bytes]] = {5: [], 7: [], 19: []}
     seen_variable_sets: dict[str, str] = {}
@@ -463,19 +497,14 @@ def _rewrite_reference_records(
             seen[set_key] = set_name
             members: list[bytes] = []
             for raw_member in parsed.members:
-                try:
-                    member = raw_member.decode("ascii")
-                    _validated_compatible_name(member)
-                except (UnicodeDecodeError, RawDictionaryError) as error:
-                    raise RawDictionaryError(
-                        f"Invalid subtype-{subtype} compatible member token."
-                    ) from error
-                member_key = member.casefold()
-                if member_key not in known:
-                    raise RawDictionaryError(
-                        f"Subtype-{subtype} references unknown compatible variable {member!r}."
-                    )
-                members.append(encoded_replacements.get(member_key, raw_member))
+                members.append(_rewrite_reference_member(
+                    raw_member,
+                    subtype=subtype,
+                    encoding=encoding,
+                    known_compatible=known,
+                    known_long=known_long,
+                    encoded_replacements=encoded_replacements,
+                ))
             updated_lines.append(_ReferenceLine(
                 set_name=parsed.set_name,
                 segments=parsed.segments,
@@ -648,7 +677,7 @@ def _very_long_string_entries(
     return _very_long_string_pairs(payload), True
 
 def _very_long_string_pairs(payload: bytes) -> list[tuple[str, bytes]]:
-    if not payload or not payload.endswith(b"\x00	"):
+    if not payload or not payload.endswith(b"	"):
         raise RawDictionaryError("Invalid SAV very-long-string record terminator.")
     raw_pairs = payload.split(b"	")
     if raw_pairs[-1] != b"":
@@ -722,6 +751,7 @@ def _assert_compatible_name_rewrite(
         encoding=encoding,
         compatible_replacements={},
         known_compatible_names=expected_primary_names,
+        known_long_names=[long_name for _short_name, long_name in pairs],
     )
     if observed_reference_payloads != expected_reference_payloads:
         raise RawDictionaryError(
@@ -844,7 +874,7 @@ def _document_record(lines: Iterable[str], *, encoding: str, byte_order: str) ->
 
 
 def _records(data: bytes) -> tuple[str, list[_Record]]:
-    if len(data) < _HEADER_SIZE or data[:4] not in {b"$FL2", b"$FL3"}:
+    if len(data) < _HEADER_SIZE or bytes(data[:4]) not in {b"$FL2", b"$FL3"}:
         raise RawDictionaryError("Not a supported SAV or ZSAV system file.")
     byte_order = _byte_order(data)
     offset = _HEADER_SIZE

src/openstatspec/sql/profiles.py

@@ -266,18 +266,6 @@ def preflight(
                     row_ordinal=row_ordinal, source_name=variable["source_name"],
                     encoded_bytes=encoded_bytes, maximum=profile.max_text_value_bytes,
                 )
-            declared_width = variable.get("string_width")
-            if declared_width is not None and encoded_bytes > declared_width:
-                raise _exceeded(
-                    "text_value_declared_width",
-                    f"row {row_ordinal} value for {variable['source_name']!r} is "
-                    f"{encoded_bytes} UTF-8 bytes; its declared SPSS width is "
-                    f"{declared_width}.",
-                    row_ordinal=row_ordinal,
-                    source_name=variable["source_name"],
-                    encoded_bytes=encoded_bytes,
-                    string_width=declared_width,
-                )
             row_bytes += 20 if profile.name in MYSQL_WIRE_PROFILES else encoded_bytes
         if profile.max_row_bytes is not None and row_bytes > profile.max_row_bytes:
             raise _exceeded(

tests/test_document_round_trip.py

@@ -1,4 +1,5 @@
 from pathlib import Path
+import re
 import shutil
 import sqlite3
 import subprocess
@@ -48,9 +49,12 @@ def _replace_expected_member_tokens(
     updated = []
     for payload in payloads:
         for old_name, new_name in replacements.items():
-            old = old_name.encode("ascii")
-            assert payload.count(old) <= 1
-            payload = payload.replace(old, new_name.encode("ascii"))
+            pattern = rb"(?i)(?<!\S)" + re.escape(old_name.encode("ascii")) + rb"(?!\S)"
+            matches = list(re.finditer(pattern, payload))
+            assert len(matches) <= 1
+            if matches:
+                match = matches[0]
+                payload = payload[:match.start()] + new_name.encode("ascii") + payload[match.end():]
         updated.append(payload)
     return tuple(updated)
 
@@ -89,8 +93,12 @@ def _replace_reference_payload_bytes(
     assert len(matches) == 1
     record = matches[0]
     payload = bytes(data[record.start + 16 : record.end])
-    assert payload.count(old) == 1
-    data[record.start + 16 : record.end] = payload.replace(old, new, 1)
+    matches = list(re.finditer(re.escape(old), payload, flags=re.IGNORECASE))
+    assert len(matches) == 1
+    match = matches[0]
+    data[record.start + 16 : record.end] = (
+        payload[:match.start()] + new + payload[match.end():]
+    )
     path.write_bytes(data)
 
 
@@ -500,7 +508,7 @@ def test_unknown_subtype_19_member_fails_without_source_change(
     )
     original = source.read_bytes()
 
-    with pytest.raises(RawDictionaryError, match="references unknown compatible variable"):
+    with pytest.raises(RawDictionaryError, match="references unknown variable"):
         write_compatible_names(
             source, {source_names[1]: "MEMBER2"}, encoding="UTF-8",
         )
@@ -526,7 +534,7 @@ def test_combined_zsav_reference_rewrite_preserves_offsets_and_semantics(
         for name in text_names:
             writer._add_var(name, 8)  # pylint: disable=protected-access
         writer._add_var(vls_name, 300)  # pylint: disable=protected-access
-        writer.var_sets = {"All analysis variables": all_names}
+        writer.var_sets = {"Analysis": all_names}
         writer.mrsets = {
             "$categories": {"label": "Categories", "variable_list": numeric_names[:2]},
             "$dichotomies": {
@@ -696,7 +704,7 @@ def test_legacy_source_unicode_export_uses_actual_output_encoding_for_raw_rewrit
     source = tmp_path / "legacy-source.sav"
     destination = tmp_path / "unicode-output.sav"
     database = f"sqlite:///{tmp_path / 'legacy-to-unicode.sqlite'}"
-    source_names = ["küsimuse_pikk_nimi", "teine_pikk_nimi"]
+    source_names = ["kusimuse_pikk_nimi", "teine_pikk_nimi"]
     mr_label = "Mitmikvalik – küsimus"
     frame = pd.DataFrame({source_names[0]: [1.0, 0.0], source_names[1]: [0.0, 1.0]})
     pyspssio.write_sav(