mihailkudravcev
91521e6604
Add virtual baseconf support for external data processors so reindex can discover and index them alongside regular src-based configurations, and update docs to use generic examples without workspace-specific names.
377 lines
13 KiB
Python
377 lines
13 KiB
Python
"""Общие аргументы CLI: --config (скрипт), --root, --baseconf (конфигурация 1С)."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import sys
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
from . import db as dbmod
|
||
from .config import cache_dir, discover_configs, project_root_from_here, resolve_config_name
|
||
|
||
|
||
class BaseconfError(Exception):
|
||
"""Ошибка выбора конфигурации 1С (--baseconf)."""
|
||
|
||
def __init__(self, message: str, *, exit_code: int = 2):
|
||
super().__init__(message)
|
||
self.exit_code = exit_code
|
||
|
||
|
||
def load_tool_config(path: Path) -> dict[str, Any]:
|
||
"""Файл настроек утилиты index (YAML; legacy JSON поддерживается)."""
|
||
ext = path.suffix.lower()
|
||
try:
|
||
raw = path.read_text(encoding="utf-8")
|
||
except OSError as e:
|
||
raise BaseconfError(f"Не удалось прочитать --config {path}: {e}", exit_code=1) from e
|
||
|
||
if ext == ".json":
|
||
try:
|
||
import json
|
||
|
||
data = json.loads(raw)
|
||
except Exception as e:
|
||
raise BaseconfError(f"Некорректный JSON в --config {path}: {e}", exit_code=1) from e
|
||
if not isinstance(data, dict):
|
||
raise BaseconfError(f"--config {path}: ожидается JSON-объект", exit_code=1)
|
||
return data
|
||
|
||
try:
|
||
import yaml # type: ignore
|
||
except Exception as e:
|
||
raise BaseconfError(
|
||
"Для чтения --config в формате YAML нужен пакет pyyaml: pip install pyyaml",
|
||
exit_code=1,
|
||
) from e
|
||
try:
|
||
data = yaml.safe_load(raw)
|
||
except Exception as e:
|
||
raise BaseconfError(f"Некорректный YAML в --config {path}: {e}", exit_code=1) from e
|
||
if not isinstance(data, dict):
|
||
raise BaseconfError(f"--config {path}: ожидается YAML-объект", exit_code=1)
|
||
return data
|
||
|
||
|
||
def discover_local_tool_config() -> Path | None:
|
||
"""Автопоиск локального конфига без --config."""
|
||
candidates = ("config.local.yml", "config.local.yaml", "config.local.json")
|
||
roots = [Path.cwd(), project_root_from_here()]
|
||
seen: set[Path] = set()
|
||
for root in roots:
|
||
for rel in (Path("tools/index"), Path(".")):
|
||
base = (root / rel).resolve()
|
||
if base in seen or not base.is_dir():
|
||
continue
|
||
seen.add(base)
|
||
for name in candidates:
|
||
p = base / name
|
||
if p.is_file():
|
||
return p
|
||
return None
|
||
|
||
|
||
def apply_tool_config(data: dict[str, Any]) -> None:
|
||
"""Применить настройки из YAML (--config): алиасы и дефолтные baseconf."""
|
||
from .config import set_config_aliases, set_default_baseconfs
|
||
|
||
aliases = data.get("baseconf_aliases")
|
||
if aliases is not None:
|
||
if not isinstance(aliases, dict):
|
||
raise BaseconfError("--config: baseconf_aliases должен быть объектом YAML")
|
||
set_config_aliases({str(k): str(v) for k, v in aliases.items()})
|
||
default_baseconfs = data.get("default_baseconfs")
|
||
if default_baseconfs is not None:
|
||
if not isinstance(default_baseconfs, list):
|
||
raise BaseconfError("--config: default_baseconfs должен быть YAML-списком")
|
||
set_default_baseconfs([str(x) for x in default_baseconfs])
|
||
else:
|
||
set_default_baseconfs([])
|
||
|
||
|
||
def resolve_project_root(args: argparse.Namespace) -> Path:
|
||
"""Корень проекта: --root → --config/auto-config → авто."""
|
||
if getattr(args, "root", None):
|
||
return Path(args.root).resolve()
|
||
cfg_path = getattr(args, "config", None)
|
||
if not cfg_path:
|
||
auto = discover_local_tool_config()
|
||
if auto is not None:
|
||
cfg_path = str(auto)
|
||
setattr(args, "config", cfg_path)
|
||
if cfg_path:
|
||
data = load_tool_config(Path(cfg_path))
|
||
apply_tool_config(data)
|
||
pr = data.get("project_root")
|
||
if pr:
|
||
return Path(pr).resolve()
|
||
return project_root_from_here()
|
||
|
||
|
||
def add_tool_args(parser: argparse.ArgumentParser) -> None:
|
||
"""Параметры утилиты (не конфигурация 1С)."""
|
||
parser.add_argument(
|
||
"--config",
|
||
"-c",
|
||
metavar="FILE",
|
||
help=(
|
||
"YAML с настройками утилиты (project_root и др.), см. config.example.yml. "
|
||
"Без параметра: автопоиск config.local.yml/.yaml/.json"
|
||
),
|
||
)
|
||
parser.add_argument(
|
||
"--root",
|
||
"-r",
|
||
metavar="DIR",
|
||
help="Корень проекта с выгрузками 1С (перекрывает project_root из --config)",
|
||
)
|
||
|
||
|
||
def add_baseconf_args(
|
||
parser: argparse.ArgumentParser,
|
||
*,
|
||
mode: str = "query",
|
||
) -> None:
|
||
"""Фильтр по выгрузке конфигурации 1С в репозитории."""
|
||
if mode == "index":
|
||
default_hint = (
|
||
"Без параметра — default_baseconfs из --config; "
|
||
"если блок не задан, индексировать все выгрузки с src/ и внешние обработки."
|
||
)
|
||
else:
|
||
default_hint = "Без параметра — поиск по всем проиндексированным."
|
||
parser.add_argument(
|
||
"--baseconf",
|
||
"-b",
|
||
action="append",
|
||
metavar="NAME",
|
||
dest="baseconf",
|
||
help=(
|
||
"Источник индексации 1С: <name>/src/ или virtual <dir>.<Обработка>. "
|
||
"Можно несколько раз или через запятую: -b cfg1,cfg2. "
|
||
"Краткие имена (алиасы) — в baseconf_aliases файла --config. "
|
||
f"{default_hint}"
|
||
),
|
||
)
|
||
|
||
|
||
def make_parent_parser(*, include_baseconf: bool = True, baseconf_mode: str = "query") -> argparse.ArgumentParser:
|
||
"""Общие опции для подкоманд (--config, --root, опционально --baseconf)."""
|
||
p = argparse.ArgumentParser(add_help=False)
|
||
add_tool_args(p)
|
||
if include_baseconf:
|
||
add_baseconf_args(p, mode=baseconf_mode)
|
||
return p
|
||
|
||
|
||
def extract_baseconf_argv(argv: list[str]) -> tuple[list[str], list[str]]:
|
||
"""
|
||
Извлечь все -b / --baseconf из argv (до или после подкоманды).
|
||
Возвращает (значения baseconf, argv без этих аргументов).
|
||
"""
|
||
baseconfs: list[str] = []
|
||
rest: list[str] = []
|
||
i = 0
|
||
while i < len(argv):
|
||
tok = argv[i]
|
||
if tok in ("-b", "--baseconf"):
|
||
if i + 1 >= len(argv):
|
||
rest.append(tok)
|
||
break
|
||
baseconfs.append(argv[i + 1])
|
||
i += 2
|
||
continue
|
||
if tok.startswith("--baseconf="):
|
||
baseconfs.append(tok.split("=", 1)[1])
|
||
i += 1
|
||
continue
|
||
if tok.startswith("-b") and len(tok) > 2:
|
||
baseconfs.append(tok[2:])
|
||
i += 1
|
||
continue
|
||
rest.append(tok)
|
||
i += 1
|
||
return baseconfs, rest
|
||
|
||
|
||
def merge_baseconf_args(args: argparse.Namespace, extra: list[str] | None) -> None:
|
||
"""Объединить baseconf из argv и argparse (in-place)."""
|
||
if not extra:
|
||
return
|
||
current = list(getattr(args, "baseconf", None) or [])
|
||
seen = set(current)
|
||
for item in extra:
|
||
if item not in seen:
|
||
seen.add(item)
|
||
current.append(item)
|
||
args.baseconf = current or None
|
||
|
||
|
||
def resolve_baseconf_names(raw: list[str] | None) -> list[str]:
|
||
"""Разрешить алиасы имён конфигураций 1С (без проверки индекса)."""
|
||
if not raw:
|
||
return []
|
||
out: list[str] = []
|
||
seen: set[str] = set()
|
||
for item in raw:
|
||
for part in item.replace(";", ",").split(","):
|
||
part = part.strip()
|
||
if not part:
|
||
continue
|
||
name = resolve_config_name(part)
|
||
if name not in seen:
|
||
seen.add(name)
|
||
out.append(name)
|
||
return out
|
||
|
||
|
||
def validate_baseconfs(
|
||
project_root: Path,
|
||
names: list[str],
|
||
*,
|
||
conn=None,
|
||
require_indexed: bool = True,
|
||
) -> list[str]:
|
||
"""
|
||
Проверить --baseconf: src-выгрузка или virtual внешняя обработка, наличие в индексе.
|
||
Возвращает канонические имена. Иначе BaseconfError с подсказками.
|
||
"""
|
||
if not names:
|
||
return []
|
||
|
||
resolved = resolve_baseconf_names(names)
|
||
available = {c.name for c in discover_configs(project_root, None)}
|
||
missing_src: list[str] = []
|
||
for name in resolved:
|
||
if discover_configs(project_root, [name]):
|
||
continue
|
||
if name not in available:
|
||
missing_src.append(name)
|
||
|
||
if missing_src:
|
||
available_list = sorted(available)
|
||
lines = [
|
||
"Источник индексации 1С не найден в репозитории "
|
||
"(нет <name>/src/ или внешней обработки <dir>/<name>.xml + <dir>/<name>/):",
|
||
*(f" - {n}" for n in missing_src),
|
||
"",
|
||
"Доступные выгрузки в проекте:",
|
||
]
|
||
if available_list:
|
||
lines.extend(f" - {n}" for n in available_list)
|
||
else:
|
||
lines.append(" (нет)")
|
||
lines.extend(
|
||
[
|
||
"",
|
||
"Краткие имена (алиасы) для -b задаются в baseconf_aliases файла --config.",
|
||
]
|
||
)
|
||
raise BaseconfError("\n".join(lines))
|
||
|
||
if not require_indexed or conn is None:
|
||
return resolved
|
||
|
||
indexed = {c["name"] for c in dbmod.list_configs(conn)}
|
||
not_indexed = [n for n in resolved if n not in indexed]
|
||
if not_indexed:
|
||
lines = [
|
||
"Конфигурация 1С есть в репозитории, но отсутствует в индексе cache/1c_meta/index.sqlite:",
|
||
*(f" - {n}" for n in not_indexed),
|
||
"",
|
||
"Проиндексировано:",
|
||
]
|
||
if indexed:
|
||
lines.extend(f" - {n}" for n in sorted(indexed))
|
||
else:
|
||
lines.append(" (индекс пуст)")
|
||
lines.extend(
|
||
[
|
||
"",
|
||
f"Переиндексация: python index_1c.py reindex --full -b {not_indexed[0]}",
|
||
]
|
||
)
|
||
raise BaseconfError("\n".join(lines))
|
||
|
||
return resolved
|
||
|
||
|
||
def get_baseconfs_for_query(
|
||
args: argparse.Namespace,
|
||
conn,
|
||
project_root: Path,
|
||
) -> list[str] | None:
|
||
"""None = все конфигурации в индексе; иначе список имён."""
|
||
raw = getattr(args, "baseconf", None)
|
||
if not raw:
|
||
return None
|
||
return validate_baseconfs(project_root, raw, conn=conn, require_indexed=True)
|
||
|
||
|
||
def format_not_found_object(
|
||
name: str,
|
||
*,
|
||
baseconfs: list[str] | None,
|
||
conn,
|
||
) -> str:
|
||
"""Подсказка, если объект не найден."""
|
||
lines = [f"Объект «{name}» не найден."]
|
||
if baseconfs:
|
||
lines.append(f"Ограничение --baseconf: {', '.join(baseconfs)}")
|
||
|
||
# похожие имена
|
||
like = f"%{name}%"
|
||
params: list[Any] = [like, like, like]
|
||
sql = """
|
||
SELECT config, full_name, synonym FROM objects
|
||
WHERE name LIKE ? OR full_name LIKE ? OR synonym LIKE ?
|
||
"""
|
||
if baseconfs:
|
||
sql += f" AND config IN ({','.join('?' * len(baseconfs))})"
|
||
params.extend(baseconfs)
|
||
sql += " ORDER BY config, full_name LIMIT 15"
|
||
rows = conn.execute(sql, params).fetchall()
|
||
if rows:
|
||
lines.append("")
|
||
lines.append("Возможно, имелось в виду:")
|
||
for r in rows:
|
||
syn = f" — {r['synonym']}" if r["synonym"] else ""
|
||
lines.append(f" [{r['config']}] {r['full_name']}{syn}")
|
||
else:
|
||
indexed = [c["name"] for c in dbmod.list_configs(conn)]
|
||
lines.append("")
|
||
lines.append("В индексе конфигурации: " + (", ".join(indexed) if indexed else "(пусто)"))
|
||
lines.append("Проверьте имя (Document.ЗаказКлиента) или выполните: query_1c.py search \"…\"")
|
||
return "\n".join(lines)
|
||
|
||
|
||
def format_not_found_modules(
|
||
query: str,
|
||
*,
|
||
baseconfs: list[str] | None,
|
||
conn,
|
||
) -> str:
|
||
lines = [f"Модули по запросу «{query}» не найдены."]
|
||
if baseconfs:
|
||
lines.append(f"Ограничение --baseconf: {', '.join(baseconfs)}")
|
||
indexed = [c["name"] for c in dbmod.list_configs(conn)]
|
||
lines.append("Проиндексировано: " + ", ".join(indexed) if indexed else "Индекс пуст.")
|
||
return "\n".join(lines)
|
||
|
||
|
||
def handle_baseconf_error(e: BaseconfError) -> int:
|
||
print(str(e), file=sys.stderr)
|
||
return e.exit_code
|
||
|
||
|
||
def index_db_missing_message(project_root: Path) -> str:
|
||
path = dbmod.global_db_path(cache_dir(project_root))
|
||
return (
|
||
f"Индекс не найден: {path}\n"
|
||
"Сначала выполните:\n"
|
||
" python index_1c.py reindex --full -j 12\n"
|
||
"или для одной конфигурации:\n"
|
||
" python index_1c.py reindex --full -b <имя-выгрузки>"
|
||
)
|