Каталогизатор документов: разбор папки, тезисы, поиск, GUI

Программа принимает папку с документами, извлекает из них текст и
метаданные и строит каталог: краткое описание, тезисы и ключевые слова
для каждого файла плюс поиск по всему собранному.

Ядро:
- 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>
This commit is contained in:
2026-08-30 19:20:29 +03:00
commit 49f91247c7
30 changed files with 3934 additions and 0 deletions

2
catalogizer/__init__.py Normal file
View File

@@ -0,0 +1,2 @@
"""Каталогизатор технической литературы."""
__version__ = "1.0.0"

75
catalogizer/analyze.py Normal file
View File

@@ -0,0 +1,75 @@
"""Второй проход: тезисы, ключевые слова, описания (этап «analyze»).
IDF считается по всей библиотеке, поэтому проход выполняется после того,
как тексты всех документов уже лежат в базе.
"""
from __future__ import annotations
import json
import time
from collections import Counter
from . import db
from .summarize import Analyzer, describe
from .textutil import stem, tokens
def build_df(con) -> tuple[dict[str, int], int]:
"""Документная частота каждой основы слова по всей библиотеке."""
df: Counter = Counter()
n = 0
for row in con.execute("SELECT terms FROM docs WHERE terms IS NOT NULL"):
try:
terms = json.loads(row["terms"])
except (TypeError, ValueError):
continue
n += 1
df.update(terms.keys())
return dict(df), n
def analyze(con, *, force: bool = False, quiet: bool = False,
progress=None) -> dict:
df, n_docs = build_df(con)
an = Analyzer(df, n_docs)
where = "" if force else " WHERE analyzed = 0"
ids = [r["id"] for r in con.execute("SELECT id FROM docs" + where)]
total = len(ids)
if not quiet:
print("Документов в базе: %d, к анализу: %d" % (n_docs, total), flush=True)
if progress:
progress(0, total, "Анализ: %d документов" % total)
started = time.time()
for i, doc_id in enumerate(ids, 1):
row = con.execute(
"SELECT title, body, toc, terms, lang, kind, pages, scanned, filename"
" FROM docs WHERE id = ?", (doc_id,)).fetchone()
body = row["body"] or ""
try:
terms = Counter(json.loads(row["terms"] or "{}"))
toc = json.loads(row["toc"] or "[]")
except ValueError:
terms, toc = Counter(), []
keywords = an.keywords(terms, body, row["title"] or "", toc)
theses = an.theses(body, terms) if not row["scanned"] else []
if not theses and toc:
# Для сканов и справок без связного текста тезисы берём из оглавления.
theses = [t for t in toc[:8] if len(t) > 8]
description = describe(row["title"] or row["filename"], row["kind"] or "",
row["lang"] or "?", row["pages"], keywords,
theses, not row["scanned"])
stems = " ".join(stem(t) for t in tokens(body[:20_000]))
db.save_summary(con, doc_id, description, theses, keywords, stems)
if i % 100 == 0:
con.commit()
if not quiet:
print(" %d/%d" % (i, total), flush=True)
if progress:
progress(i, total, "Анализ %d/%d" % (i, total))
con.commit()
return {"analyzed": total, "corpus": n_docs,
"seconds": round(time.time() - started, 1)}

58
catalogizer/app.py Normal file
View File

@@ -0,0 +1,58 @@
"""@file app.py
@brief Точка входа графического приложения каталогизатора.
Собирает QApplication и главное окно. С ключом ``--check`` окно строится,
но не показывается: так проверяется собранный exe, где traceback показать
некому.
"""
from __future__ import annotations
import sys
from pathlib import Path
from PySide6.QtCore import Qt
from PySide6.QtWidgets import QApplication
from .ui.main_window import MainWindow
def _set_windows_app_id() -> None:
"""@brief Закрепляет собственную группу приложения на панели задач Windows."""
if sys.platform != "win32":
return
try:
import ctypes
ctypes.windll.shell32.SetCurrentProcessExplicitAppUserModelID(
"SET.Catalogizer")
except (AttributeError, OSError):
pass
def main(argv: list[str] | None = None) -> int:
"""@brief Запускает окно каталогизатора.
@param argv Аргументы без имени программы: путь к папке и/или ``--check``.
@return Код выхода приложения.
"""
argv = list(sys.argv[1:] if argv is None else argv)
check = "--check" in argv
folders = [a for a in argv if not a.startswith("-")]
_set_windows_app_id()
QApplication.setHighDpiScaleFactorRoundingPolicy(
Qt.HighDpiScaleFactorRoundingPolicy.PassThrough)
application = QApplication(sys.argv[:1])
application.setApplicationName("Каталогизатор")
application.setOrganizationName("SET")
window = MainWindow(Path(folders[0]) if folders else None)
if check:
# Самопроверка сборки: окно строится целиком и приложение выходит.
application.processEvents()
titles = [window.tabs.tabText(i) for i in range(window.tabs.count())]
print("Каталогизатор: вкладок", len(titles), "-", ", ".join(titles))
return 0
window.show()
return application.exec()

383
catalogizer/cli.py Normal file
View File

@@ -0,0 +1,383 @@
"""Командный интерфейс каталогизатора."""
from __future__ import annotations
import argparse
import json
import sys
from pathlib import Path
from . import analyze as analyze_mod
from . import config, db, report, scanner, search as search_mod
def _utf8_console() -> None:
for stream in (sys.stdout, sys.stderr):
try:
stream.reconfigure(encoding="utf-8", errors="replace")
except (AttributeError, ValueError):
pass
def _fmt_size(n: int) -> str:
return report._size(n)
def resolve_root(args) -> Path:
"""Папка с документами: позиционный аргумент, --root или прошлый запуск."""
given = (getattr(args, "catalog", None) or getattr(args, "folder", None)
or getattr(args, "root", None))
return Path(given).expanduser() if given else config.current_root()
def open_catalog(args, root: Path | None = None):
"""Подключение к базе того каталога, о котором идёт речь."""
if root is None:
root = resolve_root(args)
db_path, out_dir = config.paths_for(root)
if args.db:
db_path = Path(args.db)
if not getattr(args, "out", None):
args.out = str(out_dir)
return db.connect(db_path), root
# --------------------------------------------------------------------------
# Команды
# --------------------------------------------------------------------------
def cmd_scan(args) -> int:
root = resolve_root(args)
if not root.is_dir():
print("Папка не найдена: %s" % root)
return 2
con, root = open_catalog(args, root)
config.remember(root)
if args.retry_broken:
con.execute("DELETE FROM attempts")
con.commit()
print("Папка: %s" % root)
exts = None
if args.ext:
exts = {"." + e.lower().lstrip(".") for e in args.ext}
stats = scanner.scan(con, root, max_pages=args.max_pages,
jobs=args.jobs, force=args.force, limit=args.limit,
extensions=exts)
print("Разбор завершён: обработано %(processed)d, с ошибками %(errors)d, "
"пропущено %(skipped)d, за %(seconds)s с" % stats)
con.close()
return 0
def cmd_analyze(args) -> int:
con, _ = open_catalog(args)
stats = analyze_mod.analyze(con, force=args.force)
print("Анализ завершён: %(analyzed)d документов за %(seconds)s с" % stats)
con.close()
return 0
def cmd_report(args) -> int:
con, _ = open_catalog(args)
paths = report.build_all(con, args.out)
for kind, p in paths.items():
print("%-5s %s" % (kind, p))
con.close()
return 0
def cmd_build(args) -> int:
rc = cmd_scan(args) or cmd_analyze(args) or cmd_report(args)
return rc
def cmd_search(args) -> int:
con, _ = open_catalog(args)
rows = search_mod.search(con, " ".join(args.query), limit=args.limit,
ext=args.ext_filter, folder=args.section,
lang=args.lang, year_from=args.year_from,
kind=args.kind)
if args.json:
print(json.dumps(rows, ensure_ascii=False, indent=1))
con.close()
return 0
if not rows:
print("Ничего не найдено.")
con.close()
return 1
for i, r in enumerate(rows, 1):
facts = [r["ext"].lstrip("."), r["kind"] or ""]
if r["author"]:
facts.append(r["author"])
if r["year"]:
facts.append(str(r["year"]))
if r["pages"]:
facts.append("%d с." % r["pages"])
print("\n%2d. %s" % (i, r["title"]))
print(" %s" % " · ".join(f for f in facts if f))
if r["description"]:
print(" %s" % r["description"][:300])
if r["snip"]:
print(" … %s" % " ".join(r["snip"].split())[:280])
if r["user_keywords"]:
print(" свои слова: %s" % ", ".join(r["user_keywords"]))
if r["keywords"]:
print(" ключевые слова: %s" % ", ".join(r["keywords"][:8]))
print(" %s" % r["path"])
con.close()
return 0
def cmd_show(args) -> int:
con, _ = open_catalog(args)
if args.target.isdigit():
row = con.execute("SELECT * FROM docs WHERE id = ?",
(int(args.target),)).fetchone()
else:
row = con.execute(
"SELECT * FROM docs WHERE path LIKE ? OR rel_path LIKE ?"
" OR filename LIKE ? LIMIT 1",
("%" + args.target + "%",) * 3).fetchone()
if row is None:
print("Документ не найден.")
return 1
print(row["title"])
print("=" * min(len(row["title"]), 80))
for label, key in (("Автор", "author"), ("Год", "year"), ("Тип", "kind"),
("Язык", "lang"), ("Страниц", "pages"), ("Путь", "path")):
if row[key]:
print("%-9s %s" % (label + ":", row[key]))
print("%-9s %s" % ("Размер:", _fmt_size(row["size"])))
if row["error"]:
print("%-9s %s" % ("Ошибка:", row["error"]))
print("\nОписание:\n %s" % (row["description"] or "—"))
theses = json.loads(row["theses"] or "[]")
if theses:
print("\nТезисы:")
for t in theses:
print(" • %s" % t)
kws = json.loads(row["keywords"] or "[]")
if kws:
print("\nКлючевые слова: %s" % ", ".join(kws))
toc = json.loads(row["toc"] or "[]")
if toc and args.toc:
print("\nОглавление:")
for t in toc[:60]:
print(" - %s" % t)
con.close()
return 0
def _find_doc(con, target: str):
"""Документ по числовому id или по части имени/пути."""
if target.isdigit():
return con.execute("SELECT * FROM docs WHERE id = ?",
(int(target),)).fetchone()
return con.execute(
"SELECT * FROM docs WHERE path LIKE ? OR rel_path LIKE ?"
" OR filename LIKE ? OR title LIKE ? LIMIT 1",
("%" + target + "%",) * 4).fetchone()
def cmd_tag(args) -> int:
"""Свои ключевые слова документа: показать, добавить, убрать, заменить."""
con, _ = open_catalog(args)
if args.all:
words = db.all_user_keywords(con)
print("\n".join(words) if words else "Своих ключевых слов пока нет.")
con.close()
return 0
row = _find_doc(con, args.target)
if row is None:
print("Документ не найден: %s" % args.target)
con.close()
return 1
current = db.get_user_keywords(con, row["path"])
words = [w for chunk in args.words for w in chunk.split(",") if w.strip()]
if args.remove:
drop = {w.strip().lower() for w in words}
new = [w for w in current if w.lower() not in drop]
elif args.set:
new = words
elif words:
new = current + words
else:
new = current # без слов — просто показать
if new != current:
new = db.set_user_keywords(con, row["path"], new)
print("%s\n свои ключевые слова: %s"
% (row["title"], ", ".join(new) if new else "—"))
con.close()
return 0
def cmd_stats(args) -> int:
con, _ = open_catalog(args)
total = con.execute("SELECT COUNT(*) c, SUM(size) s FROM docs").fetchone()
if not total["c"]:
print("База пуста — выполните: python catalog.py build")
return 1
print("Корень: %s" % db.get_meta(con, "root"))
print("Обновлено: %s" % db.get_meta(con, "last_scan"))
print("Документов: %d (%s)" % (total["c"], _fmt_size(total["s"] or 0)))
print("С тезисами: %d" % con.execute(
"SELECT COUNT(*) c FROM docs WHERE analyzed = 1 AND scanned = 0").fetchone()["c"])
print("Сканов без текста: %d" % con.execute(
"SELECT COUNT(*) c FROM docs WHERE scanned = 1").fetchone()["c"])
print("С ошибками: %d" % con.execute(
"SELECT COUNT(*) c FROM docs WHERE error <> ''").fetchone()["c"])
print("\nПо форматам:")
for r in con.execute("SELECT ext, COUNT(*) c, SUM(size) s FROM docs"
" GROUP BY ext ORDER BY c DESC"):
print(" %-6s %5d %s" % (r["ext"], r["c"], _fmt_size(r["s"] or 0)))
print("\nКрупнейшие разделы:")
for r in con.execute("SELECT folder, COUNT(*) c FROM docs GROUP BY folder"
" ORDER BY c DESC LIMIT 15"):
print(" %5d %s" % (r["c"], r["folder"][:70]))
print("\nЧастые темы библиотеки:")
from collections import Counter
cnt: Counter = Counter()
for r in con.execute("SELECT keywords FROM docs WHERE keywords IS NOT NULL"):
cnt.update(json.loads(r["keywords"])[:8])
print(" " + ", ".join("%s (%d)" % (w, n) for w, n in cnt.most_common(25)))
con.close()
return 0
# --------------------------------------------------------------------------
def build_parser() -> argparse.ArgumentParser:
p = argparse.ArgumentParser(
prog="catalog",
description="Каталогизатор технической библиотеки: разбор файлов, "
"краткие описания, тезисы и поиск по ключевым словам.")
p.add_argument("-c", "--catalog", default=None, metavar="ПАПКА",
help="папка каталога для search/report/stats/show "
"(по умолчанию — обработанная последней)")
p.add_argument("--db", default=None,
help="файл базы каталога (по умолчанию свой для каждой папки)")
sub = p.add_subparsers(dest="cmd", required=False)
def add_scan_args(sp):
sp.add_argument("folder", nargs="?", default=None,
help="папка с документами (по умолчанию из config.py)")
sp.add_argument("--root", default=None,
help="то же самое, что и позиционная папка")
sp.add_argument("--max-pages", type=int, default=config.MAX_PDF_PAGES,
help="сколько страниц читать из документа")
sp.add_argument("--jobs", type=int, default=4,
help="потоков разбора (HDD — 2-4, SSD — 8-16)")
sp.add_argument("--limit", type=int, default=None,
help="разобрать не более N новых файлов (для пробы)")
sp.add_argument("--ext", nargs="*", default=None,
help="ограничить форматы, например: pdf djvu")
sp.add_argument("--force", action="store_true",
help="перечитать все файлы, а не только изменившиеся")
sp.add_argument("--retry-broken", action="store_true",
help="снова попробовать файлы, помеченные как битые")
sp = sub.add_parser("scan", help="обойти библиотеку и извлечь тексты")
add_scan_args(sp)
sp.set_defaults(func=cmd_scan)
sp = sub.add_parser("analyze", help="построить описания, тезисы, ключевые слова")
sp.add_argument("--force", action="store_true", help="пересчитать всё заново")
sp.set_defaults(func=cmd_analyze)
sp = sub.add_parser("report", help="выгрузить каталог (HTML, MD, JSON, CSV)")
sp.add_argument("folder", nargs="?", default=None,
help="папка каталога (по умолчанию — обработанная последней)")
sp.add_argument("--out", default=None,
help="папка выгрузки (по умолчанию out/ этого каталога)")
sp.set_defaults(func=cmd_report)
sp = sub.add_parser("build", help="полный цикл: scan + analyze + report")
add_scan_args(sp)
sp.add_argument("--out", default=None,
help="папка выгрузки (по умолчанию out/ этого каталога)")
sp.set_defaults(func=cmd_build)
sp = sub.add_parser("search", help="поиск по каталогу")
sp.add_argument("query", nargs="+", help="слова запроса; \"фраза\" — точная")
sp.add_argument("-n", "--limit", type=int, default=20)
sp.add_argument("--ext", dest="ext_filter", default=None, help="формат: pdf")
sp.add_argument("--section", "--folder", dest="section", default=None,
help="искать только в разделе (подстрока пути)")
sp.add_argument("--lang", default=None, choices=["ru", "en"])
sp.add_argument("--kind", default=None, help="тип: книга, даташит, учебник…")
sp.add_argument("--year-from", type=int, default=None)
sp.add_argument("--json", action="store_true", help="вывод в JSON")
sp.set_defaults(func=cmd_search)
sp = sub.add_parser("show", help="карточка документа по id или имени")
sp.add_argument("target")
sp.add_argument("--toc", action="store_true", help="показать оглавление")
sp.set_defaults(func=cmd_show)
sp = sub.add_parser("tag", help="свои ключевые слова документа")
sp.add_argument("target", nargs="?", default="",
help="id документа или часть имени/названия")
sp.add_argument("words", nargs="*", default=[],
help="слова через пробел или запятую")
sp.add_argument("--remove", action="store_true", help="убрать эти слова")
sp.add_argument("--set", action="store_true",
help="заменить список целиком (без слов — очистить)")
sp.add_argument("--all", action="store_true",
help="показать все свои ключевые слова каталога")
sp.set_defaults(func=cmd_tag)
sp = sub.add_parser("gui", help="графическое окно (по умолчанию)")
sp.add_argument("folder", nargs="?", default=None,
help="папка, открытая при запуске")
sp.set_defaults(func=cmd_gui)
sp = sub.add_parser("stats", help="сводка по каталогу")
sp.add_argument("folder", nargs="?", default=None,
help="папка каталога (по умолчанию — обработанная последней)")
sp.set_defaults(func=cmd_stats)
return p
def cmd_gui(args) -> int:
"""Графическое окно каталогизатора."""
try:
from .app import main as gui_main
except ImportError:
print("Не установлен PySide6: python -m pip install PySide6")
return 2
folder = getattr(args, "folder", None)
return gui_main([folder] if folder else [])
def interactive() -> list[str]:
"""Запуск без аргументов: открываем окно, а без PySide6 — спрашиваем папку."""
try:
import PySide6 # noqa: F401 — проверка наличия графической библиотеки
except ImportError:
pass
else:
return ["gui"]
print("Каталогизатор документов.")
print("Укажите папку с документами (Enter — %s):" % config.current_root())
try:
folder = input("> ").strip().strip('"')
except (EOFError, KeyboardInterrupt):
return ["--help"]
return ["build", folder] if folder else ["build"]
def main(argv: list[str] | None = None) -> int:
_utf8_console()
argv = list(sys.argv[1:] if argv is None else argv)
if not argv:
argv = interactive()
args = build_parser().parse_args(argv)
if not getattr(args, "func", None):
build_parser().print_help()
return 2
return args.func(args)

