Andrey Kruchinkin
49f91247c7
Программа принимает папку с документами, извлекает из них текст и метаданные и строит каталог: краткое описание, тезисы и ключевые слова для каждого файла плюс поиск по всему собранному. Ядро: - extract: PDF, DJVU, DOC(X), RTF, ODT, CHM, EPUB, FB2, PPT(X), XLS(X), TXT; PDF передаётся MuPDF потоком в память (работают длинные пути) и ограничен по времени — повреждённый файл иначе чинится минутами; - summarize: экстрактивное реферирование по TF-IDF, посчитанному на самой библиотеке, с лёгким стеммером для русского и английского; - db: SQLite с полнотекстовым индексом FTS5, морфологический поиск с BM25; - scanner: инкрементальный многопоточный обход, счётчик попыток защищает от файла, обрывающего разбор; - report: автономный HTML с поиском, Markdown, JSON, CSV. Интерфейс: - окно PySide6 в тёмной теме: вкладки разбора и поиска, фоновый поток, карточка документа, правка своих ключевых слов; - командная строка: build, scan, analyze, report, search, show, tag, stats; - у каждой папки свой каталог, последняя обработанная запоминается. Сборка: scripts/build_exe.py даёт Catalogizer.exe в Windows и Catalogizer.app в macOS, иконки .ico и .icns рисуются кодом. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
384 lines
16 KiB
Python
384 lines
16 KiB
Python
"""Командный интерфейс каталогизатора."""
|
||
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)
|