Andrey Kruchinkin
49f91247c7
Программа принимает папку с документами, извлекает из них текст и метаданные и строит каталог: краткое описание, тезисы и ключевые слова для каждого файла плюс поиск по всему собранному. Ядро: - extract: PDF, DJVU, DOC(X), RTF, ODT, CHM, EPUB, FB2, PPT(X), XLS(X), TXT; PDF передаётся MuPDF потоком в память (работают длинные пути) и ограничен по времени — повреждённый файл иначе чинится минутами; - summarize: экстрактивное реферирование по TF-IDF, посчитанному на самой библиотеке, с лёгким стеммером для русского и английского; - db: SQLite с полнотекстовым индексом FTS5, морфологический поиск с BM25; - scanner: инкрементальный многопоточный обход, счётчик попыток защищает от файла, обрывающего разбор; - report: автономный HTML с поиском, Markdown, JSON, CSV. Интерфейс: - окно PySide6 в тёмной теме: вкладки разбора и поиска, фоновый поток, карточка документа, правка своих ключевых слов; - командная строка: build, scan, analyze, report, search, show, tag, stats; - у каждой папки свой каталог, последняя обработанная запоминается. Сборка: scripts/build_exe.py даёт Catalogizer.exe в Windows и Catalogizer.app в macOS, иконки .ico и .icns рисуются кодом. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
350 lines
13 KiB
Python
350 lines
13 KiB
Python
"""Извлечение текста и метаданных из файлов библиотеки."""
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
import re
|
||
import shutil
|
||
import struct
|
||
import subprocess
|
||
import time
|
||
import zipfile
|
||
from pathlib import Path
|
||
from xml.etree import ElementTree as ET
|
||
|
||
from . import config
|
||
from .textutil import clean_text, pretty_name
|
||
|
||
try: # PyMuPDF ставится как pymupdf, исторический импорт — fitz
|
||
import pymupdf as fitz
|
||
except ImportError: # pragma: no cover
|
||
try:
|
||
import fitz
|
||
except ImportError:
|
||
fitz = None
|
||
|
||
if fitz is not None:
|
||
# Повреждённые PDF иначе засыпают консоль сообщениями MuPDF о починке.
|
||
try:
|
||
fitz.TOOLS.mupdf_display_errors(False)
|
||
except AttributeError: # pragma: no cover — другая версия PyMuPDF
|
||
pass
|
||
|
||
DJVUTXT = shutil.which("djvutxt")
|
||
ANTIWORD = shutil.which("antiword")
|
||
|
||
|
||
def long_path(p: str | Path) -> str:
|
||
"""Путь в форме, которую Windows принимает при длине > 260 символов."""
|
||
s = os.fspath(p)
|
||
if os.name == "nt" and not s.startswith("\\\\?\\"):
|
||
s = "\\\\?\\" + os.path.abspath(s)
|
||
return s
|
||
|
||
|
||
def _read_bytes(path: Path, limit: int | None = None) -> bytes:
|
||
with open(long_path(path), "rb") as f:
|
||
return f.read(limit) if limit else f.read()
|
||
|
||
|
||
def _decode(data: bytes) -> str:
|
||
"""Декодирование текста без chardet: пробуем типовые кодировки."""
|
||
for enc in ("utf-8-sig", "utf-8"):
|
||
try:
|
||
return data.decode(enc)
|
||
except UnicodeDecodeError:
|
||
pass
|
||
if data[:2] in (b"\xff\xfe", b"\xfe\xff"):
|
||
try:
|
||
return data.decode("utf-16")
|
||
except UnicodeDecodeError:
|
||
pass
|
||
best, best_score = "", -1.0
|
||
for enc in ("cp1251", "koi8-r", "cp866", "cp1252", "latin-1"):
|
||
try:
|
||
t = data.decode(enc)
|
||
except UnicodeDecodeError:
|
||
continue
|
||
good = sum(ch.isalpha() or ch.isspace() or ch.isdigit() for ch in t)
|
||
score = good / max(len(t), 1)
|
||
if score > best_score:
|
||
best, best_score = t, score
|
||
return best
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# Форматы
|
||
# --------------------------------------------------------------------------
|
||
|
||
def _from_pdf(path: Path, max_pages: int) -> dict:
|
||
"""Текст и метаданные PDF.
|
||
|
||
Файл читается средствами Python и передаётся MuPDF потоком: так работают
|
||
длинные и нестандартные пути, а сам разбор ограничен по времени —
|
||
повреждённый PDF MuPDF пытается чинить страницу за страницей и может
|
||
занять минуты.
|
||
"""
|
||
if fitz is None:
|
||
return {"error": "PyMuPDF не установлен (pip install pymupdf)"}
|
||
if path.stat().st_size > config.MAX_PDF_BYTES:
|
||
return {"error": "PDF больше %d МБ — пропущен"
|
||
% (config.MAX_PDF_BYTES // 1_000_000)}
|
||
started = time.time()
|
||
out: dict = {}
|
||
data = _read_bytes(path)
|
||
with fitz.open(stream=data, filetype="pdf") as doc:
|
||
if doc.needs_pass:
|
||
return {"pages": doc.page_count, "error": "PDF защищён паролем"}
|
||
out["pages"] = doc.page_count
|
||
meta = doc.metadata or {}
|
||
out["title"] = (meta.get("title") or "").strip()
|
||
out["author"] = (meta.get("author") or "").strip()
|
||
# Оглавление — концентрированный источник о содержании книги.
|
||
try:
|
||
toc = doc.get_toc()
|
||
except Exception:
|
||
toc = []
|
||
if toc:
|
||
out["toc"] = [t[1].strip() for t in toc[:120] if t[1].strip()]
|
||
parts = []
|
||
for i in range(min(max_pages, doc.page_count)):
|
||
try:
|
||
parts.append(doc[i].get_text("text"))
|
||
except Exception:
|
||
continue
|
||
if sum(len(p) for p in parts) > config.MAX_TEXT_CHARS * 1.5:
|
||
break
|
||
out["text"] = "\n".join(parts)
|
||
return out
|
||
|
||
|
||
def _djvu_pages(path: Path) -> int | None:
|
||
"""Число страниц из заголовка DJVU (чанк DIRM) без внешних утилит."""
|
||
try:
|
||
head = _read_bytes(path, 4096)
|
||
except OSError:
|
||
return None
|
||
idx = head.find(b"DIRM")
|
||
if idx > 0 and len(head) > idx + 11:
|
||
try:
|
||
n = struct.unpack(">H", head[idx + 9: idx + 11])[0]
|
||
except struct.error:
|
||
return None
|
||
# Каталог DIRM содержит и служебные компоненты — берём только
|
||
# правдоподобные значения.
|
||
return n if 1 <= n <= 5000 else None
|
||
return 1 if b"INFO" in head else None
|
||
|
||
|
||
def _from_djvu(path: Path, max_pages: int) -> dict:
|
||
out: dict = {"pages": _djvu_pages(path)}
|
||
if DJVUTXT:
|
||
try:
|
||
r = subprocess.run(
|
||
[DJVUTXT, "--page=1-%d" % max_pages, str(path)],
|
||
capture_output=True, timeout=120,
|
||
)
|
||
out["text"] = _decode(r.stdout)
|
||
except (subprocess.SubprocessError, OSError) as e:
|
||
out["error"] = "djvutxt: %s" % e
|
||
else:
|
||
out["error"] = "нет djvutxt (DjVuLibre) — текст DJVU не извлечён"
|
||
return out
|
||
|
||
|
||
def _zip_xml_text(path: Path, members: list[str], tags: tuple[str, ...]) -> str:
|
||
parts: list[str] = []
|
||
with zipfile.ZipFile(long_path(path)) as z:
|
||
names = z.namelist()
|
||
wanted = [n for n in names if any(re.fullmatch(m, n) for m in members)]
|
||
for n in sorted(wanted):
|
||
try:
|
||
root = ET.fromstring(z.read(n))
|
||
except (ET.ParseError, KeyError):
|
||
continue
|
||
for el in root.iter():
|
||
tag = el.tag.rsplit("}", 1)[-1]
|
||
if tag in tags and el.text:
|
||
parts.append(el.text)
|
||
if tag in ("p", "br", "tab"):
|
||
parts.append("\n")
|
||
if sum(len(p) for p in parts) > config.MAX_TEXT_CHARS * 1.5:
|
||
break
|
||
return "".join(parts)
|
||
|
||
|
||
def _ooxml_meta(path: Path) -> dict:
|
||
try:
|
||
with zipfile.ZipFile(long_path(path)) as z:
|
||
root = ET.fromstring(z.read("docProps/core.xml"))
|
||
except (KeyError, zipfile.BadZipFile, ET.ParseError, OSError):
|
||
return {}
|
||
vals = {}
|
||
for el in root:
|
||
tag = el.tag.rsplit("}", 1)[-1]
|
||
if el.text:
|
||
vals[tag] = el.text.strip()
|
||
return {"title": vals.get("title", ""), "author": vals.get("creator", "")}
|
||
|
||
|
||
def _from_docx(path: Path) -> dict:
|
||
out = _ooxml_meta(path)
|
||
out["text"] = _zip_xml_text(path, [r"word/document\.xml"], ("t",))
|
||
return out
|
||
|
||
|
||
def _from_pptx(path: Path) -> dict:
|
||
out = _ooxml_meta(path)
|
||
out["text"] = _zip_xml_text(path, [r"ppt/slides/slide\d+\.xml"], ("t",))
|
||
return out
|
||
|
||
|
||
def _from_xlsx(path: Path) -> dict:
|
||
out = _ooxml_meta(path)
|
||
out["text"] = _zip_xml_text(path, [r"xl/sharedStrings\.xml"], ("t",))
|
||
return out
|
||
|
||
|
||
def _from_binary_office(path: Path) -> dict:
|
||
"""Word 97 / Excel 97 / PowerPoint 97: antiword, иначе выборка строк."""
|
||
if ANTIWORD and path.suffix.lower() == ".doc":
|
||
try:
|
||
r = subprocess.run([ANTIWORD, "-m", "UTF-8.txt", str(path)],
|
||
capture_output=True, timeout=90)
|
||
if r.stdout:
|
||
return {"text": _decode(r.stdout)}
|
||
except (subprocess.SubprocessError, OSError):
|
||
pass
|
||
data = _read_bytes(path, 2_000_000)
|
||
text = _decode(data)
|
||
chunks = re.findall(
|
||
r"[A-Za-zА-Яа-яЁё][A-Za-zА-Яа-яЁё0-9 ,.;:()\-«»\"']{25,}", text)
|
||
return {"text": "\n".join(chunks[:2000])}
|
||
|
||
|
||
def _from_rtf(path: Path) -> dict:
|
||
raw = _decode(_read_bytes(path, 4_000_000))
|
||
raw = re.sub(r"\\'([0-9a-fA-F]{2})",
|
||
lambda m: bytes([int(m.group(1), 16)]).decode("cp1251", "ignore"),
|
||
raw)
|
||
raw = re.sub(r"\\u(-?\d+)\s?\??", lambda m: chr(int(m.group(1)) % 65536), raw)
|
||
raw = re.sub(r"\{\\\*.*?\}", " ", raw, flags=re.S)
|
||
raw = re.sub(r"\\[a-zA-Z]+-?\d*\s?", " ", raw)
|
||
return {"text": raw.replace("{", " ").replace("}", " ")}
|
||
|
||
|
||
def _from_fb2(path: Path) -> dict:
|
||
raw = _decode(_read_bytes(path, 8_000_000))
|
||
title = re.search(r"<book-title>(.*?)</book-title>", raw, re.S)
|
||
author = re.findall(r"<(?:first|last)-name>(.*?)</(?:first|last)-name>", raw)
|
||
return {"title": title.group(1).strip() if title else "",
|
||
"author": " ".join(author[:2]),
|
||
"text": re.sub(r"<[^>]+>", " ", raw)}
|
||
|
||
|
||
def _from_epub(path: Path) -> dict:
|
||
parts: list[str] = []
|
||
title = author = ""
|
||
with zipfile.ZipFile(long_path(path)) as z:
|
||
for n in z.namelist():
|
||
if n.endswith(".opf"):
|
||
meta = _decode(z.read(n))
|
||
m = re.search(r"<dc:title[^>]*>(.*?)</dc:title>", meta, re.S)
|
||
a = re.search(r"<dc:creator[^>]*>(.*?)</dc:creator>", meta, re.S)
|
||
title = m.group(1).strip() if m else ""
|
||
author = a.group(1).strip() if a else ""
|
||
break
|
||
for n in sorted(z.namelist()):
|
||
if n.lower().endswith((".xhtml", ".html", ".htm")):
|
||
parts.append(re.sub(r"<[^>]+>", " ", _decode(z.read(n))))
|
||
if sum(len(p) for p in parts) > config.MAX_TEXT_CHARS * 1.5:
|
||
break
|
||
return {"title": title, "author": author, "text": "\n".join(parts)}
|
||
|
||
|
||
def _from_chm(path: Path) -> dict:
|
||
"""Из CHM без chmlib берём читаемые заголовки разделов справки."""
|
||
text = _decode(_read_bytes(path, 1_500_000))
|
||
titles = re.findall(r'<param name="Name" value="([^"]{4,120})"', text)
|
||
if not titles:
|
||
titles = re.findall(r"<title>([^<]{4,120})</title>", text, re.I)
|
||
return {"text": "\n".join(dict.fromkeys(titles))[: config.MAX_TEXT_CHARS],
|
||
"error": "" if titles else "CHM: текст не извлекается без chmlib"}
|
||
|
||
|
||
def _from_plain(path: Path) -> dict:
|
||
raw = _decode(_read_bytes(path, 4_000_000))
|
||
if path.suffix.lower() in (".html", ".htm"):
|
||
raw = re.sub(r"<(script|style)[^>]*>.*?</\1>", " ", raw, flags=re.S | re.I)
|
||
raw = re.sub(r"<[^>]+>", " ", raw)
|
||
return {"text": raw}
|
||
|
||
|
||
_HANDLERS = {
|
||
".pdf": lambda p, mp: _from_pdf(p, mp),
|
||
".djvu": lambda p, mp: _from_djvu(p, mp),
|
||
".djv": lambda p, mp: _from_djvu(p, mp),
|
||
".docx": lambda p, mp: _from_docx(p),
|
||
".doc": lambda p, mp: _from_binary_office(p),
|
||
".rtf": lambda p, mp: _from_rtf(p),
|
||
".odt": lambda p, mp: {"text": _zip_xml_text(p, [r"content\.xml"],
|
||
("p", "h", "span"))},
|
||
".pptx": lambda p, mp: _from_pptx(p),
|
||
".ppt": lambda p, mp: _from_binary_office(p),
|
||
".xlsx": lambda p, mp: _from_xlsx(p),
|
||
".xls": lambda p, mp: _from_binary_office(p),
|
||
".epub": lambda p, mp: _from_epub(p),
|
||
".fb2": lambda p, mp: _from_fb2(p),
|
||
".chm": lambda p, mp: _from_chm(p),
|
||
".txt": lambda p, mp: _from_plain(p),
|
||
".md": lambda p, mp: _from_plain(p),
|
||
}
|
||
|
||
YEAR_RE = re.compile(r"\b(19[5-9]\d|20[0-4]\d)\b")
|
||
|
||
BAD_TITLE_MARKS = ("untitled", "microsoft word", "pdfcreator", "документ",
|
||
"unknown", "print", "no title", "titul")
|
||
|
||
|
||
def guess_year(*sources: str) -> int | None:
|
||
for s in sources:
|
||
if not s:
|
||
continue
|
||
years = [int(y) for y in YEAR_RE.findall(s)]
|
||
if years:
|
||
return max(years)
|
||
return None
|
||
|
||
|
||
def extract(path: Path, max_pages: int = config.MAX_PDF_PAGES) -> dict:
|
||
"""Единая точка: текст и метаданные файла.
|
||
|
||
Никогда не бросает исключение — ошибка возвращается в поле ``error``.
|
||
"""
|
||
ext = path.suffix.lower()
|
||
res: dict = {"title": "", "author": "", "pages": None, "text": "",
|
||
"error": "", "toc": []}
|
||
handler = _HANDLERS.get(ext)
|
||
if handler is None:
|
||
res["error"] = "формат %s не поддерживается" % ext
|
||
return res
|
||
try:
|
||
got = handler(path, max_pages)
|
||
res.update({k: v for k, v in got.items() if v is not None})
|
||
except Exception as e: # noqa: BLE001 — устойчивость важнее диагностики
|
||
res["error"] = ("%s: %s" % (type(e).__name__, e))[:300]
|
||
|
||
res["text"] = clean_text(res.get("text") or "")[: config.MAX_TEXT_CHARS]
|
||
|
||
title = (res.get("title") or "").strip()
|
||
low = title.lower()
|
||
# Метаданные часто содержат мусор от конвертеров и имена исходников вёрстки —
|
||
# в таких случаях берём название из имени файла.
|
||
if (len(title) < 4 or any(b in low for b in BAD_TITLE_MARKS)
|
||
or re.search(r"\.(indb|indd|doc|docx|pdf|qxd|tex|fm|cdr|pmd|vp|rtf|"
|
||
r"ps|sla|odt)$", low)):
|
||
title = pretty_name(path.name)
|
||
res["title"] = title[:300]
|
||
res["author"] = (res.get("author") or "")[:200]
|
||
res["year"] = guess_year(path.name, res["title"], res["text"][:1500])
|
||
return res
|