119
catalogizer/config.py Normal file
View File

@@ -0,0 +1,119 @@
"""Настройки каталогизатора."""
from __future__ import annotations
import os
import sys
from pathlib import Path
def _default_root() -> Path:
"""Папка, предлагаемая при первом запуске.
Переопределяется переменной окружения CATALOG_ROOT или аргументом
команды. Жёсткий путь к библиотеке действует только там, где он есть,
иначе берётся домашняя папка «Документы» — так программа осмысленно
стартует и на macOS, и на Linux.
"""
given = os.environ.get("CATALOG_ROOT")
if given:
return Path(given).expanduser()
library = Path(r"E:\Yandex.Disk\литература")
if library.is_dir():
return library
for name in ("Documents", "Документы"):
candidate = Path.home() / name
if candidate.is_dir():
return candidate
return Path.home()
DEFAULT_ROOT = _default_root()
def _app_dir() -> Path:
"""Папка, где программа держит базы каталогов и отчёты.
Из исходников — корень проекта. В собранной программе — рядом с самим
файлом, кроме macOS: там ``.app`` считается неизменяемым (и может
запускаться из карантина), поэтому данные уходят в
``~/Library/Application Support``.
"""
if not getattr(sys, "frozen", False):
return Path(__file__).resolve().parent.parent
if sys.platform == "darwin":
target = Path.home() / "Library" / "Application Support" / "Catalogizer"
target.mkdir(parents=True, exist_ok=True)
return target
return Path(sys.executable).resolve().parent
APP_DIR = _app_dir()
DB_PATH = APP_DIR / "catalog.db" # база «главной» библиотеки
OUT_DIR = APP_DIR / "out" # отчёты по ней
CATALOGS_DIR = APP_DIR / "catalogs" # базы остальных обработанных папок
CURRENT_FILE = APP_DIR / ".current" # последняя обработанная папка
# Расширения, которые считаем «документами» библиотеки.
DOC_EXTENSIONS = {
".pdf", ".djvu", ".djv",
".doc", ".docx", ".rtf", ".odt",
".chm", ".epub", ".fb2",
".ppt", ".pptx", ".xls", ".xlsx",
".txt", ".md",
}
# Каталоги, которые не имеет смысла разбирать (SDK, репозитории, сборки).
SKIP_DIR_NAMES = {
".git", ".svn", "node_modules", "__pycache__", ".idea", ".vscode",
"Debug", "Release", "obj", "build", "cmake-build-debug",
}
# Мусорные txt/md из состава SDK и репозиториев — не тащим в каталог.
NOISE_FILENAMES = {
"readme.txt", "readme.md", "license.txt", "license.md", "licence.txt",
"changelog.md", "changelog.txt", "contributing.md", "code_of_conduct.md",
"manifest.txt", "makefile.txt", "notice.txt", "authors.md",
}
# Ограничения разбора.
MAX_PDF_PAGES = 20 # сколько страниц читаем из начала документа
MAX_TEXT_CHARS = 30_000 # сколько символов текста храним в базе
MIN_TEXT_CHARS = 200 # меньше — считаем, что текстового слоя нет (скан)
MIN_FILE_SIZE = 2_048 # файлы мельче игнорируем
MAX_PDF_BYTES = 400_000_000 # PDF крупнее не разбираем
PDF_TIME_BUDGET = 25 # секунд на один PDF: повреждённые чинятся вечно
# Сколько тезисов и ключевых слов делать на документ.
N_THESES = 5
N_KEYWORDS = 12
def _slug(root: Path) -> str:
"""Короткое имя каталога, однозначно привязанное к пути папки."""
import hashlib
import re
name = re.sub(r"[^\w\-]+", "_", root.name, flags=re.U).strip("_") or "catalog"
tail = hashlib.sha1(str(root).lower().encode("utf-8")).hexdigest()[:6]
return "%s-%s" % (name[:40], tail)
def paths_for(root: Path) -> tuple[Path, Path]:
"""База и папка отчётов для указанной папки с документами."""
root = Path(root).resolve()
if root == Path(DEFAULT_ROOT).resolve():
return DB_PATH, OUT_DIR
slug = _slug(root)
return CATALOGS_DIR / (slug + ".db"), OUT_DIR / slug
def remember(root: Path) -> None:
"""Запоминает последнюю обработанную папку — её и берут search/report."""
CURRENT_FILE.write_text(str(Path(root).resolve()), encoding="utf-8")
def current_root() -> Path:
"""Папка, с которой работали в последний раз."""
try:
value = CURRENT_FILE.read_text(encoding="utf-8").strip()
except OSError:
return Path(DEFAULT_ROOT)
return Path(value) if value else Path(DEFAULT_ROOT)

234
catalogizer/db.py Normal file
View File

