fix(windows): add legacy encoding repair tool

mempalace/encoding_repair.py

@@ -0,0 +1,151 @@
+"""Repair legacy UTF-8 text that was decoded as Windows-1252."""
+
+from __future__ import annotations
+
+
+def _byte_for_character(character: str) -> int | None:
+    try:
+        encoded = character.encode("cp1252")
+    except UnicodeEncodeError:
+        codepoint = ord(character)
+        if codepoint <= 255:
+            return codepoint
+        return None
+
+    if len(encoded) != 1:
+        return None
+
+    return encoded[0]
+
+
+def _decode_candidate(
+    text: str,
+    start: int,
+) -> tuple[str, int] | None:
+    for width in (4, 3, 2):
+        segment = text[start : start + width]
+        if len(segment) != width:
+            continue
+
+        raw_values = []
+        for character in segment:
+            value = _byte_for_character(character)
+            if value is None:
+                break
+            raw_values.append(value)
+        else:
+            raw = bytes(raw_values)
+            try:
+                decoded = raw.decode("utf-8")
+            except UnicodeDecodeError:
+                continue
+
+            if len(decoded) == 1 and ord(decoded) >= 128:
+                return decoded, width
+
+    return None
+
+
+def repair_mojibake_once(text: str) -> str:
+    """Repair one layer of UTF-8-as-Windows-1252 mojibake."""
+    output: list[str] = []
+    index = 0
+
+    while index < len(text):
+        candidate = _decode_candidate(text, index)
+        if candidate is None:
+            output.append(text[index])
+            index += 1
+            continue
+
+        decoded, width = candidate
+        output.append(decoded)
+        index += width
+
+    return "".join(output)
+
+
+def repair_mojibake(
+    text: str,
+    *,
+    max_passes: int = 3,
+) -> str:
+    """Repair repeated mojibake layers until stable."""
+    current = text
+
+    for _ in range(max_passes):
+        repaired = repair_mojibake_once(current)
+        if repaired == current:
+            break
+        current = repaired
+
+    return current
+
+
+def _result_field(result, name: str):
+    if isinstance(result, dict):
+        return result.get(name)
+    return getattr(result, name, None)
+
+
+def repair_collection(
+    collection,
+    *,
+    apply: bool = False,
+    page_size: int = 500,
+) -> dict[str, int]:
+    """Scan a collection and optionally update damaged documents."""
+    if page_size < 1:
+        raise ValueError("page_size must be at least 1")
+
+    scanned = 0
+    changed = 0
+    updated = 0
+    offset = 0
+
+    while True:
+        page = collection.get(
+            limit=page_size,
+            offset=offset,
+            include=["documents"],
+        )
+        ids = list(_result_field(page, "ids") or [])
+        documents = list(_result_field(page, "documents") or [])
+
+        if not ids:
+            break
+
+        update_ids = []
+        update_documents = []
+
+        for drawer_id, document in zip(ids, documents):
+            scanned += 1
+
+            if not isinstance(document, str):
+                continue
+
+            repaired = repair_mojibake(document)
+            if repaired == document:
+                continue
+
+            changed += 1
+            update_ids.append(drawer_id)
+            update_documents.append(repaired)
+
+        if apply and update_ids:
+            collection.update(
+                ids=update_ids,
+                documents=update_documents,
+            )
+            updated += len(update_ids)
+
+        offset += len(ids)
+
+        if len(ids) < page_size:
+            break
+
+    return {
+        "scanned": scanned,
+        "changed": changed,
+        "updated": updated,
+    }

scripts/mempalace_repair_encoding.py

