Scaffold doced: OPC package I/O, XML core, AST model, locator model, reader, LLM contract
This commit is contained in:
doced builder
@@ -0,0 +1,315 @@
|
||||
"""The reader: parse ``word/document.xml`` into the typed AST.
|
||||
|
||||
This is the *view* builder — it walks the parsed XML tree and populates a
|
||||
:class:`~doced.ast.model.Document`. It never mutates; it only reads. The
|
||||
positional ids (``pN`` / ``tN`` / ``sN`` / ``cN`` / ``tcN`` / …) are assigned
|
||||
in document order here, so they are stable within a single read and shift only
|
||||
after structural edits (see :mod:`doced.ast.locator`).
|
||||
|
||||
The reader is namespace-aware (see :mod:`doced.xmlcore`) and tolerant: unknown
|
||||
elements are skipped rather than raising, so a document with features we don't
|
||||
model still reads.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from lxml import etree as ET
|
||||
|
||||
from .. import xmlcore as xc
|
||||
from ..package import Package, PackagePartMissing
|
||||
from . import model as M
|
||||
|
||||
|
||||
def _local(el: ET._Element) -> str:
|
||||
return xc.local(el.tag)
|
||||
|
||||
|
||||
def _xn(prefix: str, name: str) -> str:
|
||||
"""Namespace-qualified attribute name, e.g. ``{w}val``."""
|
||||
return f"{{{xc.NSMAP[prefix]}}}{name}"
|
||||
|
||||
|
||||
def _opt(el: ET._Element | None, prefix: str, name: str, attr: str) -> str | None:
|
||||
"""Return ``el.{prefix}:{attr}`` if ``el`` and the attr exist, else None."""
|
||||
if el is None:
|
||||
return None
|
||||
val = el.get(_xn(prefix, attr))
|
||||
return val if val else None
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Character formatting extraction.
|
||||
# ---------------------------------------------------------------------------
|
||||
def _run_format(rPr: ET._Element | None) -> dict:
|
||||
"""Extract character formatting from a ``<w:rPr>`` element."""
|
||||
fmt: dict = {}
|
||||
if rPr is None:
|
||||
return fmt
|
||||
if xc.find(rPr, "w:b") is not None:
|
||||
fmt["bold"] = _opt(xc.find(rPr, "w:b"), "w", "b", "val") != "false"
|
||||
if xc.find(rPr, "w:i") is not None:
|
||||
fmt["italic"] = True
|
||||
u = xc.find(rPr, "w:u")
|
||||
if u is not None and u.get(_xn("w", "val")) != "none":
|
||||
fmt["underline"] = True
|
||||
if xc.find(rPr, "w:strike") is not None:
|
||||
fmt["strike"] = True
|
||||
color = xc.find(rPr, "w:color")
|
||||
col = _opt(color, "w", "color", "val")
|
||||
if col and col != "auto":
|
||||
fmt["color"] = col
|
||||
font = xc.find(rPr, "w:rFonts")
|
||||
f = _opt(font, "w", "rFonts", "ascii")
|
||||
if f:
|
||||
fmt["font"] = f
|
||||
sz = xc.find(rPr, "w:sz")
|
||||
sz_val = _opt(sz, "w", "sz", "val")
|
||||
if sz_val:
|
||||
fmt["size_half_points"] = int(sz_val)
|
||||
return fmt
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Paragraph reading.
|
||||
# ---------------------------------------------------------------------------
|
||||
def _paragraph_runs(pEl: ET._Element) -> tuple[list[M.Run], str | None]:
|
||||
"""Extract runs and the paragraph style id from a ``<w:p>`` element."""
|
||||
runs: list[M.Run] = []
|
||||
style = _opt(xc.find(pEl, "w:pPr"), "w", "pStyle", "val")
|
||||
|
||||
for child in pEl:
|
||||
tag = _local(child)
|
||||
if tag == "r":
|
||||
run = _run_element(child)
|
||||
if run is not None:
|
||||
runs.append(run)
|
||||
elif tag == "br":
|
||||
br_type = _opt(child, "w", "br", "type")
|
||||
kind = "line"
|
||||
if br_type == "page":
|
||||
kind = "page"
|
||||
elif br_type == "column":
|
||||
kind = "column"
|
||||
runs.append(M.BreakRun(kind=kind))
|
||||
elif tag == "tab":
|
||||
runs.append(M.TabRun())
|
||||
return runs, style
|
||||
|
||||
|
||||
def _run_element(rEl: ET._Element) -> M.Run | None:
|
||||
"""Build a run from a ``<w:r>`` element."""
|
||||
t = xc.find(rEl, "w:t")
|
||||
if t is not None and t.text:
|
||||
fmt = _run_format(xc.find(rEl, "w:rPr"))
|
||||
hyperlink = xc.find(rEl, "r:hyperlink")
|
||||
hl_id = _opt(hyperlink, "r", "hyperlink", "id")
|
||||
return M.TextRun(
|
||||
text=t.text,
|
||||
bold=fmt.get("bold", False),
|
||||
italic=fmt.get("italic", False),
|
||||
underline=fmt.get("underline", False),
|
||||
strike=fmt.get("strike", False),
|
||||
color=fmt.get("color"),
|
||||
font=fmt.get("font"),
|
||||
size_half_points=fmt.get("size_half_points"),
|
||||
hyperlink_id=hl_id,
|
||||
)
|
||||
return None
|
||||
|
||||
|
||||
def _paragraph_alignment(pEl: ET._Element) -> str | None:
|
||||
return _opt(xc.find(pEl, "w:pPr"), "w", "jc", "val")
|
||||
|
||||
|
||||
def _paragraph_list(pEl: ET._Element) -> tuple[int | None, str | None]:
|
||||
numPr = xc.find(pEl, "w:pPr/w:numPr")
|
||||
if numPr is None:
|
||||
return None, None
|
||||
level = _opt(numPr, "w", "ilvl", "val")
|
||||
nid = _opt(numPr, "w", "numId", "val")
|
||||
return int(level) if level else None, nid
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Table reading.
|
||||
# ---------------------------------------------------------------------------
|
||||
def _table_grid(tbl: ET._Element) -> list[int]:
|
||||
widths: list[int] = []
|
||||
grid = xc.find(tbl, "w:tblGrid")
|
||||
if grid is not None:
|
||||
for col in grid.findall(xn("w", "gridCol")):
|
||||
w = col.get(_xn("w", "w"))
|
||||
if w:
|
||||
widths.append(int(w))
|
||||
return widths
|
||||
|
||||
|
||||
def _table_cell(tc: ET._Element) -> M.TableCell:
|
||||
cell = M.TableCell()
|
||||
tcPr = xc.find(tc, "w:tcPr")
|
||||
if tcPr is not None:
|
||||
span = xc.find(tcPr, "w:gridSpan")
|
||||
if span is not None and span.get(_xn("w", "val")):
|
||||
cell.grid_span = int(span.get(_xn("w", "val")))
|
||||
vMerge = xc.find(tcPr, "w:vMerge")
|
||||
if vMerge is not None:
|
||||
val = vMerge.get(_xn("w", "val"))
|
||||
cell.v_merge = "restart" if val == "restart" else "continue"
|
||||
shading = xc.find(tcPr, "w:shd")
|
||||
if shading is not None and shading.get(_xn("w", "val")) not in (None, "clear"):
|
||||
cell.shading = shading.get(_xn("w", "fill"))
|
||||
tcW = xc.find(tcPr, "w:tcW")
|
||||
if tcW is not None and tcW.get(_xn("w", "w")):
|
||||
cell.width_twips = int(tcW.get(_xn("w", "w")))
|
||||
vAlign = xc.find(tcPr, "w:vAlign")
|
||||
if vAlign is not None:
|
||||
cell.v_align = vAlign.get(_xn("w", "val"))
|
||||
|
||||
for child in tc:
|
||||
if _local(child) == "p":
|
||||
runs, style = _paragraph_runs(child)
|
||||
para = M.Paragraph(id="", style=style, runs=runs)
|
||||
cell.blocks.append(para)
|
||||
return cell
|
||||
|
||||
|
||||
def _read_table(tbl: ET._Element, idx: int) -> M.Table:
|
||||
return M.Table(
|
||||
id=f"t{idx}",
|
||||
rows=[M.TableRow(cells=(_table_cell(tc) for tc in tbl.findall(_xn("w", "tr"))))],
|
||||
grid=_table_grid(tbl),
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Section break reading.
|
||||
# ---------------------------------------------------------------------------
|
||||
def _section_break(sid: str, sectPr: ET._Element) -> M.SectionBreak:
|
||||
sb = M.SectionBreak(id=sid)
|
||||
for child in sectPr:
|
||||
tag = _local(child)
|
||||
if tag == "cols" and child.get(_xn("w", "num")):
|
||||
sb.columns = int(child.get(_xn("w", "num")))
|
||||
elif tag == "pgSz":
|
||||
w = int(child.get(_xn("w", "w"), "0"))
|
||||
h = int(child.get(_xn("w", "h"), "0"))
|
||||
sb.page_orientation = "landscape" if w > h else "portrait"
|
||||
elif tag == "pgMar":
|
||||
sb.margin_top = _twips(child, "top")
|
||||
sb.margin_right = _twips(child, "right")
|
||||
sb.margin_bottom = _twips(child, "bottom")
|
||||
sb.margin_left = _twips(child, "left")
|
||||
return sb
|
||||
|
||||
|
||||
def _twips(el: ET._Element, name: str) -> int | None:
|
||||
val = el.get(_xn("w", name))
|
||||
return int(val) if val else None
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Side parts.
|
||||
# ---------------------------------------------------------------------------
|
||||
def _read_comments(pkg: Package) -> list[M.Comment]:
|
||||
if "word/comments.xml" not in pkg.parts:
|
||||
return []
|
||||
root = pkg.parse("word/comments.xml")
|
||||
comments: list[M.Comment] = []
|
||||
for c in root.findall(xn("w", "comment")):
|
||||
cid = c.get(_xn("w", "id")) or f"c{len(comments)}"
|
||||
author = (xc.find(c, "w:author") or ET.Element("x")).get(_xn("w", "val"), "")
|
||||
date = (xc.find(c, "w:date") or ET.Element("x")).get(_xn("w", "val"))
|
||||
text = "".join(t.text or "" for t in c.iter(xn("w", "t")))
|
||||
comments.append(M.Comment(id=cid, author=author, date=date, text=text))
|
||||
return comments
|
||||
|
||||
|
||||
def _read_notes(pkg: Package, part: str, kind: str) -> list[M.Note]:
|
||||
if part not in pkg.parts:
|
||||
return []
|
||||
root = pkg.parse(part)
|
||||
notes: list[M.Note] = []
|
||||
for n in root.findall(xn("w", kind)):
|
||||
nid = n.get(_xn("w", "id")) or f"{kind[0]}{len(notes) + 1}"
|
||||
notes.append(M.Note(id=nid, kind=kind, text=n.text or ""))
|
||||
return notes
|
||||
|
||||
|
||||
def _read_properties(pkg: Package) -> M.DocProperties:
|
||||
if "docProps/core.xml" not in pkg.parts:
|
||||
return M.DocProperties()
|
||||
root = pkg.parse("docProps/core.xml")
|
||||
props = M.DocProperties()
|
||||
for tag, attr in (
|
||||
("dc:title", "title"),
|
||||
("dc:creator", "author"),
|
||||
("dcterms:created", "created"),
|
||||
("dcterms:modified", "modified"),
|
||||
("dc:subject", "subject"),
|
||||
("dc:keywords", "keywords"),
|
||||
):
|
||||
el = xc.find(root, tag)
|
||||
if el is not None and el.text:
|
||||
setattr(props, attr, el.text)
|
||||
return props
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Document reading.
|
||||
# ---------------------------------------------------------------------------
|
||||
def _iter_body_elements(body: ET._Element) -> list[ET._Element]:
|
||||
"""Return body elements in order, separating ``<w:sectPr>`` out."""
|
||||
els: list[ET._Element] = []
|
||||
sect_pr: ET._Element | None = None
|
||||
for child in body:
|
||||
if _local(child) == "sectPr":
|
||||
sect_pr = child
|
||||
else:
|
||||
els.append(child)
|
||||
return els, sect_pr
|
||||
|
||||
|
||||
def build_document(pkg: Package) -> M.Document:
|
||||
"""Parse the package into a typed :class:`M.Document`.
|
||||
|
||||
Raises:
|
||||
PackagePartMissing: if ``word/document.xml`` is absent.
|
||||
"""
|
||||
if "word/document.xml" not in pkg.parts:
|
||||
raise PackagePartMissing("word/document.xml not found in package")
|
||||
|
||||
root = pkg.parse("word/document.xml")
|
||||
body = xc.find(root, "w:body")
|
||||
if body is None:
|
||||
body = root # degenerate document; treat root as body
|
||||
|
||||
doc = M.Document(path=pkg.path)
|
||||
p_idx, t_idx, s_idx = 0, 0, 0
|
||||
|
||||
for el in body:
|
||||
tag = _local(el)
|
||||
if tag == "p":
|
||||
runs, style = _paragraph_runs(el)
|
||||
doc.blocks.append(M.Paragraph(
|
||||
id=f"p{p_idx}",
|
||||
style=style,
|
||||
runs=runs,
|
||||
alignment=_paragraph_alignment(el),
|
||||
list_level=_paragraph_list(el)[0],
|
||||
list_id=_paragraph_list(el)[1],
|
||||
))
|
||||
p_idx += 1
|
||||
elif tag == "tbl":
|
||||
doc.blocks.append(_read_table(el, t_idx))
|
||||
t_idx += 1
|
||||
|
||||
# The trailing sectPr defines document-wide section properties.
|
||||
sect_pr = xc.find(body, "w:sectPr")
|
||||
if sect_pr is not None:
|
||||
doc.blocks.append(_section_break(f"s{s_idx}", sect_pr))
|
||||
|
||||
doc.comments = _read_comments(pkg)
|
||||
doc.footnotes = _read_notes(pkg, "word/footnotes.xml", "footnote")
|
||||
doc.endnotes = _read_notes(pkg, "word/endnotes.xml", "endnote")
|
||||
doc.properties = _read_properties(pkg)
|
||||
return doc
|
||||
Reference in New Issue
Block a user