@@ -0,0 +1,234 @@
"""Хранилище каталога: SQLite + полнотекстовый индекс FTS5."""
from __future__ import annotations
import json
import sqlite3
from pathlib import Path
from . import config
SCHEMA = """
PRAGMA journal_mode = WAL;
CREATE TABLE IF NOT EXISTS docs (
id INTEGER PRIMARY KEY,
path TEXT UNIQUE NOT NULL, -- абсолютный путь
rel_path TEXT NOT NULL, -- путь от корня библиотеки
folder TEXT NOT NULL, -- раздел (папка) библиотеки
filename TEXT NOT NULL,
ext TEXT NOT NULL,
size INTEGER NOT NULL,
mtime REAL NOT NULL,
title TEXT,
author TEXT,
year INTEGER,
pages INTEGER,
lang TEXT,
kind TEXT,
text_chars INTEGER DEFAULT 0,
scanned INTEGER DEFAULT 0, -- 1 = нет текстового слоя
error TEXT,
body TEXT, -- извлечённый текст (обрезанный)
toc TEXT, -- оглавление, JSON-список
terms TEXT, -- частоты основ, JSON
description TEXT,
theses TEXT, -- JSON-список тезисов
keywords TEXT, -- JSON-список ключевых слов
stems TEXT, -- основы слов текста для поиска
indexed_at TEXT,
analyzed INTEGER DEFAULT 0
);
CREATE INDEX IF NOT EXISTS idx_docs_folder ON docs(folder);
CREATE INDEX IF NOT EXISTS idx_docs_ext ON docs(ext);
CREATE INDEX IF NOT EXISTS idx_docs_analyzed ON docs(analyzed);
CREATE VIRTUAL TABLE IF NOT EXISTS docs_fts USING fts5(
title, keywords, description, body, stems, path,
tokenize = 'unicode61 remove_diacritics 2'
);
CREATE TABLE IF NOT EXISTS meta (key TEXT PRIMARY KEY, value TEXT);
-- Ключевые слова, добавленные вручную. Хранятся по пути файла, а не по id:
-- так они переживают перестроение каталога и повторный разбор папки.
CREATE TABLE IF NOT EXISTS user_keywords (
path TEXT PRIMARY KEY,
keywords TEXT NOT NULL, -- JSON-список
changed TEXT
);
-- Счётчик попыток разбора. Нужен, чтобы файл, на котором парсер аварийно
-- завершает весь процесс, не блокировал обход при следующем запуске.
CREATE TABLE IF NOT EXISTS attempts (path TEXT PRIMARY KEY, n INTEGER NOT NULL);
"""
#: Колонки, добавленные после первых версий базы. Старые каталоги
#: дополняются на месте, чтобы их не пришлось строить заново.
MIGRATIONS = (("docs", "stems", "TEXT"),)
def connect(db_path: Path | None = None,
create: bool = True) -> sqlite3.Connection:
"""Подключение к базе каталога.
@param create Создавать базу, если её ещё нет. Окно читает каталоги
с ``create=False``: просмотр не должен плодить пустые базы рядом
с программой.
"""
path = Path(db_path or config.DB_PATH)
if not create and not path.exists():
raise FileNotFoundError("каталог ещё не построен: %s" % path)
path.parent.mkdir(parents=True, exist_ok=True)
con = sqlite3.connect(path, timeout=60)
con.row_factory = sqlite3.Row
con.executescript(SCHEMA)
for table, column, decl in MIGRATIONS:
have = {r["name"] for r in con.execute("PRAGMA table_info(%s)" % table)}
if column not in have:
con.execute("ALTER TABLE %s ADD COLUMN %s %s" % (table, column, decl))
con.commit()
return con
def get_state(con: sqlite3.Connection) -> dict[str, tuple[float, int]]:
"""Соответствие путь -> (mtime, size) для инкрементального обхода."""
return {r["path"]: (r["mtime"], r["size"])
for r in con.execute("SELECT path, mtime, size FROM docs")}
def upsert_doc(con: sqlite3.Connection, rec: dict) -> None:
cols = ("path", "rel_path", "folder", "filename", "ext", "size", "mtime",
"title", "author", "year", "pages", "lang", "kind", "text_chars",
"scanned", "error", "body", "toc", "terms", "indexed_at")
values = [rec.get(c) for c in cols]
con.execute(
"INSERT INTO docs (%s, analyzed) VALUES (%s, 0) "
"ON CONFLICT(path) DO UPDATE SET %s, analyzed = 0"
% (", ".join(cols),
", ".join("?" * len(cols)),
", ".join("%s = excluded.%s" % (c, c) for c in cols[1:])),
values,
)
def save_summary(con: sqlite3.Connection, doc_id: int, description: str,
theses: list[str], keywords: list[str], stems: str) -> None:
con.execute(
"UPDATE docs SET description = ?, theses = ?, keywords = ?, stems = ?, "
"analyzed = 1 WHERE id = ?",
(description, json.dumps(theses, ensure_ascii=False),
json.dumps(keywords, ensure_ascii=False), stems, doc_id))
reindex_doc(con, doc_id)
def reindex_doc(con: sqlite3.Connection, doc_id: int) -> None:
"""Перестраивает строку полнотекстового индекса для одного документа.
Вызывается и после анализа, и после правки ключевых слов вручную —
иначе добавленное слово не находилось бы поиском.
"""
row = con.execute(
"SELECT d.title, d.body, d.path, d.description, d.keywords, d.stems,"
" u.keywords AS mine"
" FROM docs d LEFT JOIN user_keywords u ON u.path = d.path"
" WHERE d.id = ?", (doc_id,)).fetchone()
if row is None:
return
words = _as_list(row["keywords"]) + _as_list(row["mine"])
con.execute("DELETE FROM docs_fts WHERE rowid = ?", (doc_id,))
con.execute(
"INSERT INTO docs_fts (rowid, title, keywords, description, body, stems,"
" path) VALUES (?, ?, ?, ?, ?, ?, ?)",
(doc_id, row["title"] or "", " ".join(words), row["description"] or "",
(row["body"] or "")[:20_000], row["stems"] or "", row["path"]))
def _as_list(value: str | None) -> list[str]:
try:
return list(json.loads(value or "[]"))
except (TypeError, ValueError):
return []
def get_user_keywords(con: sqlite3.Connection, path: str) -> list[str]:
"""Ключевые слова, добавленные к документу вручную."""
row = con.execute("SELECT keywords FROM user_keywords WHERE path = ?",
(path,)).fetchone()
return _as_list(row["keywords"]) if row else []
def set_user_keywords(con: sqlite3.Connection, path: str,
keywords: list[str]) -> list[str]:
"""Заменяет свои ключевые слова документа и обновляет поисковый индекс.
@param keywords Новый список; пустой список удаляет запись.
@return Сохранённый список без повторов и пустых значений.
"""
from datetime import datetime
clean: list[str] = []
for word in keywords:
word = " ".join(str(word).split())
if word and word.lower() not in {w.lower() for w in clean}:
clean.append(word)
if clean:
con.execute(
"INSERT INTO user_keywords (path, keywords, changed)"
" VALUES (?, ?, ?) ON CONFLICT(path) DO UPDATE SET"
" keywords = excluded.keywords, changed = excluded.changed",
(path, json.dumps(clean, ensure_ascii=False),
datetime.now().isoformat(timespec="seconds")))
else:
con.execute("DELETE FROM user_keywords WHERE path = ?", (path,))
row = con.execute("SELECT id FROM docs WHERE path = ?", (path,)).fetchone()
if row:
reindex_doc(con, row["id"])
con.commit()
return clean
def all_user_keywords(con: sqlite3.Connection) -> list[str]:
"""Все свои ключевые слова каталога — для подсказок при вводе."""
seen: dict[str, None] = {}
for row in con.execute("SELECT keywords FROM user_keywords"):
for word in _as_list(row["keywords"]):
seen.setdefault(word, None)
return sorted(seen, key=str.lower)
def purge_missing(con: sqlite3.Connection, alive: set[str]) -> int:
"""Удаляет из базы записи о файлах, которых больше нет на диске."""
gone = [r["id"] for r in con.execute("SELECT id, path FROM docs")
if r["path"] not in alive]
for i in gone:
con.execute("DELETE FROM docs WHERE id = ?", (i,))
con.execute("DELETE FROM docs_fts WHERE rowid = ?", (i,))
return len(gone)
def set_meta(con: sqlite3.Connection, key: str, value: str) -> None:
con.execute("INSERT INTO meta (key, value) VALUES (?, ?) "
"ON CONFLICT(key) DO UPDATE SET value = excluded.value",
(key, value))
def get_meta(con: sqlite3.Connection, key: str, default: str = "") -> str:
row = con.execute("SELECT value FROM meta WHERE key = ?", (key,)).fetchone()
return row["value"] if row else default
def bump_attempt(con: sqlite3.Connection, path: str) -> None:
con.execute("INSERT INTO attempts (path, n) VALUES (?, 1) "
"ON CONFLICT(path) DO UPDATE SET n = n + 1", (path,))
def clear_attempt(con: sqlite3.Connection, path: str) -> None:
con.execute("DELETE FROM attempts WHERE path = ?", (path,))
def failed_paths(con: sqlite3.Connection, max_tries: int = 3) -> set[str]:
"""Файлы, разбор которых уже несколько раз обрывал процесс."""
return {r["path"] for r in
con.execute("SELECT path FROM attempts WHERE n >= ?", (max_tries,))}

349
catalogizer/extract.py Normal file
View File

@@ -0,0 +1,349 @@
"""Извлечение текста и метаданных из файлов библиотеки."""
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

301
catalogizer/report.py Normal file
View File

