doc-forge-reds/word_html_converter/converter.py

597 lines
21 KiB
Python
Executable File

#!/usr/bin/env python3
"""
word-html-converter
Open-source DOCX <-> HTML conversion wrapper focused on high visual fidelity.
Core engine: LibreOffice headless conversion.
HTML post-processing: inline CSS, keep @page rules, normalize metadata.
Commands:
python converter.py docx2html input.docx output.html
python converter.py html2docx input.html output.docx
python converter.py roundtrip input.docx output_dir
python converter.py verify original.docx converted.docx report.json
"""
from __future__ import annotations
import argparse
import base64
import hashlib
import json
import os
import re
import shutil
import subprocess
import sys
import warnings
import tempfile
from dataclasses import dataclass, asdict
from pathlib import Path
from typing import Dict, Iterable, List, Optional, Tuple
from bs4 import BeautifulSoup, XMLParsedAsHTMLWarning
import tinycss2
warnings.filterwarnings("ignore", category=XMLParsedAsHTMLWarning)
APP_NAME = "word-html-converter"
DEFAULT_DOCX_TO_HTML_FILTERS = [
# Writer/Web HTML filter preserves more page/header/footer information than the clean XHTML filter.
"html",
# Fallback clean XHTML Writer filter.
"html:XHTML Writer File:UTF8",
]
DEFAULT_HTML_TO_DOCX_FILTERS = [
'docx:"Office Open XML Text"',
"docx",
]
@dataclass
class ConvertResult:
source: str
output: str
command: List[str]
stdout: str
stderr: str
filter_name: str
class ConversionError(RuntimeError):
pass
def sha256_file(path: Path) -> str:
h = hashlib.sha256()
with path.open("rb") as f:
for chunk in iter(lambda: f.read(1024 * 1024), b""):
h.update(chunk)
return h.hexdigest()
def find_soffice(explicit: Optional[str] = None) -> str:
"""Find LibreOffice/soffice executable on Linux/macOS/Windows."""
candidates: List[str] = []
if explicit:
candidates.append(explicit)
if os.environ.get("SOFFICE_BIN"):
candidates.append(os.environ["SOFFICE_BIN"])
candidates.extend([
"soffice",
"libreoffice",
"/Applications/LibreOffice.app/Contents/MacOS/soffice",
r"C:\Program Files\LibreOffice\program\soffice.com",
r"C:\Program Files\LibreOffice\program\soffice.exe",
r"C:\Program Files (x86)\LibreOffice\program\soffice.com",
r"C:\Program Files (x86)\LibreOffice\program\soffice.exe",
])
for c in candidates:
if not c:
continue
if Path(c).exists():
return str(Path(c))
resolved = shutil.which(c)
if resolved:
return resolved
raise ConversionError(
"LibreOffice executable not found. Install LibreOffice, or set SOFFICE_BIN=/path/to/soffice."
)
def file_uri(path: Path) -> str:
return path.resolve().as_uri()
def run_soffice_convert(
source: Path,
outdir: Path,
convert_to: str,
soffice_bin: Optional[str] = None,
timeout: int = 120,
input_filter: Optional[str] = None,
) -> Tuple[subprocess.CompletedProcess[str], List[str]]:
source = source.resolve()
outdir = outdir.resolve()
outdir.mkdir(parents=True, exist_ok=True)
soffice = find_soffice(soffice_bin)
# Use an isolated LO user profile so conversion works even if desktop LO is open.
profile_dir = Path(tempfile.mkdtemp(prefix="lo-profile-"))
profile_uri = file_uri(profile_dir)
cmd = [
soffice,
f"-env:UserInstallation={profile_uri}",
"--headless",
"--nologo",
"--nofirststartwizard",
"--nodefault",
"--nolockcheck",
"--norestore",
]
if input_filter:
cmd.append(f"--infilter={input_filter}")
cmd.extend([
"--convert-to",
convert_to,
"--outdir",
str(outdir),
str(source),
])
try:
proc = subprocess.run(cmd, text=True, capture_output=True, timeout=timeout)
finally:
shutil.rmtree(profile_dir, ignore_errors=True)
return proc, cmd
def find_converted_file(before: set[Path], outdir: Path, source_stem: str, expected_ext: str) -> Optional[Path]:
expected = outdir / f"{source_stem}{expected_ext}"
if expected.exists():
return expected
after = set(outdir.iterdir())
created = [p for p in after - before if p.is_file() and p.suffix.lower() == expected_ext.lower()]
if created:
return max(created, key=lambda p: p.stat().st_mtime)
candidates = list(outdir.glob(f"{source_stem}*{expected_ext}"))
if candidates:
return max(candidates, key=lambda p: p.stat().st_mtime)
return None
def convert_with_fallbacks(
source: Path,
output: Path,
filters: Iterable[str],
expected_ext: str,
soffice_bin: Optional[str] = None,
timeout: int = 120,
input_filter: Optional[str] = None,
) -> ConvertResult:
source = source.resolve()
output = output.resolve()
output.parent.mkdir(parents=True, exist_ok=True)
last_error = ""
with tempfile.TemporaryDirectory(prefix="convert-work-") as tmp:
tmpdir = Path(tmp)
for filter_name in filters:
before = set(tmpdir.iterdir())
proc, cmd = run_soffice_convert(source, tmpdir, filter_name, soffice_bin, timeout, input_filter=input_filter)
converted = find_converted_file(before, tmpdir, source.stem, expected_ext)
if proc.returncode == 0 and converted and converted.exists():
if output.exists():
output.unlink()
shutil.move(str(converted), str(output))
# Move sidecar image/assets generated by HTML export.
for item in tmpdir.iterdir():
if item.is_file() and item.name != output.name:
dest = output.parent / item.name
if dest.exists():
dest.unlink()
shutil.move(str(item), str(dest))
return ConvertResult(
source=str(source),
output=str(output),
command=cmd,
stdout=proc.stdout,
stderr=proc.stderr,
filter_name=filter_name,
)
last_error = (
f"Filter failed: {filter_name}\n"
f"Return code: {proc.returncode}\nSTDOUT:\n{proc.stdout}\nSTDERR:\n{proc.stderr}"
)
raise ConversionError(last_error or "No LibreOffice conversion filter succeeded.")
def style_to_dict(style: str) -> Dict[str, str]:
result: Dict[str, str] = {}
if not style:
return result
declarations = tinycss2.parse_declaration_list(style, skip_comments=True, skip_whitespace=True)
for d in declarations:
if getattr(d, "type", None) == "declaration" and not d.name.startswith("--"):
value = tinycss2.serialize(d.value).strip()
if d.important:
value = f"{value} !important"
result[d.name.lower()] = value
return result
def dict_to_style(d: Dict[str, str]) -> str:
return "; ".join(f"{k}: {v}" for k, v in d.items() if v) + (";" if d else "")
def merge_inline_style(tag, declarations: Dict[str, str]) -> None:
current = style_to_dict(tag.get("style", ""))
current.update(declarations)
tag["style"] = dict_to_style(current)
def parse_css_rules(css: str) -> Tuple[List[Tuple[str, Dict[str, str]]], str]:
"""Return normal CSS rules plus preserved at-rules such as @page/@media."""
normal_rules: List[Tuple[str, Dict[str, str]]] = []
preserved_at_rules: List[str] = []
rules = tinycss2.parse_stylesheet(css, skip_comments=True, skip_whitespace=True)
for rule in rules:
if rule.type == "qualified-rule":
selector = tinycss2.serialize(rule.prelude).strip()
declarations: Dict[str, str] = {}
for d in tinycss2.parse_declaration_list(rule.content, skip_comments=True, skip_whitespace=True):
if getattr(d, "type", None) == "declaration":
value = tinycss2.serialize(d.value).strip()
if d.important:
value = f"{value} !important"
declarations[d.name.lower()] = value
if selector and declarations:
normal_rules.append((selector, declarations))
elif rule.type == "at-rule":
# @page is important for print/page-like preview; @media may contain print rules.
preserved_at_rules.append(tinycss2.serialize([rule]).strip())
return normal_rules, "\n".join(preserved_at_rules)
def inline_css(html: str, keep_style_tag: bool = True) -> str:
"""Inline style rules from <style> tags. This is intentionally conservative.
It handles most simple selectors generated by LibreOffice. Unsupported complex selectors
are skipped rather than breaking conversion.
"""
soup = BeautifulSoup(html, "lxml")
style_tags = soup.find_all("style")
preserved_css: List[str] = []
for st in style_tags:
css = st.string or st.get_text() or ""
normal_rules, at_rules = parse_css_rules(css)
if at_rules:
preserved_css.append(at_rules)
for selector, declarations in normal_rules:
selectors = [s.strip() for s in selector.split(",") if s.strip()]
for sel in selectors:
# Browser-only pseudo-selectors do not make sense for DOCX reconstruction.
if ":" in sel and not re.search(r":(first-child|last-child|nth-child)", sel):
continue
try:
matches = soup.select(sel)
except Exception:
continue
for tag in matches:
merge_inline_style(tag, declarations)
if not keep_style_tag:
st.decompose()
if keep_style_tag:
head = soup.head or soup.new_tag("head")
if not soup.head and soup.html:
soup.html.insert(0, head)
if preserved_css:
new_style = soup.new_tag("style")
new_style.string = "\n".join(preserved_css)
head.append(new_style)
return str(soup)
def ensure_html_metadata(html_path: Path, source_docx: Optional[Path] = None) -> None:
html = html_path.read_text(encoding="utf-8", errors="replace")
soup = BeautifulSoup(html, "lxml")
if soup.html is None:
new_html = soup.new_tag("html")
new_html.extend(soup.contents)
soup.append(new_html)
if soup.head is None:
head = soup.new_tag("head")
soup.html.insert(0, head)
if not soup.head.find("meta", attrs={"charset": True}):
meta = soup.new_tag("meta", charset="utf-8")
soup.head.insert(0, meta)
if not soup.head.find("meta", attrs={"name": "generator"}):
meta = soup.new_tag("meta")
meta["name"] = "generator"
meta["content"] = f"{APP_NAME}; LibreOffice headless"
soup.head.append(meta)
if source_docx and not soup.head.find("meta", attrs={"name": "docx-source-sha256"}):
meta = soup.new_tag("meta")
meta["name"] = "docx-source-sha256"
meta["content"] = sha256_file(source_docx)
soup.head.append(meta)
html_path.write_text(str(soup), encoding="utf-8")
def embed_source_docx(html_path: Path, source_docx: Path) -> None:
"""Optional: embed original docx for audit/archival round-trip fallback.
This does NOT merge arbitrary HTML edits back into the original docx. It is useful for
proving byte-exact preservation when the HTML is only used for preview/storage.
"""
html = html_path.read_text(encoding="utf-8", errors="replace")
soup = BeautifulSoup(html, "lxml")
if soup.body is None:
body = soup.new_tag("body")
if soup.html:
soup.html.append(body)
else:
soup.append(body)
old = soup.find(id="__source_docx_base64__")
if old:
old.decompose()
script = soup.new_tag("script")
script["type"] = "application/vnd.openxmlformats-officedocument.wordprocessingml.document+base64"
script["id"] = "__source_docx_base64__"
script.string = base64.b64encode(source_docx.read_bytes()).decode("ascii")
soup.body.append(script)
html_path.write_text(str(soup), encoding="utf-8")
def restore_embedded_docx_if_present(html_path: Path, output_docx: Path) -> bool:
html = html_path.read_text(encoding="utf-8", errors="replace")
soup = BeautifulSoup(html, "lxml")
script = soup.find(id="__source_docx_base64__")
if not script or not script.string:
return False
output_docx.parent.mkdir(parents=True, exist_ok=True)
output_docx.write_bytes(base64.b64decode(script.string.strip()))
return True
def docx_to_html(
source_docx: Path,
output_html: Path,
soffice_bin: Optional[str] = None,
inline: bool = True,
embed_source: bool = False,
timeout: int = 120,
) -> ConvertResult:
if source_docx.suffix.lower() != ".docx":
raise ConversionError("Input file must be .docx")
result = convert_with_fallbacks(
source=source_docx,
output=output_html,
filters=DEFAULT_DOCX_TO_HTML_FILTERS,
expected_ext=".html",
soffice_bin=soffice_bin,
timeout=timeout,
)
ensure_html_metadata(output_html, source_docx)
if inline:
html = output_html.read_text(encoding="utf-8", errors="replace")
output_html.write_text(inline_css(html, keep_style_tag=True), encoding="utf-8")
if embed_source:
embed_source_docx(output_html, source_docx)
return result
def html_to_docx(
source_html: Path,
output_docx: Path,
soffice_bin: Optional[str] = None,
timeout: int = 120,
prefer_embedded_source: bool = False,
) -> ConvertResult:
if source_html.suffix.lower() not in {".html", ".htm", ".xhtml"}:
raise ConversionError("Input file must be .html/.htm/.xhtml")
if prefer_embedded_source and restore_embedded_docx_if_present(source_html, output_docx):
return ConvertResult(
source=str(source_html.resolve()),
output=str(output_docx.resolve()),
command=["restore-embedded-docx"],
stdout="Restored embedded source DOCX.",
stderr="",
filter_name="embedded-docx",
)
return convert_with_fallbacks(
source=source_html,
output=output_docx,
filters=DEFAULT_HTML_TO_DOCX_FILTERS,
expected_ext=".docx",
soffice_bin=soffice_bin,
timeout=timeout,
input_filter="HTML (StarWriter)",
)
def convert_to_pdf(source: Path, output_pdf: Path, soffice_bin: Optional[str] = None, timeout: int = 120) -> ConvertResult:
return convert_with_fallbacks(
source=source,
output=output_pdf,
filters=['pdf:"writer_pdf_Export"', "pdf"],
expected_ext=".pdf",
soffice_bin=soffice_bin,
timeout=timeout,
)
def verify_docx_visual_similarity(
original_docx: Path,
converted_docx: Path,
report_json: Path,
soffice_bin: Optional[str] = None,
dpi: int = 120,
timeout: int = 120,
) -> Dict[str, object]:
"""Render two DOCX files to PDF, then compare page images.
Requires PyMuPDF and Pillow. The score is pragmatic, not a formal proof.
exact_page_count=true and low mean_abs_diff are good signs.
"""
try:
import fitz # type: ignore
from PIL import ImageChops, ImageStat # type: ignore
except Exception as exc:
raise ConversionError("verify requires pymupdf and pillow: pip install pymupdf pillow") from exc
with tempfile.TemporaryDirectory(prefix="verify-") as tmp:
tmpdir = Path(tmp)
pdf1 = tmpdir / "original.pdf"
pdf2 = tmpdir / "converted.pdf"
convert_to_pdf(original_docx, pdf1, soffice_bin, timeout)
convert_to_pdf(converted_docx, pdf2, soffice_bin, timeout)
doc1 = fitz.open(str(pdf1))
doc2 = fitz.open(str(pdf2))
pages = min(len(doc1), len(doc2))
page_reports: List[Dict[str, object]] = []
zoom = dpi / 72.0
matrix = fitz.Matrix(zoom, zoom)
for i in range(pages):
p1 = doc1.load_page(i).get_pixmap(matrix=matrix, alpha=False)
p2 = doc2.load_page(i).get_pixmap(matrix=matrix, alpha=False)
img1 = p1.pil_image()
img2 = p2.pil_image()
same_size = img1.size == img2.size
if not same_size:
# Compare common area, record size mismatch.
w = min(img1.width, img2.width)
h = min(img1.height, img2.height)
img1 = img1.crop((0, 0, w, h))
img2 = img2.crop((0, 0, w, h))
diff = ImageChops.difference(img1, img2)
stat = ImageStat.Stat(diff)
mean_abs_diff = sum(stat.mean) / len(stat.mean)
page_reports.append({
"page": i + 1,
"same_size": same_size,
"mean_abs_diff_0_255": round(mean_abs_diff, 4),
})
report: Dict[str, object] = {
"original": str(original_docx.resolve()),
"converted": str(converted_docx.resolve()),
"original_sha256": sha256_file(original_docx),
"converted_sha256": sha256_file(converted_docx),
"original_pages": len(doc1),
"converted_pages": len(doc2),
"exact_page_count": len(doc1) == len(doc2),
"dpi": dpi,
"pages_compared": pages,
"page_reports": page_reports,
}
report_json.parent.mkdir(parents=True, exist_ok=True)
report_json.write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8")
return report
def cmd_docx2html(args: argparse.Namespace) -> None:
r = docx_to_html(
Path(args.input),
Path(args.output),
soffice_bin=args.soffice,
inline=not args.no_inline_css,
embed_source=args.embed_source,
timeout=args.timeout,
)
print(json.dumps(asdict(r), ensure_ascii=False, indent=2))
def cmd_html2docx(args: argparse.Namespace) -> None:
r = html_to_docx(
Path(args.input),
Path(args.output),
soffice_bin=args.soffice,
timeout=args.timeout,
prefer_embedded_source=args.prefer_embedded_source,
)
print(json.dumps(asdict(r), ensure_ascii=False, indent=2))
def cmd_roundtrip(args: argparse.Namespace) -> None:
src = Path(args.input).resolve()
outdir = Path(args.output_dir).resolve()
outdir.mkdir(parents=True, exist_ok=True)
html = outdir / f"{src.stem}.html"
docx_back = outdir / f"{src.stem}.roundtrip.docx"
report = outdir / f"{src.stem}.visual-report.json"
r1 = docx_to_html(src, html, args.soffice, inline=not args.no_inline_css, embed_source=args.embed_source, timeout=args.timeout)
r2 = html_to_docx(html, docx_back, args.soffice, timeout=args.timeout, prefer_embedded_source=args.prefer_embedded_source)
payload = {"docx2html": asdict(r1), "html2docx": asdict(r2)}
if args.verify:
payload["verify"] = verify_docx_visual_similarity(src, docx_back, report, args.soffice, args.dpi, args.timeout)
print(json.dumps(payload, ensure_ascii=False, indent=2))
def cmd_verify(args: argparse.Namespace) -> None:
report = verify_docx_visual_similarity(
Path(args.original),
Path(args.converted),
Path(args.report),
soffice_bin=args.soffice,
dpi=args.dpi,
timeout=args.timeout,
)
print(json.dumps(report, ensure_ascii=False, indent=2))
def build_parser() -> argparse.ArgumentParser:
p = argparse.ArgumentParser(description="Bidirectional high-fidelity DOCX <-> HTML converter using open-source tools.")
p.add_argument("--soffice", default=None, help="Path to LibreOffice soffice. Can also use SOFFICE_BIN env var.")
p.add_argument("--timeout", type=int, default=120, help="Conversion timeout seconds. Default: 120")
sub = p.add_subparsers(dest="command", required=True)
a = sub.add_parser("docx2html", help="Convert DOCX to HTML")
a.add_argument("input")
a.add_argument("output")
a.add_argument("--no-inline-css", action="store_true", help="Do not inline CSS from <style> tags")
a.add_argument("--embed-source", action="store_true", help="Embed original DOCX in HTML as base64 for archival fallback")
a.set_defaults(func=cmd_docx2html)
a = sub.add_parser("html2docx", help="Convert HTML to DOCX")
a.add_argument("input")
a.add_argument("output")
a.add_argument("--prefer-embedded-source", action="store_true", help="If HTML contains embedded original DOCX, restore it instead of converting")
a.set_defaults(func=cmd_html2docx)
a = sub.add_parser("roundtrip", help="DOCX -> HTML -> DOCX")
a.add_argument("input")
a.add_argument("output_dir")
a.add_argument("--no-inline-css", action="store_true")
a.add_argument("--embed-source", action="store_true")
a.add_argument("--prefer-embedded-source", action="store_true")
a.add_argument("--verify", action="store_true", help="Generate PDF-render visual diff report")
a.add_argument("--dpi", type=int, default=120)
a.set_defaults(func=cmd_roundtrip)
a = sub.add_parser("verify", help="Render two DOCX files to PDF and compare page images")
a.add_argument("original")
a.add_argument("converted")
a.add_argument("report")
a.add_argument("--dpi", type=int, default=120)
a.set_defaults(func=cmd_verify)
return p
def main(argv: Optional[List[str]] = None) -> int:
args = build_parser().parse_args(argv)
try:
args.func(args)
return 0
except ConversionError as exc:
print(f"ERROR: {exc}", file=sys.stderr)
return 2
if __name__ == "__main__":
raise SystemExit(main())