Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
34 changes: 27 additions & 7 deletions packages/markitdown/src/markitdown/converters/_epub_converter.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,7 @@
import zipfile
from urllib.parse import unquote
from defusedxml import minidom
from xml.dom.minidom import Document
from xml.dom.minidom import Document, Element

from typing import BinaryIO, Any, Dict, List, Set

Expand All @@ -24,6 +24,10 @@
".xhtml": "application/xhtml+xml",
}

CONTAINER_NAMESPACE = "urn:oasis:names:tc:opendocument:xmlns:container"
PACKAGE_NAMESPACE = "http://www.idpf.org/2007/opf"
DC_NAMESPACE = "http://purl.org/dc/elements/1.1/"


class EpubConverter(HtmlConverter):
"""
Expand Down Expand Up @@ -63,9 +67,9 @@ def convert(

# Locate content.opf
container_dom = minidom.parse(z.open("META-INF/container.xml"))
opf_path = container_dom.getElementsByTagName("rootfile")[0].getAttribute(
"full-path"
)
opf_path = self._get_package_elements(
container_dom, CONTAINER_NAMESPACE, "rootfile"
)[0].getAttribute("full-path")

# Parse content.opf
opf_dom = minidom.parse(z.open(opf_path))
Expand All @@ -82,11 +86,15 @@ def convert(
# Extract manifest items (ID → href mapping)
manifest = {
item.getAttribute("id"): item.getAttribute("href")
for item in opf_dom.getElementsByTagName("item")
for item in self._get_package_elements(
opf_dom, PACKAGE_NAMESPACE, "item"
)
}

# Extract spine order (ID refs)
spine_items = opf_dom.getElementsByTagName("itemref")
spine_items = self._get_package_elements(
opf_dom, PACKAGE_NAMESPACE, "itemref"
)
spine_order = [item.getAttribute("idref") for item in spine_items]

# Convert spine order to actual file paths
Expand Down Expand Up @@ -156,6 +164,16 @@ def _resolve_manifest_href(

return candidates[0]

def _get_package_elements(
self, dom: Document, namespace: str, local_name: str
) -> List[Element]:
# Keep accepting unqualified elements from older, non-conforming files.
return [
node
for node in dom.getElementsByTagNameNS("*", local_name)
if node.namespaceURI in (namespace, None)
]

def _get_text_from_node(self, dom: Document, tag_name: str) -> str | None:
"""Convenience function to extract a single occurrence of a tag (e.g., title)."""
texts = self._get_all_texts_from_nodes(dom, tag_name)
Expand All @@ -167,7 +185,9 @@ def _get_text_from_node(self, dom: Document, tag_name: str) -> str | None:
def _get_all_texts_from_nodes(self, dom: Document, tag_name: str) -> List[str]:
"""Helper function to extract all occurrences of a tag (e.g., multiple authors)."""
texts: List[str] = []
for node in dom.getElementsByTagName(tag_name):
for node in dom.getElementsByTagNameNS(
DC_NAMESPACE, tag_name.removeprefix("dc:")
):
text_parts: List[str] = []
self._collect_node_text(node, text_parts)
text_val = "".join(text_parts).strip()
Expand Down
146 changes: 134 additions & 12 deletions packages/markitdown/tests/test_epub_converter.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,9 @@
import io
import zipfile

from markitdown import StreamInfo
import pytest

from markitdown import MarkItDown, StreamInfo
from markitdown.converters import EpubConverter

CONTAINER_XML = """<?xml version="1.0"?>
Expand All @@ -18,27 +20,57 @@
"""


def _build_epub(manifest_items, spine_ids, documents) -> io.BytesIO:
def _build_epub(
manifest_items,
spine_ids,
documents,
*,
container_prefix="",
package_prefix="",
dc_prefix="dc",
) -> io.BytesIO:
"""Assemble a minimal EPUB from manifest entries and ZIP member names."""
package_xmlns = f"xmlns:{package_prefix}" if package_prefix else "xmlns"
dc_xmlns = f"xmlns:{dc_prefix}" if dc_prefix else "xmlns"
package_tag = f"{package_prefix}:" if package_prefix else ""
dc_tag = f"{dc_prefix}:" if dc_prefix else ""
manifest = "\n".join(
f'<item id="{item_id}" href="{href}" media-type="application/xhtml+xml"/>'
f'<{package_tag}item id="{item_id}" href="{href}" media-type="application/xhtml+xml"/>'
for item_id, href in manifest_items
)
spine = "\n".join(f'<itemref idref="{item_id}"/>' for item_id in spine_ids)
spine = "\n".join(
f'<{package_tag}itemref idref="{item_id}"/>' for item_id in spine_ids
)
opf = f"""<?xml version="1.0"?>
<package xmlns="http://www.idpf.org/2007/opf" version="3.0" unique-identifier="id">
<metadata xmlns:dc="http://purl.org/dc/elements/1.1/">
<dc:title>Encoded Hrefs</dc:title>
</metadata>
<manifest>{manifest}</manifest>
<spine>{spine}</spine>
</package>
<{package_tag}package {package_xmlns}="http://www.idpf.org/2007/opf" version="3.0" unique-identifier="id">
<{package_tag}metadata {dc_xmlns}="http://purl.org/dc/elements/1.1/">
<{dc_tag}title>Encoded Hrefs</{dc_tag}title>
<{dc_tag}creator>First Author</{dc_tag}creator>
<{dc_tag}creator>Second Author</{dc_tag}creator>
</{package_tag}metadata>
<{package_tag}manifest>{manifest}</{package_tag}manifest>
<{package_tag}spine>{spine}</{package_tag}spine>
</{package_tag}package>
"""

container_xml = CONTAINER_XML
if container_prefix:
container_xml = container_xml.replace("xmlns=", f"xmlns:{container_prefix}=")
for tag in ("container", "rootfiles", "rootfile"):
container_xml = container_xml.replace(
f"<{tag} ", f"<{container_prefix}:{tag} "
)
container_xml = container_xml.replace(
f"<{tag}>", f"<{container_prefix}:{tag}>"
)
container_xml = container_xml.replace(
f"</{tag}>", f"</{container_prefix}:{tag}>"
)

buffer = io.BytesIO()
with zipfile.ZipFile(buffer, "w") as z:
z.writestr("mimetype", "application/epub+zip")
z.writestr("META-INF/container.xml", CONTAINER_XML)
z.writestr("META-INF/container.xml", container_xml)
z.writestr("OEBPS/content.opf", opf)
for name, (title, body) in documents.items():
z.writestr(name, CHAPTER_XHTML.format(title=title, body=body))
Expand Down Expand Up @@ -107,6 +139,96 @@ def test_parent_relative_href_resolves() -> None:
assert "SHARED_BODY" in _convert(stream)


@pytest.mark.parametrize("public_api", [False, True])
@pytest.mark.parametrize(
"container_prefix, package_prefix, dc_prefix",
[
("", "", "dc"),
("ocf", "", "dc"),
("", "opf", "dc"),
("", "", "meta"),
("ocf", "opf", ""),
],
)
def test_namespace_prefixes_preserve_metadata_and_spine_order(
public_api, container_prefix, package_prefix, dc_prefix
) -> None:
stream = _build_epub(
manifest_items=[("c1", "first.xhtml"), ("c2", "second.xhtml")],
spine_ids=["c2", "c1"],
documents={
"OEBPS/first.xhtml": ("First", "First chapter body"),
"OEBPS/second.xhtml": ("Second", "Second chapter body"),
},
container_prefix=container_prefix,
package_prefix=package_prefix,
dc_prefix=dc_prefix,
)
stream_info = StreamInfo(mimetype="application/epub+zip", extension=".epub")
if public_api:
result = MarkItDown().convert_stream(stream, stream_info=stream_info)
else:
result = EpubConverter().convert(stream, stream_info)

assert result.title == "Encoded Hrefs"
assert "**Authors:** First Author, Second Author" in result.markdown
assert "First chapter body" in result.markdown
assert "Second chapter body" in result.markdown
assert result.markdown.index("Second chapter body") < result.markdown.index(
"First chapter body"
)


@pytest.mark.parametrize("unqualified", [False, True])
def test_package_elements_ignore_foreign_namespaces(unqualified) -> None:
stream = _build_epub(
manifest_items=[("c1", "chapter.xhtml")],
spine_ids=["c1"],
documents={"OEBPS/chapter.xhtml": ("Chapter", "Expected chapter body")},
)
modified = io.BytesIO()
with zipfile.ZipFile(stream) as source, zipfile.ZipFile(modified, "w") as target:
for name in source.namelist():
data = source.read(name)
if name == "META-INF/container.xml":
data = data.replace(
b"<rootfiles>",
b'<rootfiles><ext:rootfile xmlns:ext="urn:example:extension" '
b'full-path="missing.opf"/>',
)
if unqualified:
data = data.replace(
b' xmlns="urn:oasis:names:tc:opendocument:xmlns:container"', b""
)
elif name == "OEBPS/content.opf":
data = (
data.replace(
b"</manifest>",
b'<ext:item xmlns:ext="urn:example:extension" '
b'id="c1" href="missing.xhtml"/></manifest>',
)
.replace(
b"<spine>",
b'<spine><ext:itemref xmlns:ext="urn:example:extension" idref="c1"/>',
)
.replace(
b"<dc:title>",
b'<ext:title xmlns:ext="urn:example:extension">Wrong title</ext:title><dc:title>',
)
)
if unqualified:
data = data.replace(b' xmlns="http://www.idpf.org/2007/opf"', b"")
target.writestr(name, data)
modified.seek(0)

result = MarkItDown().convert_stream(
modified, stream_info=StreamInfo(extension=".epub")
)
assert result.title == "Encoded Hrefs"
assert result.markdown.count("Expected chapter body") == 1
assert "Wrong title" not in result.markdown


if __name__ == "__main__":
test_percent_encoded_href_resolves_to_zip_entry()
test_non_ascii_percent_encoded_href_resolves()
Expand Down