@@ -0,0 +1,301 @@
"""Выгрузка каталога: Markdown, HTML с поиском, JSON, CSV."""
from __future__ import annotations
import csv
import html
import json
from datetime import datetime
from pathlib import Path
from . import config, db
def fetch_all(con) -> list[dict]:
rows = con.execute(
"SELECT d.id, d.title, d.author, d.year, d.pages, d.ext, d.size, d.lang,"
" d.kind, d.folder, d.rel_path, d.path, d.description, d.theses,"
" d.keywords, d.scanned, d.error, u.keywords AS user_keywords"
" FROM docs d LEFT JOIN user_keywords u ON u.path = d.path"
" ORDER BY d.folder, d.title").fetchall()
out = []
for r in rows:
d = dict(r)
d["theses"] = json.loads(d["theses"] or "[]")
d["keywords"] = json.loads(d["keywords"] or "[]")
d["user_keywords"] = json.loads(d["user_keywords"] or "[]")
out.append(d)
return out
def _size(n: int) -> str:
for unit in ("Б", "КБ", "МБ", "ГБ"):
if n < 1024:
return "%.0f %s" % (n, unit)
n /= 1024.0
return "%.1f ТБ" % n
# --------------------------------------------------------------------------
# Markdown
# --------------------------------------------------------------------------
def to_markdown(docs: list[dict], root: str, path: Path) -> Path:
by_folder: dict[str, list[dict]] = {}
for d in docs:
by_folder.setdefault(d["folder"], []).append(d)
lines = [
"# Каталог библиотеки",
"",
"Источник: `%s` " % root,
"Документов: **%d**, разделов: **%d** " % (len(docs), len(by_folder)),
"Составлен: %s" % datetime.now().strftime("%d.%m.%Y %H:%M"),
"",
"## Разделы",
"",
]
for folder in sorted(by_folder):
anchor = folder.lower().replace(" ", "-").replace("\\", "-").replace(".", "")
lines.append("- [%s](#%s) — %d" % (folder, anchor, len(by_folder[folder])))
lines.append("")
for folder in sorted(by_folder):
lines += ["", "## %s" % folder, ""]
for d in sorted(by_folder[folder], key=lambda x: (x["title"] or "").lower()):
head = "### %s" % (d["title"] or d["rel_path"])
lines.append(head)
facts = [d["ext"].lstrip("."), d["kind"] or ""]
if d["author"]:
facts.append(d["author"])
if d["year"]:
facts.append(str(d["year"]))
if d["pages"]:
facts.append("%d с." % d["pages"])
facts.append(_size(d["size"]))
if d["scanned"]:
facts.append("скан")
lines.append("*%s*" % " · ".join(f for f in facts if f))
lines.append("")
lines.append(d["description"] or "")
if d["theses"]:
lines.append("")
lines.append("**Тезисы:**")
for t in d["theses"]:
lines.append("- %s" % t)
if d["keywords"]:
lines.append("")
lines.append("**Ключевые слова:** %s" % ", ".join(d["keywords"]))
if d["user_keywords"]:
lines.append("")
lines.append("**Свои ключевые слова:** %s"
% ", ".join(d["user_keywords"]))
lines.append("")
lines.append("`%s`" % d["rel_path"])
lines.append("")
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text("\n".join(lines), encoding="utf-8")
return path
# --------------------------------------------------------------------------
# HTML с поиском
# --------------------------------------------------------------------------
HTML_TEMPLATE = """<!doctype html>
<html lang="ru"><head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>Каталог библиотеки</title>
<style>
:root{--bg:#fbfbfa;--fg:#1c1c1a;--mut:#6b6b66;--line:#e2e1dc;--card:#fff;--acc:#8a4b1f;--hl:#ffeaa7}
@media(prefers-color-scheme:dark){:root{--bg:#16171a;--fg:#e8e8e4;--mut:#9a9a94;--line:#2c2e33;--card:#1d1f23;--acc:#e0a06a;--hl:#5a4a12}}
*{box-sizing:border-box}
body{margin:0;background:var(--bg);color:var(--fg);font:15px/1.5 -apple-system,Segoe UI,Roboto,sans-serif}
header{position:sticky;top:0;background:var(--bg);border-bottom:1px solid var(--line);padding:14px 20px;z-index:5}
h1{margin:0 0 10px;font-size:19px;font-weight:650}
.bar{display:flex;gap:8px;flex-wrap:wrap;align-items:center}
input,select{padding:9px 12px;border:1px solid var(--line);border-radius:8px;background:var(--card);color:var(--fg);font-size:15px}
#q{flex:1;min-width:240px}
.stat{color:var(--mut);font-size:13px;margin-top:8px}
main{padding:16px 20px 60px;max-width:1080px;margin:0 auto}
.doc{background:var(--card);border:1px solid var(--line);border-radius:10px;padding:14px 16px;margin-bottom:12px}
.doc h2{margin:0 0 4px;font-size:16px;font-weight:620;line-height:1.35}
.meta{color:var(--mut);font-size:12.5px;margin-bottom:8px}
.desc{margin:0 0 8px}
ul.th{margin:6px 0 8px;padding-left:20px}
ul.th li{margin:3px 0;color:var(--fg)}
.kw{display:flex;flex-wrap:wrap;gap:5px;margin-top:8px}
.kw span{background:var(--bg);border:1px solid var(--line);border-radius:20px;padding:2px 10px;font-size:12px;color:var(--mut);cursor:pointer}
.kw span:hover{color:var(--acc);border-color:var(--acc)}
.kw span.mine{color:var(--acc);border-color:var(--acc);font-weight:600}
.path{font:12px ui-monospace,Consolas,monospace;color:var(--mut);word-break:break-all;margin-top:8px}
mark{background:var(--hl);color:inherit;border-radius:2px}
.folder{margin:22px 0 10px;font-size:13px;text-transform:uppercase;letter-spacing:.06em;color:var(--acc);font-weight:650}
.empty{color:var(--mut);padding:40px 0;text-align:center}
button.more{background:none;border:1px solid var(--line);border-radius:8px;padding:8px 14px;color:var(--mut);cursor:pointer}
</style></head><body>
<header>
<h1>Каталог библиотеки</h1>
<div class="bar">
<input id="q" placeholder="Поиск по названию, тезисам, ключевым словам…" autofocus>
<select id="ext"><option value="">все форматы</option>__EXTS__</select>
<select id="fold"><option value="">все разделы</option>__FOLDERS__</select>
<select id="lang"><option value="">язык</option><option value="ru">рус</option><option value="en">англ</option></select>
</div>
<div class="stat" id="stat"></div>
</header>
<main id="list"></main>
<script>
const DOCS = __DATA__;
const ROOT = __ROOT__;
// Ссылка на файл: путь приходит из каталога уже абсолютным, поэтому здесь
// достаточно привести разделители к виду file:// (работает и в Windows,
// и в macOS, и в Linux).
function fileUrl(p){
let s = String(p).replace(/\\\\/g,'/');
if(!s.startsWith('/')) s = '/' + s; // Windows: /E:/папка/файл.pdf
return 'file://' + encodeURI(s);
}
DOCS.forEach(d => { d.mk = d.mk || [];
d._s = (d.t+" "+d.a+" "+d.d+" "+d.k.join(" ")+" "+d.mk.join(" ")+" "
+d.th.join(" ")+" "+d.p).toLowerCase(); });
const q=document.getElementById('q'), ext=document.getElementById('ext'),
fold=document.getElementById('fold'), lang=document.getElementById('lang'),
list=document.getElementById('list'), stat=document.getElementById('stat');
let shown=60;
function esc(s){return (s||'').replace(/[&<>"]/g,c=>({'&':'&amp;','<':'&lt;','>':'&gt;','"':'&quot;'}[c]));}
function hl(s,terms){s=esc(s);for(const t of terms){if(t.length<2)continue;
s=s.replace(new RegExp('('+t.replace(/[.*+?^${}()|[\\]\\\\]/g,'\\\\$&')+')','gi'),'<mark>$1</mark>');}return s;}
function terms(){return q.value.toLowerCase().split(/[\\s,;]+/).filter(w=>w.length>1);}
function score(d,ts){
let sc=0;
for(const t of ts){
if(!d._s.includes(t)) return -1; // все слова обязательны
if(d.t.toLowerCase().includes(t)) sc+=10;
if(d.mk.some(k=>k.toLowerCase().includes(t))) sc+=12; // свои слова важнее
if(d.k.some(k=>k.includes(t))) sc+=6;
if(d.d.toLowerCase().includes(t)) sc+=3;
sc+=1;
}
return sc;
}
function render(){
const ts=terms(), e=ext.value, f=fold.value, l=lang.value;
let res=[];
for(const d of DOCS){
if(e && d.e!==e) continue;
if(f && d.fold!==f) continue;
if(l && d.l!==l) continue;
const sc = ts.length? score(d,ts) : 0;
if(sc<0) continue;
res.push([sc,d]);
}
if(ts.length) res.sort((a,b)=>b[0]-a[0]);
stat.textContent = 'Найдено: '+res.length+' из '+DOCS.length;
const part=res.slice(0,shown);
let html='', lastF=null;
for(const [,d] of part){
if(!ts.length && d.fold!==lastF){ html+='<div class="folder">'+esc(d.fold)+'</div>'; lastF=d.fold; }
html+='<article class="doc"><h2>'+hl(d.t,ts)+'</h2><div class="meta">'+
esc([d.e.slice(1),d.kind,d.a,d.y||'',d.pg?d.pg+' с.':'',d.sz,d.sc?'скан':''].filter(Boolean).join(' · '))+
'</div><p class="desc">'+hl(d.d,ts)+'</p>';
if(d.th.length) html+='<ul class="th">'+d.th.map(t=>'<li>'+hl(t,ts)+'</li>').join('')+'</ul>';
if(d.mk.length || d.k.length){
html+='<div class="kw">'
+ d.mk.map(k=>'<span class="mine" onclick="q.value=this.textContent;render()">'+esc(k)+'</span>').join('')
+ d.k.map(k=>'<span onclick="q.value=this.textContent;render()">'+esc(k)+'</span>').join('')
+ '</div>';
}
html+='<div class="path"><a href="'+fileUrl(d.f)+'">'+esc(d.p)+'</a></div></article>';
}
if(res.length>shown) html+='<button class="more" onclick="shown+=100;render()">Показать ещё</button>';
if(!res.length) html='<div class="empty">Ничего не найдено</div>';
list.innerHTML=html;
}
let timer; q.oninput=()=>{clearTimeout(timer);shown=60;timer=setTimeout(render,120);};
ext.onchange=fold.onchange=lang.onchange=()=>{shown=60;render();};
render();
</script></body></html>
"""
def to_html(docs: list[dict], root: str, path: Path) -> Path:
data = [{
"t": d["title"] or d["rel_path"],
"a": d["author"] or "",
"y": d["year"] or "",
"pg": d["pages"] or 0,
"e": d["ext"],
"l": d["lang"] or "",
"kind": d["kind"] or "",
"f": d["path"],
"p": d["rel_path"],
"fold": d["folder"],
"d": d["description"] or "",
"th": d["theses"],
"k": d["keywords"],
"mk": d["user_keywords"],
"sz": _size(d["size"]),
"sc": 1 if d["scanned"] else 0,
} for d in docs]
exts = sorted({d["ext"] for d in docs})
folders = sorted({d["folder"] for d in docs})
# "</" в данных закрыло бы <script> раньше времени.
payload = json.dumps(data, ensure_ascii=False).replace("</", "<\\/")
out = (HTML_TEMPLATE
.replace("__DATA__", payload)
.replace("__ROOT__", json.dumps(root))
.replace("__EXTS__", "".join(
'<option value="%s">%s</option>' % (e, html.escape(e.lstrip(".")))
for e in exts))
.replace("__FOLDERS__", "".join(
'<option value="%s">%s</option>' % (html.escape(f), html.escape(f[:70]))
for f in folders)))
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(out, encoding="utf-8")
return path
# --------------------------------------------------------------------------
# JSON / CSV
# --------------------------------------------------------------------------
def to_json(docs: list[dict], path: Path) -> Path:
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(json.dumps(docs, ensure_ascii=False, indent=1), encoding="utf-8")
return path
def to_csv(docs: list[dict], path: Path) -> Path:
path.parent.mkdir(parents=True, exist_ok=True)
cols = ("title", "author", "year", "pages", "kind", "lang", "ext", "size",
"folder", "rel_path", "description")
with open(path, "w", encoding="utf-8-sig", newline="") as f:
w = csv.writer(f, delimiter=";")
w.writerow(["Название", "Автор", "Год", "Страниц", "Тип", "Язык",
"Формат", "Размер", "Раздел", "Путь", "Описание",
"Ключевые слова", "Свои ключевые слова"])
for d in docs:
w.writerow([d.get(c, "") for c in cols]
+ [", ".join(d["keywords"]),
", ".join(d["user_keywords"])])
return path
def build_all(con, out_dir: Path | None = None) -> dict[str, Path]:
out_dir = Path(out_dir or config.OUT_DIR)
docs = fetch_all(con)
root = db.get_meta(con, "root", str(config.DEFAULT_ROOT))
return {
"html": to_html(docs, root, out_dir / "catalog.html"),
"md": to_markdown(docs, root, out_dir / "catalog.md"),
"json": to_json(docs, out_dir / "catalog.json"),
"csv": to_csv(docs, out_dir / "catalog.csv"),
}

175
catalogizer/scanner.py Normal file
View File

@@ -0,0 +1,175 @@
"""Обход библиотеки и наполнение базы (этап «scan»)."""
from __future__ import annotations
import json
import os
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import datetime
from pathlib import Path
from . import config, db
from .extract import extract, long_path
from .summarize import analyze_language, doc_terms, guess_kind
from .textutil import pretty_name
MAX_TRIES = 3 # после стольких обрывов файл помечается как непригодный
def iter_files(root: Path, extensions: set[str] | None = None):
"""Все файлы-документы библиотеки (с фильтрацией мусора и SDK-папок)."""
exts = extensions or config.DOC_EXTENSIONS
for dirpath, dirnames, filenames in os.walk(long_path(root)):
dirnames[:] = [d for d in dirnames
if d not in config.SKIP_DIR_NAMES and not d.startswith(".")]
for name in filenames:
ext = os.path.splitext(name)[1].lower()
if ext not in exts:
continue
if name.lower() in config.NOISE_FILENAMES:
continue
full = os.path.join(dirpath, name)
try:
st = os.stat(full)
except OSError:
continue
if st.st_size < config.MIN_FILE_SIZE:
continue
clean = full[4:] if full.startswith("\\\\?\\") else full
yield Path(clean), st
def _base_record(path: Path, st, root: Path) -> dict:
rel = os.path.relpath(str(path), str(root))
return {
"path": str(path),
"rel_path": rel,
"folder": os.path.dirname(rel) or ".",
"filename": path.name,
"ext": path.suffix.lower(),
"size": st.st_size,
"mtime": st.st_mtime,
"indexed_at": datetime.now().isoformat(timespec="seconds"),
}
def _build_record(path: Path, st, root: Path, max_pages: int) -> dict:
res = extract(path, max_pages)
text = res["text"]
terms = doc_terms(text, res["title"], res["toc"])
rec = _base_record(path, st, root)
rec.update({
"title": res["title"],
"author": res["author"],
"year": res["year"],
"pages": res["pages"],
"lang": analyze_language(text),
"kind": guess_kind(res["title"], text, res["pages"]),
"text_chars": len(text),
"scanned": int(len(text) < config.MIN_TEXT_CHARS),
"error": res["error"],
"body": text,
"toc": json.dumps(res["toc"][:120], ensure_ascii=False),
"terms": json.dumps(dict(terms.most_common(400)), ensure_ascii=False),
})
return rec
def _stub_record(path: Path, st, root: Path, reason: str) -> dict:
"""Запись о файле, который разобрать не удалось: он остаётся в каталоге."""
rec = _base_record(path, st, root)
rec.update({
"title": pretty_name(path.name), "author": "", "year": None,
"pages": None, "lang": "?", "kind": "документ", "text_chars": 0,
"scanned": 1, "error": reason, "body": "", "toc": "[]", "terms": "{}",
})
return rec
def scan(con, root: Path, *, max_pages: int = config.MAX_PDF_PAGES,
jobs: int = 4, force: bool = False, limit: int | None = None,
extensions: set[str] | None = None, quiet: bool = False,
progress=None, stop=None) -> dict:
"""Обходит библиотеку и обновляет записи изменившихся файлов.
Обход инкрементальный: повторный запуск читает только новые и
изменившиеся файлы, поэтому прерванный разбор можно продолжить.
"""
known = db.get_state(con)
broken = db.failed_paths(con, MAX_TRIES)
alive: set[str] = set()
todo: list[tuple[Path, os.stat_result]] = []
skipped = 0
for path, st in iter_files(root, extensions):
sp = str(path)
alive.add(sp)
prev = known.get(sp)
if not force and prev and abs(prev[0] - st.st_mtime) < 1 \
and prev[1] == st.st_size:
continue
if sp in broken:
# Файл уже несколько раз обрывал разбор — заносим без текста.
db.upsert_doc(con, _stub_record(
path, st, root,
"разбор аварийно прерывался %d раз(а)" % MAX_TRIES))
skipped += 1
continue
todo.append((path, st))
if limit and len(todo) >= limit:
break
removed = 0 if limit else db.purge_missing(con, alive)
con.commit()
total = len(todo)
msg = ("Найдено файлов: %d, к разбору: %d, пропущено битых: %d, "
"удалено из базы: %d" % (len(alive), total, skipped, removed))
if not quiet:
print(msg, flush=True)
if progress:
progress(0, total, msg)
done = errors = 0
started = time.time()
chunk = max(jobs, 10)
for offset in range(0, total, chunk):
batch = todo[offset: offset + chunk]
# Попытка отмечается до разбора: если процесс погибнет внутри
# библиотеки-парсера, при следующем запуске файл будет опознан.
for p, _ in batch:
db.bump_attempt(con, str(p))
con.commit()
with ThreadPoolExecutor(max_workers=jobs) as pool:
futures = {pool.submit(_build_record, p, s, root, max_pages): (p, s)
for p, s in batch}
for fut in as_completed(futures):
path, st = futures[fut]
try:
rec = fut.result()
except Exception as e: # noqa: BLE001
rec = _stub_record(path, st, root,
"%s: %s" % (type(e).__name__, e))
db.upsert_doc(con, rec)
db.clear_attempt(con, str(path))
done += 1
if rec["error"]:
errors += 1
con.commit()
speed = done / max(time.time() - started, 0.1)
eta = (total - done) / max(speed, 0.001)
line = ("%d/%d (%.1f файл/с, осталось ~%d мин)"
% (done, total, speed, eta / 60))
if not quiet:
print(" " + line, flush=True)
if progress:
progress(done, total, line)
if stop and stop():
break
db.set_meta(con, "root", str(root))
db.set_meta(con, "last_scan", datetime.now().isoformat(timespec="seconds"))
con.commit()
return {"total_files": len(alive), "processed": done, "errors": errors,
"removed": removed, "skipped": skipped,
"seconds": round(time.time() - started, 1)}

