Restore API

decode_document() validates and decodes printable lines. unarchive_bytes() validates every target before writing the restored tree.

glyphive.restore

Restore-side subpackage: printed/scanned glyphive pages back to bytes.

Contains the OCR orchestration layer (:mod:glyphive.restore.ocr) together with the text-transcript decode and safe unarchive path. Importing this package pulls in no heavy optional dependencies.

__all__ = ['RestoreError', 'decode_document', 'QrTransportError', 'transcript_from_images', 'unarchive_bytes', 'restore_document'] module-attribute

QrTransportError

Bases: ValueError

A QR symbol set is malformed, inconsistent, or incomplete.

RestoreError

Bases: Exception

Raised on an integrity failure the restore path refuses to paper over.

Used for whole-document SHA-256 mismatches (:func:decode_document) and path-traversal / clobber violations (:mod:glyphive.restore.unarchive). The message always names the concrete offender (expected-vs-got digest, or the offending relpath) so a failure is actionable, never silent.

decode_document(text_lines, *, char_conf=None, conf_threshold=0.6, max_suspects=6)

Turn a full page transcript back into (meta, raw_archive_bytes).

Pipeline (each stage is an already-tested module; this only composes them):

1. :func:glyphive.layout.read_pages — parse the #!glyphive header, strip page headers/footers, tolerate out-of-order pages, and detect a missing page. Propagates :class:glyphive.layout.MissingPageError (naming the absent page numbers) unchanged — that is the correct, loud behaviour for an incomplete transcript.
2. codec.get(meta["codec"]).decode — per-line CRC + Reed-Solomon recovery of the compressed payload. Propagates :class:glyphive.codec.CodecError (naming the exact failing line) unchanged when a line is unrecoverable.
3. :func:glyphive.compression.get — inverse of the archive's compression stage, selected by the header's comp= value.
4. Whole-document integrity: sha256(raw) must equal the header's sha256=. On mismatch this raises :class:RestoreError naming the expected and observed digests and returns nothing — corrupt bytes are never handed back — no silent corruption.

char_conf (plan 3, optional): per-line RAW OCR character confidence, one entry per element of text_lines in the same order (None for a line with no confidence) -- see :func:decode_document_to_spool and the module docstring's "OCR-confidence erasure hint" section in :mod:glyphive.codec.engine. Absent (the default), decode is byte- identical to a build without this feature.

Returns

(meta, raw) where raw is the archive byte stream ready for :func:glyphive.restore.unarchive.unarchive_bytes, and meta is the parsed header dict. meta still carries meta["_page_warnings"] (a list of corrupt-but-recovered-page warning strings from read_pages) so the caller can log them. The returned meta["meta"] is the resolved archive profile; old document headers without that key are accepted.

Raises

glyphive.layout.LayoutError / MissingPageError: No parseable header, or a whole page absent from the transcript. glyphive.codec.engine.CodecError: A framed line failed CRC and RS could not correct it (line named). RestoreError: The decompressed archive's SHA-256 does not match the header's.

Source code in src/glyphive/restore/decode.py

def decode_document(
    text_lines: _ty.Iterable[str],
    *,
    char_conf: "_ty.Optional[_ty.Sequence[_ty.Optional[_ty.Sequence[float]]]]" = None,
    conf_threshold: float = 0.6,
    max_suspects: int = 6,
) -> _ty.Tuple[_ty.Dict[str, _ty.Any], bytes]:
    """Turn a full page transcript back into ``(meta, raw_archive_bytes)``.

    Pipeline (each stage is an already-tested module; this only composes them):

    1. :func:`glyphive.layout.read_pages` — parse the ``#!glyphive`` header,
       strip page headers/footers, tolerate out-of-order pages, and detect a
       missing page. Propagates :class:`glyphive.layout.MissingPageError`
       (naming the absent page numbers) unchanged — that is the correct, loud
       behaviour for an incomplete transcript.
    2. ``codec.get(meta["codec"]).decode`` — per-line CRC + Reed-Solomon recovery of
       the compressed payload. Propagates :class:`glyphive.codec.CodecError`
       (naming the exact failing line) unchanged when a line is unrecoverable.
    3. :func:`glyphive.compression.get` — inverse of the archive's compression
       stage, selected by the header's ``comp=`` value.
    4. **Whole-document integrity**: ``sha256(raw)`` must equal the header's
       ``sha256=``. On mismatch this raises :class:`RestoreError` naming the
       expected and observed digests and returns nothing — corrupt bytes are
       never handed back — no silent corruption.

    ``char_conf`` (plan 3, optional): per-line RAW OCR character confidence,
    one entry per element of ``text_lines`` in the same order (``None`` for
    a line with no confidence) -- see :func:`decode_document_to_spool` and
    the module docstring's "OCR-confidence erasure hint" section in
    :mod:`glyphive.codec.engine`. Absent (the default), decode is byte-
    identical to a build without this feature.

    Returns
    -------
    ``(meta, raw)`` where ``raw`` is the archive byte stream ready for
    :func:`glyphive.restore.unarchive.unarchive_bytes`, and ``meta`` is the
    parsed header dict. ``meta`` still carries ``meta["_page_warnings"]`` (a
    list of corrupt-but-recovered-page warning strings from ``read_pages``) so
    the caller can log them. The returned ``meta["meta"]`` is the resolved
    archive profile; old document headers without that key are accepted.

    Raises
    ------
    glyphive.layout.LayoutError / MissingPageError:
        No parseable header, or a whole page absent from the transcript.
    glyphive.codec.engine.CodecError:
        A framed line failed CRC and RS could not correct it (line named).
    RestoreError:
        The decompressed archive's SHA-256 does not match the header's.
    """
    sink = io.BytesIO()
    meta = decode_document_to_spool(
        text_lines,
        sink,
        char_conf=char_conf,
        conf_threshold=conf_threshold,
        max_suspects=max_suspects,
    )
    return meta, sink.getvalue()

