diff --git a/.github/workflows/pdf-regressions.yml b/.github/workflows/pdf-regressions.yml new file mode 100644 index 0000000..20d4ab2 --- /dev/null +++ b/.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 diff --git a/pdf_metadata.py b/pdf_metadata.py index f261fb6..88af678 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 @@ -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: diff --git a/tag-ocr-pdf.py b/tag-ocr-pdf.py index 4819b4d..05ce62c 100644 --- a/tag-ocr-pdf.py +++ b/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) diff --git a/tests/test_pdf_metadata.py b/tests/test_pdf_metadata.py index 04a89c9..bc4f1a8 100644 --- a/tests/test_pdf_metadata.py +++ b/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): diff --git a/tests/test_tag_existing_text.py b/tests/test_tag_existing_text.py index 07d08a4..2cf35f3 100644 --- a/tests/test_tag_existing_text.py +++ b/tests/test_tag_existing_text.py @@ -284,6 +284,69 @@ 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_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"