1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
|
"""Text extraction from book files (TXT, PDF, EPUB) and text/HTML cleaning."""
import logging
import re
import zipfile
from html import unescape
from pathlib import Path
try:
from bs4 import BeautifulSoup
BS4_AVAILABLE = True
except ImportError:
BS4_AVAILABLE = False
logger = logging.getLogger(__name__)
def extract_text(file_path: Path) -> str:
"""Extract text from a book file based on its extension."""
extension = file_path.suffix.lower()
if extension == ".txt":
return _extract_txt(file_path)
if extension == ".pdf":
return _extract_pdf(file_path)
if extension == ".epub":
return extract_epub(file_path)
raise ValueError(f"Unsupported file format: {extension}")
def clean_text(text: str) -> str:
"""Normalize whitespace and strip standalone page numbers.
Page numbers are removed only when they appear as a short number alone on
its own line (before whitespace collapsing), so inline numbers like
"42 years", "1,000" or "3.5" are preserved.
"""
if not text:
return ""
# Standalone page numbers (digits alone on a line) must go BEFORE the
# newline-collapsing step below.
text = re.sub(r"(?m)^\s*\d{1,4}\s*$", " ", text)
text = re.sub(r"\s+", " ", text)
return text.strip()
def clean_html(html_content: str) -> str:
"""Strip markup, scripts and styles from HTML content."""
if not html_content:
return ""
if BS4_AVAILABLE:
try:
soup = BeautifulSoup(html_content, "html.parser")
for tag in soup(["script", "style"]):
tag.decompose()
text = soup.get_text()
lines = (line.strip() for line in text.splitlines())
chunks = (phrase.strip() for line in lines for phrase in line.split(" "))
return " ".join(chunk for chunk in chunks if chunk)
except Exception as exc:
logger.debug("BeautifulSoup cleaning failed, falling back to regex: %s", exc)
# Fallback regex cleaning
html_content = re.sub(r"<style[^>]*>.*?</style>", "", html_content, flags=re.DOTALL | re.IGNORECASE)
html_content = re.sub(r"<script[^>]*>.*?</script>", "", html_content, flags=re.DOTALL | re.IGNORECASE)
html_content = re.sub(r"<[^>]+>", " ", html_content)
html_content = unescape(html_content)
html_content = re.sub(r"\s+", " ", html_content)
return html_content.strip()
def extract_epub(file_path: Path) -> str:
"""Extract text from EPUB, trying several methods in order."""
methods = [
_extract_epub_ebooklib,
_extract_epub_zipfile,
_extract_epub_manual,
]
for method in methods:
try:
text = method(file_path)
if text and text.strip():
logger.info("EPUB extraction successful (%s): %d characters", method.__name__, len(text))
return text
except Exception as exc:
logger.warning("EPUB method %s failed: %s", method.__name__, exc)
raise RuntimeError("All EPUB extraction methods failed")
def _extract_epub_ebooklib(file_path: Path) -> str:
"""Extract using ebooklib, following the spine (reading) order."""
import ebooklib
from ebooklib import epub
book = epub.read_epub(str(file_path))
text_parts = []
for entry in book.spine:
item_id = entry[0] if isinstance(entry, (tuple, list)) else entry
try:
item = book.get_item_with_id(item_id)
if item and item.get_type() == ebooklib.ITEM_DOCUMENT:
content = item.get_body_content()
if content:
if isinstance(content, bytes):
content = content.decode("utf-8", errors="ignore")
cleaned = clean_html(str(content))
if cleaned.strip():
text_parts.append(cleaned)
except Exception as exc:
logger.debug("Skipping EPUB spine item %r: %s", item_id, exc)
return "\n\n".join(text_parts)
def _extract_epub_zipfile(file_path: Path) -> str:
"""Extract by parsing HTML members of the EPUB zip directly."""
text_parts = []
with zipfile.ZipFile(file_path, "r") as epub_zip:
for file_name in sorted(epub_zip.namelist()):
if file_name.lower().endswith((".html", ".xhtml", ".htm")):
try:
content = epub_zip.read(file_name).decode("utf-8", errors="ignore")
cleaned = clean_html(content)
if cleaned.strip():
text_parts.append(cleaned)
except Exception as exc:
logger.debug("Skipping EPUB member %r: %s", file_name, exc)
return "\n\n".join(text_parts)
def _extract_epub_manual(file_path: Path) -> str:
"""Last-resort extraction from any markup-looking EPUB member."""
skipped_extensions = (".jpg", ".jpeg", ".png", ".gif", ".css", ".js")
text_parts = []
with zipfile.ZipFile(file_path, "r") as epub_zip:
for file_name in sorted(epub_zip.namelist()):
if file_name.lower().endswith(skipped_extensions):
continue
try:
content = epub_zip.read(file_name).decode("utf-8", errors="ignore")
if "<" in content and len(content.strip()) > 100:
cleaned = clean_html(content)
if cleaned:
text_parts.append(cleaned)
except Exception as exc:
logger.debug("Skipping EPUB member %r: %s", file_name, exc)
return "\n\n".join(text_parts)
def _extract_txt(file_path: Path) -> str:
"""Extract from TXT, trying common encodings (latin-1 is the catch-all)."""
for encoding in ("utf-8", "utf-16", "cp1252", "latin-1"):
try:
with open(file_path, "r", encoding=encoding) as f:
return clean_text(f.read())
except UnicodeError:
continue
raise ValueError(f"Could not decode text file: {file_path}")
def _extract_pdf(file_path: Path) -> str:
"""Extract from PDF."""
from pypdf import PdfReader
text = ""
with open(file_path, "rb") as file:
pdf_reader = PdfReader(file)
total_pages = len(pdf_reader.pages)
logger.info("PDF has %d pages", total_pages)
for page_num, page in enumerate(pdf_reader.pages, 1):
try:
page_text = page.extract_text() or ""
if page_text.strip():
text += f"\n\n{page_text}"
if page_num % 10 == 0:
logger.debug("Extracted %d/%d pages", page_num, total_pages)
except Exception as exc:
logger.warning("Failed to extract page %d: %s", page_num, exc)
logger.info("Extracted text from %d pages, %d characters total", total_pages, len(text))
return clean_text(text)
|
