index/query_1c.py

#!/usr/bin/env python3
"""Запросы к единому индексу конфигураций 1С (cache/1c_meta/index.sqlite).

Параметры:
  --config, -c   файл настроек утилиты (JSON), не конфигурация 1С
  --baseconf, -b выгрузка конфигурации 1С в репозитории (crm3-26, target, …)

Примеры:

  python tools/index/query_1c.py search \"ЗаказКлиента\"
  python tools/index/query_1c.py search \"Партнер\" --type Document -b crm3-26
  python tools/index/query_1c.py show ЗаказКлиента -b target
  python tools/index/query_1c.py show Document.ЗаказКлиента -b crm3-26
  python tools/index/query_1c.py modules \"РВС_Transfer\" -b ws-rhana
  python tools/index/query_1c.py refs Catalog.Партнеры -b source
  python tools/index/query_1c.py module CommonModules/РВС_TransferAPIДокументы
"""

from __future__ import annotations

import argparse
import json
import sys
from pathlib import Path

HERE = Path(__file__).resolve().parent
if str(HERE) not in sys.path:
    sys.path.insert(0, str(HERE))

from index1c.cli import (  # noqa: E402
    BaseconfError,
    add_baseconf_args,
    add_tool_args,
    format_not_found_modules,
    format_not_found_object,
    get_baseconfs_for_query,
    handle_baseconf_error,
    index_db_missing_message,
    resolve_project_root,
)
from index1c.config import resolve_object_type  # noqa: E402
from index1c.search import (  # noqa: E402
    find_refs_to,
    get_module,
    get_object,
    open_index,
    search_modules,
    search_objects,
)


def _open_or_exit(root: Path):
    try:
        return open_index(root)
    except FileNotFoundError:
        print(index_db_missing_message(root), file=sys.stderr)
        sys.exit(1)


def cmd_search(args: argparse.Namespace) -> int:
    root = resolve_project_root(args)
    conn = _open_or_exit(root)
    try:
        try:
            baseconfs = get_baseconfs_for_query(args, conn, root)
        except BaseconfError as e:
            return handle_baseconf_error(e)
        otype = resolve_object_type(args.type) if args.type else None
        rows = search_objects(
            conn,
            args.query,
            configs=baseconfs,
            object_type=otype,
            limit=args.limit,
        )
        if not rows:
            scope = f" (baseconf: {', '.join(baseconfs)})" if baseconfs else ""
            print(f"Ничего не найдено{scope}.")
            return 2
        for r in rows:
            syn = f" — {r['synonym']}" if r.get("synonym") else ""
            print(f"- [{r['config']}] `{r['full_name']}`{syn}")
            if r.get("snip"):
                print(f"  {r['snip']}")
        return 0
    finally:
        conn.close()


def cmd_modules(args: argparse.Namespace) -> int:
    root = resolve_project_root(args)
    conn = _open_or_exit(root)
    try:
        try:
            baseconfs = get_baseconfs_for_query(args, conn, root)
        except BaseconfError as e:
            return handle_baseconf_error(e)
        rows = search_modules(conn, args.query, configs=baseconfs, limit=args.limit)
        if not rows:
            print(format_not_found_modules(args.query, baseconfs=baseconfs, conn=conn))
            return 2
        for r in rows:
            print(f"- [{r['config']}] `{r['rel_path']}` ({r['module_kind']}, owner={r['owner']})")
            if r.get("snip"):
                print(f"  {r['snip']}")
        return 0
    finally:
        conn.close()


def cmd_refs(args: argparse.Namespace) -> int:
    root = resolve_project_root(args)
    conn = _open_or_exit(root)
    try:
        try:
            baseconfs = get_baseconfs_for_query(args, conn, root)
        except BaseconfError as e:
            return handle_baseconf_error(e)
        rows = find_refs_to(conn, args.target, configs=baseconfs, limit=args.limit)
        if not rows:
            scope = f" в {', '.join(baseconfs)}" if baseconfs else ""
            print(f"Ссылок на «{args.target}» не найдено{scope}.")
            return 2
        for r in rows:
            print(
                f"- [{r['config']}] `{r['from_full_name']}.{r['from_field']}` "
                f"→ {r['to_kind']}.{r['to_name']}"
            )
        return 0
    finally:
        conn.close()


def cmd_show(args: argparse.Namespace) -> int:
    root = resolve_project_root(args)
    conn = _open_or_exit(root)
    try:
        try:
            baseconfs = get_baseconfs_for_query(args, conn, root)
        except BaseconfError as e:
            return handle_baseconf_error(e)
        obj = get_object(conn, args.name, configs=baseconfs)
        if not obj:
            print(format_not_found_object(args.name, baseconfs=baseconfs, conn=conn), file=sys.stderr)
            return 2
        print(json.dumps(obj, ensure_ascii=False, indent=2))
        return 0
    finally:
        conn.close()


def cmd_module(args: argparse.Namespace) -> int:
    root = resolve_project_root(args)
    conn = _open_or_exit(root)
    try:
        try:
            baseconfs = get_baseconfs_for_query(args, conn, root)
        except BaseconfError as e:
            return handle_baseconf_error(e)
        rows = get_module(conn, args.path, configs=baseconfs)
        if not rows:
            print(format_not_found_modules(args.path, baseconfs=baseconfs, conn=conn))
            return 2
        for r in rows:
            print(json.dumps(dict(r), ensure_ascii=False, indent=2))
        return 0
    finally:
        conn.close()


def build_parser() -> argparse.ArgumentParser:
    p = argparse.ArgumentParser(
        description="Поиск по cache/1c_meta/index.sqlite (метаданные и BSL-модули).",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog=(
            "Конфигурация 1С — параметр --baseconf (-b): папка выгрузки в репозитории.\n"
            "Файл настроек утилиты — --config (-c): см. tools/index/config.example.json"
        ),
    )
    add_tool_args(p)
    sub = p.add_subparsers(dest="cmd", required=True)

    s = sub.add_parser("search", help="Полнотекстовый поиск объектов метаданных")
    s.add_argument("query", help="Имя, синоним, фрагмент поля")
    add_baseconf_args(s)
    s.add_argument("--type", help="Тип объекта: Document, Catalog, рс, …")
    s.add_argument("--limit", type=int, default=30)
    s.set_defaults(func=cmd_search)

    m = sub.add_parser("modules", help="Поиск по BSL: процедуры, тело модуля, путь")
    m.add_argument("query")
    add_baseconf_args(m)
    m.add_argument("--limit", type=int, default=30)
    m.set_defaults(func=cmd_modules)

    ref = sub.add_parser("refs", help="Обратный индекс: кто ссылается на объект")
    ref.add_argument("target", help="Catalog.Партнеры | CatalogRef.Партнеры | Партнеры")
    add_baseconf_args(ref)
    ref.add_argument("--limit", type=int, default=100)
    ref.set_defaults(func=cmd_refs)

    sh = sub.add_parser("show", help="Карточка объекта (JSON: поля, типы, ссылки)")
    sh.add_argument("name", help="ЗаказКлиента | Document.ЗаказКлиента")
    add_baseconf_args(sh)
    sh.set_defaults(func=cmd_show)

    md = sub.add_parser("module", help="Метаданные модуля по пути или owner")
    md.add_argument("path", help="CommonModules/… или CommonModule.Имя")
    add_baseconf_args(md)
    md.set_defaults(func=cmd_module)

    return p


def main(argv: list[str] | None = None) -> int:
    parser = build_parser()
    args = parser.parse_args(argv)
    return int(args.func(args))


if __name__ == "__main__":
    raise SystemExit(main())