86
catalogizer/search.py Normal file
View File

@@ -0,0 +1,86 @@
"""Поиск по каталогу: полнотекстовый запрос с учётом морфологии."""
from __future__ import annotations
import json
import re
from .textutil import stem
SAFE = re.compile(r"[^\w\-Ѐ-ӿ]+", re.U)
# Веса колонок для bm25: заголовок и ключевые слова важнее тела документа.
BM25_WEIGHTS = (10.0, 8.0, 4.0, 1.0, 0.8, 2.0)
def build_query(text: str) -> str:
"""Превращает пользовательский запрос в выражение FTS5.
Каждое слово ищется и как есть, и по основе с префиксом — чтобы
«инвертор» находил «инверторы» и «инвертора».
"""
phrases = re.findall(r'"([^"]+)"', text)
rest = re.sub(r'"[^"]*"', " ", text)
parts = ['"%s"' % SAFE.sub(" ", p).strip() for p in phrases if p.strip()]
for raw in rest.split():
neg = raw.startswith("-")
word = SAFE.sub("", raw.lstrip("-+")).strip()
if len(word) < 2:
continue
st = stem(word)
variants = {'"%s"' % word.lower()}
if st and st != word.lower():
variants.add('"%s"*' % st)
else:
variants.add('"%s"*' % word.lower())
clause = "(" + " OR ".join(sorted(variants)) + ")"
parts.append(("NOT " + clause) if neg else clause)
return " AND ".join(p for p in parts if p).replace("AND NOT", "NOT")
def search(con, query: str, *, limit: int = 20, ext: str | None = None,
folder: str | None = None, lang: str | None = None,
year_from: int | None = None, kind: str | None = None) -> list[dict]:
fts_query = build_query(query)
if not fts_query:
return []
sql = [
"SELECT d.id, d.title, d.author, d.year, d.pages, d.ext, d.lang, d.kind,",
" d.rel_path, d.path, d.folder, d.description, d.keywords, d.theses,",
" d.scanned, u.keywords AS user_keywords,",
" snippet(docs_fts, 3, '«', '»', ' … ', 14) AS snip,",
" bm25(docs_fts, %s) AS rank" % ", ".join(str(w) for w in BM25_WEIGHTS),
"FROM docs_fts JOIN docs d ON d.id = docs_fts.rowid",
" LEFT JOIN user_keywords u ON u.path = d.path",
"WHERE docs_fts MATCH ?",
]
params: list = [fts_query]
if ext:
sql.append("AND d.ext = ?")
params.append("." + ext.lower().lstrip("."))
if folder:
sql.append("AND d.folder LIKE ?")
params.append("%" + folder + "%")
if lang:
sql.append("AND d.lang = ?")
params.append(lang)
if kind:
sql.append("AND d.kind = ?")
params.append(kind)
if year_from:
sql.append("AND d.year >= ?")
params.append(year_from)
sql.append("ORDER BY rank LIMIT ?")
params.append(limit)
rows = con.execute("\n".join(sql), params).fetchall()
out = []
for r in rows:
d = dict(r)
d["keywords"] = json.loads(d.get("keywords") or "[]")
d["theses"] = json.loads(d.get("theses") or "[]")
d["user_keywords"] = json.loads(d.get("user_keywords") or "[]")
d["score"] = round(-d.pop("rank"), 2)
out.append(d)
return out

178
catalogizer/summarize.py Normal file
View File

@@ -0,0 +1,178 @@
"""Ключевые слова, тезисы и краткое описание документа.
Без обращения к внешним сервисам: экстрактивное реферирование по TF-IDF,
рассчитанному на самой библиотеке.
"""
from __future__ import annotations
import math
import re
from collections import Counter
from . import config
from .textutil import (STOP, detect_lang, sentences, similar, stem, tokens)
# Общие для любой техлитературы слова: сами по себе ничего не говорят.
GENERIC = {
"устройство", "система", "систем", "работа", "работы", "значение", "число",
"рисунок", "таблица", "формула", "пример", "случай", "результат", "метод",
"задача", "часть", "время", "уровень", "величина", "процесс", "качество",
"device", "system", "value", "number", "example", "case", "result",
"method", "task", "part", "time", "level", "process", "quality", "user",
"chapter", "figure", "section", "information", "application", "data",
}
GENERIC_STEMS = {stem(w) for w in GENERIC}
WORD_OK = re.compile(r"^[a-zа-я][a-zа-я0-9\-]{2,}$")
def doc_terms(text: str, title: str = "", toc: list[str] | None = None) -> Counter:
"""Частоты основ слов документа с усилением заголовка и оглавления."""
c: Counter = Counter()
for w in tokens(text):
if WORD_OK.match(w):
c[stem(w)] += 1
for w in tokens(title):
if WORD_OK.match(w):
c[stem(w)] += 6
for line in (toc or [])[:120]:
for w in tokens(line):
if WORD_OK.match(w):
c[stem(w)] += 3
return c
def _surface_forms(text: str, title: str, toc: list[str] | None) -> dict[str, str]:
"""Для каждой основы — самая частая словоформа (её и показываем)."""
forms: dict[str, Counter] = {}
for src in (title, "\n".join(toc or []), text[:15_000]):
for w in tokens(src):
if WORD_OK.match(w):
forms.setdefault(stem(w), Counter())[w] += 1
best = {}
for st, c in forms.items():
top = c.most_common(1)[0][1]
# Среди частых словоформ берём самую короткую: «вход», а не «входами».
good = [w for w, n in c.items() if n >= top * 0.5]
best[st] = min(good, key=lambda w: (len(w), w))
return best
class Analyzer:
"""Считает ключевые слова и тезисы с учётом IDF по всей библиотеке."""
def __init__(self, df: dict[str, int], n_docs: int):
self.df = df
self.n_docs = max(n_docs, 1)
def idf(self, term: str) -> float:
return math.log((self.n_docs + 1) / (self.df.get(term, 0) + 1)) + 1.0
def weights(self, terms: Counter) -> dict[str, float]:
"""Вес основы = сглаженный TF × IDF, без слишком общих слов."""
total = sum(terms.values()) or 1
w = {}
for t, n in terms.items():
if t in GENERIC_STEMS or t in STOP or len(t) < 3:
continue
if n < 2 and total > 400:
continue
w[t] = (1 + math.log(n)) * self.idf(t) / math.log(total + 10)
return w
def keywords(self, terms: Counter, text: str, title: str,
toc: list[str] | None, limit: int = config.N_KEYWORDS) -> list[str]:
w = self.weights(terms)
forms = _surface_forms(text, title, toc)
best = sorted(w.items(), key=lambda kv: -kv[1])
out: list[str] = []
for st, _ in best:
word = forms.get(st, st)
if any(word in o or o in word for o in out):
continue
out.append(word)
if len(out) >= limit:
break
return out
def theses(self, text: str, terms: Counter,
limit: int = config.N_THESES) -> list[str]:
"""Экстрактивные тезисы: самые «плотные» по ключевым словам фразы."""
w = self.weights(terms)
cands = sentences(text)
if not cands:
return []
scored = []
for pos, s in enumerate(cands):
ws = [w.get(stem(t), 0.0) for t in tokens(s)]
if len(ws) < 5:
continue
score = sum(ws) / math.sqrt(len(ws))
# Начало документа (аннотация, введение) — обычно самое ценное.
score *= 1.0 + 0.4 * max(0.0, 1.0 - pos / 40.0)
if re.search(r"(предназначен|рассматрива|описыва|посвящ|позволяет"
r"|книга|пособие|руководство|describes|presents"
r"|provides|introduces|this (book|guide|manual))",
s, re.I):
score *= 1.35
scored.append((score, pos, s))
scored.sort(key=lambda x: -x[0])
picked: list[tuple[int, str]] = []
for _, pos, s in scored:
if any(similar(s, p) > 0.55 for _, p in picked):
continue
picked.append((pos, s))
if len(picked) >= limit:
break
picked.sort(key=lambda x: x[0])
return [s for _, s in picked]
TYPE_HINTS = (
(r"datasheet|data sheet|техническое описание", "даташит"),
(r"reference manual|user manual|руководство пользователя", "руководство"),
(r"application note|прикладн", "апноут"),
(r"учебн|пособие|курс лекций|лекци", "учебник"),
(r"диссертац|автореферат", "диссертация"),
(r"гост|стандарт|iec |ieee std", "стандарт"),
(r"статья|journal|proceedings", "статья"),
(r"errata|release notes", "заметки к выпуску"),
)
def guess_kind(title: str, text: str, pages: int | None) -> str:
"""Тип издания — по названию, первым страницам и объёму."""
probe = (title + " " + text[:3000]).lower()
for pat, kind in TYPE_HINTS:
if re.search(pat, probe):
return kind
if pages and pages > 120:
return "книга"
if pages and pages <= 20:
return "заметка"
return "документ"
def describe(title: str, kind: str, lang: str, pages: int | None,
keywords: list[str], theses: list[str], has_text: bool) -> str:
"""Краткое описание в одну-две фразы."""
head = kind.capitalize()
if lang in ("ru", "en"):
head += " на %s языке" % ("русском" if lang == "ru" else "английском")
if pages:
head += ", %d с" % pages
if keywords:
head += ". Темы: " + ", ".join(keywords[:6])
if not has_text:
return head + ". Текстовый слой отсутствует (скан) — описание по имени файла."
if theses:
first = theses[0]
if len(first) > 220:
first = first[:217].rsplit(" ", 1)[0] + "…"
head += ". " + first
return head + ("" if head.endswith((".", "…")) else ".")
def analyze_language(text: str) -> str:
return detect_lang(text)

156
catalogizer/textutil.py Normal file
View File

