1917 lines
80 KiB
Python
1917 lines
80 KiB
Python
"""
|
|
app/modules/editor/pdf_controller.py
|
|
|
|
Block extraction -> pdfplumber (text/font/color/bbox) + pypdfium2 (image
|
|
region rasterization) + pypdf (embedded font byte access)
|
|
Save / apply edits -> pypdf (content-stream redaction, page structure) +
|
|
reportlab (new text/image rendering, font embedding) +
|
|
pypdf (merge overlay content back onto the page)
|
|
|
|
Pure Python PDF native block processing. All coordinates are in
|
|
PHYSICAL (unrotated) PDF-point space — the same convention PyMuPDF's
|
|
``get_text("dict")`` always used regardless of a page's ``/Rotate`` value
|
|
(text-extraction coordinates are rotation-invariant; only pixel rendering is
|
|
rotation-aware). ``pdfplumber``, by contrast, reports rotation-*adjusted*
|
|
coordinates, so every bbox pulled from it is converted back to physical
|
|
space via :mod:`app.modules.editor.pdf_geometry` before use, keeping this
|
|
module's output identical in convention to the original PyMuPDF-based one
|
|
(verified empirically against PyMuPDF's own output — see migration notes).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import io
|
|
import logging
|
|
import os
|
|
import re
|
|
from typing import Any, Dict, List, Optional, Tuple
|
|
|
|
import pdfplumber
|
|
import pypdfium2 as pdfium
|
|
from pypdf import PdfReader, PdfWriter
|
|
from pypdf.generic import ContentStream
|
|
from reportlab.pdfgen import canvas as rl_canvas
|
|
from reportlab.lib.utils import ImageReader
|
|
from reportlab.pdfbase import pdfmetrics
|
|
from reportlab.pdfbase.ttfonts import TTFont
|
|
|
|
from PIL import Image as PILImage, ImageStat
|
|
|
|
from app.modules.editor.pdf_geometry import (
|
|
Rect as PdfRect,
|
|
visual_to_physical_rect,
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# ─── Color / font helpers ──────────────────────────────────────────────────────
|
|
|
|
def _color_to_hex(color: Any) -> str:
|
|
if color is None:
|
|
return "#000000"
|
|
if isinstance(color, (list, tuple)):
|
|
if len(color) == 3:
|
|
r, g, b = [max(0, min(255, int(round(c * 255)))) for c in color]
|
|
return f"#{r:02x}{g:02x}{b:02x}"
|
|
if len(color) == 4:
|
|
# CMYK -> RGB
|
|
cc, mm, yy, kk = color
|
|
r = 255 * (1 - cc) * (1 - kk)
|
|
g = 255 * (1 - mm) * (1 - kk)
|
|
b = 255 * (1 - yy) * (1 - kk)
|
|
return f"#{int(round(r)):02x}{int(round(g)):02x}{int(round(b)):02x}"
|
|
if len(color) == 1:
|
|
v = max(0, min(255, int(round(color[0] * 255))))
|
|
return f"#{v:02x}{v:02x}{v:02x}"
|
|
if isinstance(color, (int, float)):
|
|
v = max(0, min(255, int(round(color * 255))))
|
|
return f"#{v:02x}{v:02x}{v:02x}"
|
|
return "#000000"
|
|
|
|
|
|
def _hex_to_rgb(hex_color: str) -> Tuple[float, float, float]:
|
|
h = (hex_color or "").lstrip("#")
|
|
if len(h) == 3:
|
|
h = "".join(c * 2 for c in h)
|
|
try:
|
|
r, g, b = int(h[0:2], 16), int(h[2:4], 16), int(h[4:6], 16)
|
|
return r / 255, g / 255, b / 255
|
|
except Exception:
|
|
return (0.0, 0.0, 0.0)
|
|
|
|
|
|
def _clean_font(font: str) -> str:
|
|
if font and "+" in font:
|
|
font = font.split("+", 1)[1]
|
|
return font or ""
|
|
|
|
|
|
def _infer_flags(fontname: str) -> int:
|
|
"""Name-only fallback signal, used when a font has no PDF FontDescriptor
|
|
to read real metadata from (see ``_describe_font`` for the primary path,
|
|
which prefers actual ``/Flags``, ``/FontWeight``, ``/ItalicAngle`` PDF
|
|
metadata over guessing from the name).
|
|
"""
|
|
n = (fontname or "").lower()
|
|
flags = 0
|
|
if "italic" in n or "oblique" in n:
|
|
flags |= 2
|
|
if any(k in n for k in ("times", "georgia", "garamond", "cambria", "serif", "roman")) and "sans" not in n:
|
|
flags |= 4
|
|
if any(k in n for k in ("courier", "mono", "consolas")):
|
|
flags |= 8
|
|
if "bold" in n or "black" in n or "heavy" in n:
|
|
flags |= 16
|
|
return flags
|
|
|
|
|
|
def _parse_flags(flags: int) -> Dict[str, bool]:
|
|
return {
|
|
"superscript": bool(flags & 1),
|
|
"italic": bool(flags & 2),
|
|
"serif": bool(flags & 4),
|
|
"monospace": bool(flags & 8),
|
|
"bold": bool(flags & 16),
|
|
}
|
|
|
|
|
|
_BLOCK_ID_RE = re.compile(r"^p(\d+)_(\d+)(?:_l(\d+))?(?:_s(\d+))?$")
|
|
|
|
|
|
# ─── Font metadata catalog (pypdf FontDescriptor introspection) ────────────────
|
|
#
|
|
# PDF spec (32000-1:2008 §9.8.2, Table 123) FontDescriptor /Flags bits:
|
|
_FLAG_FIXED_PITCH = 1
|
|
_FLAG_SERIF = 1 << 1
|
|
_FLAG_SYMBOLIC = 1 << 2
|
|
_FLAG_NONSYMBOLIC = 1 << 5
|
|
_FLAG_ITALIC = 1 << 6
|
|
_FLAG_FORCE_BOLD = 1 << 18
|
|
|
|
_WEIGHT_KEYWORDS: List[Tuple[int, Tuple[str, ...]]] = [
|
|
(900, ("black", "heavy")),
|
|
(800, ("extrabold", "ultrabold")),
|
|
(700, ("bold",)),
|
|
(600, ("semibold", "demibold")),
|
|
(500, ("medium",)),
|
|
(350, ("book",)),
|
|
(300, ("light",)),
|
|
(200, ("extralight", "ultralight")),
|
|
(100, ("thin", "hairline")),
|
|
]
|
|
|
|
|
|
def _weight_from_name(name_lower: str) -> Optional[int]:
|
|
for weight, keywords in _WEIGHT_KEYWORDS:
|
|
if any(k in name_lower for k in keywords):
|
|
return weight
|
|
return None
|
|
|
|
|
|
def _weight_name_from_number(weight: int) -> str:
|
|
if weight >= 850:
|
|
return "black"
|
|
if weight >= 750:
|
|
return "extrabold"
|
|
if weight >= 650:
|
|
return "bold"
|
|
if weight >= 550:
|
|
return "semibold"
|
|
if weight >= 450:
|
|
return "medium"
|
|
if weight >= 350:
|
|
return "regular"
|
|
if weight >= 250:
|
|
return "light"
|
|
if weight >= 150:
|
|
return "extralight"
|
|
return "thin"
|
|
|
|
|
|
def _repair_unsupported_cmap(font_bytes: bytes) -> bytes:
|
|
"""PDF font embedders (observed with reportlab's TrueType embedding, in
|
|
a synthetic-PDF test — likely others too) can ship a font whose only
|
|
``cmap`` subtable is a legacy Macintosh one (platform 1, e.g. format 6),
|
|
which is *not* Unicode-BMP. Chromium's OTS sanitizer only accepts
|
|
Windows-Unicode (platform 3, encoding 1 or 10) or platform-0 Unicode
|
|
cmap subtables and silently drops anything else — verified empirically:
|
|
``console.warn`` shows "OTS parsing error: cmap: no supported subtables
|
|
were found", and with no usable cmap left, the browser maps the wrong
|
|
glyph (or none) to each character, rendering garbled/overlapping text
|
|
even though the font "loaded" without throwing.
|
|
|
|
This adds a Windows-Unicode-BMP (platform 3, encoding 1, format 4)
|
|
subtable built from the existing subtable(s)' own (code -> glyph)
|
|
mapping — correct whenever those codes are already ASCII/Unicode
|
|
codepoints, which holds for the common case of a Latin-range PDF font;
|
|
the original subtable(s) are left in place, not replaced. No-op if a
|
|
browser-supported subtable already exists.
|
|
"""
|
|
try:
|
|
from fontTools.ttLib import TTFont
|
|
from fontTools.ttLib.tables._c_m_a_p import cmap_format_4
|
|
except Exception:
|
|
return font_bytes
|
|
|
|
try:
|
|
tt = TTFont(io.BytesIO(font_bytes), fontNumber=0, lazy=True)
|
|
except Exception:
|
|
return font_bytes
|
|
|
|
if "cmap" not in tt or not ("glyf" in tt or "CFF " in tt):
|
|
return font_bytes
|
|
|
|
try:
|
|
cmap_table = tt["cmap"]
|
|
has_supported = any(
|
|
(st.platformID == 3 and st.platEncID in (1, 10)) or st.platformID == 0
|
|
for st in cmap_table.tables
|
|
)
|
|
if has_supported:
|
|
return font_bytes
|
|
|
|
merged: Dict[int, str] = {}
|
|
for st in cmap_table.tables:
|
|
for code, glyph in st.cmap.items():
|
|
merged.setdefault(code, glyph)
|
|
if not merged:
|
|
return font_bytes
|
|
|
|
new_subtable = cmap_format_4(4)
|
|
new_subtable.platformID = 3
|
|
new_subtable.platEncID = 1
|
|
new_subtable.language = 0
|
|
new_subtable.cmap = merged
|
|
cmap_table.tables.append(new_subtable)
|
|
|
|
out = io.BytesIO()
|
|
tt.save(out)
|
|
repaired = out.getvalue()
|
|
logger.info(
|
|
"Repaired unsupported cmap in an embedded font: added Windows-Unicode subtable (%d code points, %d -> %d bytes)",
|
|
len(merged), len(font_bytes), len(repaired),
|
|
)
|
|
return repaired
|
|
except Exception as e:
|
|
logger.debug(f"cmap repair failed, using original font bytes: {e}")
|
|
return font_bytes
|
|
|
|
|
|
def _repair_missing_os2_table(font_bytes: bytes) -> bytes:
|
|
"""Some PDF font-subsetting tools (observed with GoNotoKurrent, likely
|
|
others) strip the OpenType ``OS/2`` table since PDF rendering itself
|
|
never needs it. fontTools and reportlab happily load such a font, but
|
|
browsers reject it outright (Chromium's OTS sanitizer: "OS/2: missing
|
|
required table"), so the frontend's real-embedded-font FontFace.load()
|
|
fails and every block using that font silently falls back to a generic
|
|
substitute — for a broad-coverage font like Noto that can mean missing
|
|
glyphs (tofu boxes) for whatever it was covering, not just a style
|
|
mismatch. Synthesizing a minimal, valid OS/2 table fixes browser loading
|
|
without altering any glyph outlines or metrics tables already present.
|
|
Returns the original bytes unchanged if the font already has an OS/2
|
|
table, isn't a TrueType/CFF font, or repair fails for any reason.
|
|
"""
|
|
try:
|
|
from fontTools.ttLib import TTFont, newTable
|
|
from fontTools.ttLib.tables.O_S_2f_2 import Panose
|
|
except Exception:
|
|
return font_bytes
|
|
|
|
try:
|
|
tt = TTFont(io.BytesIO(font_bytes), fontNumber=0, lazy=True)
|
|
except Exception:
|
|
return font_bytes
|
|
|
|
if "OS/2" in tt or not ("glyf" in tt or "CFF " in tt):
|
|
return font_bytes
|
|
|
|
try:
|
|
units_per_em = tt["head"].unitsPerEm if "head" in tt else 1000
|
|
hhea = tt["hhea"] if "hhea" in tt else None
|
|
ascent = hhea.ascent if hhea else int(units_per_em * 0.8)
|
|
descent = hhea.descent if hhea else -int(units_per_em * 0.2)
|
|
line_gap = hhea.lineGap if hhea else 0
|
|
|
|
os2 = newTable("OS/2")
|
|
os2.version = 4
|
|
os2.xAvgCharWidth = 0
|
|
os2.usWeightClass = 400
|
|
os2.usWidthClass = 5
|
|
os2.fsType = 0
|
|
os2.ySubscriptXSize = int(units_per_em * 0.65)
|
|
os2.ySubscriptYSize = int(units_per_em * 0.6)
|
|
os2.ySubscriptXOffset = 0
|
|
os2.ySubscriptYOffset = int(units_per_em * 0.075)
|
|
os2.ySuperscriptXSize = int(units_per_em * 0.65)
|
|
os2.ySuperscriptYSize = int(units_per_em * 0.6)
|
|
os2.ySuperscriptXOffset = 0
|
|
os2.ySuperscriptYOffset = int(units_per_em * 0.35)
|
|
os2.yStrikeoutSize = int(units_per_em * 0.05)
|
|
os2.yStrikeoutPosition = int(units_per_em * 0.22)
|
|
os2.sFamilyClass = 0
|
|
os2.panose = Panose()
|
|
os2.ulUnicodeRange1 = 1
|
|
os2.ulUnicodeRange2 = 0
|
|
os2.ulUnicodeRange3 = 0
|
|
os2.ulUnicodeRange4 = 0
|
|
os2.achVendID = "NONE"
|
|
os2.fsSelection = 0x40
|
|
os2.usFirstCharIndex = 0x20
|
|
os2.usLastCharIndex = 0xFFFF
|
|
os2.sTypoAscender = ascent
|
|
os2.sTypoDescender = descent
|
|
os2.sTypoLineGap = line_gap
|
|
os2.usWinAscent = max(ascent, 0)
|
|
os2.usWinDescent = abs(descent)
|
|
os2.ulCodePageRange1 = 1
|
|
os2.ulCodePageRange2 = 0
|
|
os2.sxHeight = int(units_per_em * 0.5)
|
|
os2.sCapHeight = int(units_per_em * 0.7)
|
|
os2.usDefaultChar = 0
|
|
os2.usBreakChar = 0x20
|
|
os2.usMaxContext = 0
|
|
tt["OS/2"] = os2
|
|
|
|
out = io.BytesIO()
|
|
tt.save(out)
|
|
repaired = out.getvalue()
|
|
logger.info("Repaired missing OS/2 table in an embedded font (%d -> %d bytes)", len(font_bytes), len(repaired))
|
|
return repaired
|
|
except Exception as e:
|
|
logger.debug(f"OS/2 table repair failed, using original font bytes: {e}")
|
|
return font_bytes
|
|
|
|
|
|
def _describe_font(base_font: str, subtype: str = "", descriptor: Any = None) -> Dict[str, Any]:
|
|
"""Build one font-metadata entry from real PDF data where available
|
|
(FontDescriptor ``/Flags``, ``/FontWeight``, ``/ItalicAngle``, embedded
|
|
font program), falling back to name-based inference only for whatever
|
|
the descriptor doesn't supply — e.g. non-embedded Base-14 references
|
|
normally carry no FontDescriptor at all.
|
|
"""
|
|
clean = base_font.split("+")[-1] if "+" in base_font else base_font
|
|
subset_prefix = base_font.split("+")[0] if "+" in base_font else ""
|
|
name_lower = clean.lower()
|
|
|
|
flags = 0
|
|
font_weight_meta: Optional[float] = None
|
|
italic_angle = 0.0
|
|
font_bytes: Optional[bytes] = None
|
|
font_program_type: Optional[str] = None
|
|
|
|
if descriptor is not None:
|
|
try:
|
|
descriptor = descriptor.get_object()
|
|
try:
|
|
flags = int(descriptor.get("/Flags", 0) or 0)
|
|
except Exception:
|
|
flags = 0
|
|
try:
|
|
fw = descriptor.get("/FontWeight")
|
|
font_weight_meta = float(fw) if fw is not None else None
|
|
except Exception:
|
|
font_weight_meta = None
|
|
try:
|
|
italic_angle = float(descriptor.get("/ItalicAngle", 0) or 0)
|
|
except Exception:
|
|
italic_angle = 0.0
|
|
for key, ptype in (("/FontFile2", "TrueType"), ("/FontFile3", "CFF/OpenType"), ("/FontFile", "Type1")):
|
|
stream_ref = descriptor.get(key)
|
|
if stream_ref is not None:
|
|
try:
|
|
data = stream_ref.get_object().get_data()
|
|
if data and len(data) > 64:
|
|
if ptype in ("TrueType", "CFF/OpenType"):
|
|
data = _repair_missing_os2_table(data)
|
|
data = _repair_unsupported_cmap(data)
|
|
font_bytes = data
|
|
font_program_type = ptype
|
|
except Exception:
|
|
pass
|
|
break
|
|
except Exception:
|
|
pass
|
|
|
|
name_says_bold = any(k in name_lower for k in ("bold", "black", "heavy"))
|
|
name_says_italic = "italic" in name_lower
|
|
name_says_oblique = "oblique" in name_lower
|
|
is_force_bold = bool(flags & _FLAG_FORCE_BOLD)
|
|
|
|
weight = font_weight_meta if font_weight_meta else _weight_from_name(name_lower)
|
|
if weight is None:
|
|
weight = 700 if (is_force_bold or name_says_bold) else 400
|
|
weight = int(weight)
|
|
|
|
bold = weight >= 600 or is_force_bold or name_says_bold
|
|
italic = bool(flags & _FLAG_ITALIC) or italic_angle != 0 or name_says_italic or name_says_oblique
|
|
oblique = name_says_oblique and not name_says_italic
|
|
|
|
is_symbolic = bool(flags & _FLAG_SYMBOLIC) and not bool(flags & _FLAG_NONSYMBOLIC)
|
|
monospace = bool(flags & _FLAG_FIXED_PITCH) or any(k in name_lower for k in ("courier", "mono", "consolas"))
|
|
serif = bool(flags & _FLAG_SERIF) or (
|
|
any(k in name_lower for k in ("times", "georgia", "garamond", "cambria", "roman", "serif")) and "sans" not in name_lower
|
|
)
|
|
symbolic = is_symbolic or any(k in name_lower for k in ("symbol", "wingdings", "zapfdingbats", "dingbat"))
|
|
|
|
return {
|
|
"raw_name": base_font,
|
|
"clean_name": clean,
|
|
"subset_prefix": subset_prefix,
|
|
"family": _font_family_base(clean),
|
|
"subtype": subtype or "Unknown",
|
|
"embedded": font_bytes is not None,
|
|
"font_program_type": font_program_type,
|
|
"bytes": font_bytes,
|
|
"weight": weight,
|
|
"weight_name": _weight_name_from_number(weight),
|
|
"bold": bold,
|
|
"italic": italic,
|
|
"oblique": oblique,
|
|
"serif": serif,
|
|
"monospace": monospace,
|
|
"symbolic": symbolic,
|
|
}
|
|
|
|
|
|
def _build_font_catalog(reader: PdfReader) -> Dict[str, Dict[str, Any]]:
|
|
"""Build a per-document font metadata catalog keyed by the raw
|
|
``/BaseFont`` name, covering Type1, TrueType, CFF/OpenType (``/FontFile3``),
|
|
and CID/Type0 composite fonts (descriptor read from the descendant font).
|
|
Replaces the old bytes-only ``_extract_all_fonts`` with real PDF
|
|
FontDescriptor metadata (weight/italic/serif/monospace/subtype/embedded)
|
|
in addition to the font program bytes.
|
|
"""
|
|
catalog: Dict[str, Dict[str, Any]] = {}
|
|
for page in reader.pages:
|
|
try:
|
|
resources = page.get("/Resources")
|
|
if not resources:
|
|
continue
|
|
resources = resources.get_object()
|
|
font_dict = resources.get("/Font")
|
|
if not font_dict:
|
|
continue
|
|
font_dict = font_dict.get_object()
|
|
for _, font_ref in font_dict.items():
|
|
try:
|
|
font_obj = font_ref.get_object()
|
|
base_font = str(font_obj.get("/BaseFont", "")).lstrip("/")
|
|
if not base_font or base_font in catalog:
|
|
continue
|
|
subtype = str(font_obj.get("/Subtype", "")).lstrip("/")
|
|
descriptor = font_obj.get("/FontDescriptor")
|
|
effective_subtype = subtype
|
|
if descriptor is None and subtype == "Type0":
|
|
descendants = font_obj.get("/DescendantFonts")
|
|
if descendants:
|
|
desc_font = descendants.get_object()[0].get_object()
|
|
descriptor = desc_font.get("/FontDescriptor")
|
|
effective_subtype = str(desc_font.get("/Subtype", subtype)).lstrip("/") or subtype
|
|
catalog[base_font] = _describe_font(base_font, effective_subtype, descriptor)
|
|
except Exception as e:
|
|
logger.debug(f"Font catalog: skipped one font resource: {e}")
|
|
except Exception as e:
|
|
logger.debug(f"Font catalog: failed for a page: {e}")
|
|
return catalog
|
|
|
|
|
|
_registered_font_names: Dict[str, str] = {}
|
|
_font_registration_counter = 0
|
|
|
|
|
|
def _font_family_base(clean_name: str) -> str:
|
|
"""Strip a trailing weight/style suffix ("Georgia-BoldItalic" -> "georgia")
|
|
so different variants of the same embedded font can be grouped."""
|
|
return re.split(r"[-,]", clean_name)[0].lower().replace(" ", "")
|
|
|
|
|
|
def _register_reportlab_font(font_bytes: bytes) -> Optional[str]:
|
|
"""Register raw TTF/OTF bytes with reportlab's font cache and return the
|
|
registered font name, or None if the bytes can't be parsed as a font
|
|
(e.g. Type1/CFF programs reportlab's TTFont can't load).
|
|
"""
|
|
global _font_registration_counter
|
|
key = str(hash(font_bytes))
|
|
if key in _registered_font_names:
|
|
return _registered_font_names[key]
|
|
try:
|
|
_font_registration_counter += 1
|
|
name = f"DQF{_font_registration_counter}"
|
|
pdfmetrics.registerFont(TTFont(name, io.BytesIO(font_bytes)))
|
|
_registered_font_names[key] = name
|
|
return name
|
|
except Exception as e:
|
|
logger.debug(f"Could not register embedded font with reportlab: {e}")
|
|
return None
|
|
|
|
|
|
# ─── System font discovery (stage 4 of the fallback pipeline) ─────────────────
|
|
|
|
_SYSTEM_FONT_DIRS = [
|
|
r"C:\Windows\Fonts",
|
|
"/usr/share/fonts",
|
|
"/usr/local/share/fonts",
|
|
os.path.expanduser("~/.fonts"),
|
|
"/System/Library/Fonts",
|
|
"/Library/Fonts",
|
|
]
|
|
|
|
_system_font_index: Optional[Dict[str, str]] = None
|
|
_registered_font_files: Dict[str, str] = {}
|
|
|
|
# Metric-compatible substitutes for common commercial fonts that are rarely
|
|
# actually installed on a Linux server, so "compatible locally available
|
|
# font" degrades gracefully (e.g. Arial -> Liberation Sans / DejaVu Sans,
|
|
# the same substitution LibreOffice/fontconfig make by convention).
|
|
_FAMILY_SUBSTITUTES: Dict[str, Tuple[str, ...]] = {
|
|
"arial": ("arial", "liberationsans", "dejavusans"),
|
|
"helvetica": ("arial", "helvetica", "liberationsans", "dejavusans"),
|
|
"timesnewroman": ("timesnewroman", "times", "liberationserif", "dejavuserif"),
|
|
"times": ("times", "timesnewroman", "liberationserif", "dejavuserif"),
|
|
"couriernew": ("couriernew", "courier", "liberationmono", "dejavusansmono"),
|
|
"courier": ("courier", "couriernew", "liberationmono", "dejavusansmono"),
|
|
"calibri": ("calibri", "carlito", "liberationsans"),
|
|
"cambria": ("cambria", "caladea", "liberationserif"),
|
|
"verdana": ("verdana", "dejavusans"),
|
|
"georgia": ("georgia", "gelasio", "liberationserif"),
|
|
"tahoma": ("tahoma", "dejavusans"),
|
|
}
|
|
|
|
_WEIGHT_STYLE_SUFFIXES: Dict[Tuple[bool, bool], Tuple[str, ...]] = {
|
|
# "z" alone covers Windows's single-letter BoldItalic convention used by
|
|
# e.g. Georgia (georgiaz.ttf), Comic Sans, Verdana.
|
|
(True, True): ("bolditalic", "bi", "boldoblique", "bz", "z"),
|
|
(True, False): ("bold", "bd", "b"),
|
|
(False, True): ("italic", "oblique", "i", "it"),
|
|
(False, False): ("regular", "", "mt", "psmt"),
|
|
}
|
|
|
|
|
|
def _get_system_font_index() -> Dict[str, str]:
|
|
"""Lazily build (once per process) an index of every TTF/OTF/TTC file
|
|
under the common OS font directories, keyed by lowercase filename stem.
|
|
"""
|
|
global _system_font_index
|
|
if _system_font_index is not None:
|
|
return _system_font_index
|
|
index: Dict[str, str] = {}
|
|
for base_dir in _SYSTEM_FONT_DIRS:
|
|
if not base_dir or not os.path.isdir(base_dir):
|
|
continue
|
|
try:
|
|
for root, _, files in os.walk(base_dir):
|
|
for fname in files:
|
|
if fname.lower().endswith((".ttf", ".otf", ".ttc")):
|
|
stem = re.sub(r"\.(ttf|otf|ttc)$", "", fname.lower()).replace(" ", "").replace("_", "")
|
|
index.setdefault(stem, os.path.join(root, fname))
|
|
except Exception as e:
|
|
logger.debug(f"System font scan failed for {base_dir}: {e}")
|
|
_system_font_index = index
|
|
return index
|
|
|
|
|
|
def _find_system_font_file(family_hint: str, want_bold: bool, want_italic: bool) -> Optional[str]:
|
|
"""Best-effort discovery of a locally installed font file matching
|
|
``family_hint`` at the desired weight/style. Returns a filesystem path,
|
|
or None if nothing plausible was found.
|
|
"""
|
|
index = _get_system_font_index()
|
|
if not index or not family_hint:
|
|
return None
|
|
|
|
base = _font_family_base(family_hint.split("+")[-1])
|
|
if not base:
|
|
return None
|
|
|
|
candidates = _FAMILY_SUBSTITUTES.get(base, (base,))
|
|
suffixes = _WEIGHT_STYLE_SUFFIXES[(want_bold, want_italic)]
|
|
|
|
for fam in candidates:
|
|
fam_key = fam.replace(" ", "")
|
|
for stem, path in index.items():
|
|
if not stem.startswith(fam_key):
|
|
continue
|
|
remainder = stem[len(fam_key):].lstrip("-")
|
|
if not remainder:
|
|
# Bare family file (e.g. "georgia.ttf") only counts as a
|
|
# match for the plain regular/upright request — an empty
|
|
# remainder must never satisfy a bold/italic suffix check.
|
|
if not want_bold and not want_italic:
|
|
return path
|
|
continue
|
|
if any(remainder == suf for suf in suffixes if suf):
|
|
return path
|
|
|
|
return None
|
|
|
|
|
|
def _register_reportlab_font_file(path: str) -> Optional[str]:
|
|
global _font_registration_counter
|
|
if path in _registered_font_files:
|
|
return _registered_font_files[path]
|
|
try:
|
|
_font_registration_counter += 1
|
|
name = f"DQFS{_font_registration_counter}"
|
|
pdfmetrics.registerFont(TTFont(name, path))
|
|
_registered_font_files[path] = name
|
|
return name
|
|
except Exception as e:
|
|
logger.debug(f"Could not register system font {path}: {e}")
|
|
return None
|
|
|
|
|
|
def _font_supports_text(font_name: str, text: str) -> bool:
|
|
"""Check if reportlab font has valid glyphs for all non-whitespace characters in text."""
|
|
if not text or not font_name:
|
|
return True
|
|
try:
|
|
font = pdfmetrics.getFont(font_name)
|
|
if hasattr(font, "face") and hasattr(font.face, "charToGlyph"):
|
|
c2g = font.face.charToGlyph
|
|
for ch in text:
|
|
code = ord(ch)
|
|
if code <= 32:
|
|
continue
|
|
gid = c2g.get(code, 0)
|
|
if gid == 0:
|
|
return False
|
|
return True
|
|
except Exception:
|
|
return True
|
|
|
|
|
|
def _resolve_font_for_render(
|
|
raw_font: str, want_bold: bool, want_italic: bool, font_catalog: Dict[str, Dict[str, Any]],
|
|
text: str = "",
|
|
) -> Tuple[str, str]:
|
|
"""Resolve the best available font for rendering ``raw_font`` at the
|
|
desired (bold, italic) style. Six-stage fallback pipeline, returning
|
|
``(reportlab_font_name, stage)``:
|
|
|
|
1. exact_embedded — this PDF's own embedded font, same family,
|
|
exact weight/style match, supports all chars in text
|
|
2. exact_system — a locally installed font, same family
|
|
(or metric-compatible substitute), exact
|
|
weight/style match, complete character set
|
|
3. compatible_embedded — same family embedded in this PDF, but no
|
|
variant matches the desired style exactly
|
|
4/5. category_system — a system font matching just the broad
|
|
category (serif/sans/monospace), exact style
|
|
6. base14 — always available; always renders the
|
|
requested weight/style correctly
|
|
"""
|
|
clean = raw_font.split("+")[-1] if raw_font else ""
|
|
base_key = _font_family_base(clean)
|
|
own_entry = font_catalog.get(raw_font, {}) if raw_font else {}
|
|
serif_hint = bool(own_entry.get("serif"))
|
|
mono_hint = bool(own_entry.get("monospace"))
|
|
|
|
def style_score(entry: Dict[str, Any]) -> int:
|
|
return (2 if bool(entry.get("bold")) == want_bold else 0) + (2 if bool(entry.get("italic")) == want_italic else 0)
|
|
|
|
best_embedded: Optional[Dict[str, Any]] = None
|
|
best_score = -1
|
|
if base_key:
|
|
for entry in font_catalog.values():
|
|
if not entry.get("embedded") or not entry.get("bytes"):
|
|
continue
|
|
fam = entry.get("family", "")
|
|
if not (fam == base_key or fam.startswith(base_key) or base_key.startswith(fam)):
|
|
continue
|
|
serif_hint = serif_hint or bool(entry.get("serif"))
|
|
mono_hint = mono_hint or bool(entry.get("monospace"))
|
|
s = style_score(entry)
|
|
if s > best_score:
|
|
best_score, best_embedded = s, entry
|
|
|
|
# Stage 1: exact embedded family + exact style (only if it supports all characters in text).
|
|
if best_embedded is not None and best_score == 4:
|
|
reg = _register_reportlab_font(best_embedded["bytes"])
|
|
if reg and _font_supports_text(reg, text):
|
|
return reg, "exact_embedded"
|
|
|
|
# Stage 2: exact system font (same family / metric-compatible substitute).
|
|
sys_path = _find_system_font_file(clean or raw_font, want_bold, want_italic)
|
|
if sys_path:
|
|
reg = _register_reportlab_font_file(sys_path)
|
|
if reg and _font_supports_text(reg, text):
|
|
return reg, "exact_system"
|
|
|
|
# Stage 3: same family embedded, best available (imperfect) variant.
|
|
if best_embedded is not None and best_score >= 2:
|
|
reg = _register_reportlab_font(best_embedded["bytes"])
|
|
if reg and _font_supports_text(reg, text):
|
|
return reg, "compatible_embedded"
|
|
|
|
# Stage 4/5: category-compatible system font (serif/sans/monospace).
|
|
category_name = "Courier New" if mono_hint else ("Georgia" if serif_hint else "Arial")
|
|
sys_path = _find_system_font_file(category_name, want_bold, want_italic)
|
|
if sys_path:
|
|
reg = _register_reportlab_font_file(sys_path)
|
|
if reg:
|
|
return reg, "category_system"
|
|
|
|
# Stage 6: Base-14 — always available, always renders the requested style.
|
|
flags = 0
|
|
if want_bold:
|
|
flags |= 16
|
|
if want_italic:
|
|
flags |= 2
|
|
if serif_hint:
|
|
flags |= 4
|
|
if mono_hint:
|
|
flags |= 8
|
|
return _resolve_base14(clean or raw_font, flags), "base14"
|
|
|
|
|
|
def _resolve_base14(font_name: str, flags: int = 0) -> str:
|
|
"""Map an arbitrary font name (+ inferred flags) to a Base-14 reportlab
|
|
font name, mirroring the original ``_resolve_base14``/``_resolve_builtin_font``
|
|
heuristics in the fitz-based implementation.
|
|
"""
|
|
n = (font_name or "").lower()
|
|
is_bold = bool(flags & 16) or "bold" in n or "black" in n or "heavy" in n
|
|
is_italic = bool(flags & 2) or "italic" in n or "oblique" in n
|
|
is_mono = bool(flags & 8) or any(k in n for k in ("courier", "mono", "consolas"))
|
|
is_serif = bool(flags & 4) or (
|
|
any(k in n for k in ("times", "georgia", "garamond", "cambria", "roman", "serif")) and "sans" not in n
|
|
)
|
|
|
|
if is_mono:
|
|
if is_bold and is_italic:
|
|
return "Courier-BoldOblique"
|
|
if is_bold:
|
|
return "Courier-Bold"
|
|
if is_italic:
|
|
return "Courier-Oblique"
|
|
return "Courier"
|
|
if is_serif:
|
|
if is_bold and is_italic:
|
|
return "Times-BoldItalic"
|
|
if is_bold:
|
|
return "Times-Bold"
|
|
if is_italic:
|
|
return "Times-Italic"
|
|
return "Times-Roman"
|
|
if is_bold and is_italic:
|
|
return "Helvetica-BoldOblique"
|
|
if is_bold:
|
|
return "Helvetica-Bold"
|
|
if is_italic:
|
|
return "Helvetica-Oblique"
|
|
return "Helvetica"
|
|
|
|
|
|
def _text_width(text: str, fontname: str, size: float) -> float:
|
|
try:
|
|
return pdfmetrics.stringWidth(text, fontname, size)
|
|
except Exception:
|
|
return len(text) * size * 0.55
|
|
|
|
|
|
# ─── Underline detection (pdfplumber rects/lines, replaces fitz get_drawings) ──
|
|
|
|
def _detect_underlines(spans: List[Dict[str, Any]], drawings: List[Dict[str, float]]) -> None:
|
|
"""Heuristic: a thin horizontal drawn line/rect positioned just below a
|
|
span's baseline is treated as an underline, same approach as the
|
|
original fitz-based implementation.
|
|
"""
|
|
for span in spans:
|
|
sb = span.get("bbox")
|
|
if not sb:
|
|
continue
|
|
sb_width = sb[2] - sb[0]
|
|
fs = span.get("size", 12) or 12
|
|
baseline_approx = sb[3] - fs * 0.15 # bbox bottom sits slightly below the true baseline
|
|
v_thresh = max(3, fs * 0.25)
|
|
|
|
for d in drawings:
|
|
dx0, dy0, dx1, dy1 = d["x0"], d["y0"], d["x1"], d["y1"]
|
|
width_d = dx1 - dx0
|
|
height_d = dy1 - dy0
|
|
if width_d < 2 or height_d > 2.5:
|
|
continue
|
|
ix0 = max(sb[0], dx0)
|
|
ix1 = min(sb[2], dx1)
|
|
overlap_width = ix1 - ix0
|
|
if overlap_width < 0.5 * sb_width and overlap_width < 5:
|
|
continue
|
|
dist_to_baseline = dy0 - baseline_approx
|
|
dist_to_bottom = dy0 - sb[3]
|
|
if (-2 <= dist_to_baseline <= v_thresh) or (-2 <= dist_to_bottom <= v_thresh):
|
|
span.setdefault("style", {})["underline"] = True
|
|
break
|
|
|
|
|
|
# ─── Block extraction ───────────────────────────────────────────────────────────
|
|
|
|
def _chars_to_spans(chars: List[Dict[str, Any]], font_catalog: Dict[str, Dict[str, Any]]) -> List[Dict[str, Any]]:
|
|
"""Group characters into spans of uniform (font, size, color).
|
|
|
|
pdfplumber's char list omits literal space glyphs entirely (it
|
|
reconstructs word gaps heuristically in its own higher-level text
|
|
methods), so a horizontal-gap heuristic re-inserts inferred spaces here
|
|
— otherwise words would run together with no whitespace at all.
|
|
|
|
Style metadata (bold/italic/serif/monospace/...) is looked up from
|
|
``font_catalog`` — real PDF FontDescriptor data — falling back to
|
|
``_infer_flags``'s name-only heuristic only when the font isn't in the
|
|
catalog (e.g. a non-embedded Base-14 reference with no descriptor).
|
|
"""
|
|
spans: List[Dict[str, Any]] = []
|
|
cur: Optional[Dict[str, Any]] = None
|
|
cur_key = None
|
|
prev_x1: Optional[float] = None
|
|
prev_text: str = ""
|
|
|
|
for ch in chars:
|
|
text = ch.get("text", "")
|
|
font = ch.get("fontname", "") or ""
|
|
size = round(float(ch.get("size", 12) or 12), 2)
|
|
color = _color_to_hex(ch.get("non_stroking_color"))
|
|
key = (font, size, color)
|
|
|
|
# The gap after a numbered heading's period (e.g. "2. Job Title...")
|
|
# was empirically measured (real document) at ~0.144x font size —
|
|
# tighter than a genuine inter-word space (>=0.207x in the same
|
|
# document) so the general 0.22 threshold below missed it. A flat
|
|
# lower threshold isn't safe though: measured against a math-heavy
|
|
# PDF, plain intra-expression kerning before an open-paren (e.g.
|
|
# "L(z)", "d(...)") sits at ~0.109-0.124x, right in that same gap,
|
|
# so it would misfire there. Since every real case of the narrow
|
|
# heading-gap was specifically a period followed by a letter, that
|
|
# combination alone gets the lower, more sensitive threshold;
|
|
# everything else keeps the original, already-verified-safe 0.22.
|
|
is_period_then_letter = (
|
|
prev_text == "." and text[:1].isalpha() if prev_x1 is not None else False
|
|
)
|
|
threshold = 0.10 if is_period_then_letter else 0.22
|
|
if prev_x1 is not None and ch["x0"] - prev_x1 > size * threshold:
|
|
text = " " + text
|
|
prev_x1 = ch["x1"]
|
|
prev_text = ch.get("text", "")
|
|
|
|
if cur is not None and cur_key == key:
|
|
cur["text"] += text
|
|
cur["bbox"][0] = min(cur["bbox"][0], ch["x0"])
|
|
cur["bbox"][1] = min(cur["bbox"][1], ch["top"])
|
|
cur["bbox"][2] = max(cur["bbox"][2], ch["x1"])
|
|
cur["bbox"][3] = max(cur["bbox"][3], ch["bottom"])
|
|
else:
|
|
if cur is not None:
|
|
spans.append(cur)
|
|
info = font_catalog.get(font)
|
|
if info is not None:
|
|
bold, italic = info["bold"], info["italic"]
|
|
flags = (16 if bold else 0) | (2 if italic else 0) | (4 if info["serif"] else 0) | (8 if info["monospace"] else 0)
|
|
else:
|
|
flags = _infer_flags(font)
|
|
bold, italic = bool(flags & 16), bool(flags & 2)
|
|
cur = {
|
|
"text": text,
|
|
"font": _clean_font(font),
|
|
"raw_font": font,
|
|
"size": size,
|
|
"color": color,
|
|
"flags": flags,
|
|
"style": _parse_flags(flags),
|
|
"ascender": 0.8,
|
|
"origin": [round(ch["x0"], 2), round(ch["bottom"] - size * 0.2, 2)],
|
|
"bbox": [ch["x0"], ch["top"], ch["x1"], ch["bottom"]],
|
|
# Richer, PDF-metadata-driven fields (additive — existing
|
|
# consumers of flags/style are unaffected).
|
|
"weight": info["weight"] if info else (700 if bold else 400),
|
|
"weight_name": info["weight_name"] if info else ("bold" if bold else "regular"),
|
|
"subtype": info["subtype"] if info else "Unknown",
|
|
"embedded": bool(info and info["embedded"]),
|
|
"font_program_type": info["font_program_type"] if info else None,
|
|
"oblique": bool(info and info["oblique"]),
|
|
"symbolic": bool(info and info["symbolic"]),
|
|
}
|
|
cur_key = key
|
|
|
|
if cur is not None:
|
|
spans.append(cur)
|
|
|
|
for s in spans:
|
|
s["bbox"] = [round(v, 2) for v in s["bbox"]]
|
|
return spans
|
|
|
|
|
|
def _extract_page_lines(
|
|
pl_page, phys_w: float, phys_h: float, rotation: int, font_catalog: Dict[str, Dict[str, Any]],
|
|
) -> List[Dict[str, Any]]:
|
|
def to_physical(x0: float, top: float, x1: float, bottom: float) -> List[float]:
|
|
r = visual_to_physical_rect(PdfRect(x0, top, x1, bottom), phys_w, phys_h, rotation)
|
|
return [round(r.x0, 2), round(r.y0, 2), round(r.x1, 2), round(r.y1, 2)]
|
|
|
|
drawings: List[Dict[str, float]] = []
|
|
for d in list(pl_page.rects) + list(pl_page.lines):
|
|
try:
|
|
bbox = to_physical(d["x0"], d["top"], d["x1"], d["bottom"])
|
|
drawings.append({"x0": bbox[0], "y0": bbox[1], "x1": bbox[2], "y1": bbox[3]})
|
|
except Exception:
|
|
continue
|
|
|
|
try:
|
|
raw_lines = pl_page.extract_text_lines(return_chars=True, strip=False)
|
|
except Exception as e:
|
|
logger.warning(f"extract_text_lines failed for a page: {e}")
|
|
raw_lines = []
|
|
|
|
# pdfplumber's extract_text_lines(return_chars=True) has a real bug (seen
|
|
# with real-world, professionally-typeset PDFs) where the per-line
|
|
# `chars` list it returns can be corrupted for fonts using ligature
|
|
# glyphs ("ff", "fi", "fl"): spaces are dropped and the ligature's
|
|
# characters come out duplicated (e.g. "different" -> "difffferent").
|
|
# `rl["text"]` and the page's raw `page.chars` are both unaffected, so
|
|
# each line's character list is rebuilt here from `page.chars` instead
|
|
# of trusting `rl["chars"]` — scoped to that line's own vertical AND
|
|
# horizontal span (taken from the otherwise-position-correct buggy list)
|
|
# so multi-column layouts don't bleed text across columns.
|
|
all_page_chars = pl_page.chars
|
|
|
|
def _line_chars(rl: Dict[str, Any]) -> List[Dict[str, Any]]:
|
|
buggy_chars = rl.get("chars") or []
|
|
if not buggy_chars:
|
|
return []
|
|
top, bottom = rl.get("top"), rl.get("bottom")
|
|
if top is None or bottom is None:
|
|
return buggy_chars
|
|
pad = 0.75
|
|
x0b = min(c["x0"] for c in buggy_chars) - pad
|
|
x1b = max(c["x1"] for c in buggy_chars) + pad
|
|
selected = [
|
|
c for c in all_page_chars
|
|
if top - pad <= c["top"] <= bottom + pad and x0b <= c["x0"] and c["x1"] <= x1b
|
|
]
|
|
selected.sort(key=lambda c: c["x0"])
|
|
return selected or buggy_chars
|
|
|
|
out_lines: List[Dict[str, Any]] = []
|
|
for rl in raw_lines:
|
|
chars = _line_chars(rl)
|
|
if not chars:
|
|
continue
|
|
# Convert each char's coordinates to physical space before grouping into spans.
|
|
phys_chars = []
|
|
for ch in chars:
|
|
if not (ch.get("text") or "").strip() and ch.get("text") != " ":
|
|
continue
|
|
bbox = to_physical(ch["x0"], ch["top"], ch["x1"], ch["bottom"])
|
|
phys_chars.append({**ch, "x0": bbox[0], "top": bbox[1], "x1": bbox[2], "bottom": bbox[3]})
|
|
if not phys_chars:
|
|
continue
|
|
|
|
spans = _chars_to_spans(phys_chars, font_catalog)
|
|
if not spans:
|
|
continue
|
|
_detect_underlines(spans, drawings)
|
|
|
|
x0 = min(s["bbox"][0] for s in spans)
|
|
y0 = min(s["bbox"][1] for s in spans)
|
|
x1 = max(s["bbox"][2] for s in spans)
|
|
y1 = max(s["bbox"][3] for s in spans)
|
|
out_lines.append({"bbox": [round(x0, 2), round(y0, 2), round(x1, 2), round(y1, 2)], "spans": spans})
|
|
|
|
return out_lines
|
|
|
|
|
|
def _dominant_span(line: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""The span covering the most (non-whitespace) text in a line — used as
|
|
that line's representative font for block-grouping decisions.
|
|
"""
|
|
spans = line.get("spans") or []
|
|
if not spans:
|
|
return {}
|
|
return max(spans, key=lambda s: len((s.get("text") or "").strip()))
|
|
|
|
|
|
def _group_lines_into_blocks(lines: List[Dict[str, Any]]) -> List[List[Dict[str, Any]]]:
|
|
"""Group text lines into logical editable blocks (paragraphs).
|
|
|
|
A new block starts whenever the vertical gap is too large, the lines
|
|
don't share horizontal extent (a likely column boundary), OR the
|
|
dominant font size/family changes noticeably between consecutive lines.
|
|
The size/family checks are what keep a heading from being swallowed
|
|
into a paragraph that happens to sit close beneath it — a fixed
|
|
vertical-gap threshold alone can't distinguish "next line of this
|
|
paragraph" from "a different, closely-set block" when font sizes differ.
|
|
"""
|
|
if not lines:
|
|
return []
|
|
ordered = sorted(lines, key=lambda l: l["bbox"][1])
|
|
blocks: List[List[Dict[str, Any]]] = [[ordered[0]]]
|
|
|
|
for prev, this in zip(ordered, ordered[1:]):
|
|
prev_bottom = prev["bbox"][3]
|
|
this_top = this["bbox"][1]
|
|
prev_dom = _dominant_span(prev)
|
|
this_dom = _dominant_span(this)
|
|
prev_size = prev_dom.get("size") or max((s.get("size", 12) for s in prev["spans"]), default=12)
|
|
this_size = this_dom.get("size") or prev_size
|
|
gap = this_top - prev_bottom
|
|
|
|
prev_x0, prev_x1 = prev["bbox"][0], prev["bbox"][2]
|
|
this_x0, this_x1 = this["bbox"][0], this["bbox"][2]
|
|
h_overlap = min(prev_x1, this_x1) - max(prev_x0, this_x0)
|
|
|
|
gap_ok = gap <= max(4.0, prev_size * 0.9) and (h_overlap > 0 or gap <= prev_size * 0.35)
|
|
|
|
size_ratio = max(prev_size, this_size, 0.1) / max(min(prev_size, this_size), 0.1)
|
|
size_compatible = size_ratio <= 1.25
|
|
|
|
prev_family = _font_family_base(prev_dom.get("font") or "")
|
|
this_family = _font_family_base(this_dom.get("font") or "")
|
|
family_compatible = not prev_family or not this_family or prev_family == this_family
|
|
|
|
same_block = gap_ok and size_compatible and family_compatible
|
|
if same_block:
|
|
blocks[-1].append(this)
|
|
else:
|
|
blocks.append([this])
|
|
|
|
return blocks
|
|
|
|
|
|
def _render_page_image(pdfium_page) -> Tuple[Any, float]:
|
|
zoom = 2.0 # matrix(2,2)-equivalent default; used only as a base for crops
|
|
bitmap = pdfium_page.render(scale=zoom, fill_color=(0, 0, 0, 0))
|
|
return bitmap.to_pil(), zoom
|
|
|
|
|
|
def _crop_to_base64_png(page_image, zoom: float, bbox: List[float]) -> Optional[str]:
|
|
try:
|
|
x0, y0, x1, y1 = bbox
|
|
px0, py0 = max(0, int(x0 * zoom)), max(0, int(y0 * zoom))
|
|
px1, py1 = min(page_image.width, int(x1 * zoom)), min(page_image.height, int(y1 * zoom))
|
|
if px1 <= px0 or py1 <= py0:
|
|
return None
|
|
crop = page_image.crop((px0, py0, px1, py1))
|
|
|
|
# Filter out blank white regions (e.g. erased/covered image areas)
|
|
grayscale = crop.convert("L")
|
|
stat = ImageStat.Stat(grayscale)
|
|
if stat.stddev[0] < 2.0 and stat.mean[0] > 250:
|
|
return None
|
|
|
|
buf = io.BytesIO()
|
|
crop.save(buf, format="PNG")
|
|
return f"data:image/png;base64,{base64.b64encode(buf.getvalue()).decode()}"
|
|
except Exception as e:
|
|
logger.debug(f"Image crop extraction failed: {e}")
|
|
return None
|
|
|
|
|
|
def _tight_alpha_content_rect(pil_img: "PILImage.Image") -> Optional[Tuple[int, int, int, int]]:
|
|
"""Return the pixel bbox (left, top, right, bottom, in the image's own
|
|
coordinate space) of its actually-visible (non-transparent) content.
|
|
|
|
Many decorative/background image XObjects are placed at a much larger
|
|
rectangle than their visible artwork actually occupies — the rest of
|
|
that rectangle is transparent padding baked into the PNG itself. Since
|
|
``extract_blocks`` currently uses the XObject's full placement rect as
|
|
the block's bbox, that padding gets attributed to the image block too,
|
|
letting it visually swallow unrelated content (e.g. a text block) that
|
|
happens to sit in the padded, transparent area. Only images with a real
|
|
alpha channel are tightened here — an opaque photo/JPEG has no reliable
|
|
"background" to detect, so it's left untouched rather than risk cropping
|
|
real photo content.
|
|
"""
|
|
if pil_img.mode not in ("RGBA", "LA") and not (pil_img.mode == "P" and "transparency" in pil_img.info):
|
|
return None
|
|
try:
|
|
alpha = pil_img.convert("RGBA").split()[-1]
|
|
return alpha.getbbox()
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def _tight_opaque_content_rect(pil_img: "PILImage.Image") -> Optional[Tuple[int, int, int, int]]:
|
|
"""Same idea as :func:`_tight_alpha_content_rect`, for images with no
|
|
alpha channel at all. Decorative graphics are very often exported
|
|
pre-flattened onto a plain white background rather than kept
|
|
transparent (a common side-effect of "save as picture" round-trips
|
|
from PowerPoint/Canva-style tools), so the same oversized-placement
|
|
problem shows up without any alpha to key off of. This finds the
|
|
bbox of pixels that actually differ from solid white, which is a
|
|
standard, well-behaved "trim whitespace" operation — it naturally
|
|
returns the full image (no trim) for anything that isn't bordered by
|
|
a genuinely blank margin, which is the common case for real photos.
|
|
"""
|
|
try:
|
|
from PIL import ImageChops
|
|
rgb = pil_img.convert("RGB")
|
|
bg = PILImage.new("RGB", rgb.size, (255, 255, 255))
|
|
diff = ImageChops.difference(rgb, bg)
|
|
# Ignore faint compression/anti-aliasing noise near white.
|
|
diff = diff.point(lambda p: 255 if p > 8 else 0)
|
|
return diff.getbbox()
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def _tighten_image_placement(
|
|
raw_rect: Tuple[float, float, float, float],
|
|
pypdf_image,
|
|
) -> Tuple[float, float, float, float]:
|
|
"""Shrink ``raw_rect`` (x0, top, x1, bottom, in page units, top-down) to
|
|
just the visible content of ``pypdf_image``, if it has real transparency.
|
|
Falls back to the original rect on any mismatch or decode failure.
|
|
"""
|
|
x0, top, x1, bottom = raw_rect
|
|
try:
|
|
pil_img = pypdf_image.image
|
|
if pil_img is None:
|
|
return raw_rect
|
|
pw, ph = pil_img.size
|
|
if pw <= 0 or ph <= 0:
|
|
return raw_rect
|
|
|
|
content = _tight_alpha_content_rect(pil_img)
|
|
min_savings = 0.0
|
|
if content is None:
|
|
# No alpha to key off — fall back to trimming a plain white
|
|
# border, but only trust it if it saves a real, deliberate
|
|
# margin (not a couple of anti-aliased edge pixels), since an
|
|
# opaque photo legitimately touching white content is a much
|
|
# more ambiguous signal than a real alpha channel.
|
|
content = _tight_opaque_content_rect(pil_img)
|
|
min_savings = 0.03
|
|
if content is None:
|
|
return raw_rect
|
|
|
|
cb_left, cb_top, cb_right, cb_bottom = content
|
|
if cb_right <= cb_left or cb_bottom <= cb_top:
|
|
return raw_rect
|
|
if min_savings:
|
|
saved_w = 1.0 - (cb_right - cb_left) / pw
|
|
saved_h = 1.0 - (cb_bottom - cb_top) / ph
|
|
if max(saved_w, saved_h) < min_savings:
|
|
return raw_rect
|
|
|
|
scale_x = (x1 - x0) / pw
|
|
scale_y = (bottom - top) / ph
|
|
tight = (
|
|
x0 + cb_left * scale_x,
|
|
top + cb_top * scale_y,
|
|
x0 + cb_right * scale_x,
|
|
top + cb_bottom * scale_y,
|
|
)
|
|
# Sanity guard: only accept a tightening that's a real, meaningful
|
|
# shrink — never accept something wider/taller than the original,
|
|
# and never collapse to a sliver (avoids acting on a bad match).
|
|
if tight[0] < x0 - 0.5 or tight[2] > x1 + 0.5 or tight[1] < top - 0.5 or tight[3] > bottom + 0.5:
|
|
return raw_rect
|
|
if (tight[2] - tight[0]) < 2.0 or (tight[3] - tight[1]) < 2.0:
|
|
return raw_rect
|
|
return tight
|
|
except Exception as e:
|
|
logger.debug(f"Image placement tightening failed, using raw rect: {e}")
|
|
return raw_rect
|
|
|
|
|
|
def _rect_gap(a: List[float], b: List[float]) -> float:
|
|
"""Minimum distance between two axis-aligned rects (0 if they touch or
|
|
overlap). Diagonal separation uses the true Euclidean distance so a
|
|
rect offset both horizontally and vertically from another isn't
|
|
under-counted by looking at only one axis.
|
|
"""
|
|
ax0, ay0, ax1, ay1 = a
|
|
bx0, by0, bx1, by1 = b
|
|
dx = max(0.0, max(ax0, bx0) - min(ax1, bx1))
|
|
dy = max(0.0, max(ay0, by0) - min(ay1, by1))
|
|
if dx == 0.0 or dy == 0.0:
|
|
return max(dx, dy)
|
|
return (dx * dx + dy * dy) ** 0.5
|
|
|
|
|
|
def _group_nearby_image_rects(rects: List[List[float]], tolerance: float = 8.0) -> List[List[int]]:
|
|
"""Cluster image placement rects that are touching or within a small
|
|
gap of each other, returning each cluster as a list of source indices.
|
|
|
|
Some DOCX/design-tool -> PDF converters rasterize one visual graphic
|
|
(e.g. a custom letterhead logo with a stylised wordmark) as several
|
|
small, separately-placed raster image XObjects instead of one —
|
|
verified directly against a real letterhead where a single logo came
|
|
through as 10 separate image blocks (an icon plus individual letter
|
|
fragments), each independently selectable/editable instead of the
|
|
graphic as a whole. Clustering by proximity re-unifies those into one
|
|
block. A tolerance of ~8pt reliably joins tightly-packed fragments
|
|
like that (gaps of a few points) while leaving genuinely distinct,
|
|
deliberately-spaced small images (page-corner marks, separate icons
|
|
tens of points apart) untouched — verified against a real multi-icon
|
|
page where icons ~78pt apart correctly stayed separate.
|
|
"""
|
|
n = len(rects)
|
|
parent = list(range(n))
|
|
|
|
def find(i: int) -> int:
|
|
while parent[i] != i:
|
|
parent[i] = parent[parent[i]]
|
|
i = parent[i]
|
|
return i
|
|
|
|
def union(i: int, j: int) -> None:
|
|
ri, rj = find(i), find(j)
|
|
if ri != rj:
|
|
parent[ri] = rj
|
|
|
|
for i in range(n):
|
|
for j in range(i + 1, n):
|
|
if _rect_gap(rects[i], rects[j]) <= tolerance:
|
|
union(i, j)
|
|
|
|
groups: Dict[int, List[int]] = {}
|
|
for i in range(n):
|
|
groups.setdefault(find(i), []).append(i)
|
|
return list(groups.values())
|
|
|
|
|
|
def _compute_layout_bbox(content_bbox: List[float], all_line_bboxes: List[List[float]], phys_w: float) -> List[float]:
|
|
"""Estimate how far a text block could grow to the right before it would
|
|
need to wrap, as a distinct "layout constraint" separate from the
|
|
block's own tight content bounds.
|
|
|
|
Heuristic: look at every line on the page whose left edge starts at
|
|
roughly the same x as this block (a proxy for "same column") and use
|
|
the furthest right edge among them as the constraint — this keeps a
|
|
narrow column in a multi-column page from being handed the whole page's
|
|
width, while still giving a single-column paragraph room to grow up to
|
|
where text on that page actually reaches, rather than an arbitrary cap.
|
|
"""
|
|
x0 = content_bbox[0]
|
|
tolerance = 30.0
|
|
same_column = [lb[2] for lb in all_line_bboxes if abs(lb[0] - x0) <= tolerance]
|
|
right = max(same_column) if same_column else max((lb[2] for lb in all_line_bboxes), default=content_bbox[2])
|
|
right = min(right + 4.0, phys_w - 4.0)
|
|
right = max(right, content_bbox[2])
|
|
return [content_bbox[0], content_bbox[1], round(right, 2), content_bbox[3]]
|
|
|
|
|
|
def extract_blocks(pdf_bytes: bytes) -> dict:
|
|
"""Extract text and image blocks from every page, in the same JSON shape
|
|
PdfBlockEditor.tsx (and this module's own edit-resolution) already expect.
|
|
|
|
Each text block additionally carries ``layout_bbox`` — the same top/left/
|
|
bottom as ``bbox`` (the tight content bounds) but extended rightward to a
|
|
page-derived column-width estimate. This is purely additive: existing
|
|
consumers that only read ``bbox`` are unaffected.
|
|
"""
|
|
results: Dict[str, Any] = {"pages": []}
|
|
|
|
reader = PdfReader(io.BytesIO(pdf_bytes))
|
|
font_catalog = _build_font_catalog(reader)
|
|
results["fonts"] = {
|
|
k: base64.b64encode(v["bytes"]).decode("utf-8")
|
|
for k, v in font_catalog.items() if v.get("bytes")
|
|
}
|
|
# Additive alongside `fonts` (which stays exactly as before for any
|
|
# existing consumer): real numeric weight/style per font name, straight
|
|
# from the PDF's own FontDescriptor. The frontend's font registration
|
|
# previously guessed weight from a regex on the font name — a name like
|
|
# "Calibri-Light" doesn't match /bold|black|heavy|semibold/, so it fell
|
|
# back to a plain 400, the same weight as regular "Calibri". Both then
|
|
# tried to register under the same shared family+weight slot, whichever
|
|
# loaded first silently won it, and the other was never actually
|
|
# rendered with — confirmed directly against a real document's "Light"
|
|
# heading rendering as full-weight Calibri once clicked into.
|
|
results["font_meta"] = {
|
|
k: {"weight": v.get("weight", 400), "italic": bool(v.get("italic"))}
|
|
for k, v in font_catalog.items() if v.get("bytes")
|
|
}
|
|
|
|
pdfium_doc = pdfium.PdfDocument(pdf_bytes)
|
|
try:
|
|
with pdfplumber.open(io.BytesIO(pdf_bytes)) as pl_pdf:
|
|
logger.info(f"Extracting blocks from {len(pl_pdf.pages)} pages")
|
|
|
|
for page_idx, pl_page in enumerate(pl_pdf.pages):
|
|
pypdf_page = reader.pages[page_idx]
|
|
phys_w = float(pypdf_page.mediabox.width)
|
|
phys_h = float(pypdf_page.mediabox.height)
|
|
rotation = int(pypdf_page.rotation or 0) % 360
|
|
|
|
page_entry: Dict[str, Any] = {
|
|
"page_number": page_idx,
|
|
"width": round(phys_w, 2),
|
|
"height": round(phys_h, 2),
|
|
"blocks": [],
|
|
}
|
|
|
|
text_lines = _extract_page_lines(pl_page, phys_w, phys_h, rotation, font_catalog)
|
|
text_block_groups = _group_lines_into_blocks(text_lines)
|
|
all_line_bboxes = [ln["bbox"] for ln in text_lines]
|
|
|
|
page_image, zoom = _render_page_image(pdfium_doc[page_idx])
|
|
|
|
# pdfplumber gives accurate page-space placement rects but no
|
|
# easy access to each image's own alpha channel; pypdf decodes
|
|
# the actual pixels (SMask composited into RGBA where present).
|
|
# Match the two lists by position — both walk the page content
|
|
# stream's Do operators in the same document order — and only
|
|
# trust the match when the counts agree, so a divergence (e.g.
|
|
# one library recursing into a nested Form XObject the other
|
|
# doesn't) safely falls back to the untightened rect instead of
|
|
# tightening against the wrong image.
|
|
pl_images = list(pl_page.images)
|
|
pypdf_images: List[Any] = []
|
|
if pl_images:
|
|
try:
|
|
pypdf_images = list(pypdf_page.images)
|
|
except Exception:
|
|
pypdf_images = []
|
|
images_aligned = bool(pl_images) and len(pypdf_images) == len(pl_images)
|
|
|
|
image_bboxes: List[List[float]] = []
|
|
for i, img in enumerate(pl_images):
|
|
raw_rect = (img["x0"], img["top"], img["x1"], img["bottom"])
|
|
if images_aligned:
|
|
# Matching counts alone doesn't guarantee the i-th entry
|
|
# in each list is the same image — a real, deliberate
|
|
# size check confirms it before trusting the pairing;
|
|
# a stray mismatch would otherwise tighten this image's
|
|
# placement against a completely unrelated image's
|
|
# content and produce a nonsensical crop.
|
|
pypdf_img = pypdf_images[i]
|
|
try:
|
|
same_image = pypdf_img.image is not None and tuple(pypdf_img.image.size) == tuple(img.get("srcsize") or ())
|
|
except Exception:
|
|
same_image = False
|
|
if same_image:
|
|
raw_rect = _tighten_image_placement(raw_rect, pypdf_img)
|
|
r = visual_to_physical_rect(
|
|
PdfRect(*raw_rect), phys_w, phys_h, rotation
|
|
)
|
|
image_bboxes.append([round(r.x0, 2), round(r.y0, 2), round(r.x1, 2), round(r.y1, 2)])
|
|
|
|
if len(image_bboxes) > 1:
|
|
grouped_bboxes = []
|
|
for idxs in _group_nearby_image_rects(image_bboxes):
|
|
x0 = min(image_bboxes[k][0] for k in idxs)
|
|
y0 = min(image_bboxes[k][1] for k in idxs)
|
|
x1 = max(image_bboxes[k][2] for k in idxs)
|
|
y1 = max(image_bboxes[k][3] for k in idxs)
|
|
grouped_bboxes.append([round(x0, 2), round(y0, 2), round(x1, 2), round(y1, 2)])
|
|
image_bboxes = grouped_bboxes
|
|
|
|
# Interleave text blocks and image blocks by vertical position,
|
|
# matching the original's document-order block_id assignment.
|
|
candidates: List[Tuple[float, str, Any]] = []
|
|
for grp in text_block_groups:
|
|
top = min(l["bbox"][1] for l in grp)
|
|
candidates.append((top, "text", grp))
|
|
for bbox in image_bboxes:
|
|
candidates.append((bbox[1], "image", bbox))
|
|
candidates.sort(key=lambda c: c[0])
|
|
|
|
for b_idx, (_, kind, payload) in enumerate(candidates):
|
|
block_id = f"p{page_idx}_{b_idx}"
|
|
if kind == "text":
|
|
grp = payload
|
|
x0 = min(l["bbox"][0] for l in grp)
|
|
y0 = min(l["bbox"][1] for l in grp)
|
|
x1 = max(l["bbox"][2] for l in grp)
|
|
y1 = max(l["bbox"][3] for l in grp)
|
|
content_bbox = [round(x0, 2), round(y0, 2), round(x1, 2), round(y1, 2)]
|
|
full_text = "\n".join(
|
|
"".join(s["text"] for s in ln["spans"]) for ln in grp
|
|
)
|
|
page_entry["blocks"].append({
|
|
"block_id": block_id,
|
|
"bbox": content_bbox,
|
|
"layout_bbox": _compute_layout_bbox(content_bbox, all_line_bboxes, phys_w),
|
|
"type": "text",
|
|
"text": full_text,
|
|
"lines": grp,
|
|
})
|
|
else:
|
|
bbox = payload
|
|
img_src = _crop_to_base64_png(page_image, zoom, bbox)
|
|
if img_src:
|
|
page_entry["blocks"].append({
|
|
"block_id": block_id,
|
|
"bbox": bbox,
|
|
"type": "image",
|
|
"src": img_src,
|
|
})
|
|
|
|
results["pages"].append(page_entry)
|
|
logger.debug(f"Page {page_idx}: built {len(page_entry['blocks'])} editable blocks")
|
|
finally:
|
|
pdfium_doc.close()
|
|
|
|
return results
|
|
|
|
|
|
# ─── Edit resolution ─────────────────────────────────────────────────────────
|
|
|
|
def _resolve_target(full_block_data: dict, block_id: str) -> Optional[Dict[str, Any]]:
|
|
"""Resolve a ``p{page}_{block}[_l{line}][_s{span}]`` id against a prior
|
|
:func:`extract_blocks` result, replacing the old fitz-based
|
|
``_resolve_bbox`` re-parse (this looks up the *same* extraction instead
|
|
of re-deriving it, guaranteeing the two stay consistent by construction).
|
|
"""
|
|
m = _BLOCK_ID_RE.match(block_id)
|
|
if not m:
|
|
for pg in full_block_data.get("pages", []):
|
|
for b in pg.get("blocks", []):
|
|
if b.get("block_id") == block_id:
|
|
return {"page": pg["page_number"], "bbox": list(b["bbox"]), "type": b.get("type", "text"), "block": b}
|
|
return None
|
|
p_idx = int(m.group(1))
|
|
b_idx = int(m.group(2))
|
|
l_idx = int(m.group(3)) if m.group(3) is not None else None
|
|
s_idx = int(m.group(4)) if m.group(4) is not None else None
|
|
|
|
page_entry = next((pg for pg in full_block_data.get("pages", []) if pg["page_number"] == p_idx), None)
|
|
if not page_entry:
|
|
return None
|
|
blocks = page_entry.get("blocks", [])
|
|
if b_idx >= len(blocks):
|
|
return None
|
|
block = blocks[b_idx]
|
|
bbox = block["bbox"]
|
|
|
|
if l_idx is not None and block.get("type") == "text":
|
|
lines = block.get("lines", [])
|
|
if l_idx < len(lines):
|
|
line = lines[l_idx]
|
|
if s_idx is not None and s_idx < len(line.get("spans", [])):
|
|
span = line["spans"][s_idx]
|
|
bbox = span.get("bbox") or line["bbox"]
|
|
return {"page": p_idx, "bbox": list(bbox), "type": "text", "block": block, "line": line, "span": span}
|
|
else:
|
|
bbox = line["bbox"]
|
|
return {"page": p_idx, "bbox": list(bbox), "type": "text", "block": block, "line": line}
|
|
|
|
return {"page": p_idx, "bbox": list(bbox), "type": block.get("type", "text"), "block": block}
|
|
|
|
|
|
# ─── Redaction: strip content-stream ops overlapping target rects ─────────────
|
|
|
|
def _mat_mul(m1: tuple, m2: tuple) -> tuple:
|
|
a1, b1, c1, d1, e1, f1 = m1
|
|
a2, b2, c2, d2, e2, f2 = m2
|
|
return (
|
|
a1 * a2 + b1 * c2, a1 * b2 + b1 * d2,
|
|
c1 * a2 + d1 * c2, c1 * b2 + d1 * d2,
|
|
e1 * a2 + f1 * c2 + e2, e1 * b2 + f1 * d2 + f2,
|
|
)
|
|
|
|
|
|
def _apply_mat(m: tuple, x: float, y: float) -> Tuple[float, float]:
|
|
a, b, c, d, e, f = m
|
|
return (a * x + c * y + e, b * x + d * y + f)
|
|
|
|
|
|
def _redact_regions_on_page(page, target_rects: List[PdfRect]) -> None:
|
|
"""Strip text-showing, path-fill, and image-draw operators whose
|
|
device-space bbox falls within any of ``target_rects`` (physical,
|
|
top-left/y-down convention) from ``page``'s content stream, then write
|
|
the filtered stream back — the same "true removal" semantics as
|
|
PyMuPDF's ``add_redact_annot`` + ``apply_redactions``, reimplemented at
|
|
the operator level since pypdf has no built-in equivalent.
|
|
|
|
Handles the common, dominant case (simple ``cm``/``Tm``-positioned text,
|
|
``re``-drawn rectangles, ``Do``-drawn XObjects) without recursing into
|
|
nested Form XObjects — see migration notes for this scope limitation.
|
|
"""
|
|
if not target_rects:
|
|
return
|
|
ph = float(page.mediabox.height)
|
|
native_rects = [(r.x0, ph - r.y1, r.x1, ph - r.y0) for r in target_rects]
|
|
|
|
def overlaps(bx0: float, by0: float, bx1: float, by1: float) -> bool:
|
|
if bx1 < bx0:
|
|
bx0, bx1 = bx1, bx0
|
|
if by1 < by0:
|
|
by0, by1 = by1, by0
|
|
for rx0, ry0, rx1, ry1 in native_rects:
|
|
if bx1 <= rx0 or rx1 <= bx0:
|
|
continue
|
|
if by1 <= ry0 or ry1 <= by0:
|
|
continue
|
|
return True
|
|
return False
|
|
|
|
try:
|
|
contents = page.get_contents()
|
|
if contents is None:
|
|
return
|
|
cs = ContentStream(contents, page)
|
|
except Exception as e:
|
|
logger.warning(f"Redaction: failed to parse content stream: {e}")
|
|
return
|
|
|
|
new_ops = []
|
|
ctm_stack: List[tuple] = []
|
|
ctm = (1.0, 0.0, 0.0, 1.0, 0.0, 0.0)
|
|
tm = None
|
|
cur_font_size = 12.0
|
|
|
|
for operands, operator in cs.operations:
|
|
op = operator.decode("latin1") if isinstance(operator, (bytes, bytearray)) else str(operator)
|
|
drop = False
|
|
|
|
try:
|
|
if op == "q":
|
|
ctm_stack.append(ctm)
|
|
elif op == "Q":
|
|
if ctm_stack:
|
|
ctm = ctm_stack.pop()
|
|
elif op == "cm":
|
|
m = tuple(float(v) for v in operands)
|
|
ctm = _mat_mul(m, ctm)
|
|
elif op == "BT":
|
|
tm = (1.0, 0.0, 0.0, 1.0, 0.0, 0.0)
|
|
elif op == "ET":
|
|
tm = None
|
|
elif op == "Tm":
|
|
tm = tuple(float(v) for v in operands)
|
|
elif op in ("Td", "TD"):
|
|
tx, ty = float(operands[0]), float(operands[1])
|
|
if tm is None:
|
|
tm = (1.0, 0.0, 0.0, 1.0, 0.0, 0.0)
|
|
tm = _mat_mul((1.0, 0.0, 0.0, 1.0, tx, ty), tm)
|
|
elif op == "Tf":
|
|
try:
|
|
cur_font_size = float(operands[1])
|
|
except Exception:
|
|
pass
|
|
elif op in ("Tj", "TJ", "'", '"'):
|
|
if tm is not None:
|
|
combined = _mat_mul(tm, ctm)
|
|
ox, oy = _apply_mat(combined, 0, 0)
|
|
text_chars = 0
|
|
if op == "TJ" and operands and isinstance(operands[0], list):
|
|
for item in operands[0]:
|
|
if isinstance(item, (bytes, str)):
|
|
text_chars += len(item)
|
|
elif operands and isinstance(operands[0], (bytes, str)):
|
|
text_chars = len(operands[0])
|
|
approx_w = max(text_chars * cur_font_size * 0.65, cur_font_size * 0.3)
|
|
approx_h = cur_font_size * 1.15
|
|
bx0, by0 = ox, oy - approx_h * 0.3
|
|
bx1, by1 = ox + approx_w, oy + approx_h * 0.85
|
|
if overlaps(bx0, by0, bx1, by1):
|
|
drop = True
|
|
elif op == "re":
|
|
x, y, w, h = (float(v) for v in operands)
|
|
corners = [
|
|
_apply_mat(ctm, x, y), _apply_mat(ctm, x + w, y),
|
|
_apply_mat(ctm, x, y + h), _apply_mat(ctm, x + w, y + h),
|
|
]
|
|
xs = [c[0] for c in corners]
|
|
ys = [c[1] for c in corners]
|
|
if overlaps(min(xs), min(ys), max(xs), max(ys)):
|
|
drop = True
|
|
elif op == "Do":
|
|
corners = [
|
|
_apply_mat(ctm, 0, 0), _apply_mat(ctm, 1, 0),
|
|
_apply_mat(ctm, 0, 1), _apply_mat(ctm, 1, 1),
|
|
]
|
|
xs = [c[0] for c in corners]
|
|
ys = [c[1] for c in corners]
|
|
if overlaps(min(xs), min(ys), max(xs), max(ys)):
|
|
drop = True
|
|
except Exception as e:
|
|
logger.debug(f"Redaction: skipping malformed operator {op}: {e}")
|
|
|
|
if not drop:
|
|
new_ops.append((operands, operator))
|
|
|
|
cs.operations = new_ops
|
|
try:
|
|
page.replace_contents(cs)
|
|
except Exception as e:
|
|
logger.warning(f"Redaction: failed to write filtered content stream: {e}")
|
|
|
|
|
|
# ─── Overlay rendering (new text/images onto a page, replaces insert_textbox/insert_image) ─
|
|
|
|
def _wrap_text(text: str, max_width: float, fontname: str, fontsize: float) -> List[str]:
|
|
lines: List[str] = []
|
|
for raw_line in text.split("\n"):
|
|
words = raw_line.split(" ")
|
|
cur = ""
|
|
for w in words:
|
|
candidate = f"{cur} {w}".strip()
|
|
if not cur or _text_width(candidate, fontname, fontsize) <= max_width:
|
|
cur = candidate
|
|
else:
|
|
lines.append(cur)
|
|
cur = w
|
|
lines.append(cur)
|
|
return lines
|
|
|
|
|
|
def _merge_overlay_onto_page(page, box_w: float, box_h: float, target_rect: PdfRect, draw_fn) -> None:
|
|
"""Build a ``box_w`` x ``box_h`` overlay page (draw_fn receives the
|
|
reportlab canvas), then merge it onto ``page`` positioned at
|
|
``target_rect`` (physical, top-left/y-down space).
|
|
"""
|
|
if box_w <= 0 or box_h <= 0:
|
|
return
|
|
buf = io.BytesIO()
|
|
c = rl_canvas.Canvas(buf, pagesize=(box_w, box_h))
|
|
draw_fn(c)
|
|
c.save()
|
|
buf.seek(0)
|
|
overlay_page = PdfReader(buf).pages[0]
|
|
|
|
ph = float(page.mediabox.height)
|
|
tx = target_rect.x0
|
|
ty = ph - target_rect.y1
|
|
from pypdf import Transformation
|
|
transform = Transformation().translate(tx, ty)
|
|
page.merge_transformed_page(overlay_page, transform, over=True)
|
|
|
|
|
|
def _render_text_block(page, target_rect: PdfRect, text: str, font_size: float, raw_font: str,
|
|
color_hex: str, align: int, font_catalog: Dict[str, Dict[str, Any]],
|
|
bold: bool = False, italic: bool = False, underline: bool = False) -> None:
|
|
reportlab_font, _stage = _resolve_font_for_render(raw_font, bold, italic, font_catalog, text=text)
|
|
|
|
color_rgb = _hex_to_rgb(color_hex)
|
|
box_w, box_h = target_rect.width, target_rect.height
|
|
if box_w <= 0 or box_h <= 0:
|
|
return
|
|
|
|
# When editing a line or paragraph, ensure box_w is wide enough for the text without shrinking font
|
|
text_w = _text_width(text, reportlab_font, font_size)
|
|
if "\n" not in text and text_w > box_w:
|
|
box_w = text_w + 4.0
|
|
|
|
lines = _wrap_text(text, box_w, reportlab_font, font_size)
|
|
line_height = font_size * 1.15
|
|
needed_height = len(lines) * line_height
|
|
if needed_height > box_h:
|
|
box_h = needed_height + 2.0
|
|
|
|
def draw(c):
|
|
c.setFont(reportlab_font, font_size)
|
|
c.setFillColorRGB(*color_rgb)
|
|
underline_offset = max(1.0, font_size * 0.08)
|
|
underline_width = max(0.5, font_size * 0.05)
|
|
baseline = box_h - font_size
|
|
for line in lines:
|
|
if baseline < -line_height:
|
|
break
|
|
w = _text_width(line, reportlab_font, font_size)
|
|
x = 0.0
|
|
if align in (1, 2):
|
|
if align == 1:
|
|
x = max(0.0, (box_w - w) / 2)
|
|
elif align == 2:
|
|
x = max(0.0, box_w - w)
|
|
c.drawString(x, baseline, line)
|
|
if underline and line.strip():
|
|
c.setLineWidth(underline_width)
|
|
c.setStrokeColorRGB(*color_rgb)
|
|
y = baseline - underline_offset
|
|
c.line(x, y, x + w, y)
|
|
baseline -= line_height
|
|
|
|
_merge_overlay_onto_page(page, box_w, box_h, target_rect, draw)
|
|
|
|
|
|
def _render_image_block(page, target_rect: PdfRect, image_bytes: bytes) -> None:
|
|
box_w, box_h = target_rect.width, target_rect.height
|
|
if box_w <= 0 or box_h <= 0 or not image_bytes:
|
|
return
|
|
|
|
def draw(c):
|
|
try:
|
|
c.drawImage(
|
|
ImageReader(io.BytesIO(image_bytes)), 0, 0, width=box_w, height=box_h,
|
|
mask="auto", preserveAspectRatio=False,
|
|
)
|
|
except Exception as e:
|
|
logger.error(f"Failed to draw image block: {e}")
|
|
|
|
_merge_overlay_onto_page(page, box_w, box_h, target_rect, draw)
|
|
|
|
|
|
def _erase_rect_on_page(page, rect: PdfRect, fill_color: tuple = (1.0, 1.0, 1.0)) -> None:
|
|
"""Draw an opaque rectangle over rect to ensure any underlying content
|
|
(including XObjects, inline images, and vector paths) is fully covered.
|
|
"""
|
|
box_w, box_h = rect.width, rect.height
|
|
if box_w <= 0 or box_h <= 0:
|
|
return
|
|
|
|
def draw(c):
|
|
c.setFillColorRGB(*fill_color)
|
|
c.rect(0, 0, box_w, box_h, fill=1, stroke=0)
|
|
|
|
_merge_overlay_onto_page(page, box_w, box_h, rect, draw)
|
|
|
|
|
|
# ─── Page structure ops ─────────────────────────────────────────────────────────
|
|
|
|
def _apply_page_ops(reader: PdfReader, page_ops: List[dict]) -> PdfWriter:
|
|
descriptors: List[tuple] = [("existing", i) for i in range(len(reader.pages))]
|
|
|
|
for op in page_ops or []:
|
|
t = op.get("type")
|
|
idx = op.get("page_index")
|
|
target = op.get("target_index")
|
|
try:
|
|
if t == "add":
|
|
where = target if target is not None else len(descriptors)
|
|
where = max(0, min(where, len(descriptors)))
|
|
w = op.get("width") or 595.0
|
|
h = op.get("height") or 842.0
|
|
descriptors.insert(where, ("blank", w, h))
|
|
elif t == "delete":
|
|
if idx is not None and 0 <= idx < len(descriptors):
|
|
descriptors.pop(idx)
|
|
elif t == "move":
|
|
if idx is not None and target is not None and 0 <= idx < len(descriptors):
|
|
item = descriptors.pop(idx)
|
|
target = max(0, min(target, len(descriptors)))
|
|
descriptors.insert(target, item)
|
|
except Exception as e:
|
|
logger.error(f"Page operation '{t}' failed: {e}")
|
|
|
|
writer = PdfWriter()
|
|
for desc in descriptors:
|
|
if desc[0] == "existing":
|
|
writer.add_page(reader.pages[desc[1]])
|
|
else:
|
|
writer.add_blank_page(width=desc[1], height=desc[2])
|
|
return writer
|
|
|
|
|
|
# ─── Orchestration ───────────────────────────────────────────────────────────
|
|
|
|
async def apply_edits(
|
|
pdf_bytes: bytes, edits: List[dict], additions: List[dict], page_ops: List[dict] = None,
|
|
) -> bytes:
|
|
"""Apply block-level text/image edits, new additions, and page structure
|
|
operations to a PDF and return the modified bytes.
|
|
|
|
Kept ``async`` for call-site compatibility (the router awaits it); no
|
|
actual async I/O is needed now that Stirling's HTTP calls are gone.
|
|
"""
|
|
if page_ops is None:
|
|
page_ops = []
|
|
if additions is None:
|
|
additions = []
|
|
|
|
reader = PdfReader(io.BytesIO(pdf_bytes))
|
|
|
|
if page_ops:
|
|
po_writer = _apply_page_ops(reader, page_ops)
|
|
buf = io.BytesIO()
|
|
po_writer.write(buf)
|
|
current_pdf_bytes = buf.getvalue()
|
|
reader = PdfReader(io.BytesIO(current_pdf_bytes))
|
|
else:
|
|
current_pdf_bytes = pdf_bytes
|
|
|
|
full_block_data = extract_blocks(current_pdf_bytes)
|
|
font_catalog = _build_font_catalog(reader)
|
|
logger.info(f"Extracted {len(font_catalog)} font(s) from PDF "
|
|
f"({sum(1 for v in font_catalog.values() if v['embedded'])} embedded)")
|
|
|
|
pdfium_doc = pdfium.PdfDocument(current_pdf_bytes)
|
|
|
|
writer = PdfWriter()
|
|
writer.append(reader)
|
|
|
|
try:
|
|
edits_by_page: Dict[int, List[dict]] = {}
|
|
for edit in edits or []:
|
|
target = _resolve_target(full_block_data, edit.get("block_id", ""))
|
|
if not target:
|
|
logger.debug(f"Could not resolve edit target: {edit.get('block_id')}")
|
|
continue
|
|
edits_by_page.setdefault(target["page"], []).append((edit, target))
|
|
|
|
additions_by_page: Dict[int, List[dict]] = {}
|
|
for add in additions:
|
|
p_idx = (add.get("page", 1) or 1) - 1
|
|
if 0 <= p_idx < len(writer.pages):
|
|
additions_by_page.setdefault(p_idx, []).append(add)
|
|
|
|
all_page_nums = set(edits_by_page.keys()) | set(additions_by_page.keys())
|
|
|
|
for page_idx in all_page_nums:
|
|
page = writer.pages[page_idx]
|
|
|
|
page_image, zoom = None, 2.0
|
|
if page_idx < len(pdfium_doc):
|
|
page_image, zoom = _render_page_image(pdfium_doc[page_idx])
|
|
|
|
# 1. Capture bytes for any image blocks being edited/moved, before redacting.
|
|
image_captures: Dict[str, bytes] = {}
|
|
for edit, target in edits_by_page.get(page_idx, []):
|
|
if target["type"] == "image" and page_image is not None:
|
|
bbox = target["bbox"]
|
|
src = _crop_to_base64_png(page_image, zoom, bbox)
|
|
if src and "," in src:
|
|
try:
|
|
image_captures[edit.get("block_id", "")] = base64.b64decode(src.split(",", 1)[1])
|
|
except Exception as e:
|
|
logger.error(f"Failed to capture image block for edit: {e}")
|
|
|
|
# 2. Collect redaction regions: original block bbox for each edit,
|
|
# and each addition's target bbox.
|
|
redact_rects: List[PdfRect] = []
|
|
for edit, target in edits_by_page.get(page_idx, []):
|
|
redact_rects.append(PdfRect(*target["bbox"]))
|
|
|
|
pw = float(page.mediabox.width)
|
|
ph = float(page.mediabox.height)
|
|
for add in additions_by_page.get(page_idx, []):
|
|
ax, ay = float(add.get("x", 0) or 0), float(add.get("y", 0) or 0)
|
|
aw, ah = float(add.get("width", 0.2) or 0.2), float(add.get("height", 0.05) or 0.05)
|
|
redact_rects.append(PdfRect(ax * pw, ay * ph, (ax + aw) * pw, (ay + ah) * ph))
|
|
|
|
_redact_regions_on_page(page, redact_rects)
|
|
|
|
# 2b. Erase original regions for all edits to guarantee underlying content is fully wiped
|
|
for edit, target in edits_by_page.get(page_idx, []):
|
|
_erase_rect_on_page(page, PdfRect(*target["bbox"]))
|
|
|
|
# 3. Draw replacement content for edits.
|
|
for edit, target in edits_by_page.get(page_idx, []):
|
|
bid = edit.get("block_id", "")
|
|
target_bbox = edit.get("bbox") or target["bbox"]
|
|
target_rect = PdfRect(*target_bbox)
|
|
|
|
if target["type"] == "image":
|
|
if edit.get("remove") or edit.get("is_deleted"):
|
|
continue
|
|
img_bytes = image_captures.get(bid)
|
|
if not img_bytes and edit.get("imageSrc") and "," in str(edit.get("imageSrc")):
|
|
try:
|
|
img_bytes = base64.b64decode(str(edit["imageSrc"]).split(",", 1)[1])
|
|
except Exception:
|
|
pass
|
|
if img_bytes:
|
|
_render_image_block(page, target_rect, img_bytes)
|
|
continue
|
|
|
|
new_text = (edit.get("new_text") or "").strip()
|
|
if not new_text:
|
|
continue
|
|
|
|
list_type = edit.get("list_type", "none")
|
|
if list_type == "bullet":
|
|
new_text = "\n".join(f"• {ln}" for ln in new_text.splitlines())
|
|
elif list_type == "number":
|
|
new_text = "\n".join(f"{i + 1}. {ln}" for i, ln in enumerate(new_text.splitlines()))
|
|
|
|
block = target["block"]
|
|
first_span = target.get("span")
|
|
if not first_span and target.get("line"):
|
|
spans = target["line"].get("spans", [])
|
|
if spans:
|
|
first_span = spans[0]
|
|
if not first_span:
|
|
for ln in block.get("lines", []):
|
|
if ln.get("spans"):
|
|
first_span = ln["spans"][0]
|
|
break
|
|
first_span = first_span or {}
|
|
|
|
raw_font = edit.get("font") or first_span.get("raw_font") or first_span.get("font", "")
|
|
size = edit.get("size") or first_span.get("size", 12)
|
|
color = edit.get("color") or first_span.get("color", "#000000")
|
|
align = int(edit.get("align") or 0)
|
|
|
|
original_style = first_span.get("style", {}) or {}
|
|
bold = bool(edit["bold"]) if edit.get("bold") is not None else bool(original_style.get("bold"))
|
|
italic = bool(edit["italic"]) if edit.get("italic") is not None else bool(original_style.get("italic"))
|
|
underline = (
|
|
bool(edit["underline"]) if edit.get("underline") is not None
|
|
else bool(original_style.get("underline"))
|
|
)
|
|
|
|
_render_text_block(
|
|
page, target_rect, new_text, float(size), raw_font, color, align, font_catalog,
|
|
bold=bold, italic=italic, underline=underline,
|
|
)
|
|
|
|
# 4. Draw additions.
|
|
for add in additions_by_page.get(page_idx, []):
|
|
ax, ay = float(add.get("x", 0) or 0), float(add.get("y", 0) or 0)
|
|
aw, ah = float(add.get("width", 0.2) or 0.2), float(add.get("height", 0.05) or 0.05)
|
|
add_rect = PdfRect(ax * pw, ay * ph, (ax + aw) * pw, (ay + ah) * ph)
|
|
|
|
if add.get("type") == "image" and add.get("content"):
|
|
b64 = add["content"]
|
|
if "," in b64:
|
|
b64 = b64.split(",", 1)[1]
|
|
try:
|
|
img_bytes = base64.b64decode(b64)
|
|
_render_image_block(page, add_rect, img_bytes)
|
|
except Exception as e:
|
|
logger.error(f"Failed to insert image addition: {e}")
|
|
|
|
elif add.get("type") == "text" and add.get("content"):
|
|
content = add["content"]
|
|
list_type = add.get("list_type", "none")
|
|
if list_type == "bullet":
|
|
content = "\n".join(f"• {ln}" for ln in content.splitlines())
|
|
elif list_type == "number":
|
|
content = "\n".join(f"{i + 1}. {ln}" for i, ln in enumerate(content.splitlines()))
|
|
fs = float(add.get("font_size") or 14)
|
|
color = add.get("color") or "#000000"
|
|
align = int(add.get("align") or 0)
|
|
raw_font = add.get("font") or ""
|
|
_render_text_block(
|
|
page, add_rect, content, fs, raw_font, color, align, font_catalog,
|
|
bold=bool(add.get("bold")), italic=bool(add.get("italic")),
|
|
)
|
|
|
|
out = io.BytesIO()
|
|
writer.write(out)
|
|
return out.getvalue()
|
|
finally:
|
|
pdfium_doc.close()
|