@@ -0,0 +1,75 @@
+#!/usr/bin/env python3
+"""Repair legacy Windows mojibake in a MemPalace collection."""
+
+from __future__ import annotations
+
+import argparse
+
+from mempalace.config import MempalaceConfig
+from mempalace.encoding_repair import repair_collection
+from mempalace.palace import get_collection
+
+
+def build_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(
+        description=(
+            "Detect and repair UTF-8 text that legacy Windows paths "
+            "stored as Windows-1252 mojibake."
+        )
+    )
+    parser.add_argument(
+        "--palace",
+        help="Palace path; defaults to the configured palace.",
+    )
+    parser.add_argument(
+        "--collection",
+        help="Collection name; defaults to the configured collection.",
+    )
+    parser.add_argument(
+        "--page-size",
+        type=int,
+        default=500,
+        help="Rows scanned per page (default: 500).",
+    )
+    parser.add_argument(
+        "--apply",
+        action="store_true",
+        help="Write repaired documents. Without this flag, run read-only.",
+    )
+    return parser
+
+
+def main() -> int:
+    args = build_parser().parse_args()
+    config = MempalaceConfig()
+
+    palace_path = args.palace or config.palace_path
+    collection_name = args.collection or getattr(config, "collection_name", "mempalace_drawers")
+
+    collection = get_collection(
+        palace_path,
+        collection_name=collection_name,
+        create=False,
+    )
+
+    report = repair_collection(
+        collection,
+        apply=args.apply,
+        page_size=args.page_size,
+    )
+
+    mode = "APPLY" if args.apply else "DRY RUN"
+    print(f"Mode: {mode}")
+    print(f"Rows scanned: {report['scanned']}")
+    print(f"Documents needing repair: {report['changed']}")
+    print(f"Documents updated: {report['updated']}")
+
+    if not args.apply and report["changed"]:
+        print()
+        print("Run again with --apply to write the repairs.")
+
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

tests/test_encoding_repair.py

@@ -0,0 +1,115 @@
+from mempalace.encoding_repair import (
+    repair_collection,
+    repair_mojibake,
+)
+
+
+def mojibake(text):
+    return text.encode("utf-8").decode("cp1252")
+
+
+def test_repairs_common_accented_text():
+    assert repair_mojibake("café") == "café"
+    assert repair_mojibake("naïve") == "naïve"
+
+
+def test_repairs_dash_arrow_and_emoji():
+    damaged = mojibake("Plan → result — ✅")
+    assert repair_mojibake(damaged) == "Plan → result — ✅"
+
+
+def test_preserves_clean_text():
+    clean = "Already clean: café → ✅"
+    assert repair_mojibake(clean) == clean
+
+
+def test_repairs_mixed_clean_and_damaged_text():
+    text = "Clean prefix, café, clean suffix."
+    assert repair_mojibake(text) == "Clean prefix, café, clean suffix."
+
+
+def test_repairs_double_encoded_text():
+    once = mojibake("café")
+    twice = mojibake(once)
+
+    assert repair_mojibake(twice) == "café"
+
+
+def test_repair_is_idempotent():
+    repaired = repair_mojibake("café → done")
+    assert repair_mojibake(repaired) == repaired
+
+
+class FakeCollection:
+    def __init__(self, documents):
+        self.ids = [f"drawer-{index}" for index in range(len(documents))]
+        self.documents = list(documents)
+        self.updates = []
+
+    def get(self, *, limit, offset, include):
+        del include
+        end = offset + limit
+        return {
+            "ids": self.ids[offset:end],
+            "documents": self.documents[offset:end],
+        }
+
+    def update(self, *, ids, documents):
+        self.updates.append(
+            {
+                "ids": list(ids),
+                "documents": list(documents),
+            }
+        )
+
+
+def test_collection_dry_run_reports_without_writing():
+    collection = FakeCollection(["café", "plain", "arrow →"])
+
+    report = repair_collection(
+        collection,
+        apply=False,
+        page_size=2,
+    )
+
+    assert report == {
+        "scanned": 3,
+        "changed": 2,
+        "updated": 0,
+    }
+    assert collection.updates == []
+
+
+def test_collection_apply_updates_only_changed_documents():
+    collection = FakeCollection(["café", "plain", "arrow →"])
+
+    report = repair_collection(
+        collection,
+        apply=True,
+        page_size=2,
+    )
+
+    assert report == {
+        "scanned": 3,
+        "changed": 2,
+        "updated": 2,
+    }
+
+    updated_ids = [drawer_id for batch in collection.updates for drawer_id in batch["ids"]]
+    updated_documents = [
+        document for batch in collection.updates for document in batch["documents"]
+    ]
+
+    assert updated_ids == ["drawer-0", "drawer-2"]
+    assert updated_documents == ["café", "arrow →"]
+
+
+def test_collection_rejects_invalid_page_size():
+    collection = FakeCollection([])
+
+    try:
+        repair_collection(collection, page_size=0)
+    except ValueError as exc:
+        assert "page_size" in str(exc)
+    else:
+        raise AssertionError("Expected ValueError")