@@ -0,0 +1,156 @@
"""Работа с текстом: язык, токены, стемминг, предложения."""
from __future__ import annotations
import re
import unicodedata
CYR = re.compile(r"[а-яёА-ЯЁ]")
LAT = re.compile(r"[a-zA-Z]")
TOKEN_RE = re.compile(r"[a-zA-Zа-яёА-ЯЁ][a-zA-Zа-яёА-ЯЁ0-9\-]{1,30}")
RU_STOP = set("""
а без более больше будет будто бы был была были было быть в вам вас ведь весь
вдруг вот впрочем все всегда всего всех всю вы где да даже два для до другой
его ее её ей ему если есть еще ещё же за зачем здесь и из или им иметь их к
как какой когда конечно кто куда ли лишь между меня мне может можно мой моя мы
на над надо наш не него неё нее ней нельзя нет ни нибудь никогда ним них ничего
но ну о об один он она они оно от очень перед по под после потом потому почти
при про просто раз разве с сам свое своей свои свой себе себя сейчас сказать
со совсем так такой там те тебя тем теперь то тогда того тоже той только том
тот тут ты у уж уже хорошо хоть чего чей чем через что чтобы чуть эта эти это
этой этом этот эту я является является является том числе также т е т д т п
рис таблица глава раздел стр страница пример примера данной данного данные
можно должен должна должны следует таким образом однако если этом при этом
""".split())
EN_STOP = set("""
a about above after again against all am an and any are as at be because been
before being below between both but by can cannot could did do does doing down
during each few for from further had has have having he her here hers him his
how i if in into is it its itself just me more most must my no nor not of off
on once only or other our out over own same she should so some such than that
the their them then there these they this those through to too under until up
use used using very was we were what when where which while who whom why will
with would you your figure table chapter section page fig eq ref see note
""".split())
STOP = RU_STOP | EN_STOP
# Мусорные токены, характерные для технической документации.
JUNK = set("""
www http https com ru org net pdf doc docx page pages copyright reserved rights
inc ltd corp all datasheet rev revision version document documents note notes
""".split())
def detect_lang(text: str) -> str:
"""Грубое определение языка по соотношению кириллицы и латиницы."""
sample = text[:20_000]
cyr = len(CYR.findall(sample))
lat = len(LAT.findall(sample))
if cyr + lat < 30:
return "?"
return "ru" if cyr > lat * 0.5 else "en"
_RU_ENDINGS = (
"ированием", "ированию", "ирования", "ированный", "ирование",
"ами", "ями", "ого", "ему", "ому", "ыми", "ими", "ая", "ое", "ые", "ый",
"ой", "ий", "ин", "ов", "ев", "ам", "ям", "ах", "ях", "ею", "ью", "ия",
"ии", "ие", "ем", "ом", "ах", "ешь", "ете", "ает", "ают", "ить", "ать",
"ешь", "ся", "сь", "а", "е", "и", "й", "о", "у", "ы", "ь", "ю", "я",
)
def stem(word: str) -> str:
"""Лёгкий стеммер: RU — отсечение окончаний, EN — отсечение суффиксов.
Нужен не для лингвистической точности, а чтобы «инвертор», «инвертора»
и «инверторы» попадали в один индексный ключ.
"""
w = word.lower().replace("ё", "е")
if CYR.search(w):
for end in _RU_ENDINGS:
if len(w) - len(end) >= 4 and w.endswith(end):
return w[: -len(end)]
return w
for end in ("ingly", "ations", "ation", "ings", "edly", "ies", "ing",
"ers", "er", "es", "ed", "s"):
if len(w) - len(end) >= 4 and w.endswith(end):
base = w[: -len(end)]
if end == "ies":
base += "y"
return base
return w
def tokens(text: str) -> list[str]:
"""Значимые слова текста в нижнем регистре."""
out = []
for m in TOKEN_RE.finditer(text):
w = m.group(0).lower().replace("ё", "е")
if len(w) < 3 or w in STOP or w in JUNK:
continue
if w.strip("-") != w:
w = w.strip("-")
if len(w) < 3:
continue
out.append(w)
return out
def clean_text(raw: str) -> str:
"""Убирает переносы строк внутри слов, колонтитулы и лишние пробелы."""
t = unicodedata.normalize("NFKC", raw)
t = t.replace("­", "")
t = re.sub(r"-\n(?=[a-zа-яё])", "", t) # перенос по слогам
t = re.sub(r"(?<=[^\n\.\!\?:;])\n(?=[a-zа-яё])", " ", t) # разрыв строки
t = re.sub(r"[ \t ]+", " ", t)
t = re.sub(r"\n{3,}", "\n\n", t)
return t.strip()
SENT_SPLIT = re.compile(r"(?<=[.!?…])\s+(?=[«\"'(\[]?[A-ZА-ЯЁ0-9])|\n{2,}")
def sentences(text: str) -> list[str]:
"""Разбивает текст на предложения, отбрасывая заведомый мусор."""
out = []
for chunk in SENT_SPLIT.split(text):
if not chunk:
continue
s = " ".join(chunk.split())
if not (40 <= len(s) <= 400):
continue
letters = sum(ch.isalpha() for ch in s)
if letters < len(s) * 0.55: # формулы, таблицы, номера страниц
continue
if s.count("|") > 2 or s.count("...") > 1 or s.count("__") > 0:
continue
words = s.split()
if len(words) < 6:
continue
upper = sum(w.isupper() for w in words)
if upper > len(words) * 0.6: # заголовок капсом
continue
out.append(s)
return out
def similar(a: str, b: str) -> float:
"""Мера пересечения двух предложений по множеству основ слов."""
sa = {stem(w) for w in tokens(a)}
sb = {stem(w) for w in tokens(b)}
if not sa or not sb:
return 0.0
return len(sa & sb) / len(sa | sb)
def pretty_name(filename: str) -> str:
"""Читаемое название из имени файла."""
name = re.sub(r"\.[A-Za-z0-9]{2,5}$", "", filename)
name = name.replace("_", " ").replace("%20", " ")
name = re.sub(r"[.\-]{2,}", " ", name)
name = re.sub(r"\s{2,}", " ", name).strip(" -.")
return name or filename

View File

@@ -0,0 +1,3 @@
"""@file __init__.py
@brief Слой представления: тёмное окно PySide6 поверх ядра каталогизатора.
"""

View File

@@ -0,0 +1,378 @@
"""@file catalog_tab.py
@brief Вкладка каталога: поиск по ключевым словам и карточка документа.
Пустой запрос показывает весь каталог по разделам; непустой уходит в
полнотекстовый индекс с учётом морфологии и ранжированием BM25.
"""
from __future__ import annotations
import html
import json
import os
import subprocess
import sys
from pathlib import Path
from PySide6.QtCore import Qt
from PySide6.QtGui import QKeySequence, QShortcut
from PySide6.QtWidgets import (
QAbstractItemView,
QCompleter,
QComboBox,
QHBoxLayout,
QHeaderView,
QLabel,
QLineEdit,
QMessageBox,
QPushButton,
QSplitter,
QTableWidget,
QTableWidgetItem,
QTextBrowser,
QVBoxLayout,
QWidget,
)
from .. import config, db
from .. import search as search_mod
from .theme import CARD_CSS
LANGS = (("любой язык", None), ("русский", "ru"), ("английский", "en"))
COLUMNS = (("Название", 430), ("Тип", 100), ("Год", 60), ("Стр.", 60),
("Формат", 75), ("Раздел", 280))
class CatalogTab(QWidget):
"""@brief Поиск по каталогу и просмотр карточки выбранного документа."""
def __init__(self, parent: QWidget | None = None) -> None:
super().__init__(parent)
self._rows: list[dict] = []
self._root = config.current_root()
layout = QVBoxLayout(self)
layout.setContentsMargins(12, 12, 12, 12)
layout.setSpacing(8)
layout.addLayout(self._build_query_row())
splitter = QSplitter(Qt.Orientation.Vertical)
splitter.addWidget(self._build_table())
splitter.addWidget(self._build_card())
splitter.setSizes([420, 260])
layout.addWidget(splitter, 1)
layout.addLayout(self._build_keyword_editor())
layout.addLayout(self._build_actions())
# ------------------------------------------------------------------
# Построение интерфейса
# ------------------------------------------------------------------
def _build_query_row(self) -> QHBoxLayout:
row = QHBoxLayout()
self.query = QLineEdit()
self.query.setPlaceholderText(
"Поиск по названию, тезисам, ключевым словам и тексту… "
"(\"точная фраза\", -исключить)")
self.query.returnPressed.connect(self.refresh)
row.addWidget(self.query, 1)
find = QPushButton("Найти")
find.setProperty("primary", True)
find.clicked.connect(self.refresh)
row.addWidget(find)
self.ext_box = QComboBox()
self.ext_box.addItem("все форматы", None)
self.ext_box.currentIndexChanged.connect(self.refresh)
row.addWidget(self.ext_box)
self.lang_box = QComboBox()
for title, code in LANGS:
self.lang_box.addItem(title, code)
self.lang_box.currentIndexChanged.connect(self.refresh)
row.addWidget(self.lang_box)
self.section_box = QComboBox()
self.section_box.addItem("все разделы", None)
self.section_box.currentIndexChanged.connect(self.refresh)
self.section_box.setMinimumWidth(220)
row.addWidget(self.section_box)
return row
def _build_table(self) -> QTableWidget:
self.table = QTableWidget(0, len(COLUMNS))
self.table.setHorizontalHeaderLabels([c[0] for c in COLUMNS])
self.table.verticalHeader().setVisible(False)
self.table.setAlternatingRowColors(True)
self.table.setSelectionBehavior(
QAbstractItemView.SelectionBehavior.SelectRows)
self.table.setSelectionMode(
QAbstractItemView.SelectionMode.SingleSelection)
self.table.setEditTriggers(
QAbstractItemView.EditTrigger.NoEditTriggers)
for index, (_, width) in enumerate(COLUMNS):
self.table.setColumnWidth(index, width)
self.table.horizontalHeader().setSectionResizeMode(
0, QHeaderView.ResizeMode.Stretch)
self.table.itemSelectionChanged.connect(self._show_card)
self.table.doubleClicked.connect(self.open_document)
return self.table
def _build_card(self) -> QTextBrowser:
self.card = QTextBrowser()
self.card.setOpenExternalLinks(False)
self.card.document().setDefaultStyleSheet(CARD_CSS)
return self.card
def _build_keyword_editor(self) -> QHBoxLayout:
"""@brief Строка правки своих ключевых слов выбранного документа."""
row = QHBoxLayout()
label = QLabel("Свои ключевые слова:")
label.setProperty("muted", True)
row.addWidget(label)
self.my_keywords = QLineEdit()
self.my_keywords.setPlaceholderText(
"через запятую — сохраняются в каталоге и участвуют в поиске")
self.my_keywords.setEnabled(False)
self.my_keywords.returnPressed.connect(self.save_keywords)
row.addWidget(self.my_keywords, 1)
self.save_kw = QPushButton("Сохранить")
self.save_kw.setEnabled(False)
self.save_kw.clicked.connect(self.save_keywords)
row.addWidget(self.save_kw)
QShortcut(QKeySequence("Ctrl+S"), self, self.save_keywords)
return row
def _build_actions(self) -> QHBoxLayout:
row = QHBoxLayout()
open_doc = QPushButton("Открыть документ")
open_doc.clicked.connect(self.open_document)
row.addWidget(open_doc)
show_dir = QPushButton("Показать в папке")
show_dir.clicked.connect(self.open_containing_folder)
row.addWidget(show_dir)
row.addStretch(1)
self.found = QLabel("")
self.found.setProperty("muted", True)
row.addWidget(self.found)
return row
# ------------------------------------------------------------------
# Данные
# ------------------------------------------------------------------
def set_root(self, root: Path) -> None:
"""@brief Переключает вкладку на каталог другой папки."""
self._root = Path(root)
self.reload_filters()
self._reload_completer()
self.refresh()
def _connect(self, create: bool = False):
db_path, _ = config.paths_for(self._root)
return db.connect(db_path, create=create)
def _reload_completer(self) -> None:
"""@brief Подсказки при вводе — из уже заведённых своих слов."""
try:
con = self._connect()
words = db.all_user_keywords(con)
con.close()
except Exception: # noqa: BLE001
return
completer = QCompleter(words, self)
completer.setCaseSensitivity(Qt.CaseSensitivity.CaseInsensitive)
completer.setFilterMode(Qt.MatchFlag.MatchContains)
self.my_keywords.setCompleter(completer)
def reload_filters(self) -> None:
"""@brief Обновляет списки форматов и разделов по содержимому базы."""
exts, folders = [], []
try:
con = self._connect()
exts = [r["ext"] for r in
con.execute("SELECT DISTINCT ext FROM docs ORDER BY ext")]
folders = [r["folder"] for r in con.execute(
"SELECT folder, COUNT(*) c FROM docs GROUP BY folder"
" ORDER BY c DESC LIMIT 200")]
con.close()
except Exception: # noqa: BLE001 — базы может ещё не быть
pass
for box, items, first in ((self.ext_box, exts, "все форматы"),
(self.section_box, folders, "все разделы")):
box.blockSignals(True)
box.clear()
box.addItem(first, None)
for item in items:
box.addItem(item, item)
box.blockSignals(False)
def refresh(self) -> None:
"""@brief Выполняет поиск (или показ всего каталога) и заполняет таблицу."""
text = self.query.text().strip()
ext = self.ext_box.currentData()
lang = self.lang_box.currentData()
section = self.section_box.currentData()
try:
con = self._connect()
if text:
self._rows = search_mod.search(con, text, limit=500, ext=ext,
lang=lang, folder=section)
else:
self._rows = self._browse(con, ext, lang, section)
con.close()
except FileNotFoundError:
self._rows = [] # каталог этой папки ещё не построен
except Exception as e: # noqa: BLE001
self._rows = []
QMessageBox.warning(self, "Каталогизатор", str(e))
self.table.setRowCount(len(self._rows))
for row, doc in enumerate(self._rows):
values = (doc["title"], doc.get("kind") or "",
str(doc.get("year") or ""), str(doc.get("pages") or ""),
doc["ext"].lstrip("."), doc.get("folder") or "")
for col, value in enumerate(values):
item = QTableWidgetItem(value)
if col in (2, 3, 4):
item.setTextAlignment(Qt.AlignmentFlag.AlignCenter)
self.table.setItem(row, col, item)
self.found.setText("Найдено: %d" % len(self._rows))
if self._rows:
self.table.selectRow(0)
else:
self.card.setHtml("<p class='facts'>Ничего не найдено.</p>")
self.my_keywords.clear()
self.my_keywords.setEnabled(False)
self.save_kw.setEnabled(False)
@staticmethod
def _browse(con, ext, lang, section) -> list[dict]:
"""@brief Весь каталог по разделам — когда строка запроса пуста."""
sql = ["SELECT d.id, d.title, d.author, d.year, d.pages, d.ext, d.lang,",
" d.kind, d.folder, d.rel_path, d.path, d.description,",
" d.keywords, d.theses, d.scanned, d.error,",
" u.keywords AS user_keywords",
" FROM docs d LEFT JOIN user_keywords u ON u.path = d.path",
" WHERE 1 = 1"]
params: list = []
for column, value in (("ext", ext), ("lang", lang), ("folder", section)):
if value:
sql.append("AND d.%s = ?" % column)
params.append(value)
sql.append("ORDER BY d.folder, d.title LIMIT 3000")
rows = []
for r in con.execute("\n".join(sql), params):
doc = dict(r)
doc["keywords"] = json.loads(doc["keywords"] or "[]")
doc["theses"] = json.loads(doc["theses"] or "[]")
doc["user_keywords"] = json.loads(doc["user_keywords"] or "[]")
doc["snip"] = ""
rows.append(doc)
return rows
def _current(self) -> dict | None:
row = self.table.currentRow()
if row < 0 or row >= len(self._rows):
return None
return self._rows[row]
# ------------------------------------------------------------------
# Карточка и действия
# ------------------------------------------------------------------
def _show_card(self) -> None:
doc = self._current()
if doc is None:
return
esc = html.escape
facts = [doc["ext"].lstrip("."), doc.get("kind") or ""]
if doc.get("author"):
facts.append(doc["author"])
if doc.get("year"):
facts.append(str(doc["year"]))
if doc.get("pages"):
facts.append("%s с." % doc["pages"])
if doc.get("scanned"):
facts.append("скан без текстового слоя")
if doc.get("error"):
facts.append(doc["error"])
parts = ["<h2>%s</h2>" % esc(doc["title"]),
"<div class='facts'>%s</div>"
% esc(" · ".join(f for f in facts if f)),
"<div class='desc'>%s</div>" % esc(doc.get("description") or "")]
if doc.get("theses"):
parts.append("<div class='label'>Тезисы</div><ul>")
parts += ["<li>%s</li>" % esc(t) for t in doc["theses"]]
parts.append("</ul>")
if doc.get("user_keywords"):
parts.append("<div class='label'>Свои ключевые слова</div>"
"<div class='mine'>%s</div>"
% esc(", ".join(doc["user_keywords"])))
if doc.get("keywords"):
parts.append("<div class='label'>Ключевые слова</div>"
"<div class='kw'>%s</div>"
% esc(", ".join(doc["keywords"])))
if doc.get("snip"):
snippet = esc(" ".join(doc["snip"].split()))
snippet = snippet.replace("«", "<mark>").replace("»", "</mark>")
parts.append("<div class='label'>Из текста</div><div>%s</div>"
% snippet)
parts.append("<div class='path'>%s</div>" % esc(doc["path"]))
self.card.setHtml("".join(parts))
self.my_keywords.setText(", ".join(doc.get("user_keywords") or []))
self.my_keywords.setEnabled(True)
self.save_kw.setEnabled(True)
def save_keywords(self) -> None:
"""@brief Записывает свои ключевые слова выбранного документа."""
doc = self._current()
if doc is None:
return
words = [w.strip() for w in self.my_keywords.text().split(",")]
try:
con = self._connect(create=True)
saved = db.set_user_keywords(con, doc["path"], words)
con.close()
except Exception as e: # noqa: BLE001
QMessageBox.warning(self, "Каталогизатор", str(e))
return
doc["user_keywords"] = saved
self.my_keywords.setText(", ".join(saved))
self._show_card()
self._reload_completer()
self.found.setText("Ключевые слова сохранены")
def open_document(self) -> None:
"""@brief Открывает выбранный документ системным приложением."""
doc = self._current()
if doc:
self._open(Path(doc["path"]))
def open_containing_folder(self) -> None:
"""@brief Показывает документ в проводнике."""
doc = self._current()
if not doc:
return
path = Path(doc["path"])
if os.name == "nt":
subprocess.Popen(["explorer", "/select,", str(path)])
elif sys.platform == "darwin":
subprocess.Popen(["open", "-R", str(path)]) # показать в Finder
else:
self._open(path.parent)
def _open(self, path: Path) -> None:
try:
if os.name == "nt":
os.startfile(str(path)) # noqa: S606 — открываем через оболочку
elif sys.platform == "darwin":
subprocess.Popen(["open", str(path)])
else:
subprocess.Popen(["xdg-open", str(path)])
except OSError as e:
QMessageBox.warning(self, "Каталогизатор",
"Не удалось открыть:\n%s\n%s" % (path, e))

