Kjell Tore Guttormsen
1c958ab8d6
Chose the side channel over neutralising pointer-shaped document text, because the second fix changes what every document SAYS in order to defend a tool outside the build: a source quoting a bundle listing would come out altered and existing bundles would move bytes. This reads a file the run already writes. `assets.conversion` names the pair, `DocumentAssets.conversions` carries it out of the run, `DocumentAccount.conversions` books it, and the accounting JSON states it per document. `_declared_conversions` reads it; `_conversions` now believes a pair only when the RUN booked it AND a pointer block confirms it for the asset it names. The confirmation can be forged and the ledger cannot, which is why the ledger decides. Measured through the real `okf build`: the three arms PM reproduced (two `<p>`, one `<p>` with `<br>`, a markdown note beside the carrier) go forged -> refused, 3 of 3, with the known-positive True in all three. The text-level regression guard goes 3 arms to 13, the two new ones being a perfectly written pointer block the run never booked. R761, rebuilt: 25 BMP sources, 19 held, 19 of 19 conversions confirmed against 19 declared, 50 assets (29 JPEG + 21 PNG, 0 BMP), SHY 71, u = 0, d = 0, exit 0. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
1045 lines
38 KiB
Python
1045 lines
38 KiB
Python
"""Content accounting for `okf build`: what the SOURCE held, and where each part went.
|
|
|
|
`okf build`'s conservation identity counts FILES. A file can be merged while
|
|
text inside it is gone, and nothing about the identity can say so. This module
|
|
adds the element level:
|
|
|
|
1. an INVENTORY per document, read from the source bytes by the format's own
|
|
rules, independent of whether extraction or the persist gate later succeed
|
|
-- so a refused document still says what it held;
|
|
2. an ACCOUNT after the run, giving every inventoried element exactly one
|
|
fate: `carried`, `pointer` or a coded rejection. What has none is
|
|
UNACCOUNTED, what has two is DOUBLE-BOOKED, and a build asked for the
|
|
account fails on either.
|
|
|
|
**The vocabulary is the gate's.** `tools/okf_accounting_gate.py` compares the
|
|
inventory here against `tools/okf_witness.py`, an implementation that imports
|
|
nothing from this package. The two implement ONE set of definitions (named per
|
|
format below) twice; agreeing on the fixtures is evidence that the definitions
|
|
are implementable as written, and disagreeing is a finding about one of them.
|
|
|
|
**`carried` is checked, not declared.** A persisted document's element is
|
|
carried when every piece of its text is found in the concept bodies written
|
|
for that document, compared on letters and digits alone (case-folded), so
|
|
markdown escapes, table pipes and whitespace do not count as loss. A container
|
|
(a table, a section, a slide) is carried when everything inside it is. An
|
|
element with no text of its own is carried with its document. This finds text
|
|
that the extractor, the segmentation, the gate or the writer lost; it cannot
|
|
find text the extractor never saw AND no inventory counts -- that is what the
|
|
per-format vocabulary bounds, and what the gate's exceptions name.
|
|
|
|
**Images** are booked from what the reader did with each placement, never
|
|
from the bundle: carried (bytes in `assets/`), `pointer` (a remote reference,
|
|
which extraction never fetches) or rejected with the reader's code. A format
|
|
whose reader does not handle images at all books an image as `pointer` when
|
|
its reference is present in the carried text (markdown), and leaves it
|
|
unaccounted otherwise.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import csv
|
|
import io
|
|
import json
|
|
import posixpath
|
|
import re
|
|
import zipfile
|
|
from collections.abc import Callable, Iterable, Mapping, Sequence
|
|
from dataclasses import dataclass, field
|
|
from html.parser import HTMLParser
|
|
from pathlib import Path, PurePosixPath
|
|
from typing import Any
|
|
from xml.etree import ElementTree
|
|
|
|
from .inbox import DocumentAssets, InboxResult, relative_source
|
|
|
|
ACCOUNTING_VERSION = 1
|
|
|
|
CARRIED = "carried"
|
|
POINTER = "pointer"
|
|
MERGED = "merged"
|
|
REJECTED = "rejected"
|
|
PERSISTED = "persisted"
|
|
|
|
#: The reader's code for an image it deliberately did not fetch.
|
|
_REMOTE_CODE = "asset_remote"
|
|
|
|
_REMOTE = re.compile(r"^(?:[a-zA-Z][a-zA-Z0-9+.-]*:|//)")
|
|
_NOT_ALNUM = re.compile(r"[\W_]+")
|
|
|
|
|
|
def _norm(text: str) -> str:
|
|
return _NOT_ALNUM.sub("", text.casefold())
|
|
|
|
|
|
# --- inventory ---------------------------------------------------------------
|
|
|
|
|
|
@dataclass
|
|
class Inventory:
|
|
"""What one source document holds, in the gate's element vocabulary.
|
|
|
|
`chunks` is the text of the document in pieces; each element names the
|
|
pieces it consists of. `refs` holds, per image element, what the source
|
|
points at: an inbox-relative path, a remote reference, or None for an
|
|
embedded picture.
|
|
"""
|
|
|
|
source_file: str
|
|
vocabulary: tuple[str, ...]
|
|
elements: list[tuple[str, tuple[int, ...]]] = field(default_factory=list)
|
|
chunks: list[str] = field(default_factory=list)
|
|
refs: list[tuple[str, str | None]] = field(default_factory=list)
|
|
|
|
def chunk(self, text: str) -> int:
|
|
self.chunks.append(text)
|
|
return len(self.chunks) - 1
|
|
|
|
def add(self, kind: str, chunks: Iterable[int] = ()) -> None:
|
|
if kind not in self.vocabulary:
|
|
raise ValueError(f"{kind!r} is not in this format's vocabulary")
|
|
self.elements.append((kind, tuple(chunks)))
|
|
|
|
def counts(self) -> dict[str, int]:
|
|
result = dict.fromkeys(self.vocabulary, 0)
|
|
for kind, _ in self.elements:
|
|
result[kind] += 1
|
|
return result
|
|
|
|
def pointed_files(self) -> set[str]:
|
|
return {target for kind, target in self.refs if kind == "local" and target}
|
|
|
|
|
|
def _resolve(inbox: Path, document: Path, ref: str, *, sts: bool) -> str | None:
|
|
"""The inbox-relative file a local reference names, contained in the
|
|
document's own directory, or None."""
|
|
directory = document.parent
|
|
tries = [ref] + ([f"graphics/{PurePosixPath(ref).name}"] if sts else [])
|
|
for candidate in tries:
|
|
normal = posixpath.normpath(candidate)
|
|
if normal.startswith(("/", "..")) or normal == ".":
|
|
continue
|
|
target = directory / normal
|
|
if target.is_file():
|
|
return relative_source(target, inbox)
|
|
return None
|
|
|
|
|
|
def _add_image(inv: Inventory, inbox: Path, document: Path, ref: str | None, sts: bool) -> None:
|
|
if ref is None:
|
|
inv.refs.append(("embedded", None))
|
|
inv.add("image")
|
|
return
|
|
if not ref or _REMOTE.match(ref):
|
|
inv.refs.append(("remote", ref or None))
|
|
inv.add("image", (inv.chunk(ref),) if ref else ())
|
|
return
|
|
inv.refs.append(("local", _resolve(inbox, document, ref, sts=sts)))
|
|
inv.add("image", (inv.chunk(ref),))
|
|
|
|
|
|
# markdown -- fences as the proposer reads them, so the two cannot disagree
|
|
_ATX_LINE = re.compile(r"^ {0,3}#{1,6}(?:[ \t]|$)")
|
|
_TABLE_DELIMITER = re.compile(r"^\s*\|?(?:\s*:?-{3,}:?\s*\|)*\s*:?-{3,}:?\s*\|?\s*$")
|
|
_IMAGE_REF = re.compile(r"!\[[^\]]*\]\(\s*<?([^)\s>]+)>?[^)]*\)")
|
|
|
|
_MARKDOWN = ("heading", "paragraph", "table", "table_row", "image", "code_block")
|
|
|
|
|
|
def _inventory_markdown(inv: Inventory, inbox: Path, path: Path, text: str) -> None:
|
|
from .propose import fenced_lines
|
|
|
|
lines = text.split("\n")
|
|
fenced = fenced_lines(lines)
|
|
index = 0
|
|
while index < len(lines):
|
|
line = lines[index]
|
|
if index in fenced:
|
|
start = index
|
|
while index < len(lines) and index in fenced:
|
|
index += 1
|
|
inv.add("code_block", (inv.chunk("\n".join(lines[start:index])),))
|
|
continue
|
|
if not line.strip():
|
|
index += 1
|
|
continue
|
|
if _ATX_LINE.match(line):
|
|
inv.add("heading", (inv.chunk(line),))
|
|
index += 1
|
|
continue
|
|
nxt = lines[index + 1] if index + 1 < len(lines) else ""
|
|
if "|" in line and index + 1 not in fenced and _TABLE_DELIMITER.match(nxt):
|
|
header = inv.chunk(line)
|
|
index += 2
|
|
rows: list[int] = []
|
|
while index < len(lines) and index not in fenced and "|" in lines[index]:
|
|
row = inv.chunk(lines[index])
|
|
inv.add("table_row", (row,))
|
|
rows.append(row)
|
|
index += 1
|
|
inv.add("table", (header, *rows))
|
|
continue
|
|
chunks: list[int] = []
|
|
while (
|
|
index < len(lines)
|
|
and index not in fenced
|
|
and lines[index].strip()
|
|
and not _ATX_LINE.match(lines[index])
|
|
):
|
|
current = lines[index]
|
|
for ref in _IMAGE_REF.findall(current):
|
|
_add_image(inv, inbox, path, ref, sts=False)
|
|
rest = _IMAGE_REF.sub("", current)
|
|
if rest.strip():
|
|
chunks.append(inv.chunk(rest))
|
|
index += 1
|
|
if chunks:
|
|
inv.add("paragraph", chunks)
|
|
|
|
|
|
_TEXT = ("paragraph", "line")
|
|
|
|
|
|
def _inventory_text(inv: Inventory, text: str) -> None:
|
|
block: list[int] = []
|
|
for line in [*text.split("\n"), ""]:
|
|
if line.strip():
|
|
piece = inv.chunk(line)
|
|
inv.add("line", (piece,))
|
|
block.append(piece)
|
|
elif block:
|
|
inv.add("paragraph", block)
|
|
block = []
|
|
|
|
|
|
_CSV = ("header_cell", "row", "cell")
|
|
|
|
|
|
def _inventory_csv(inv: Inventory, text: str) -> None:
|
|
rows = [row for row in csv.reader(io.StringIO(text)) if row]
|
|
for position, row in enumerate(rows):
|
|
pieces = [inv.chunk(value) for value in row]
|
|
if position == 0:
|
|
for piece in pieces:
|
|
inv.add("header_cell", (piece,))
|
|
continue
|
|
for piece in pieces:
|
|
inv.add("cell", (piece,))
|
|
inv.add("row", pieces)
|
|
|
|
|
|
_JSON = ("key", "value")
|
|
|
|
|
|
def _inventory_json(inv: Inventory, text: str) -> None:
|
|
"""Fenced verbatim by the reader, so every key and value is carried exactly
|
|
when the document's own text is: each names the whole text."""
|
|
whole = (inv.chunk(text),)
|
|
stack: list[Any] = [json.loads(text)]
|
|
while stack:
|
|
node = stack.pop()
|
|
if isinstance(node, dict):
|
|
for key, value in node.items():
|
|
inv.add("key", whole)
|
|
stack.append(value)
|
|
elif isinstance(node, list):
|
|
stack.extend(node)
|
|
else:
|
|
inv.add("value", whole)
|
|
|
|
|
|
_HTML = ("cell", "heading", "image", "list_item", "paragraph", "table")
|
|
_HTML_KIND = {
|
|
"p": "paragraph",
|
|
"li": "list_item",
|
|
"table": "table",
|
|
"td": "cell",
|
|
"th": "cell",
|
|
**{f"h{n}": "heading" for n in range(1, 7)},
|
|
}
|
|
_HTML_VOID = {"img", "br", "hr", "meta", "link", "input", "col", "area", "base", "wbr", "source"}
|
|
|
|
|
|
class _HtmlInventory(HTMLParser):
|
|
def __init__(self, inv: Inventory, inbox: Path, path: Path) -> None:
|
|
super().__init__(convert_charrefs=True)
|
|
self.inv, self.inbox, self.path = inv, inbox, path
|
|
self.open: list[tuple[str, list[int]]] = []
|
|
self.skip = 0
|
|
|
|
def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None:
|
|
if tag == "img":
|
|
_add_image(self.inv, self.inbox, self.path, dict(attrs).get("src") or "", sts=False)
|
|
return
|
|
if tag in ("script", "style"):
|
|
self.skip += 1
|
|
if tag not in _HTML_VOID:
|
|
self.open.append((tag, []))
|
|
|
|
def handle_endtag(self, tag: str) -> None:
|
|
if tag in ("script", "style") and self.skip:
|
|
self.skip -= 1
|
|
for position in range(len(self.open) - 1, -1, -1):
|
|
if self.open[position][0] == tag:
|
|
break
|
|
else:
|
|
return
|
|
while len(self.open) > position:
|
|
self._close(*self.open.pop())
|
|
|
|
def _close(self, tag: str, pieces: list[int]) -> None:
|
|
kind = _HTML_KIND.get(tag)
|
|
if kind is not None:
|
|
self.inv.add(kind, pieces)
|
|
if self.open:
|
|
self.open[-1][1].extend(pieces)
|
|
|
|
def handle_data(self, data: str) -> None:
|
|
if self.skip or not data.strip():
|
|
return
|
|
if self.open:
|
|
self.open[-1][1].append(self.inv.chunk(data))
|
|
|
|
def finish(self) -> None:
|
|
self.close()
|
|
while self.open:
|
|
self._close(*self.open.pop())
|
|
|
|
|
|
# xml / sts
|
|
_STS = (
|
|
"cell",
|
|
"footnote",
|
|
"image",
|
|
"list_item",
|
|
"paragraph",
|
|
"section",
|
|
"section_label",
|
|
"table",
|
|
"table_label",
|
|
"title",
|
|
)
|
|
_STS_KIND = {
|
|
"sec": "section",
|
|
"p": "paragraph",
|
|
"table-wrap": "table",
|
|
"td": "cell",
|
|
"th": "cell",
|
|
"list-item": "list_item",
|
|
"fn": "footnote",
|
|
}
|
|
|
|
|
|
def _bare(tag: str) -> str:
|
|
return tag.rpartition("}")[2]
|
|
|
|
|
|
def _inventory_xml(inv: Inventory, inbox: Path, path: Path, data: bytes) -> None:
|
|
if b"<!DOCTYPE" in data:
|
|
raise ValueError("a DOCTYPE is not read")
|
|
root = ElementTree.fromstring(data)
|
|
sts = _bare(root.tag) == "standard" or any(_bare(e.tag) == "sec" for e in root.iter())
|
|
if not sts:
|
|
inv.vocabulary = ("element",)
|
|
inv.add("element", (inv.chunk("".join(root.itertext())),))
|
|
return
|
|
|
|
def walk(node: ElementTree.Element, parent: str) -> list[int]:
|
|
tag = _bare(node.tag)
|
|
pieces: list[int] = []
|
|
if node.text and node.text.strip():
|
|
pieces.append(inv.chunk(node.text))
|
|
if tag in ("graphic", "inline-graphic"):
|
|
href = next((v for k, v in node.attrib.items() if _bare(k) == "href"), "")
|
|
_add_image(inv, inbox, path, href, sts=True)
|
|
for child in node:
|
|
pieces.extend(walk(child, tag))
|
|
if child.tail and child.tail.strip():
|
|
pieces.append(inv.chunk(child.tail))
|
|
kind = _STS_KIND.get(tag)
|
|
if tag == "title" and parent == "sec":
|
|
kind = "title"
|
|
elif tag == "label" and parent == "sec":
|
|
kind = "section_label"
|
|
elif tag == "label" and parent == "table-wrap":
|
|
kind = "table_label"
|
|
if kind is not None:
|
|
inv.add(kind, pieces)
|
|
return pieces
|
|
|
|
walk(root, "")
|
|
|
|
|
|
# office containers
|
|
_WNS = "{http://schemas.openxmlformats.org/wordprocessingml/2006/main}"
|
|
_ANS = "{http://schemas.openxmlformats.org/drawingml/2006/main}"
|
|
_PNS = "{http://schemas.openxmlformats.org/presentationml/2006/main}"
|
|
_SNS = "{http://schemas.openxmlformats.org/spreadsheetml/2006/main}"
|
|
_XDRNS = "{http://schemas.openxmlformats.org/drawingml/2006/spreadsheetDrawing}"
|
|
_TNS = "{urn:oasis:names:tc:opendocument:xmlns:text:1.0}"
|
|
_TBNS = "{urn:oasis:names:tc:opendocument:xmlns:table:1.0}"
|
|
_DNS = "{urn:oasis:names:tc:opendocument:xmlns:drawing:1.0}"
|
|
|
|
_DOCX = ("cell", "footnote", "heading", "image", "paragraph", "table")
|
|
_HEADING_STYLES = re.compile(r"(?i)^(?:heading|overskrift|title|tittel)\s*\d*$")
|
|
|
|
|
|
def _joined(node: ElementTree.Element, text_tag: str) -> str:
|
|
return "".join(t.text or "" for t in node.iter(text_tag))
|
|
|
|
|
|
def _docx_lines(para: ElementTree.Element) -> list[str]:
|
|
"""A paragraph's text, split where it breaks a line (`w:br`, `w:cr`).
|
|
|
|
A break is a line boundary in the source, and the converter keeps it one:
|
|
inside a grid-table cell the two halves land on different rows, with other
|
|
cells' text between them (measured on K2).
|
|
"""
|
|
lines = [""]
|
|
for node in para.iter():
|
|
if node.tag == f"{_WNS}t":
|
|
lines[-1] += node.text or ""
|
|
elif node.tag in (f"{_WNS}br", f"{_WNS}cr"):
|
|
lines.append("")
|
|
return lines
|
|
|
|
|
|
def _inventory_docx(inv: Inventory, data: bytes) -> None:
|
|
with zipfile.ZipFile(io.BytesIO(data)) as archive:
|
|
body = ElementTree.fromstring(archive.read("word/document.xml"))
|
|
notes = (
|
|
ElementTree.fromstring(archive.read("word/footnotes.xml"))
|
|
if "word/footnotes.xml" in archive.namelist()
|
|
else None
|
|
)
|
|
paragraph_pieces: dict[int, list[int]] = {}
|
|
for para in body.iter(f"{_WNS}p"):
|
|
pieces = [inv.chunk(line) for line in _docx_lines(para) if line.strip()]
|
|
paragraph_pieces[id(para)] = pieces
|
|
style = para.find(f"{_WNS}pPr/{_WNS}pStyle")
|
|
if style is not None and _HEADING_STYLES.match(style.get(f"{_WNS}val", "")):
|
|
inv.add("heading", pieces)
|
|
elif pieces:
|
|
inv.add("paragraph", pieces)
|
|
for kind, tag in (("table", "tbl"), ("cell", "tc")):
|
|
for node in body.iter(f"{_WNS}{tag}"):
|
|
inv.add(
|
|
kind,
|
|
[c for p in node.iter(f"{_WNS}p") for c in paragraph_pieces.get(id(p), [])],
|
|
)
|
|
for _ in body.iter(f"{_ANS}blip"):
|
|
inv.refs.append(("embedded", None))
|
|
inv.add("image")
|
|
if notes is not None:
|
|
for note in notes.iter(f"{_WNS}footnote"):
|
|
if int(note.get(f"{_WNS}id", "0")) > 0:
|
|
text = _joined(note, f"{_WNS}t")
|
|
inv.add("footnote", (inv.chunk(text),) if text.strip() else ())
|
|
|
|
|
|
_PPTX = ("cell", "image", "paragraph", "slide", "table", "title")
|
|
_SLIDE = re.compile(r"ppt/slides/slide\d+\.xml")
|
|
|
|
|
|
def _inventory_pptx(inv: Inventory, data: bytes) -> None:
|
|
with zipfile.ZipFile(io.BytesIO(data)) as archive:
|
|
slides = [
|
|
ElementTree.fromstring(archive.read(n))
|
|
for n in archive.namelist()
|
|
if _SLIDE.fullmatch(n)
|
|
]
|
|
for slide in slides:
|
|
pieces: list[int] = []
|
|
for frame in slide.iter(f"{_ANS}tbl"):
|
|
cells: list[int] = []
|
|
for cell in frame.iter(f"{_ANS}tc"):
|
|
text = _joined(cell, f"{_ANS}t")
|
|
own = (inv.chunk(text),) if text.strip() else ()
|
|
inv.add("cell", own)
|
|
cells.extend(own)
|
|
inv.add("table", cells)
|
|
pieces.extend(cells)
|
|
for shape in slide.iter(f"{_PNS}sp"):
|
|
holder = shape.find(f"{_PNS}nvSpPr/{_PNS}nvPr/{_PNS}ph")
|
|
texts = [
|
|
inv.chunk(t)
|
|
for t in (_joined(p, f"{_ANS}t") for p in shape.iter(f"{_ANS}p"))
|
|
if t.strip()
|
|
]
|
|
if holder is not None and holder.get("type") in ("title", "ctrTitle") and texts:
|
|
inv.add("title", texts)
|
|
else:
|
|
for piece in texts:
|
|
inv.add("paragraph", (piece,))
|
|
pieces.extend(texts)
|
|
for _ in slide.iter(f"{_PNS}pic"):
|
|
inv.refs.append(("embedded", None))
|
|
inv.add("image")
|
|
inv.add("slide", pieces)
|
|
|
|
|
|
_XLSX = ("cell", "image", "row", "sheet")
|
|
_SHEET = re.compile(r"xl/worksheets/sheet\d+\.xml")
|
|
_DRAWING = re.compile(r"xl/drawings/drawing\d+\.xml")
|
|
|
|
|
|
def _inventory_xlsx(inv: Inventory, data: bytes) -> None:
|
|
with zipfile.ZipFile(io.BytesIO(data)) as archive:
|
|
names = archive.namelist()
|
|
shared: list[str] = []
|
|
if "xl/sharedStrings.xml" in names:
|
|
table = ElementTree.fromstring(archive.read("xl/sharedStrings.xml"))
|
|
shared = ["".join(t.text or "" for t in si.iter(f"{_SNS}t")) for si in table]
|
|
sheets = [ElementTree.fromstring(archive.read(n)) for n in names if _SHEET.fullmatch(n)]
|
|
drawings = [ElementTree.fromstring(archive.read(n)) for n in names if _DRAWING.fullmatch(n)]
|
|
for sheet in sheets:
|
|
sheet_pieces: list[int] = []
|
|
for row in sheet.iter(f"{_SNS}row"):
|
|
row_pieces: list[int] = []
|
|
valued = False
|
|
for cell in row.iter(f"{_SNS}c"):
|
|
value = cell.find(f"{_SNS}v")
|
|
inline = cell.find(f"{_SNS}is")
|
|
if value is None and inline is None:
|
|
continue
|
|
valued = True
|
|
if inline is not None:
|
|
text = "".join(t.text or "" for t in inline.iter(f"{_SNS}t"))
|
|
elif cell.get("t") == "s" and value is not None and value.text is not None:
|
|
text = shared[int(value.text)]
|
|
else:
|
|
text = (value.text or "") if value is not None else ""
|
|
own = (inv.chunk(text),) if text.strip() else ()
|
|
inv.add("cell", own)
|
|
row_pieces.extend(own)
|
|
if valued:
|
|
inv.add("row", row_pieces)
|
|
sheet_pieces.extend(row_pieces)
|
|
inv.add("sheet", sheet_pieces)
|
|
for drawing in drawings:
|
|
for _ in drawing.iter(f"{_XDRNS}pic"):
|
|
inv.refs.append(("embedded", None))
|
|
inv.add("image")
|
|
|
|
|
|
_ODT = ("cell", "heading", "image", "list_item", "paragraph", "table")
|
|
|
|
|
|
def _inventory_odt(inv: Inventory, data: bytes) -> None:
|
|
with zipfile.ZipFile(io.BytesIO(data)) as archive:
|
|
content = ElementTree.fromstring(archive.read("content.xml"))
|
|
pieces: dict[int, int] = {}
|
|
in_cells: set[int] = set()
|
|
for cell in content.iter(f"{_TBNS}table-cell"):
|
|
in_cells.update(id(p) for p in cell.iter(f"{_TNS}p"))
|
|
for tag in (f"{_TNS}p", f"{_TNS}h"):
|
|
for node in content.iter(tag):
|
|
text = "".join(node.itertext())
|
|
if text.strip():
|
|
pieces[id(node)] = inv.chunk(text)
|
|
|
|
def within(node: ElementTree.Element) -> list[int]:
|
|
return [pieces[id(n)] for n in node.iter() if id(n) in pieces]
|
|
|
|
for node in content.iter(f"{_TNS}h"):
|
|
inv.add("heading", within(node))
|
|
for node in content.iter(f"{_TNS}p"):
|
|
if id(node) not in in_cells and id(node) in pieces:
|
|
inv.add("paragraph", (pieces[id(node)],))
|
|
for kind, tag in (
|
|
("table", f"{_TBNS}table"),
|
|
("cell", f"{_TBNS}table-cell"),
|
|
("list_item", f"{_TNS}list-item"),
|
|
):
|
|
for node in content.iter(tag):
|
|
inv.add(kind, within(node))
|
|
for _ in content.iter(f"{_DNS}image"):
|
|
inv.refs.append(("embedded", None))
|
|
inv.add("image")
|
|
|
|
|
|
_RTF = ("cell", "image", "paragraph", "table_row")
|
|
_RTF_TOKEN = re.compile(
|
|
r"\\([a-zA-Z]+)(-?\d+)? ?|\\'([0-9a-fA-F]{2})|\\([^a-zA-Z])|([{}])|([^\\{}\r\n]+)|[\r\n]"
|
|
)
|
|
_RTF_SKIP = {"fonttbl", "colortbl", "stylesheet", "info", "pict", "header", "footer", "*"}
|
|
|
|
|
|
def _inventory_rtf(inv: Inventory, data: bytes) -> None:
|
|
"""Paragraph: `\\par`. Table row: `\\row`. Cell: `\\cell`. Image: `\\pict`.
|
|
|
|
The text between two such words is the element's text: `\\'hh` is read as
|
|
cp1252, `\\uN` as the code point N with its one fallback character
|
|
skipped, and destination groups (font table, pictures, ...) carry none.
|
|
"""
|
|
raw = data.decode("latin-1")
|
|
depth = 0
|
|
skip_below: int | None = None
|
|
skip_fallback = 0
|
|
buffer: list[str] = []
|
|
row: list[int] = []
|
|
for match in _RTF_TOKEN.finditer(raw):
|
|
word, number, hexa, symbol, brace, text = match.groups()
|
|
if brace == "{":
|
|
depth += 1
|
|
continue
|
|
if brace == "}":
|
|
if skip_below is not None and depth <= skip_below:
|
|
skip_below = None
|
|
depth -= 1
|
|
continue
|
|
if word == "pict":
|
|
inv.refs.append(("embedded", None))
|
|
inv.add("image")
|
|
if skip_below is not None:
|
|
continue
|
|
if word in _RTF_SKIP or symbol == "*":
|
|
skip_below = depth
|
|
continue
|
|
if word == "u" and number is not None:
|
|
value = int(number)
|
|
buffer.append(chr(value + 65536 if value < 0 else value))
|
|
skip_fallback = 1
|
|
continue
|
|
if hexa is not None:
|
|
if skip_fallback:
|
|
skip_fallback -= 1
|
|
else:
|
|
buffer.append(bytes([int(hexa, 16)]).decode("cp1252", errors="replace"))
|
|
continue
|
|
if text is not None:
|
|
if skip_fallback:
|
|
text = text[1:]
|
|
skip_fallback = 0
|
|
buffer.append(text)
|
|
continue
|
|
if symbol is not None:
|
|
buffer.append({"~": " ", "-": "", "_": "-"}.get(symbol, symbol))
|
|
continue
|
|
if word in ("par", "cell", "row"):
|
|
piece = inv.chunk("".join(buffer))
|
|
buffer = []
|
|
if word == "par":
|
|
inv.add("paragraph", (piece,))
|
|
elif word == "cell":
|
|
inv.add("cell", (piece,))
|
|
row.append(piece)
|
|
else:
|
|
inv.add("table_row", row)
|
|
row = []
|
|
|
|
|
|
_PDF = ("image", "page")
|
|
|
|
|
|
def _inventory_pdf(inv: Inventory, data: bytes) -> None:
|
|
"""Page: one per page. Image: one per placement on a page, as pdfplumber
|
|
lists them. A page's own text is not inventoried here: a PDF without a
|
|
structure tree declares no headings or paragraphs, which is one of the
|
|
gate's proposed exceptions, and its pages are checked through the images
|
|
they carry and the document's persist outcome alone."""
|
|
import pdfplumber
|
|
|
|
with pdfplumber.open(io.BytesIO(data)) as pdf:
|
|
for page in pdf.pages:
|
|
inv.add("page")
|
|
for _ in page.images:
|
|
inv.refs.append(("embedded", None))
|
|
inv.add("image")
|
|
page.close()
|
|
|
|
|
|
Reader = Callable[[Inventory, Path, Path, bytes], None]
|
|
|
|
|
|
def _text_reader(function: Callable[[Inventory, str], None]) -> Reader:
|
|
def read(inv: Inventory, inbox: Path, path: Path, data: bytes) -> None:
|
|
function(inv, data.decode("utf-8-sig"))
|
|
|
|
return read
|
|
|
|
|
|
def _bytes_reader(function: Callable[[Inventory, bytes], None]) -> Reader:
|
|
def read(inv: Inventory, inbox: Path, path: Path, data: bytes) -> None:
|
|
function(inv, data)
|
|
|
|
return read
|
|
|
|
|
|
def _markdown(inv: Inventory, inbox: Path, path: Path, data: bytes) -> None:
|
|
_inventory_markdown(inv, inbox, path, data.decode("utf-8-sig"))
|
|
|
|
|
|
def _html(inv: Inventory, inbox: Path, path: Path, data: bytes) -> None:
|
|
parser = _HtmlInventory(inv, inbox, path)
|
|
parser.feed(data.decode("utf-8-sig"))
|
|
parser.finish()
|
|
|
|
|
|
_READERS: dict[str, tuple[tuple[str, ...], Reader]] = {
|
|
".md": (_MARKDOWN, _markdown),
|
|
".txt": (_TEXT, _text_reader(_inventory_text)),
|
|
".csv": (_CSV, _text_reader(_inventory_csv)),
|
|
".json": (_JSON, _text_reader(_inventory_json)),
|
|
".html": (_HTML, _html),
|
|
".htm": (_HTML, _html),
|
|
".xml": (_STS, _inventory_xml),
|
|
".docx": (_DOCX, _bytes_reader(_inventory_docx)),
|
|
".pptx": (_PPTX, _bytes_reader(_inventory_pptx)),
|
|
".xlsx": (_XLSX, _bytes_reader(_inventory_xlsx)),
|
|
".odt": (_ODT, _bytes_reader(_inventory_odt)),
|
|
".rtf": (_RTF, _bytes_reader(_inventory_rtf)),
|
|
".pdf": (_PDF, _bytes_reader(_inventory_pdf)),
|
|
}
|
|
|
|
|
|
def inventory(inbox: Path, path: Path, data: bytes | None = None) -> Inventory:
|
|
"""What the source file at `path` holds. A file of a type this module does
|
|
not inventory has an empty vocabulary; one it cannot read raises."""
|
|
reader = _READERS.get(path.suffix.lower())
|
|
inv = Inventory(relative_source(path, inbox), reader[0] if reader else ())
|
|
if reader is not None:
|
|
reader[1](inv, inbox, path, path.read_bytes() if data is None else data)
|
|
return inv
|
|
|
|
|
|
# --- the account -------------------------------------------------------------
|
|
|
|
|
|
@dataclass
|
|
class Fate:
|
|
carried: int = 0
|
|
pointer: int = 0
|
|
rejected: dict[str, int] = field(default_factory=dict)
|
|
|
|
@property
|
|
def total(self) -> int:
|
|
return self.carried + self.pointer + sum(self.rejected.values())
|
|
|
|
def to_json(self) -> dict[str, Any]:
|
|
return {
|
|
CARRIED: self.carried,
|
|
POINTER: self.pointer,
|
|
REJECTED: dict(sorted(self.rejected.items())),
|
|
}
|
|
|
|
|
|
@dataclass
|
|
class DocumentAccount:
|
|
source_file: str
|
|
status: str
|
|
code: str | None
|
|
counts: dict[str, int]
|
|
fates: dict[str, Fate]
|
|
error: str | None = None
|
|
#: The `(source digest, asset digest)` pairs for the images this run
|
|
#: REWROTE -- a BMP that reaches the bundle as a PNG, today's only case.
|
|
#: Booked because the bundle states the same pairs only as prose on a
|
|
#: pointer's second line, and measured by PM 2026-09-19 an ordinary HTML
|
|
#: document with two `<p>` elements writes exactly that prose. A reader
|
|
#: proving a conversion off the bundle text is reading the document; this
|
|
#: is the same fact written by the run.
|
|
conversions: tuple[tuple[str, str], ...] = ()
|
|
#: How many U+00AD the normalisation door removed from this document's
|
|
#: text before the persist gate saw it (operator decision 2026-09-18).
|
|
#: Booked rather than silently applied: a door that changes a source's
|
|
#: bytes and says nothing is the same class of absence this whole module
|
|
#: exists to close.
|
|
normalised_soft_hyphen: int = 0
|
|
|
|
@property
|
|
def unaccounted(self) -> dict[str, int]:
|
|
return {
|
|
kind: count - self.fates[kind].total
|
|
for kind, count in self.counts.items()
|
|
if count > self.fates[kind].total
|
|
}
|
|
|
|
@property
|
|
def double_booked(self) -> dict[str, int]:
|
|
return {
|
|
kind: self.fates[kind].total - count
|
|
for kind, count in self.counts.items()
|
|
if self.fates[kind].total > count
|
|
}
|
|
|
|
@property
|
|
def total(self) -> int:
|
|
return sum(self.counts.values())
|
|
|
|
def to_json(self) -> dict[str, Any]:
|
|
entry: dict[str, Any] = {
|
|
"source_file": self.source_file,
|
|
"status": self.status,
|
|
"code": self.code,
|
|
"normalised_soft_hyphen": self.normalised_soft_hyphen,
|
|
"conversions": [{"from": before, "to": after} for before, after in self.conversions],
|
|
"inventory": dict(self.counts),
|
|
"fates": {kind: self.fates[kind].to_json() for kind in self.counts},
|
|
"unaccounted": self.unaccounted,
|
|
"double_booked": self.double_booked,
|
|
}
|
|
if self.error is not None:
|
|
entry["inventory_error"] = self.error
|
|
return entry
|
|
|
|
|
|
@dataclass
|
|
class FileAccount:
|
|
source_file: str
|
|
fate: str
|
|
code: str | None
|
|
|
|
def to_json(self) -> dict[str, Any]:
|
|
return {"source_file": self.source_file, "fate": self.fate, "code": self.code}
|
|
|
|
|
|
@dataclass
|
|
class Accounting:
|
|
documents: list[DocumentAccount]
|
|
files: list[FileAccount]
|
|
|
|
@property
|
|
def unaccounted(self) -> int:
|
|
return sum(sum(d.unaccounted.values()) for d in self.documents) + sum(
|
|
1 for f in self.files if not f.fate
|
|
)
|
|
|
|
@property
|
|
def double_booked(self) -> int:
|
|
return sum(sum(d.double_booked.values()) for d in self.documents)
|
|
|
|
@property
|
|
def refused(self) -> int:
|
|
"""Documents the build read and persisted NOTHING of.
|
|
|
|
Their elements are booked honestly, as coded rejections, so neither
|
|
`unaccounted` nor `double_booked` moves and the loss is invisible in
|
|
the two numbers a reader looks at. A run that persisted at least one
|
|
document exits 0 -- the exit code belongs to the whole run, and a
|
|
corpus holding one unreadable file is the ordinary case -- so this
|
|
count is what keeps a partial refusal from being silent (H1).
|
|
"""
|
|
return sum(1 for d in self.documents if d.status == REJECTED)
|
|
|
|
@property
|
|
def normalised_soft_hyphen(self) -> int:
|
|
"""Soft hyphens the normalisation door removed across the whole run."""
|
|
return sum(d.normalised_soft_hyphen for d in self.documents)
|
|
|
|
@property
|
|
def images_found(self) -> int:
|
|
return sum(d.counts.get("image", 0) for d in self.documents)
|
|
|
|
@property
|
|
def elements(self) -> int:
|
|
return sum(d.total for d in self.documents)
|
|
|
|
def to_json(self) -> dict[str, Any]:
|
|
return {
|
|
"accounting_version": ACCOUNTING_VERSION,
|
|
"unaccounted": self.unaccounted,
|
|
"double_booked": self.double_booked,
|
|
"refused": self.refused,
|
|
"normalised_soft_hyphen": self.normalised_soft_hyphen,
|
|
"documents": [d.to_json() for d in self.documents],
|
|
"files": [f.to_json() for f in self.files],
|
|
}
|
|
|
|
def log_lines(self) -> list[str]:
|
|
lines = [
|
|
f"* **Accounting**: {len(self.documents)} document(s) and {len(self.files)} "
|
|
f"other file(s); {self.elements} elements found in the sources; "
|
|
f"{self.unaccounted} unaccounted, {self.double_booked} double-booked; "
|
|
f"{self.refused} of {len(self.documents)} document(s) refused whole."
|
|
]
|
|
touched = sum(1 for d in self.documents if d.normalised_soft_hyphen)
|
|
lines.append(
|
|
f"* **Normalisation**: {self.normalised_soft_hyphen} soft hyphen(s) (U+00AD) "
|
|
f"removed from {touched} of {len(self.documents)} document(s) before the persist "
|
|
"gate. No other character is touched. The count is the door's own, read off the "
|
|
"run and not recounted from the source."
|
|
)
|
|
for doc in self.documents:
|
|
if doc.status == REJECTED:
|
|
lines.append(
|
|
f"* {doc.source_file}: {doc.total} elements found in the source, 0 carried: "
|
|
f"document rejected `{doc.code}`."
|
|
)
|
|
for kind, count in sorted(doc.unaccounted.items()):
|
|
lines.append(f"* {doc.source_file}: {kind} {count} unaccounted.")
|
|
for kind, count in sorted(doc.double_booked.items()):
|
|
lines.append(f"* {doc.source_file}: {kind} {count} double-booked.")
|
|
return lines
|
|
|
|
|
|
def _bodies(paths: Sequence[Path]) -> str:
|
|
parts = []
|
|
for path in paths:
|
|
text = path.read_text(encoding="utf-8")
|
|
if text.startswith("---\n"):
|
|
end = text.find("\n---\n", 4)
|
|
text = text[end + 5 :] if end >= 0 else text
|
|
parts.append(text)
|
|
return "\n".join(parts)
|
|
|
|
|
|
#: A converter attribute block such as `{.mark}` or `{#slide-1}`. Its letters
|
|
#: stand between the words of text that WAS carried (`[Sted]{.mark}`), so it is
|
|
#: removed before comparing -- measured on K2, where a highlighted placeholder
|
|
#: otherwise read as lost.
|
|
_CONVERTER_ATTRIBUTE = re.compile(r"\{[#.][^{}\n]*\}")
|
|
|
|
|
|
class _Finder:
|
|
"""Is a piece of text in the carried text? Searched from where the last
|
|
piece was found first, because readers keep the source order."""
|
|
|
|
def __init__(self, text: str) -> None:
|
|
self.text = _norm(_CONVERTER_ATTRIBUTE.sub("", text))
|
|
self.cursor = 0
|
|
|
|
def __call__(self, piece: str) -> bool:
|
|
needle = _norm(piece)
|
|
if not needle:
|
|
return True
|
|
at = self.text.find(needle, self.cursor)
|
|
if at < 0:
|
|
at = self.text.find(needle)
|
|
if at < 0:
|
|
return False
|
|
self.cursor = at + len(needle)
|
|
return True
|
|
|
|
|
|
def _book_images(
|
|
inv: Inventory, fate: Fate, assets: DocumentAssets | None, found: Callable[[str], bool]
|
|
) -> None:
|
|
placements = [chunks for kind, chunks in inv.elements if kind == "image"]
|
|
handled = assets is not None and (assets.carried or assets.rejected)
|
|
if assets is not None and handled:
|
|
fate.carried += assets.carried
|
|
for rejection in assets.rejected:
|
|
if rejection.code == _REMOTE_CODE:
|
|
fate.pointer += 1
|
|
else:
|
|
fate.rejected[rejection.code] = fate.rejected.get(rejection.code, 0) + 1
|
|
return
|
|
# The reader did not handle images for this document: a reference kept in
|
|
# the carried text is a pointer; anything else has no fate.
|
|
for chunks in placements:
|
|
if chunks and all(found(inv.chunks[c]) for c in chunks):
|
|
fate.pointer += 1
|
|
|
|
|
|
def _persisted_account(
|
|
inv: Inventory, bodies: str, assets: DocumentAssets | None
|
|
) -> DocumentAccount:
|
|
finder = _Finder(bodies)
|
|
seen: dict[int, bool] = {}
|
|
|
|
def found(index: int) -> bool:
|
|
if index not in seen:
|
|
seen[index] = finder(inv.chunks[index])
|
|
return seen[index]
|
|
|
|
counts = inv.counts()
|
|
fates = {kind: Fate() for kind in counts}
|
|
for kind, chunks in inv.elements:
|
|
if kind == "image":
|
|
continue
|
|
if all(found(c) for c in chunks):
|
|
fates[kind].carried += 1
|
|
if "image" in fates:
|
|
_book_images(inv, fates["image"], assets, finder)
|
|
return DocumentAccount(
|
|
inv.source_file,
|
|
PERSISTED,
|
|
None,
|
|
counts,
|
|
fates,
|
|
conversions=() if assets is None else assets.conversions,
|
|
)
|
|
|
|
|
|
def _refused_account(inv: Inventory, code: str) -> DocumentAccount:
|
|
counts = inv.counts()
|
|
fates = {kind: Fate(rejected={code: n} if n else {}) for kind, n in counts.items()}
|
|
return DocumentAccount(inv.source_file, REJECTED, code, counts, fates)
|
|
|
|
|
|
def account_run(inbox: Path, walked: Sequence[Path], result: InboxResult) -> Accounting:
|
|
"""Every walked file's account after one `process_inbox` run."""
|
|
persisted = {item.source_file for item in result.persisted}
|
|
concepts: dict[str, list[Path]] = {}
|
|
for item in result.concepts:
|
|
concepts.setdefault(item.source_file, []).append(item.path)
|
|
refused = {item.source_file: item.disposition for item in result.quarantined + result.rejected}
|
|
failed = {item.source_file: item.error.code for item in result.failed}
|
|
assets = {item.source_file: item for item in result.document_assets}
|
|
carried = set(result.carried_files)
|
|
documents: list[DocumentAccount] = []
|
|
files: list[FileAccount] = []
|
|
for path in walked:
|
|
name = relative_source(path, inbox)
|
|
if path.suffix.lower() not in _READERS:
|
|
if name in carried:
|
|
files.append(FileAccount(name, CARRIED, None))
|
|
elif name in persisted:
|
|
files.append(FileAccount(name, MERGED, None))
|
|
else:
|
|
files.append(
|
|
FileAccount(name, REJECTED if name in failed else "", failed.get(name))
|
|
)
|
|
continue
|
|
try:
|
|
inv = inventory(inbox, path)
|
|
# Broad on purpose, and it diagnoses nothing: the row carries the
|
|
# exception's own type and text, never a guessed cause.
|
|
except Exception as exc:
|
|
error = f"{type(exc).__name__}: {exc}"
|
|
code = refused.get(name) or failed.get(name) or "inventory_unreadable"
|
|
documents.append(
|
|
DocumentAccount(
|
|
name,
|
|
REJECTED if name not in persisted else PERSISTED,
|
|
code,
|
|
{},
|
|
{},
|
|
error=error,
|
|
)
|
|
)
|
|
continue
|
|
if name in persisted:
|
|
documents.append(
|
|
_persisted_account(inv, _bodies(concepts.get(name, [])), assets.get(name))
|
|
)
|
|
elif name in refused:
|
|
documents.append(_refused_account(inv, refused[name]))
|
|
elif name in failed:
|
|
documents.append(_refused_account(inv, failed[name]))
|
|
else:
|
|
documents.append(
|
|
DocumentAccount(name, "", None, inv.counts(), {k: Fate() for k in inv.counts()})
|
|
)
|
|
# Read off the run, never recounted from the source: a second count would
|
|
# be a second reader, and the number this publishes has to be the number
|
|
# the normalisation door acted on. It is also the one number here with no
|
|
# independent denominator behind it, and the door acts on the EXTRACTED
|
|
# text -- so a second counter over the source bytes would disagree by
|
|
# construction on every type extraction does not carry verbatim. The
|
|
# provenance is stated in `log_lines` instead of hidden.
|
|
removed = {item.source_file: item.soft_hyphens for item in result.normalised}
|
|
for document in documents:
|
|
document.normalised_soft_hyphen = removed.get(document.source_file, 0)
|
|
return Accounting(documents, files)
|
|
|
|
|
|
def summary(accounting: Accounting) -> Mapping[str, int]:
|
|
return {
|
|
"documents": len(accounting.documents),
|
|
"files": len(accounting.files),
|
|
"elements": accounting.elements,
|
|
"unaccounted": accounting.unaccounted,
|
|
"double_booked": accounting.double_booked,
|
|
}
|