🔩 Add outlines to preserved tagged PDFs and normalize document dates

.github/workflows/pdf-regressions.yml

@@ -0,0 +1,25 @@
+name: PDF metadata and preserved-text regressions
+
+on:
+  push:
+    branches: [snails]
+  pull_request:
+    paths:
+      - "pdf_metadata.py"
+      - "tag-ocr-pdf.py"
+      - "tests/test_pdf_metadata.py"
+      - "tests/test_tag_existing_text.py"
+      - ".github/workflows/pdf-regressions.yml"
+
+jobs:
+  metadata-and-headings:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-python@v5
+        with:
+          python-version: "3.12"
+      - name: Install regression dependencies
+        run: python -m pip install pikepdf pdfminer.six
+      - name: Run metadata and tagged PDF tests
+        run: python -m unittest discover -s tests -p 'test_pdf_metadata.py' -v && python -m unittest discover -s tests -p 'test_tag_existing_text.py' -v

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
@@ -155,6 +156,31 @@ def _keywords(value):
     return result
 
 
+def extract_keyword_dates(value):
+    """Separate valid labeled document dates from descriptive keyword tokens."""
+    raw = sorted(value) if isinstance(value, set) else (
+        value if isinstance(value, (list, tuple)) else [value]
+    )
+    keywords = []
+    dates = []
+    for entry in raw:
+        for token in re.split(r"[;,\n]", str(entry or "")):
+            token = clean_text(token)
+            match = re.fullmatch(r"Date:\s*(\d{4}-\d{2}-\d{2})", token, re.I)
+            if match:
+                try:
+                    date.fromisoformat(match.group(1))
+                except ValueError:
+                    pass
+                else:
+                    if match.group(1) not in dates:
+                        dates.append(match.group(1))
+                    continue
+            if token:
+                keywords.append(token)
+    return _keywords(keywords), dates
+
+
 def metadata_details(pdf):
     """Read reusable descriptive metadata, excluding derivative-specific IDs/dates."""
     pikepdf.models.PdfMetadata.register_xml_namespace(SOURCE_NAMESPACE, "pdfmagic")
@@ -174,6 +200,29 @@ def metadata_details(pdf):
         keywords = metadata.get("dc:subject") or []
         if not keywords:
             keywords = metadata.get("pdf:Keywords") or pdf.docinfo.get("/Keywords") or ""
