"""Командный интерфейс каталогизатора.""" 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)