View File

@@ -0,0 +1,124 @@
"""@file main_window.py
@brief Главное окно каталогизатора: шапка, вкладки разбора и поиска.
Обе вкладки работают с одной и той же папкой: смена папки на вкладке
разбора сразу переключает каталог, по которому идёт поиск.
"""
from __future__ import annotations
from pathlib import Path
from PySide6.QtWidgets import (
QFrame,
QHBoxLayout,
QLabel,
QMainWindow,
QTabWidget,
QVBoxLayout,
QWidget,
)
from .. import config, db
from .catalog_tab import CatalogTab
from .scan_tab import ScanTab
from .theme import APP_STYLESHEET
class MainWindow(QMainWindow):
"""@brief Окно с вкладками «Каталог и поиск» и «Разбор папки»."""
def __init__(self, root: Path | None = None,
parent: QWidget | None = None) -> None:
super().__init__(parent)
self.setWindowTitle("Каталогизатор документов")
self.resize(1280, 820)
self.setMinimumSize(1000, 640)
self.setStyleSheet(APP_STYLESHEET)
self.catalog_tab = CatalogTab()
self.scan_tab = ScanTab()
central = QWidget()
layout = QVBoxLayout(central)
layout.setContentsMargins(0, 0, 0, 0)
layout.setSpacing(0)
layout.addWidget(self._build_header())
layout.addWidget(self._build_tabs(), 1)
self.setCentralWidget(central)
self.scan_tab.catalog_ready.connect(self._on_catalog_ready)
self.scan_tab.root_changed.connect(self._on_root_changed)
self.set_root(Path(root) if root else config.current_root())
# ------------------------------------------------------------------
def _build_header(self) -> QFrame:
header = QFrame()
header.setObjectName("header")
row = QHBoxLayout(header)
row.setContentsMargins(18, 12, 18, 12)
titles = QVBoxLayout()
title = QLabel("Каталогизатор")
title.setProperty("title", True)
subtitle = QLabel("Описания, тезисы и поиск по папке с документами")
subtitle.setProperty("muted", True)
titles.addWidget(title)
titles.addWidget(subtitle)
row.addLayout(titles)
row.addStretch(1)
self.summary = QLabel("")
self.summary.setProperty("muted", True)
row.addWidget(self.summary)
return header
def _build_tabs(self) -> QTabWidget:
self.tabs = QTabWidget()
self.tabs.addTab(self.catalog_tab, "Каталог и поиск")
self.tabs.addTab(self.scan_tab, "Разбор папки")
return self.tabs
# ------------------------------------------------------------------
def set_root(self, root: Path) -> None:
"""@brief Переключает обе вкладки на указанную папку."""
self.scan_tab.set_root(root)
self.catalog_tab.set_root(root)
self._update_summary(root)
def _on_root_changed(self, text: str) -> None:
root = Path(text.strip('" '))
self.catalog_tab.set_root(root)
self._update_summary(root)
def _on_catalog_ready(self, text: str) -> None:
root = Path(text)
self.catalog_tab.set_root(root)
self._update_summary(root)
self.tabs.setCurrentWidget(self.catalog_tab)
def _update_summary(self, root: Path) -> None:
"""@brief Показывает в шапке, что уже есть в каталоге этой папки."""
try:
db_path, _ = config.paths_for(root)
con = db.connect(db_path, create=False)
total = con.execute("SELECT COUNT(*) c FROM docs").fetchone()["c"]
scans = con.execute(
"SELECT COUNT(*) c FROM docs WHERE scanned = 1").fetchone()["c"]
when = db.get_meta(con, "last_scan")
con.close()
except Exception: # noqa: BLE001 — база может быть ещё не создана
self.summary.setText("")
return
if not total:
self.summary.setText("Каталог пуст — постройте его на вкладке «Разбор папки»")
return
self.summary.setText(
"В каталоге %d документов, из них %d сканов · обновлён %s"
% (total, scans, when or "—"))
def closeEvent(self, event) -> None: # noqa: N802 — имя из Qt
if self.scan_tab.busy():
self.scan_tab.stop_build()
event.accept()

244
catalogizer/ui/scan_tab.py Normal file
View File