+        # Capture dates in every metadata keyword field, even when another
+        # field supplies the preferred descriptive keywords.
+        keyword_dates = []
+        for source in (pdf.docinfo.get("/Keywords"), metadata.get("pdf:Keywords"),
+                       metadata.get("dc:subject")):
+            _, extracted = extract_keyword_dates(source)
+            keyword_dates.extend(item for item in extracted if item not in keyword_dates)
+        existing_dates = metadata.get("dc:date") or []
+        if not isinstance(existing_dates, (list, tuple, set)):
+            existing_dates = [existing_dates]
+        dates = list(dict.fromkeys([
+            *(clean_text(item) for item in existing_dates if clean_text(item)),
+            *keyword_dates,
+        ]))
+        if keyword_dates:
+            # When keyword dates are removed, retain other keywords from all
+            # Info/XMP sources instead of losing a non-preferred source.
+            merged_keywords = []
+            for source in (pdf.docinfo.get("/Keywords"), metadata.get("pdf:Keywords"),
+                           metadata.get("dc:subject")):
+                ordinary, _ = extract_keyword_dates(source)
+                merged_keywords.extend(ordinary)
+            keywords = _keywords(merged_keywords)
         return {
             "author": author if author and not SOFTWARE.search(author) else None,
             "creator_tool": info_creator or xmp_tool,
@@ -183,7 +232,8 @@ def metadata_details(pdf):
             "xmp_producer": xmp_producer,
             "title": clean_text(pdf.docinfo.get("/Title")) or _first_text(metadata.get("dc:title")),
             "description": clean_text(pdf.docinfo.get("/Subject")) or _first_text(metadata.get("dc:description")),
-            "keywords": _keywords(keywords),
+            "keywords": extract_keyword_dates(keywords)[0],
+            "dates": dates,
             "publisher": _first_text(metadata.get("dc:publisher")),
             "rights": _first_text(metadata.get("dc:rights")),
             "language": _first_text(metadata.get("dc:language")) or clean_text(pdf.Root.get("/Lang")),
@@ -242,7 +292,8 @@ def apply_metadata(pdf, author, source_url="", original=None, overrides=None):
     xmp_creator_tool = clean_text(original.get("xmp_creator_tool")) or creator_tool
     info_producer = enhanced_producer(original.get("info_producer") or original.get("xmp_producer"))
     xmp_producer = enhanced_producer(original.get("xmp_producer") or original.get("info_producer"))
-    keywords = _keywords(values.get("keywords") or [])
+    keywords, supplied_dates = extract_keyword_dates(values.get("keywords") or [])
+    dates = list(dict.fromkeys([*(original.get("dates") or []), *supplied_dates]))
     pikepdf.models.PdfMetadata.register_xml_namespace(SOURCE_NAMESPACE, "pdfmagic")
     with pdf.open_metadata(set_pikepdf_as_editor=False, update_docinfo=False) as metadata:
         if author:
@@ -266,6 +317,12 @@ def apply_metadata(pdf, author, source_url="", original=None, overrides=None):
         if keywords:
             metadata["dc:subject"] = set(keywords)
             metadata["pdf:Keywords"] = ", ".join(keywords)
+        elif dates:
+            for prop in ("dc:subject", "pdf:Keywords"):
+                if prop in metadata:
+                    del metadata[prop]
+        if dates:
+            metadata["dc:date"] = dates
         if source_url:
             metadata["pdfmagic:href"] = source_url
             metadata["dc:source"] = source_url
@@ -282,6 +339,8 @@ def apply_metadata(pdf, author, source_url="", original=None, overrides=None):
             pdf.docinfo[prop] = values[field]
     if keywords:
         pdf.docinfo["/Keywords"] = ", ".join(keywords)
+    elif dates and "/Keywords" in pdf.docinfo:
+        del pdf.docinfo["/Keywords"]
     if values.get("language"):
         pdf.Root["/Lang"] = pikepdf.String(values["language"])
     if "/PDFMagicSourceURL" in pdf.docinfo:

tag-ocr-pdf.py

@@ -884,6 +884,134 @@ 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 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.
+    """
+    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
+            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, 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": 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("/Subtype") in {Name("/Type1"), Name("/TrueType")} and 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,
@@ -979,6 +1107,11 @@ def tag_pdf(
         report["href"] = reference_url or None
         if not any(lines for _, lines in extracted):
             report["preserved_existing_text"] = True
+            if (not retag_existing and pdf.Root.get("/StructTreeRoot") is not None
+                    and not existing_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"])
             if not retag_existing or metadata_overrides:
                 apply_metadata(pdf, creator, reference_url, original=original, overrides=metadata_overrides)
             pdf.save(output_path, fix_metadata_version=not retag_existing)

tests/test_pdf_metadata.py

@@ -21,6 +21,59 @@ 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, securities")
+            self.assertEqual(str(pdf.docinfo.CreationDate), "D:20090908180705")
+            with pdf.open_metadata(set_pikepdf_as_editor=False, update_docinfo=False) as xmp:
+                self.assertEqual(xmp["pdf:Keywords"], "transfer agents, securities")
+                self.assertEqual(set(xmp["dc:subject"]), {"transfer agents", "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_date_normalization_retains_source_metadata_and_previous_dates(self):
+        with tempfile.TemporaryDirectory() as folder:
+            source = Path(folder) / "source.pdf"
+            output = Path(folder) / "output.pdf"
+            with pikepdf.Pdf.new() as pdf:
+                pdf.add_blank_page()
+                pdf.docinfo.Author = "Original Author"
+                pdf.docinfo.Creator = "Original Creator"
+                pdf.docinfo.Producer = "Original Producer"
+                pdf.docinfo.CreationDate = "D:20090908180705"
+                pdf.docinfo.Keywords = "securities; Date: 2009-08-04"
+                with pdf.open_metadata(set_pikepdf_as_editor=False, update_docinfo=False) as xmp:
+                    xmp["dc:date"] = ["2001-01-01"]
+                    xmp["dc:subject"] = ["existing keyword"]
+                pdf.save(source)
+            tagger.tag_pdf(source, output, Path(folder) / "report.json",
+                           metadata_overrides={"keywords": "Date: 2009-08-04, updated"})
+            with pikepdf.open(output) as pdf:
+                self.assertEqual(str(pdf.docinfo.Creator), "Original Creator")
+                self.assertEqual(str(pdf.docinfo.Producer), "Original Producer; PDF Magic Enhancer")
+                self.assertEqual(str(pdf.docinfo.CreationDate), "D:20090908180705")
+                self.assertEqual(str(pdf.docinfo.Keywords), "updated")
+                with pdf.open_metadata(set_pikepdf_as_editor=False, update_docinfo=False) as xmp:
+                    self.assertEqual(xmp["dc:date"], ["2001-01-01", "2009-08-04"])
+                    self.assertEqual(set(xmp["dc:subject"]), {"updated"})
+                    self.assertEqual(xmp["pdf:Keywords"], "updated")
+
     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),
@@ -208,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):