torbenandClaude Code
a581271165
Files changed: - CHANGES.md - VERSION - docs/how-to/ingest-a-source.md - docs/reference/commands/raw-material-and-uploads.md - raw/CONTRACT.md - tools/CONTRACT.md - tools/chemenu/commands/raw/derive.py - tools/chemenu/filters/pdf.py - tools/chemenu/filters/pdf_layout.py - tools/chemenu/tests/binary_formats.py - tools/chemenu/tests/test_raw_derive_binary.py Co-Authored-By: Claude Code <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01SnAJ7Z3CpVD3PRbN73QtU2
402 lines
17 KiB
Python
402 lines
17 KiB
Python
"""Builders for the two binary formats `raw derive` reads with a package: a PDF and an
|
|
Outlook `.msg` (a compound file). Built here, byte by byte, so no binary sits in the
|
|
repository and every fixture is readable in the test that uses it. Whether each one is a
|
|
valid document is settled by the package that reads it in the tests."""
|
|
from __future__ import annotations
|
|
|
|
import struct
|
|
from dataclasses import dataclass, field
|
|
from typing import Optional, Union
|
|
|
|
END, FREE, FATSECT = 0xFFFFFFFE, 0xFFFFFFFF, 0xFFFFFFFD
|
|
|
|
|
|
def build_pdf(pages: list[list[str]], *, title: Optional[str] = None) -> bytes:
|
|
"""A PDF with one text line per list item; an empty page is an empty content stream."""
|
|
objects: dict[int, bytes] = {}
|
|
page_ids = [3 + 2 * n for n in range(len(pages))]
|
|
font_id = 3 + 2 * len(pages)
|
|
info_id = font_id + 1
|
|
objects[1] = b"<< /Type /Catalog /Pages 2 0 R >>"
|
|
kids = " ".join(f"{page} 0 R" for page in page_ids)
|
|
objects[2] = f"<< /Type /Pages /Kids [{kids}] /Count {len(pages)} >>".encode()
|
|
for page_id, lines in zip(page_ids, pages):
|
|
ops = ["BT /F1 12 Tf 72 720 Td 14 TL"]
|
|
for line in lines:
|
|
escaped = line.replace("\\", "\\\\").replace("(", "\\(").replace(")", "\\)")
|
|
ops.append(f"({escaped}) Tj T*")
|
|
ops.append("ET")
|
|
stream = "\n".join(ops).encode("latin-1") if lines else b""
|
|
objects[page_id] = (
|
|
f"<< /Type /Page /Parent 2 0 R /MediaBox [0 0 612 792] /Contents {page_id + 1} 0 R "
|
|
f"/Resources << /Font << /F1 {font_id} 0 R >> >> >>"
|
|
).encode()
|
|
objects[page_id + 1] = b"<< /Length %d >>\nstream\n" % len(stream) + stream + b"\nendstream"
|
|
objects[font_id] = b"<< /Type /Font /Subtype /Type1 /BaseFont /Helvetica >>"
|
|
trailer = "/Root 1 0 R"
|
|
if title is not None:
|
|
objects[info_id] = f"<< /Title ({title}) >>".encode("latin-1")
|
|
trailer += f" /Info {info_id} 0 R"
|
|
out = bytearray(b"%PDF-1.4\n")
|
|
offsets = {}
|
|
for number in sorted(objects):
|
|
offsets[number] = len(out)
|
|
out += f"{number} 0 obj\n".encode() + objects[number] + b"\nendobj\n"
|
|
xref = len(out)
|
|
size = max(objects) + 1
|
|
out += f"xref\n0 {size}\n".encode() + b"0000000000 65535 f \n"
|
|
for number in range(1, size):
|
|
out += f"{offsets[number]:010d} 00000 n \n".encode()
|
|
out += f"trailer\n<< /Size {size} {trailer} >>\nstartxref\n{xref}\n%%EOF\n".encode()
|
|
return bytes(out)
|
|
|
|
|
|
Run = Union[str, tuple[str, dict]] # text, or text and options: color, highlight, strike, link, bold, italic, size, rotate
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Picture:
|
|
"""An image drawn at `x`, `y` (lower left) in `width` x `height` points from a
|
|
`pixels` x `pixels` raster; `gray` makes the pixel content, so equal grays are the same image."""
|
|
|
|
x: float
|
|
y: float
|
|
width: float
|
|
height: float
|
|
pixels: int = 8
|
|
gray: int = 0x80
|
|
|
|
|
|
@dataclass
|
|
class StyledPage:
|
|
"""Lines of runs. A run's options: `color` (a gray level, an RGB or a CMYK tuple),
|
|
`highlight` (an RGB filled behind the text), `strike` (a line through it), `link`
|
|
(a /Link annotation over it), `bold`, `italic`, `size` (points, default 10) and
|
|
`rotate` (90: the text runs upwards; no highlight or strike then).
|
|
|
|
`ops` are raw content-stream operators drawn before the text (rectangles, lines,
|
|
paths), `pictures` are drawn after it, `image` is the 1 x 1 pixel picture of the
|
|
first tests, and `size` is the page size in points."""
|
|
|
|
lines: list[list[Run]] = field(default_factory=list)
|
|
image: bool = False
|
|
ops: list[str] = field(default_factory=list)
|
|
pictures: list[Picture] = field(default_factory=list)
|
|
size: tuple[float, float] = (612, 792)
|
|
|
|
|
|
CHAR_WIDTH, LINE_HEIGHT, LEFT, TOP = 6, 14, 72, 720 # Courier 10 pt
|
|
FONTS = {(False, False): "Courier", (True, False): "Courier-Bold", (False, True): "Courier-Oblique", (True, True): "Courier-BoldOblique"}
|
|
|
|
|
|
def _operator(color: "float | tuple[float, ...]", fill: bool) -> str:
|
|
values = (color,) if isinstance(color, (int, float)) else tuple(color)
|
|
name = {1: "g", 3: "rg", 4: "k"}[len(values)]
|
|
return " ".join(f"{v:g}" for v in values) + " " + (name if fill else name.upper())
|
|
|
|
|
|
def build_styled_pdf(
|
|
pages: list[StyledPage],
|
|
*,
|
|
outline: "list[tuple[str, int, Optional[int]]]" = (),
|
|
title: Optional[str] = None,
|
|
) -> bytes:
|
|
"""A PDF in Courier 10 pt where each run can be coloured, highlighted, struck through
|
|
or linked, a page can carry an image, and `outline` is a list of `(title, level,
|
|
page index from 0 or None for an entry that goes nowhere)`. Character cells are 6 pt wide, so the geometry is exact."""
|
|
objects: dict[int, bytes] = {}
|
|
next_id = [3]
|
|
|
|
def allocate() -> int:
|
|
number = next_id[0]
|
|
next_id[0] += 1
|
|
return number
|
|
|
|
page_ids = [allocate() for _ in pages]
|
|
font_ids = {key: allocate() for key in FONTS}
|
|
font_names = {key: f"F{index}" for index, key in enumerate(FONTS, 1)}
|
|
image_ids: dict[tuple[int, int], int] = {}
|
|
|
|
def image_object(pixels: int, gray: int) -> int:
|
|
if (pixels, gray) not in image_ids:
|
|
number = image_ids[(pixels, gray)] = allocate()
|
|
data = bytes([gray]) * (pixels * pixels)
|
|
objects[number] = (
|
|
f"<< /Type /XObject /Subtype /Image /Width {pixels} /Height {pixels} /ColorSpace /DeviceGray "
|
|
f"/BitsPerComponent 8 /Length {len(data)} >>\nstream\n"
|
|
).encode() + data + b"\nendstream"
|
|
return image_ids[(pixels, gray)]
|
|
|
|
for number, page in zip(page_ids, pages):
|
|
ops, annots, xobjects = list(page.ops), [], {}
|
|
for row, line in enumerate(page.lines):
|
|
y, x = TOP - LINE_HEIGHT * row, LEFT
|
|
for run in line:
|
|
text, options = (run, {}) if isinstance(run, str) else run
|
|
size = options.get("size", 10)
|
|
width = 0.6 * size * len(text)
|
|
font = font_names[(bool(options.get("bold")), bool(options.get("italic")))]
|
|
if "highlight" in options:
|
|
ops.append(f"{_operator(options['highlight'], True)} {x} {y - 3} {width} {size * 1.2} re f 0 g")
|
|
if "color" in options:
|
|
ops.append(_operator(options["color"], True))
|
|
escaped = text.replace("\\", "\\\\").replace("(", "\\(").replace(")", "\\)")
|
|
if options.get("rotate") == 90:
|
|
ops.append(f"BT /{font} {size} Tf 0 1 -1 0 {x} {y} Tm ({escaped}) Tj ET")
|
|
else:
|
|
ops.append(f"BT /{font} {size} Tf {x} {y} Td ({escaped}) Tj ET")
|
|
if "color" in options:
|
|
ops.append("0 g")
|
|
if options.get("strike"):
|
|
ops.append(f"0 G 0.7 w {x} {y + 3.2} m {x + width} {y + 3.2} l S")
|
|
if options.get("link"):
|
|
annots.append(
|
|
f"<< /Type /Annot /Subtype /Link /Rect [{x} {y - 3} {x + width} {y + 9}] /Border [0 0 0] "
|
|
"/A << /S /URI /URI (https://example.org/) >> >>"
|
|
)
|
|
x += width
|
|
pictures = list(page.pictures)
|
|
if page.image:
|
|
pictures.append(Picture(72, 400, 40, 40, pixels=8))
|
|
for index, picture in enumerate(pictures):
|
|
xobjects[f"Im{index}"] = image_object(picture.pixels, picture.gray)
|
|
ops.append(f"q {picture.width} 0 0 {picture.height} {picture.x} {picture.y} cm /Im{index} Do Q")
|
|
stream = "\n".join(ops).encode("latin-1")
|
|
content_id = allocate()
|
|
objects[content_id] = b"<< /Length %d >>\nstream\n" % len(stream) + stream + b"\nendstream"
|
|
xobject = "/XObject << " + " ".join(f"/{name} {ref} 0 R" for name, ref in xobjects.items()) + " >>" if xobjects else ""
|
|
fonts = " ".join(f"/{font_names[key]} {font_ids[key]} 0 R" for key in FONTS)
|
|
objects[number] = (
|
|
f"<< /Type /Page /Parent 2 0 R /MediaBox [0 0 {page.size[0]:g} {page.size[1]:g}] /Contents {content_id} 0 R "
|
|
f"/Resources << /Font << {fonts} >> {xobject} >> /Annots [{' '.join(annots)}] >>"
|
|
).encode()
|
|
for key, number in font_ids.items():
|
|
objects[number] = f"<< /Type /Font /Subtype /Type1 /BaseFont /{FONTS[key]} >>".encode()
|
|
objects[1] = b"<< /Type /Catalog /Pages 2 0 R >>"
|
|
objects[2] = f"<< /Type /Pages /Kids [{' '.join(f'{n} 0 R' for n in page_ids)}] /Count {len(pages)} >>".encode()
|
|
if outline:
|
|
root = allocate()
|
|
ids = [allocate() for _ in outline]
|
|
parent_of: list[Optional[int]] = []
|
|
stack: list[tuple[int, int]] = [] # (level, index)
|
|
for index, (_title, level, _page) in enumerate(outline):
|
|
while stack and stack[-1][0] >= level:
|
|
stack.pop()
|
|
parent_of.append(stack[-1][1] if stack else None)
|
|
stack.append((level, index))
|
|
for index, (entry_title, _level, page_index) in enumerate(outline):
|
|
siblings = [i for i, parent in enumerate(parent_of) if parent == parent_of[index]]
|
|
children = [i for i, parent in enumerate(parent_of) if parent == index]
|
|
position = siblings.index(index)
|
|
parts = [
|
|
f"/Title ({entry_title})",
|
|
f"/Parent {(ids[parent_of[index]] if parent_of[index] is not None else root)} 0 R",
|
|
]
|
|
if page_index is not None:
|
|
parts.append(f"/Dest [{page_ids[page_index]} 0 R /Fit]")
|
|
if position > 0:
|
|
parts.append(f"/Prev {ids[siblings[position - 1]]} 0 R")
|
|
if position < len(siblings) - 1:
|
|
parts.append(f"/Next {ids[siblings[position + 1]]} 0 R")
|
|
if children:
|
|
parts += [f"/First {ids[children[0]]} 0 R", f"/Last {ids[children[-1]]} 0 R", f"/Count {len(children)}"]
|
|
objects[ids[index]] = ("<< " + " ".join(parts) + " >>").encode("latin-1")
|
|
top = [i for i, parent in enumerate(parent_of) if parent is None]
|
|
objects[root] = f"<< /Type /Outlines /First {ids[top[0]]} 0 R /Last {ids[top[-1]]} 0 R /Count {len(top)} >>".encode()
|
|
objects[1] = f"<< /Type /Catalog /Pages 2 0 R /Outlines {root} 0 R >>".encode()
|
|
trailer = "/Root 1 0 R"
|
|
if title is not None:
|
|
info_id = allocate()
|
|
objects[info_id] = f"<< /Title ({title}) >>".encode("latin-1")
|
|
trailer += f" /Info {info_id} 0 R"
|
|
return _serialize(objects, trailer)
|
|
|
|
|
|
def _serialize(objects: dict[int, bytes], trailer: str) -> bytes:
|
|
out = bytearray(b"%PDF-1.4\n")
|
|
offsets = {}
|
|
for number in sorted(objects):
|
|
offsets[number] = len(out)
|
|
out += f"{number} 0 obj\n".encode() + objects[number] + b"\nendobj\n"
|
|
xref = len(out)
|
|
size = max(objects) + 1
|
|
out += f"xref\n0 {size}\n".encode() + b"0000000000 65535 f \n"
|
|
for number in range(1, size):
|
|
out += f"{offsets[number]:010d} 00000 n \n".encode()
|
|
out += f"trailer\n<< /Size {size} {trailer} >>\nstartxref\n{xref}\n%%EOF\n".encode()
|
|
return bytes(out)
|
|
|
|
|
|
class _Node:
|
|
def __init__(self, name: str, kind: int, data: bytes = b"") -> None:
|
|
self.name, self.kind, self.data = name, kind, data
|
|
self.children: dict[str, "_Node"] = {}
|
|
self.index = 0
|
|
self.start = END
|
|
self.size = 0
|
|
|
|
|
|
def build_cfb(files: dict[str, bytes]) -> bytes:
|
|
"""A compound file (version 3) holding `files` (path with `/` -> bytes). Streams under
|
|
4096 bytes go into the mini stream, as the format requires."""
|
|
root = _Node("Root Entry", 5)
|
|
for path, data in files.items():
|
|
*folders, leaf = path.split("/")
|
|
node = root
|
|
for folder in folders:
|
|
node = node.children.setdefault(folder, _Node(folder, 1))
|
|
node.children[leaf] = _Node(leaf, 2, data)
|
|
order: list[_Node] = []
|
|
|
|
def number(node: _Node) -> None:
|
|
node.index = len(order)
|
|
order.append(node)
|
|
for child in node.children.values():
|
|
number(child)
|
|
|
|
number(root)
|
|
streams = [n for n in order if n.kind == 2]
|
|
small = [n for n in streams if 0 < len(n.data) < 4096]
|
|
big = [n for n in streams if len(n.data) >= 4096]
|
|
|
|
mini = bytearray()
|
|
mini_chain: list[int] = []
|
|
for node in small:
|
|
node.start = len(mini) // 64
|
|
node.size = len(node.data)
|
|
blocks = -(-len(node.data) // 64)
|
|
mini += node.data.ljust(blocks * 64, b"\x00")
|
|
mini_chain += [node.start + i + 1 for i in range(blocks - 1)] + [END]
|
|
for node in streams:
|
|
node.size = len(node.data)
|
|
|
|
n_minifat = -(-len(mini_chain) // 128)
|
|
n_dir = -(-len(order) // 4)
|
|
n_mini = -(-len(mini) // 512)
|
|
big_sectors = {id(n): -(-len(n.data) // 512) for n in big}
|
|
total = 1 + n_minifat + n_dir + n_mini + sum(big_sectors.values())
|
|
assert total <= 128, "the builder writes one FAT sector"
|
|
|
|
fat = [FREE] * 128
|
|
fat[0] = FATSECT
|
|
cursor = 1
|
|
first_minifat = cursor if n_minifat else END
|
|
for i in range(n_minifat):
|
|
fat[cursor + i] = cursor + i + 1 if i < n_minifat - 1 else END
|
|
cursor += n_minifat
|
|
first_dir = cursor
|
|
for i in range(n_dir):
|
|
fat[cursor + i] = cursor + i + 1 if i < n_dir - 1 else END
|
|
cursor += n_dir
|
|
root.start = cursor if n_mini else END
|
|
root.size = len(mini)
|
|
for i in range(n_mini):
|
|
fat[cursor + i] = cursor + i + 1 if i < n_mini - 1 else END
|
|
cursor += n_mini
|
|
for node in big:
|
|
node.start = cursor
|
|
count = big_sectors[id(node)]
|
|
for i in range(count):
|
|
fat[cursor + i] = cursor + i + 1 if i < count - 1 else END
|
|
cursor += count
|
|
|
|
def entry(node: Optional[_Node], right: int = FREE) -> bytes:
|
|
if node is None:
|
|
return struct.pack("<64sHBBIII16sIQQIQ", b"", 0, 0, 0, FREE, FREE, FREE, b"", 0, 0, 0, 0, 0)
|
|
name = node.name.encode("utf-16-le")
|
|
kids = list(node.children.values())
|
|
return struct.pack(
|
|
"<64sHBBIII16sIQQIQ", name.ljust(64, b"\x00"), len(name) + 2, node.kind, 1, FREE, right,
|
|
kids[0].index if kids else FREE, b"", 0, 0, 0, node.start, node.size,
|
|
)
|
|
|
|
rights: dict[int, int] = {}
|
|
for node in order:
|
|
kids = list(node.children.values())
|
|
for current, following in zip(kids, kids[1:]):
|
|
rights[current.index] = following.index
|
|
directory = b"".join(entry(n, rights.get(n.index, FREE)) for n in order)
|
|
empty_entry = entry(None)
|
|
directory = directory[: len(order) * 128] + empty_entry * (n_dir * 4 - len(order))
|
|
|
|
header = struct.pack(
|
|
"<8s16sHHHHH6sIIIIIIIII", b"\xd0\xcf\x11\xe0\xa1\xb1\x1a\xe1", b"", 0x3E, 3, 0xFFFE, 9, 6, b"", 0, 1,
|
|
first_dir, 0, 4096, first_minifat, n_minifat, END, 0,
|
|
) + struct.pack("<109I", 0, *([FREE] * 108))
|
|
|
|
minifat = struct.pack(f"<{n_minifat * 128}I", *(mini_chain + [FREE] * (n_minifat * 128 - len(mini_chain))))
|
|
out = bytearray(header)
|
|
out += struct.pack("<128I", *fat)
|
|
out += minifat
|
|
out += directory
|
|
out += bytes(mini).ljust(n_mini * 512, b"\x00")
|
|
for node in big:
|
|
out += node.data.ljust(big_sectors[id(node)] * 512, b"\x00")
|
|
return bytes(out)
|
|
|
|
|
|
def _utf16(text: str) -> bytes:
|
|
return text.encode("utf-16-le")
|
|
|
|
|
|
def build_msg(
|
|
*,
|
|
subject: str = "",
|
|
body: Optional[str] = None,
|
|
html: Optional[str] = None,
|
|
rtf: Optional[bytes] = None,
|
|
sender: str = "",
|
|
sender_address: str = "",
|
|
to: str = "",
|
|
cc: str = "",
|
|
attachments: Optional[list[tuple[str, bytes]]] = None,
|
|
inline_images: Optional[list[tuple[str, bytes]]] = None,
|
|
embedded: Optional[list[str]] = None,
|
|
sent: Optional[int] = None,
|
|
) -> bytes:
|
|
"""An Outlook message. `sent` is a FILETIME (100 ns ticks since 1601)."""
|
|
files: dict[str, bytes] = {}
|
|
|
|
def text(prefix: str, tag: str, value: str) -> None:
|
|
if value:
|
|
files[f"{prefix}__substg1.0_{tag}001F"] = _utf16(value)
|
|
|
|
text("", "0037", subject)
|
|
text("", "0C1A", sender)
|
|
text("", "5D01", sender_address)
|
|
text("", "0E04", to)
|
|
text("", "0E03", cc)
|
|
if body is not None:
|
|
text("", "1000", body)
|
|
if html is not None:
|
|
files["__substg1.0_10130102"] = html.encode("utf-8")
|
|
if rtf is not None:
|
|
files["__substg1.0_10090102"] = rtf
|
|
props = b"\x00" * 32
|
|
if sent is not None:
|
|
props += struct.pack("<IIQ", 0x00390040, 6, sent)
|
|
files["__properties_version1.0"] = props
|
|
|
|
number = 0
|
|
for name, data in attachments or []:
|
|
prefix = f"__attach_version1.0_#{number:08X}/"
|
|
text(prefix, "3707", name)
|
|
text(prefix, "370E", "application/octet-stream")
|
|
files[f"{prefix}__substg1.0_37010102"] = data
|
|
files[f"{prefix}__properties_version1.0"] = b"\x00" * 8 + struct.pack("<IIQ", 0x37050003, 6, 1)
|
|
number += 1
|
|
for name, data in inline_images or []:
|
|
prefix = f"__attach_version1.0_#{number:08X}/"
|
|
text(prefix, "3707", name)
|
|
text(prefix, "370E", "image/png")
|
|
text(prefix, "3712", "cid-" + name)
|
|
files[f"{prefix}__substg1.0_37010102"] = data
|
|
files[f"{prefix}__properties_version1.0"] = b"\x00" * 8 + struct.pack("<IIQ", 0x37050003, 6, 1)
|
|
number += 1
|
|
for name in embedded or []:
|
|
prefix = f"__attach_version1.0_#{number:08X}/"
|
|
text(prefix, "3707", name)
|
|
files[f"{prefix}__properties_version1.0"] = b"\x00" * 8 + struct.pack("<IIQ", 0x37050003, 6, 5)
|
|
number += 1
|
|
return build_cfb(files)
|