@@ -0,0 +1,244 @@
"""@file scan_tab.py
@brief Вкладка разбора: выбор папки, параметры, ход работы и журнал.
Здесь запускается полный цикл — обход папки, извлечение текста, тезисы и
выгрузка отчётов. Работа идёт в BuildWorker, окно остаётся живым.
"""
from __future__ import annotations
import os
import subprocess
import sys
from pathlib import Path
from PySide6.QtCore import Qt, Signal
from PySide6.QtWidgets import (
QCheckBox,
QFileDialog,
QFrame,
QGridLayout,
QHBoxLayout,
QLabel,
QLineEdit,
QMessageBox,
QPlainTextEdit,
QProgressBar,
QPushButton,
QSpinBox,
QVBoxLayout,
QWidget,
)
from .. import config, db, report
from .worker import BuildWorker
class ScanTab(QWidget):
"""@brief Управление разбором папки и показ хода работы."""
#: каталог перестроен — вкладке поиска пора перечитать данные
catalog_ready = Signal(str)
#: пользователь выбрал другую папку
root_changed = Signal(str)
def __init__(self, parent: QWidget | None = None) -> None:
super().__init__(parent)
self._worker: BuildWorker | None = None
layout = QVBoxLayout(self)
layout.setContentsMargins(12, 12, 12, 12)
layout.setSpacing(10)
layout.addWidget(self._build_source_panel())
layout.addWidget(self._build_progress_panel())
journal = QLabel("Журнал")
journal.setProperty("section", True)
layout.addWidget(journal)
self.log = QPlainTextEdit()
self.log.setReadOnly(True)
layout.addWidget(self.log, 1)
# ------------------------------------------------------------------
def _build_source_panel(self) -> QFrame:
panel = QFrame()
panel.setProperty("panel", True)
grid = QGridLayout(panel)
grid.setContentsMargins(14, 12, 14, 12)
grid.setHorizontalSpacing(8)
title = QLabel("Папка с документами")
title.setProperty("section", True)
grid.addWidget(title, 0, 0, 1, 4)
self.folder = QLineEdit(str(config.current_root()))
self.folder.editingFinished.connect(
lambda: self.root_changed.emit(self.folder.text()))
grid.addWidget(self.folder, 1, 0, 1, 3)
browse = QPushButton("Обзор…")
browse.clicked.connect(self.pick_folder)
grid.addWidget(browse, 1, 3)
options = QHBoxLayout()
options.addWidget(QLabel("потоков чтения:"))
self.jobs = QSpinBox()
self.jobs.setRange(1, 32)
self.jobs.setValue(4)
self.jobs.setToolTip("Жёсткий диск — 24, SSD — 816.")
options.addWidget(self.jobs)
options.addSpacing(16)
options.addWidget(QLabel("страниц на документ:"))
self.max_pages = QSpinBox()
self.max_pages.setRange(3, 200)
self.max_pages.setValue(config.MAX_PDF_PAGES)
self.max_pages.setToolTip(
"Сколько первых страниц читать: больше — точнее тезисы, но дольше.")
options.addWidget(self.max_pages)
options.addSpacing(16)
self.force = QCheckBox("перечитать всё заново")
self.force.setToolTip(
"Обычно читаются только новые и изменившиеся файлы.")
options.addWidget(self.force)
options.addStretch(1)
grid.addLayout(options, 2, 0, 1, 4)
buttons = QHBoxLayout()
self.start = QPushButton("Построить каталог")
self.start.setProperty("primary", True)
self.start.clicked.connect(self.start_build)
buttons.addWidget(self.start)
self.stop = QPushButton("Остановить")
self.stop.setEnabled(False)
self.stop.clicked.connect(self.stop_build)
buttons.addWidget(self.stop)
buttons.addStretch(1)
reports = QPushButton("Пересобрать отчёты")
reports.clicked.connect(self.rebuild_reports)
buttons.addWidget(reports)
html_button = QPushButton("Открыть HTML-каталог")
html_button.clicked.connect(self.open_html)
buttons.addWidget(html_button)
grid.addLayout(buttons, 3, 0, 1, 4)
return panel
def _build_progress_panel(self) -> QFrame:
panel = QFrame()
panel.setProperty("panel", True)
box = QVBoxLayout(panel)
box.setContentsMargins(14, 12, 14, 12)
self.bar = QProgressBar()
self.bar.setRange(0, 100)
self.bar.setValue(0)
box.addWidget(self.bar)
self.state = QLabel("Готово. Выберите папку и нажмите «Построить каталог».")
self.state.setProperty("muted", True)
box.addWidget(self.state)
return panel
# ------------------------------------------------------------------
def root(self) -> Path:
return Path(self.folder.text().strip('" '))
def set_root(self, root: Path) -> None:
self.folder.setText(str(root))
def say(self, text: str) -> None:
"""@brief Добавляет строку в журнал."""
self.log.appendPlainText(text)
def pick_folder(self) -> None:
chosen = QFileDialog.getExistingDirectory(
self, "Папка с документами", str(self.root()))
if chosen:
self.folder.setText(str(Path(chosen)))
self.root_changed.emit(self.folder.text())
# ------------------------------------------------------------------
def start_build(self) -> None:
"""@brief Запускает полный цикл в фоновом потоке."""
root = self.root()
if not root.is_dir():
QMessageBox.warning(self, "Каталогизатор",
"Папка не найдена:\n%s" % root)
return
self.start.setEnabled(False)
self.stop.setEnabled(True)
self.bar.setRange(0, 0) # неопределённый ход: идёт обход папки
self.state.setText("Обход папки…")
self._worker = BuildWorker(root, self.jobs.value(),
self.max_pages.value(), self.force.isChecked())
self._worker.progress.connect(self._on_progress)
self._worker.message.connect(self.say)
self._worker.finished_ok.connect(self._on_done)
self._worker.failed.connect(self._on_failed)
self._worker.start()
def stop_build(self) -> None:
if self._worker:
self._worker.request_stop()
self.state.setText("Останавливаюсь после текущей порции…")
def _on_progress(self, done: int, total: int, text: str) -> None:
if total > 0:
self.bar.setRange(0, total)
self.bar.setValue(done)
else:
self.bar.setRange(0, 0)
self.state.setText(text)
def _on_done(self, text: str) -> None:
self.bar.setRange(0, 100)
self.bar.setValue(100)
self.state.setText(text)
self.say(text)
self.start.setEnabled(True)
self.stop.setEnabled(False)
self.catalog_ready.emit(str(self.root()))
def _on_failed(self, text: str) -> None:
self.bar.setRange(0, 100)
self.bar.setValue(0)
self.state.setText("Ошибка разбора")
self.say(text)
self.start.setEnabled(True)
self.stop.setEnabled(False)
QMessageBox.critical(self, "Каталогизатор",
text.strip().splitlines()[-1])
# ------------------------------------------------------------------
def out_dir(self) -> Path:
return config.paths_for(self.root())[1]
def rebuild_reports(self) -> None:
"""@brief Пересобирает HTML, Markdown, JSON и CSV из готовой базы."""
try:
db_path, out_dir = config.paths_for(self.root())
con = db.connect(db_path, create=False)
paths = report.build_all(con, out_dir)
con.close()
except Exception as e: # noqa: BLE001
QMessageBox.critical(self, "Каталогизатор", str(e))
return
for kind, path in paths.items():
self.say("%-5s %s" % (kind, path))
self.state.setText("Отчёты сохранены в %s" % out_dir)
def open_html(self) -> None:
"""@brief Открывает автономную HTML-страницу каталога."""
path = self.out_dir() / "catalog.html"
if not path.exists():
QMessageBox.information(
self, "Каталогизатор",
"HTML-каталог ещё не собран.\nНажмите «Построить каталог».")
return
if os.name == "nt":
os.startfile(str(path)) # noqa: S606
elif sys.platform == "darwin":
subprocess.Popen(["open", str(path)])
else:
subprocess.Popen(["xdg-open", str(path)])
def busy(self) -> bool:
return bool(self._worker and self._worker.isRunning())

140
catalogizer/ui/theme.py Normal file
View File

@@ -0,0 +1,140 @@
"""@file theme.py
@brief Единая спокойная тёмная тема каталогизатора без внешних ресурсов.
Палитра и правила оформления совпадают со стилем SETGUI, чтобы обе
программы выглядели частями одного инструментария.
"""
APP_STYLESHEET = """
QWidget {
background: #10151d;
color: #dce6f2;
font-family: "Segoe UI", "SF Pro Text", "Helvetica Neue", "Noto Sans",
sans-serif;
font-size: 10pt;
}
QMainWindow { background: #0d1219; }
QFrame#header {
background: #151d28;
border-bottom: 1px solid #263345;
}
QFrame[panel="true"] {
background: #151d28;
border: 1px solid #263345;
border-radius: 8px;
}
QLabel[muted="true"] { color: #8291a5; }
QLabel[title="true"] {
color: #f4f8fc;
font-size: 15pt;
font-weight: 600;
}
QLabel[section="true"] {
color: #f4f8fc;
font-size: 11pt;
font-weight: 600;
}
QPushButton {
background: #243247;
border: 1px solid #344762;
border-radius: 5px;
padding: 7px 14px;
}
QPushButton:hover { background: #2d405b; }
QPushButton:pressed { background: #1c293b; }
QPushButton:disabled { color: #627086; background: #192230; }
QPushButton[primary="true"] {
background: #1570d8;
border-color: #2584ec;
font-weight: 600;
}
QPushButton[primary="true"]:hover { background: #2584ec; }
QLineEdit, QComboBox, QSpinBox, QDoubleSpinBox {
background: #0d131c;
border: 1px solid #34445a;
border-radius: 5px;
padding: 6px;
selection-background-color: #1570d8;
}
QComboBox::drop-down { border: 0; width: 24px; }
QComboBox QAbstractItemView {
background: #0d131c;
border: 1px solid #34445a;
selection-background-color: #1570d8;
}
QTabWidget::pane {
border: 1px solid #263345;
background: #10151d;
}
QTabBar::tab {
background: #151d28;
color: #91a0b4;
border: 1px solid #263345;
padding: 9px 15px;
}
QTabBar::tab:selected {
color: #ffffff;
background: #1c2838;
border-bottom: 2px solid #2f91ff;
}
QHeaderView::section {
background: #1b2635;
color: #b9c5d5;
border: 0;
border-right: 1px solid #2b394c;
padding: 6px;
}
QTableWidget, QTreeWidget {
background: #0d131c;
alternate-background-color: #111a25;
gridline-color: #263345;
border: 1px solid #263345;
selection-background-color: #1570d8;
selection-color: #ffffff;
}
QTextBrowser, QPlainTextEdit {
background: #090e14;
color: #aabbd0;
border: 1px solid #263345;
font-family: "Cascadia Mono", "SF Mono", Menlo, "DejaVu Sans Mono",
monospace;
font-size: 9pt;
}
QProgressBar {
background: #0d131c;
border: 1px solid #34445a;
border-radius: 5px;
text-align: center;
}
QProgressBar::chunk { background: #1570d8; border-radius: 4px; }
QScrollBar:vertical {
background: #10151d;
width: 11px;
margin: 0;
}
QScrollBar::handle:vertical {
background: #34445a;
min-height: 24px;
border-radius: 5px;
}
QSplitter::handle { background: #1b2635; }
QStatusBar { background: #151d28; color: #8291a5; }
"""
#: Цвета карточки документа: HTML-разметка QTextBrowser темы не наследует.
CARD_CSS = """
body { color: #dce6f2;
font-family: "Segoe UI", "SF Pro Text", "Helvetica Neue", sans-serif;
font-size: 10pt; }
h2 { color: #f4f8fc; font-size: 12pt; margin: 0 0 4px 0; }
.facts { color: #8291a5; font-size: 9pt; margin-bottom: 10px; }
.desc { margin-bottom: 10px; }
.label { color: #2f91ff; font-weight: 600; }
li { margin-bottom: 4px; }
.kw { color: #b9c5d5; }
.mine { color: #2f91ff; font-weight: 600; }
.path { color: #627086;
font-family: "Cascadia Mono", "SF Mono", Menlo, monospace;
font-size: 8pt; }
mark { background: #1570d8; color: #ffffff; }
"""

78
catalogizer/ui/worker.py Normal file
View File

@@ -0,0 +1,78 @@
"""@file worker.py
@brief Фоновый поток построения каталога.
Разбор библиотеки идёт минутами и часами, поэтому выполняется в отдельном
потоке QThread: окно остаётся отзывчивым, а ход работы приходит сигналами.
"""
from __future__ import annotations
import traceback
from pathlib import Path
from PySide6.QtCore import QThread, Signal
from .. import analyze as analyze_mod
from .. import config, db, report, scanner
class BuildWorker(QThread):
"""@brief Полный цикл: обход папки, анализ, выгрузка отчётов."""
#: сделано, всего, поясняющая строка
progress = Signal(int, int, str)
#: строка в журнал
message = Signal(str)
#: итоговая строка состояния
finished_ok = Signal(str)
#: текст ошибки с трассировкой
failed = Signal(str)
def __init__(self, root: Path, jobs: int, max_pages: int,
force: bool = False, parent=None) -> None:
super().__init__(parent)
self._root = Path(root)
self._jobs = jobs
self._max_pages = max_pages
self._force = force
self._stop = False
def request_stop(self) -> None:
"""@brief Просит остановиться после текущей порции файлов.
Разобранное уже записано в базу, поэтому следующий запуск продолжит
с того же места.
"""
self._stop = True
def run(self) -> None: # noqa: D102 — контракт QThread
try:
config.remember(self._root)
db_path, out_dir = config.paths_for(self._root)
con = db.connect(db_path)
self.message.emit("Папка: %s" % self._root)
stats = scanner.scan(
con, self._root, max_pages=self._max_pages, jobs=self._jobs,
force=self._force, quiet=True,
progress=lambda d, t, m: self.progress.emit(d, t, m),
stop=lambda: self._stop)
self.message.emit(
"Разобрано %(processed)d, с ошибками %(errors)d, "
"пропущено %(skipped)d, за %(seconds)s с" % stats)
self.progress.emit(0, 0, "Анализ текстов…")
analyze_mod.analyze(
con, quiet=True,
progress=lambda d, t, m: self.progress.emit(d, t, m))
self.progress.emit(0, 0, "Выгрузка отчётов…")
for kind, path in report.build_all(con, out_dir).items():
self.message.emit("%-5s %s" % (kind, path))
total = con.execute("SELECT COUNT(*) c FROM docs").fetchone()["c"]
con.close()
self.finished_ok.emit(
"Каталог готов: %d документов%s"
% (total, " (остановлено)" if self._stop else ""))
except Exception: # noqa: BLE001 — окно должно показать причину
self.failed.emit(traceback.format_exc())