restore_document(text_lines, dest, *, overwrite=False)

Full text-transcript → tree convenience: decode then unarchive.

Equivalent to :func:glyphive.restore.decode.decode_document followed by :func:unarchive_bytes. This is the single call the CLI uses for extract from a text transcript. All the integrity guarantees of both stages apply: a missing page, an unrecoverable line, a whole-document hash mismatch, or a path-traversal record each raises loudly and nothing corrupt is written.

Returns the list of relpaths written under dest.

Source code in src/glyphive/restore/unarchive.py

def restore_document(
    text_lines: _ty.Iterable[str],
    dest: DestLike,
    *,
    overwrite: bool = False,
) -> _ty.List[str]:
    """Full text-transcript → tree convenience: decode then unarchive.

    Equivalent to :func:`glyphive.restore.decode.decode_document` followed by
    :func:`unarchive_bytes`. This is the single call the CLI uses for
    ``extract`` from a text transcript. All the integrity guarantees of both
    stages apply: a missing page, an unrecoverable line, a whole-document hash
    mismatch, or a path-traversal record each raises loudly and nothing corrupt
    is written.

    Returns the list of relpaths written under ``dest``.
    """
    _meta, written = restore_document_spooled(
        text_lines, dest, overwrite=overwrite
    )
    return written

transcript_from_images(path)

Decode a page image or sorted direct-child image directory.

Source code in src/glyphive/restore/qr.py

def transcript_from_images(path: _ty.Union[str, Path]) -> bytes:
    """Decode a page image or sorted direct-child image directory."""
    try:
        from PIL import Image
    except ImportError as exc:  # pragma: no cover - exercised by base-install probe
        raise RuntimeError("QR decoding requires 'pip install glyphive[qr]'") from exc
    source = Path(path)
    paths = sorted(item for item in source.iterdir() if item.is_file()) if source.is_dir() else [source]
    symbols: _ty.List[bytes] = []
    for image_path in paths:
        with image_path.open("rb") as stream:
            prefix = stream.read(16)
        image_magic = (
            prefix.startswith(b"\x89PNG\r\n\x1a\n")
            or prefix.startswith(b"\xff\xd8\xff")
            or prefix.startswith((b"GIF87a", b"GIF89a"))
            or prefix.startswith(b"BM")
            or prefix.startswith((b"II*\x00", b"MM\x00*"))
            or (prefix.startswith(b"RIFF") and prefix[8:12] == b"WEBP")
        )
        if not image_magic:
            raise QrTransportError(
                f"QR input is not a supported page image by magic bytes: {image_path}"
            )
        with Image.open(image_path) as image:
            decoded = symbols_from_image(image)
        if not decoded:
            raise QrTransportError(f"no QR symbols found in {image_path}")
        symbols.extend(decoded)
    return reassemble_symbols(symbols)

unarchive_bytes(raw, dest, *, overwrite=False)

Write the archive byte stream raw into dest; return relpaths written.

The archive header and all records are validated before the destination is created. Metadata is applied only when the stream explicitly carries it.

Source code in src/glyphive/restore/unarchive.py

def unarchive_bytes(
    raw: bytes,
    dest: DestLike,
    *,
    overwrite: bool = False,
) -> _ty.List[str]:
    """Write the archive byte stream ``raw`` into ``dest``; return relpaths written.

    The archive header and all records are validated before the destination is
    created. Metadata is applied only when the stream explicitly carries it.
    """
    dest_path = _coerce_dest(dest)
    if dest_path.exists() and not dest_path.is_dir():
        raise RestoreError(f"destination {dest_path!s} exists and is not a directory")
    stream_meta = archive.stream_metadata(raw)
    records = list(archive.iter_records(raw))
    dest_resolved = dest_path.resolve()
    targets = _record_targets(dest_resolved, records)
    apply_metadata = stream_meta.metadata == "basic"
    dest_path.mkdir(parents=True, exist_ok=True)

    written: _ty.List[str] = []

    for record, target in targets:
        if record.type == archive.REC_EMPTY_DIR:
            target.mkdir(parents=True, exist_ok=True)
            _restore_metadata(
                target, record.mode, record.mtime, enabled=apply_metadata
            )
            written.append(record.path)
            continue

        # Ensure the parent directory chain exists.
        parent = target.parent
        parent.mkdir(parents=True, exist_ok=True)

        if target.exists() and not overwrite:
            # Idempotent if identical; otherwise refuse to silently destroy.
            try:
                existing = target.read_bytes()
            except OSError as exc:
                raise RestoreError(
                    f"cannot verify existing target {record.path!r} before "
                    f"writing (overwrite=False): {exc}"
                ) from exc
            if existing != record.content:
                raise RestoreError(
                    f"target {record.path!r} already exists with different "
                    "content; refusing to overwrite (pass overwrite=True to "
                    "replace)"
                )
            # Identical content already present: count it as written, skip I/O.
            _restore_metadata(
                target, record.mode, record.mtime, enabled=apply_metadata
            )
            written.append(record.path)
            continue

        target.write_bytes(record.content)
        _restore_metadata(target, record.mode, record.mtime, enabled=apply_metadata)
        written.append(record.path)

    return written