From af400090460ef06fc3c7552edc0dd69892616e35 Mon Sep 17 00:00:00 2001 From: John Wooten Date: Thu, 15 Oct 2026 19:50:42 -0400 Subject: [PATCH 1/5] =?UTF-8?q?=E2=9E=95=20Add=20function=20to=20preserve?= =?UTF-8?q?=20text=20headings=20without=20rewriting=20existing=20tags?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-authored-by: Codex --- tag-ocr-pdf.py | 62 +++++++++++++++++++++++++++++++++ tests/test_tag_existing_text.py | 24 +++++++++++++ 2 files changed, 86 insertions(+) diff --git a/tag-ocr-pdf.py b/tag-ocr-pdf.py index 484588c..5aaf96c 100755 --- a/tag-ocr-pdf.py +++ b/tag-ocr-pdf.py @@ -717,6 +717,61 @@ def write_heading_outline(pdf: pikepdf.Pdf, headings: list[dict[str, object]]) - return count +def preserved_text_headings(input_path: Path) -> list[dict[str, object]]: + """Find centered, isolated printed titles without rewriting existing tags. + + pdfminer is supplied by OCRmyPDF. Layout extraction handles PDFs that put + an entire page inside one BT/ET object, unlike the OCR tagging parser. + """ + from pdfminer.high_level import extract_pages + from pdfminer.layout import LTChar, LTTextLine + + headings = [] + for page_number, page in enumerate(extract_pages(input_path), start=1): + lines = [] + + def collect(item): + if isinstance(item, LTTextLine): + lines.append(item) + elif hasattr(item, "__iter__"): + for child in item: + collect(child) + + collect(page) + body_lines = [line for line in lines if line.width > page.width * .65] + if not body_lines: + continue + body_left = statistics.median(line.x0 for line in body_lines) + body_right = statistics.median(line.x1 for line in body_lines) + candidates = [] + for line in lines: + text = re.sub(r"\s+", " ", line.get_text()).strip() + words = [word for word in text.split() if word[0].isalpha()] + chars = [char for char in line if isinstance(char, LTChar)] + if (not words or not chars or not 4 <= len(text) <= 180 + or len(words) > 18 or WORD_END.search(text) + or text.endswith((",", ";", ":")) + or not page.height * .15 < line.y0 < page.height * .90 + or abs((line.x0 + line.x1) / 2 - page.width / 2) > page.width * .025 + or line.x0 < body_left + 8 or line.x1 > body_right - 8 + or sum(word[0].isupper() for word in words) / len(words) < .55): + continue + # Reject fragments sharing a row with prose or another column. + if any(other is not line and min(line.y1, other.y1) > max(line.y0, other.y0) + for other in lines): + continue + candidates.append((line, text)) + candidates.sort(key=lambda item: -item[0].y1) + previous = None + for line, text in candidates: + if previous is not None and 0 <= previous.y0 - line.y1 <= line.height * .6: + headings[-1]["text"] += " " + text + else: + headings.append({"level": "H1", "text": text, "page": page_number}) + previous = line + return headings + + def tag_pdf( input_path: Path, output_path: Path, @@ -785,6 +840,13 @@ def tag_pdf( report["href"] = reference_url or None if not any(lines for _, lines in extracted): report["preserved_existing_text"] = True + # Existing paragraph tags do not imply navigable section bookmarks. + # Keep the original structure and content streams untouched. + with pdf.open_outline() as outline: + has_outline = bool(outline.root) + if pdf.Root.get("/StructTreeRoot") is not None and not has_outline: + report["headings"] = preserved_text_headings(input_path) + report["outline_entries"] = write_heading_outline(pdf, report["headings"]) apply_metadata(pdf, creator, reference_url) pdf.save(output_path) report_path.write_text(json.dumps(report, indent=2, ensure_ascii=False) + "\n") diff --git a/tests/test_tag_existing_text.py b/tests/test_tag_existing_text.py index 691ebbf..ead2afa 100644 --- a/tests/test_tag_existing_text.py +++ b/tests/test_tag_existing_text.py @@ -99,6 +99,30 @@ def test_no_recognized_headings_preserves_existing_outline(self): with pdf.open_outline() as outline: self.assertEqual(outline.root[0].title, "Existing section") + def test_existing_tags_gain_outline_without_rewriting_page(self): + with tempfile.TemporaryDirectory() as folder: + source, output = (Path(folder) / name for name in ("source.pdf", "output.pdf")) + content = (b"BT /F1 12 Tf 1 0 0 1 276 600 Tm (Overview) Tj " + b"1 0 0 1 72 570 Tm (This is ordinary body prose filling the full measure across the page to its right margin.) Tj ET") + with pikepdf.Pdf.new() as pdf: + page = pdf.add_blank_page(page_size=(612, 792)) + page.obj.Resources = pikepdf.Dictionary(Font=pikepdf.Dictionary( + F1=pikepdf.Dictionary(Type=pikepdf.Name.Font, Subtype=pikepdf.Name.Type1, + BaseFont=pikepdf.Name("/Times-Roman"), Encoding=pikepdf.Name.WinAnsiEncoding))) + page.obj.Contents = pdf.make_stream(content) + pdf.Root.StructTreeRoot = pdf.make_indirect(pikepdf.Dictionary( + Type=pikepdf.Name.StructTreeRoot, TestMarker="Keep original tags")) + pdf.save(source) + report = tagger.tag_pdf(source, output, Path(folder) / "report.json") + self.assertEqual(report["tagged_pages"], 0) + self.assertEqual(report["outline_entries"], 1) + with pikepdf.open(output) as pdf: + self.assertEqual(pdf.pages[0].Contents.read_bytes(), content) + self.assertEqual(str(pdf.Root.StructTreeRoot.TestMarker), "Keep original tags") + with pdf.open_outline() as outline: + self.assertEqual(outline.root[0].title, "Overview") + self.assertEqual(outline.root[0].destination[0].objgen, pdf.pages[0].obj.objgen) + def test_no_new_ocr_preserves_existing_content_and_structure(self): with tempfile.TemporaryDirectory() as folder: source = Path(folder) / "source.pdf" From d75d942a0ec53b22f713b2712e62d8f1b6498240 Mon Sep 17 00:00:00 2001 From: John Wooten Date: Thu, 15 Oct 2026 19:52:54 -0400 Subject: [PATCH 2/5] =?UTF-8?q?=E2=9E=95=20Add=20function=20to=20promote?= =?UTF-8?q?=20preserved=20headings=20with=20inline=20styles?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-authored-by: Codex --- tag-ocr-pdf.py | 82 +++++++++++++++++++++++++++++++-- tests/test_tag_existing_text.py | 39 ++++++++++++++++ 2 files changed, 117 insertions(+), 4 deletions(-) diff --git a/tag-ocr-pdf.py b/tag-ocr-pdf.py index 5aaf96c..a141264 100755 --- a/tag-ocr-pdf.py +++ b/tag-ocr-pdf.py @@ -760,18 +760,91 @@ def collect(item): if any(other is not line and min(line.y1, other.y1) > max(line.y0, other.y0) for other in lines): continue - candidates.append((line, text)) + letters = [char for char in chars if char.get_text().isalpha()] + italic = sum(any(style in char.fontname.lower() for style in ("italic", "oblique")) + for char in letters) > len(letters) / 2 + candidates.append((line, text, "H3" if italic else "H2")) candidates.sort(key=lambda item: -item[0].y1) previous = None - for line, text in candidates: - if previous is not None and 0 <= previous.y0 - line.y1 <= line.height * .6: + for line, text, level in candidates: + if (previous is not None and headings[-1]["level"] == level + and 0 <= previous.y0 - line.y1 <= line.height * .6): headings[-1]["text"] += " " + text else: - headings.append({"level": "H1", "text": text, "page": page_number}) + headings.append({"level": level, "text": text, "page": page_number}) previous = line return headings +def promote_preserved_headings(pdf: pikepdf.Pdf, headings: list[dict[str, object]]) -> int: + """Promote exact matching paragraph elements; retain MCIDs and inline styles.""" + def key(text): + return re.sub(r"\s+", "", text) + + promoted = 0 + for page_number in sorted({heading["page"] for heading in headings}): + page = pdf.pages[page_number - 1] + fonts = page.obj.get("/Resources", Dictionary()).get("/Font", Dictionary()) + mappings = {} + for name, font in fonts.items(): + if font.get("/ToUnicode") is not None: + mappings[str(name)] = unicode_map(font) + elif font.get("/Encoding") == Name("/WinAnsiEncoding"): + mappings[str(name)] = {bytes([code]): bytes([code]).decode("cp1252", errors="replace") + for code in range(256)} + text_by_mcid = {} + stack = [] + font_name = None + for instruction in pikepdf.parse_content_stream(page): + op, args = str(instruction.operator), instruction.operands + if op == "Tf": + font_name = str(args[0]) + elif op in {"BDC", "BMC"}: + properties = args[1] if op == "BDC" else None + stack.append(int(properties.MCID) if isinstance(properties, Dictionary) + and "/MCID" in properties else None) + elif op == "EMC": + if stack: + stack.pop() + elif op in {"Tj", "TJ", "'", '"'} and stack and stack[-1] is not None: + mapping = mappings.get(font_name) + if mapping: + operands = args if op in {"Tj", "TJ"} else [args[-1]] + text_by_mcid.setdefault(stack[-1], []) + text_by_mcid[stack[-1]].append(decode_show_text(operands, mapping)) + wanted = {key(heading["text"]): heading["level"] for heading in headings + if heading["page"] == page_number} + + def element_text(item): + if isinstance(item, int): + return "".join(text_by_mcid.get(item, [])) + if isinstance(item, Array): + return "".join(element_text(child) for child in item) + if isinstance(item, Dictionary): + if "/MCID" in item: + return "".join(text_by_mcid.get(int(item.MCID), [])) + return element_text(item.get("/K")) + return "" + + def visit(item, inherited_page=None): + nonlocal promoted + if isinstance(item, Array): + for child in item: + visit(child, inherited_page) + elif isinstance(item, Dictionary): + owner = item.get("/Pg", inherited_page) + if (owner is not None and owner.objgen == page.obj.objgen + and item.get("/S") == Name("/P")): + level = wanted.get(key(element_text(item.get("/K")))) + if level: + item["/S"] = Name("/" + level) + promoted += 1 + visit(item.get("/K"), owner) + + visit(pdf.Root.StructTreeRoot.get("/K")) + return promoted + + def tag_pdf( input_path: Path, output_path: Path, @@ -847,6 +920,7 @@ def tag_pdf( if pdf.Root.get("/StructTreeRoot") is not None and not has_outline: report["headings"] = preserved_text_headings(input_path) report["outline_entries"] = write_heading_outline(pdf, report["headings"]) + report["promoted_heading_tags"] = promote_preserved_headings(pdf, report["headings"]) apply_metadata(pdf, creator, reference_url) pdf.save(output_path) report_path.write_text(json.dumps(report, indent=2, ensure_ascii=False) + "\n") diff --git a/tests/test_tag_existing_text.py b/tests/test_tag_existing_text.py index ead2afa..2953f35 100644 --- a/tests/test_tag_existing_text.py +++ b/tests/test_tag_existing_text.py @@ -123,6 +123,45 @@ def test_existing_tags_gain_outline_without_rewriting_page(self): self.assertEqual(outline.root[0].title, "Overview") self.assertEqual(outline.root[0].destination[0].objgen, pdf.pages[0].obj.objgen) + def test_mixed_italic_heading_becomes_h3_with_inline_style_preserved(self): + with tempfile.TemporaryDirectory() as folder: + source, output = (Path(folder) / name for name in ("source.pdf", "output.pdf")) + content = (b"BT /F1 12 Tf /P << /MCID 0 >> BDC " + b"1 0 0 1 276 650 Tm (Overview) Tj EMC " + b"/P << /MCID 1 >> BDC /F2 12 Tf " + b"1 0 0 1 258 600 Tm (The Rule Filing) Tj EMC " + b"/Span << /MCID 2 >> BDC /F1 12 Tf ( 1975) Tj EMC " + b"/P << /MCID 3 >> BDC 1 0 0 1 72 570 Tm " + b"(This is ordinary body prose filling the full measure across the page to its right margin.) Tj EMC ET") + with pikepdf.Pdf.new() as pdf: + page = pdf.add_blank_page(page_size=(612, 792)) + def font(name): + return pikepdf.Dictionary(Type=pikepdf.Name.Font, Subtype=pikepdf.Name.Type1, + BaseFont=pikepdf.Name('/' + name), Encoding=pikepdf.Name.WinAnsiEncoding) + page.obj.Resources = pikepdf.Dictionary(Font=pikepdf.Dictionary( + F1=font("Times-Roman"), F2=font("Times-Italic"))) + page.obj.Contents = pdf.make_stream(content) + style = pdf.make_indirect(pikepdf.Dictionary(S=pikepdf.Name.Span, K=2)) + title = pdf.make_indirect(pikepdf.Dictionary(S=pikepdf.Name.P, Pg=page.obj, + K=pikepdf.Array([1, style]))) + pdf.Root.StructTreeRoot = pdf.make_indirect(pikepdf.Dictionary( + Type=pikepdf.Name.StructTreeRoot, K=pikepdf.Array([ + pdf.make_indirect(pikepdf.Dictionary(S=pikepdf.Name.P, Pg=page.obj, K=0)), + title]))) + pdf.save(source) + report = tagger.tag_pdf(source, output, Path(folder) / "report.json") + self.assertEqual([(h["text"], h["level"]) for h in report["headings"]], + [("Overview", "H2"), ("The Rule Filing 1975", "H3")]) + self.assertEqual(report["promoted_heading_tags"], 2) + with pikepdf.open(output) as pdf: + self.assertEqual(pdf.pages[0].Contents.read_bytes(), content) + title = pdf.Root.StructTreeRoot.K[1] + self.assertEqual(title.S, pikepdf.Name.H3) + self.assertEqual(title.K[1].S, pikepdf.Name.Span) + self.assertEqual(title.K[1].K, 2) + with pdf.open_outline() as outline: + self.assertEqual(outline.root[0].children[0].title, "The Rule Filing 1975") + def test_no_new_ocr_preserves_existing_content_and_structure(self): with tempfile.TemporaryDirectory() as folder: source = Path(folder) / "source.pdf" From f6e9f779d30639064e8dd41c060ac0159031c69e Mon Sep 17 00:00:00 2001 From: John Wooten Date: Thu, 15 Oct 2026 20:02:13 -0400 Subject: [PATCH 3/5] =?UTF-8?q?=F0=9F=94=A7=20Refine=20text=20extraction?= =?UTF-8?q?=20and=20preservation=20in=20tag-ocr-pdf.py?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-authored-by: Codex --- tag-ocr-pdf.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/tag-ocr-pdf.py b/tag-ocr-pdf.py index a141264..01fe59f 100755 --- a/tag-ocr-pdf.py +++ b/tag-ocr-pdf.py @@ -718,7 +718,7 @@ def write_heading_outline(pdf: pikepdf.Pdf, headings: list[dict[str, object]]) - def preserved_text_headings(input_path: Path) -> list[dict[str, object]]: - """Find centered, isolated printed titles without rewriting existing tags. + """Find centered, isolated printed titles without rewriting page content. pdfminer is supplied by OCRmyPDF. Layout extraction handles PDFs that put an entire page inside one BT/ET object, unlike the OCR tagging parser. @@ -914,7 +914,8 @@ def tag_pdf( if not any(lines for _, lines in extracted): report["preserved_existing_text"] = True # Existing paragraph tags do not imply navigable section bookmarks. - # Keep the original structure and content streams untouched. + # Retain the existing structure and content streams, promoting only + # paragraph elements that exactly match recognized headings. with pdf.open_outline() as outline: has_outline = bool(outline.root) if pdf.Root.get("/StructTreeRoot") is not None and not has_outline: From 961830716c6b4b5cc8e7dc27575763cb6125b413 Mon Sep 17 00:00:00 2001 From: John Wooten Date: Thu, 15 Oct 2026 20:02:21 -0400 Subject: [PATCH 4/5] =?UTF-8?q?=E2=9E=95=20Add=20date=20normalization=20to?= =?UTF-8?q?=20PDF=20metadata?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-authored-by: Codex --- pdf_metadata.py | 51 ++++++++++++++++++++++++++++++++++++++ tests/test_pdf_metadata.py | 26 +++++++++++++++++++ 2 files changed, 77 insertions(+) diff --git a/pdf_metadata.py b/pdf_metadata.py index f9d02f0..a9bcf6a 100644 --- a/pdf_metadata.py +++ b/pdf_metadata.py @@ -1,6 +1,7 @@ """Identify a document creator and keep PDF Magic metadata consistent.""" from collections import Counter +from datetime import date import getpass import os import re @@ -163,9 +164,59 @@ def choose_source_url(existing, supplied): return supplied or existing +def normalize_keyword_dates(pdf, metadata): + """Move explicitly labeled ISO document dates out of keyword fields.""" + dates = [] + + def clean_keywords(value): + kept = [] + removed = False + for token in re.split(r"[;,\n]", str(value or "")): + token = token.strip() + match = re.fullmatch(r"Date:\s*(\d{4}-\d{2}-\d{2})", token, re.I) + if match: + try: + date.fromisoformat(match[1]) + except ValueError: + pass + else: + removed = True + if match[1] not in dates: + dates.append(match[1]) + continue + if token: + kept.append(token) + # Preserve original punctuation and spacing when nothing was removed. + return "; ".join(kept) if removed else value + + for container, field in ((pdf.docinfo, "/Keywords"), (metadata, "pdf:Keywords")): + if field in container: + original = str(container[field]) + cleaned = clean_keywords(original) + if cleaned != original: + if cleaned: + container[field] = cleaned + else: + del container[field] + subjects = metadata.get("dc:subject") + if subjects: + subjects = [subjects] if isinstance(subjects, str) else subjects + cleaned = [clean_keywords(value) for value in subjects] + cleaned = [value for value in cleaned if value] + if cleaned: + metadata["dc:subject"] = cleaned + else: + del metadata["dc:subject"] + if dates: + existing = metadata.get("dc:date") or [] + existing = [existing] if isinstance(existing, str) else list(existing) + metadata["dc:date"] = existing + [value for value in dates if value not in existing] + + def apply_metadata(pdf, creator, source_url=""): pikepdf.models.PdfMetadata.register_xml_namespace(SOURCE_NAMESPACE, "pdfmagic") with pdf.open_metadata(set_pikepdf_as_editor=False, update_docinfo=False) as metadata: + normalize_keyword_dates(pdf, metadata) metadata["dc:creator"] = [creator] metadata["xmp:CreatorTool"] = creator metadata["pdf:Producer"] = PRODUCER diff --git a/tests/test_pdf_metadata.py b/tests/test_pdf_metadata.py index 323c23b..492ef21 100644 --- a/tests/test_pdf_metadata.py +++ b/tests/test_pdf_metadata.py @@ -21,6 +21,32 @@ def line(text, y, page=1, size=12, x=72): class MetadataTests(unittest.TestCase): + def test_labeled_keyword_date_moves_to_document_date(self): + with pikepdf.Pdf.new() as pdf: + pdf.add_blank_page() + pdf.docinfo.Keywords = "transfer agents; Date: 2009-08-04" + pdf.docinfo.CreationDate = "D:20090908180705" + with pdf.open_metadata(set_pikepdf_as_editor=False, update_docinfo=False) as xmp: + xmp["pdf:Keywords"] = "Date: 2009-08-04" + xmp["dc:subject"] = ["Date: 2009-08-04", "securities"] + metadata.apply_metadata(pdf, "Document Author") + self.assertEqual(str(pdf.docinfo.Keywords), "transfer agents") + self.assertEqual(str(pdf.docinfo.CreationDate), "D:20090908180705") + with pdf.open_metadata(set_pikepdf_as_editor=False, update_docinfo=False) as xmp: + self.assertNotIn("pdf:Keywords", xmp) + self.assertEqual(set(xmp["dc:subject"]), {"securities"}) + self.assertEqual(xmp["dc:date"], ["2009-08-04"]) + + def test_ordinary_and_invalid_date_keywords_are_preserved(self): + for value in ("Date: 2009-02-30", "2009-08-04", "securities, transfer agents"): + with pikepdf.Pdf.new() as pdf: + pdf.add_blank_page() + pdf.docinfo.Keywords = value + metadata.apply_metadata(pdf, "Document Author") + self.assertEqual(str(pdf.docinfo.Keywords), value) + with pdf.open_metadata(set_pikepdf_as_editor=False, update_docinfo=False) as xmp: + self.assertNotIn("dc:date", xmp) + def test_signature_beats_recipient_and_letterhead(self): records = [line("Dr. Recipient Name", 700, size=28), line("Dear Dr. Recipient Name:", 600), line("Body paragraph here.", 550), line("Respectfully submitted,", 200), From 46cfc9b7b5360451b637d138301f3850d0070802 Mon Sep 17 00:00:00 2001 From: John Wooten <60260750+JFWooten4@users.noreply.github.com> Date: Sun, 18 Oct 2026 16:43:31 -0400 Subject: [PATCH 5/5] =?UTF-8?q?=F0=9F=90=9E=20Accept=20Dublin=20Core=20lan?= =?UTF-8?q?guage=20collections=20across=20pikepdf=20versions?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Compare Dublin Core language values as a set because recent pikepdf versions return a set for rdf:Bag properties. Preserve the exact expected en-US value while allowing the metadata and preserved-text regression jobs to complete. Co-authored-by: Codex Web --- tests/test_pdf_metadata.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/test_pdf_metadata.py b/tests/test_pdf_metadata.py index bdcfc79..bc4f1a8 100644 --- a/tests/test_pdf_metadata.py +++ b/tests/test_pdf_metadata.py @@ -261,7 +261,7 @@ def test_optional_metadata_and_keyword_tags_preserve_existing_creator(self): self.assertEqual(xmp["pdf:Keywords"], "Markets, stocks, ETFs") self.assertEqual(xmp["dc:publisher"], {"Example Publisher"}) self.assertEqual(xmp["dc:rights"], "CC BY-SA 4.0") - self.assertEqual(xmp["dc:language"], ["en-US"]) + self.assertEqual(set(xmp["dc:language"]), {"en-US"}) self.assertEqual(xmp["dc:identifier"], "978-1-2345") def test_original_author_metadata_survives_ocr_creator_overwrite(self):