1c/index
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Детализирована документация. Небольшие изменения имен параметров.

Browse Source
This commit is contained in:
mihailkudravcev
2026-07-16 15:38:16 +03:00
parent 7f9d198f79
commit 24200b401a
22 changed files with 2986 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files
index1c/__init__.py
+3
View File
@@ -0,0 +1,3 @@
"""Индексация метаданных конфигураций 1С (выгрузки EDT/XML) для быстрого поиска."""
__version__ = "0.1.0"
index1c/__pycache__/__init__.cpython-311.pyc
BIN
View File
Binary file not shown.
index1c/__pycache__/build.cpython-311.pyc
BIN
View File
Binary file not shown.
index1c/__pycache__/cli.cpython-311.pyc
BIN
View File
Binary file not shown.
index1c/__pycache__/config.cpython-311.pyc
BIN
View File
Binary file not shown.
index1c/__pycache__/db.cpython-311.pyc
BIN
View File
Binary file not shown.
index1c/__pycache__/parse_meta.cpython-311.pyc
BIN
View File
Binary file not shown.
index1c/__pycache__/parse_modules.cpython-311.pyc
BIN
View File
Binary file not shown.
index1c/__pycache__/search.cpython-311.pyc
BIN
View File
Binary file not shown.
index1c/__pycache__/writers.cpython-311.pyc
BIN
View File
Binary file not shown.
index1c/build.py
+349
View File
@@ -0,0 +1,349 @@
"""Построение единого индекса проекта (метаданные + модули, full/incremental)."""
from __future__ import annotations
import json
import os
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from pathlib import Path
from typing import Any
from .config import DEFAULT_SKIP_TYPES, OBJECT_TYPE_FOLDERS, ConfigPaths, cache_dir
from . import db as dbmod
from .parse_meta import FieldRec, ObjectRec, iter_object_xml_files, parse_metadata_xml
from .parse_modules import ModuleRec, iter_bsl_files, parse_bsl_file
from . import writers
def default_workers() -> int:
return max(4, min(16, (os.cpu_count() or 4)))
def _file_stamp(path: Path) -> tuple[int, int]:
st = path.stat()
mtime_ns = getattr(st, "st_mtime_ns", int(st.st_mtime * 1e9))
return mtime_ns, st.st_size
def _parse_meta_job(args: tuple[str, str, str]) -> tuple[str, ObjectRec | None, tuple[int, int]]:
config, otype, path_s = args
path = Path(path_s)
stamp = _file_stamp(path)
rec = parse_metadata_xml(path, config=config, object_type=otype)
return path_s, rec, stamp
def _parse_module_job(args: tuple[str, str, str]) -> ModuleRec | None:
config, src_s, path_s = args
return parse_bsl_file(Path(path_s), config=config, src_root=Path(src_s))
def _fields_text(obj: ObjectRec) -> str:
return " ".join(
" ".join(
filter(
None,
[
f.kind,
f.name,
f.synonym,
f.comment,
f.tabular_section,
" ".join(f.types),
],
)
)
for f in obj.fields
)
def _run_pool(jobs: list, worker, workers: int) -> list:
if not jobs:
return []
if workers <= 1 or len(jobs) < 8:
return [worker(j) for j in jobs]
out: list = []
with ThreadPoolExecutor(max_workers=workers) as pool:
futs = [pool.submit(worker, j) for j in jobs]
for fut in as_completed(futs):
try:
out.append(fut.result())
except Exception:
out.append(None)
return out
def build_project_index(
configs: list[ConfigPaths],
project_root: Path,
*,
full: bool = False,
wipe_db: bool = False,
types: set[str] | None = None,
skip_types: set[str] | None = None,
index_modules: bool = True,
workers: int | None = None,
write_md: bool = False,
) -> dict[str, Any]:
"""Индексирует конфигурации в единый cache/1c_meta/index.sqlite.
full=True — для каждой указанной конфигурации удалить её данные и пересобрать.
wipe_db=True — удалить весь index.sqlite и создать заново (обычно вместе с full
по всем конфигам).
Иначе — инкремент по mtime/size файлов.
"""
workers = workers if workers is not None else default_workers()
skip = set(skip_types if skip_types is not None else DEFAULT_SKIP_TYPES)
cache_root = cache_dir(project_root)
cache_root.mkdir(parents=True, exist_ok=True)
db_path = dbmod.global_db_path(cache_root)
t0 = time.perf_counter()
if wipe_db or not db_path.exists():
conn = dbmod.reset_db(db_path)
mode = "full"
force_config_full = True
else:
try:
conn = dbmod.ensure_db(db_path)
mode = "full" if full else "incremental"
force_config_full = full
except RuntimeError:
conn = dbmod.reset_db(db_path)
mode = "full"
force_config_full = True
summary: dict[str, Any] = {
"mode": mode,
"db": str(db_path),
"workers": workers,
"configs": [],
"meta_parsed": 0,
"meta_skipped": 0,
"modules_parsed": 0,
"modules_skipped": 0,
"files_removed": 0,
"errors": 0,
}
try:
for cfg in configs:
cfg_stats = _index_one_config(
conn,
cfg,
full=force_config_full,
types=types,
skip_types=skip,
index_modules=index_modules,
workers=workers,
summary=summary,
)
summary["configs"].append(cfg_stats)
if write_md:
_export_md_for_config(conn, cfg, cache_root)
dbmod.set_meta(conn, "last_reindex_at", dbmod.utcnow())
dbmod.set_meta(conn, "last_reindex_mode", mode)
conn.commit()
finally:
conn.close()
summary["elapsed_sec"] = round(time.perf_counter() - t0, 2)
writers.write_global_index_md(cache_root, summary)
return summary
def _index_one_config(
conn,
cfg: ConfigPaths,
*,
full: bool,
types: set[str] | None,
skip_types: set[str],
index_modules: bool,
workers: int,
summary: dict[str, Any],
) -> dict[str, Any]:
if full:
dbmod.delete_config(conn, cfg.name)
existing = dbmod.get_file_index(conn, cfg.name) if not full else {}
seen_paths: set[str] = set()
meta_files = iter_object_xml_files(cfg.src, types=types, skip_types=skip_types)
meta_jobs: list[tuple[str, str, str]] = []
meta_skipped = 0
for otype, path in meta_files:
rel = path.relative_to(cfg.src).as_posix()
seen_paths.add(rel)
mtime_ns, size = _file_stamp(path)
prev = existing.get(rel)
if (
prev
and prev["mtime_ns"] == mtime_ns
and prev["size"] == size
and prev["kind"] == "meta_xml"
):
meta_skipped += 1
continue
if prev:
dbmod.delete_file_cascade(conn, prev["id"])
meta_jobs.append((cfg.name, otype, str(path)))
meta_results = _run_pool(meta_jobs, _parse_meta_job, workers)
meta_parsed = 0
for item in meta_results:
if not item:
summary["errors"] += 1
continue
path_s, rec, stamp = item
if rec is None:
summary["errors"] += 1
continue
path = Path(path_s)
rel = path.relative_to(cfg.src).as_posix()
file_id = dbmod.upsert_file(
conn,
config=cfg.name,
rel_path=rel,
kind="meta_xml",
mtime_ns=stamp[0],
size=stamp[1],
object_type=rec.object_type,
)
dbmod.upsert_object(
conn, rec.to_dict(), file_id=file_id, fields_text=_fields_text(rec)
)
meta_parsed += 1
mod_parsed = 0
mod_skipped = 0
if index_modules:
mod_jobs: list[tuple[str, str, str]] = []
for path in iter_bsl_files(cfg.src):
rel = path.relative_to(cfg.src).as_posix()
seen_paths.add(rel)
mtime_ns, size = _file_stamp(path)
prev = existing.get(rel)
if (
prev
and prev["mtime_ns"] == mtime_ns
and prev["size"] == size
and prev["kind"] == "module_bsl"
):
mod_skipped += 1
continue
if prev:
dbmod.delete_file_cascade(conn, prev["id"])
mod_jobs.append((cfg.name, str(cfg.src), str(path)))
mod_results = _run_pool(mod_jobs, _parse_module_job, workers)
for rec in mod_results:
if rec is None:
summary["errors"] += 1
continue
file_id = dbmod.upsert_file(
conn,
config=cfg.name,
rel_path=rec.rel_path,
kind="module_bsl",
mtime_ns=rec.mtime_ns,
size=rec.size,
)
dbmod.upsert_module(
conn,
config=cfg.name,
rel_path=rec.rel_path,
owner=rec.owner,
module_kind=rec.module_kind,
symbols=rec.symbols,
body=rec.body,
line_count=rec.line_count,
size=rec.size,
file_id=file_id,
)
mod_parsed += 1
removed = 0
if not full:
for rel, info in existing.items():
if rel in seen_paths:
continue
if types is not None and info["kind"] == "meta_xml":
folder = rel.split("/", 1)[0]
otype = OBJECT_TYPE_FOLDERS.get(folder)
if otype and otype not in types:
continue
if not index_modules and info["kind"] == "module_bsl":
continue
dbmod.delete_file_cascade(conn, info["id"])
removed += 1
stats = dbmod.refresh_config_stats(conn, cfg.name, str(cfg.src))
summary["meta_parsed"] += meta_parsed
summary["meta_skipped"] += meta_skipped
summary["modules_parsed"] += mod_parsed
summary["modules_skipped"] += mod_skipped
summary["files_removed"] += removed
print(
f"→ {cfg.name}: meta +{meta_parsed}/skip {meta_skipped}, "
f"modules +{mod_parsed}/skip {mod_skipped}, removed {removed}, "
f"total objects={stats['objects']} modules={stats['modules']}"
)
return stats
def _export_md_for_config(conn, cfg: ConfigPaths, cache_root: Path) -> None:
rows = conn.execute(
"SELECT json FROM objects WHERE config=? ORDER BY object_type, name",
(cfg.name,),
).fetchall()
objects: list[ObjectRec] = []
for r in rows:
d = json.loads(r["json"])
fields = [FieldRec(**f) for f in d.get("fields") or []]
objects.append(
ObjectRec(
config=d["config"],
object_type=d["object_type"],
name=d["name"],
synonym=d.get("synonym") or "",
comment=d.get("comment") or "",
tooltip=d.get("tooltip") or "",
explanation=d.get("explanation") or "",
uuid=d.get("uuid") or "",
path=d.get("path") or "",
fields=fields,
owners=d.get("owners") or [],
register_records=d.get("register_records") or [],
based_on=d.get("based_on") or [],
input_by_string=d.get("input_by_string") or [],
)
)
out = cache_root / "md" / cfg.name
out.mkdir(parents=True, exist_ok=True)
writers.write_objects_markdown(out / "objects.md", objects, title=f"Метаданные: {cfg.name}")
writers.write_refs_markdown(out / "refs.md", objects, title=f"Ссылки: {cfg.name}")
writers.write_type_summaries(out, objects)
def build_config_index(*args, **kwargs):
raise RuntimeError(
"Устарело: используйте build_project_index() и единый cache/1c_meta/index.sqlite"
)
def write_global_index(project_root: Path, manifests: list[dict[str, Any]]) -> Path:
cache_root = cache_dir(project_root)
summary = {
"mode": "legacy",
"db": str(dbmod.global_db_path(cache_root)),
"configs": manifests,
"elapsed_sec": 0,
"meta_parsed": 0,
"modules_parsed": 0,
}
return writers.write_global_index_md(cache_root, summary)
index1c/cli.py
+242
View File
@@ -0,0 +1,242 @@
"""Общие аргументы CLI: --config (скрипт), --root, --baseconf (конфигурация 1С)."""
from __future__ import annotations
import argparse
import json
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 (JSON). Не путать с конфигурацией 1С."""
try:
data = json.loads(path.read_text(encoding="utf-8"))
except OSError as e:
raise BaseconfError(f"Не удалось прочитать --config {path}: {e}", exit_code=1) from e
except json.JSONDecodeError 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
def resolve_project_root(args: argparse.Namespace) -> Path:
"""Корень проекта: --root → project_root из --config → авто."""
if getattr(args, "root", None):
return Path(args.root).resolve()
cfg_path = getattr(args, "config", None)
if cfg_path:
data = load_tool_config(Path(cfg_path))
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="JSON с настройками утилиты (project_root и др.), см. config.example.json",
)
parser.add_argument(
"--root",
"-r",
metavar="DIR",
help="Корень проекта CRM3-26 (перекрывает project_root из --config)",
)
def add_baseconf_args(parser: argparse.ArgumentParser) -> None:
"""Фильтр по выгрузке конфигурации 1С в репозитории."""
parser.add_argument(
"--baseconf",
"-b",
action="append",
metavar="NAME",
dest="baseconf",
help=(
"Конфигурация 1С в репозитории (папка <name>/src/). "
"Можно несколько раз. Алиасы: target→crm3-26, source→crm3-dev. "
"Без параметра — поиск по всем проиндексированным."
),
)
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/, наличие в индексе.
Возвращает канонические имена. Иначе BaseconfError с подсказками.
"""
if not names:
return []
resolved = resolve_baseconf_names(names)
missing_src: list[str] = []
for name in resolved:
if not (project_root / name / "src").is_dir():
missing_src.append(name)
if missing_src:
available = [c.name for c in discover_configs(project_root, None)]
lines = [
"Конфигурация 1С не найдена в репозитории (нет каталога <name>/src/):",
*(f" - {n}" for n in missing_src),
"",
"Доступные выгрузки в проекте:",
]
if available:
lines.extend(f" - {n}" for n in available)
else:
lines.append(" (нет)")
lines.extend(
[
"",
"Алиасы: target→crm3-26, source→crm3-dev, crm3_old→crm3-dev",
]
)
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 tools/index/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 tools/index/index_1c.py reindex --full -j 12\n"
"или для одной конфигурации:\n"
" python tools/index/index_1c.py reindex --full -b crm3-26"
)
index1c/config.py
+231
View File
@@ -0,0 +1,231 @@
"""Каталоги конфигураций проекта и типы объектов метаданных 1С."""
from __future__ import annotations
from dataclasses import dataclass
from pathlib import Path
# Папки выгрузок относительно корня проекта (есть src/).
DEFAULT_CONFIGS: tuple[str, ...] = (
"crm3-26",
"crm3-26.rhana",
"crm3-dev",
"crm3-dev.rhana",
"crm3-dev.docs",
"ws-rhana",
"bu-corp",
"bu-corp.rhana",
"diadoc.ext.rhana",
)
# Алиасы имён конфигураций 1С для CLI (--baseconf / -b).
CONFIG_ALIASES: dict[str, str] = {
"crm3_26": "crm3-26",
"crm3_old": "crm3-dev",
"crm3-old": "crm3-dev",
"crm3_dev": "crm3-dev",
"target": "crm3-26",
"source": "crm3-dev",
}
# Папка EDT → канонический тип (английский идентификатор выгрузки).
OBJECT_TYPE_FOLDERS: dict[str, str] = {
"AccumulationRegisters": "AccumulationRegister",
"BusinessProcesses": "BusinessProcess",
"Catalogs": "Catalog",
"ChartsOfAccounts": "ChartOfAccounts",
"ChartsOfCalculationTypes": "ChartOfCalculationTypes",
"ChartsOfCharacteristicTypes": "ChartOfCharacteristicTypes",
"CommandGroups": "CommandGroup",
"CommonAttributes": "CommonAttribute",
"CommonCommands": "CommonCommand",
"CommonForms": "CommonForm",
"CommonModules": "CommonModule",
"CommonPictures": "CommonPicture",
"CommonTemplates": "CommonTemplate",
"Constants": "Constant",
"DataProcessors": "DataProcessor",
"DefinedTypes": "DefinedType",
"DocumentJournals": "DocumentJournal",
"DocumentNumerators": "DocumentNumerator",
"Documents": "Document",
"Enums": "Enum",
"EventSubscriptions": "EventSubscription",
"ExchangePlans": "ExchangePlan",
"FilterCriteria": "FilterCriterion",
"FunctionalOptions": "FunctionalOption",
"FunctionalOptionsParameters": "FunctionalOptionsParameter",
"HTTPServices": "HTTPService",
"InformationRegisters": "InformationRegister",
"IntegrationServices": "IntegrationService",
"Reports": "Report",
"Roles": "Role",
"ScheduledJobs": "ScheduledJob",
"SessionParameters": "SessionParameter",
"SettingsStorages": "SettingsStorage",
"StyleItems": "StyleItem",
"Subsystems": "Subsystem",
"Tasks": "Task",
"WebServices": "WebService",
"WSReferences": "WSReference",
"XDTOPackages": "XDTOPackage",
"Bots": "Bot",
"ExternalDataSources": "ExternalDataSource",
"Languages": "Language",
}
# Алиасы типов для --types (русский / короткий / мн.ч.).
TYPE_ALIASES: dict[str, str] = {
"catalog": "Catalog",
"catalogs": "Catalog",
"справочник": "Catalog",
"справочники": "Catalog",
"document": "Document",
"documents": "Document",
"документ": "Document",
"документы": "Document",
"informationregister": "InformationRegister",
"informationregisters": "InformationRegister",
"регистрсведений": "InformationRegister",
"регистрысведений": "InformationRegister",
"рс": "InformationRegister",
"accumulationregister": "AccumulationRegister",
"accumulationregisters": "AccumulationRegister",
"регистрнакопления": "AccumulationRegister",
"регистрынакопления": "AccumulationRegister",
"рн": "AccumulationRegister",
"constant": "Constant",
"constants": "Constant",
"константа": "Constant",
"константы": "Constant",
"enum": "Enum",
"enums": "Enum",
"перечисление": "Enum",
"перечисления": "Enum",
"commonmodule": "CommonModule",
"commonmodules": "CommonModule",
"общиймодуль": "CommonModule",
"общиемодули": "CommonModule",
"dataprocessor": "DataProcessor",
"dataprocessors": "DataProcessor",
"обработка": "DataProcessor",
"обработки": "DataProcessor",
"report": "Report",
"reports": "Report",
"отчет": "Report",
"отчеты": "Report",
"exchangeplan": "ExchangePlan",
"exchangeplans": "ExchangePlan",
"планыобмена": "ExchangePlan",
"definedtype": "DefinedType",
"definedtypes": "DefinedType",
"определяемыетипы": "DefinedType",
"chartofcharacteristictypes": "ChartOfCharacteristicTypes",
"пвх": "ChartOfCharacteristicTypes",
"role": "Role",
"roles": "Role",
"роль": "Role",
"роли": "Role",
"subsystem": "Subsystem",
"subsystems": "Subsystem",
"подсистема": "Subsystem",
"подсистемы": "Subsystem",
"httpservice": "HTTPService",
"httpservices": "HTTPService",
"documentjournal": "DocumentJournal",
"documentjournals": "DocumentJournal",
"журналыдокументов": "DocumentJournal",
"task": "Task",
"tasks": "Task",
"задача": "Task",
"задачи": "Task",
"businessprocess": "BusinessProcess",
"businessprocesses": "BusinessProcess",
"бизнеспроцесс": "BusinessProcess",
"бизнеспроцессы": "BusinessProcess",
"модуль": "CommonModule",
"модули": "CommonModule",
"bsl": "CommonModule",
}
# Типы с разбором реквизитов / измерений / ссылок.
FIELD_RICH_TYPES: frozenset[str] = frozenset(
{
"Catalog",
"Document",
"InformationRegister",
"AccumulationRegister",
"ChartOfCharacteristicTypes",
"ChartOfAccounts",
"ChartOfCalculationTypes",
"ExchangePlan",
"BusinessProcess",
"Task",
"DocumentJournal",
"Constant",
"DefinedType",
"CommonAttribute",
"Enum",
}
)
# Типы, которые по умолчанию пропускаем (шум / картинки).
DEFAULT_SKIP_TYPES: frozenset[str] = frozenset(
{
"CommonPicture",
"StyleItem",
"XDTOPackage",
"CommonTemplate",
"Language",
"Bot",
}
)
FOLDER_BY_TYPE: dict[str, str] = {v: k for k, v in OBJECT_TYPE_FOLDERS.items()}
@dataclass(frozen=True)
class ConfigPaths:
name: str
root: Path
src: Path
def project_root_from_here() -> Path:
"""tools/index/index1c → корень проекта."""
return Path(__file__).resolve().parents[3]
def resolve_config_name(name: str) -> str:
key = name.strip()
return CONFIG_ALIASES.get(key, CONFIG_ALIASES.get(key.lower(), key))
def resolve_object_type(name: str) -> str:
raw = name.strip()
if raw in FOLDER_BY_TYPE:
return raw
if raw in OBJECT_TYPE_FOLDERS:
return OBJECT_TYPE_FOLDERS[raw]
compact = raw.replace(" ", "").replace("_", "").replace("-", "").casefold()
if compact in TYPE_ALIASES:
return TYPE_ALIASES[compact]
for folder, otype in OBJECT_TYPE_FOLDERS.items():
if folder.casefold() == raw.casefold() or otype.casefold() == raw.casefold():
return otype
raise ValueError(f"Неизвестный тип объекта метаданных: {name!r}")
def discover_configs(project_root: Path, names: list[str] | None = None) -> list[ConfigPaths]:
wanted = [resolve_config_name(n) for n in names] if names else list(DEFAULT_CONFIGS)
found: list[ConfigPaths] = []
for name in wanted:
root = project_root / name
src = root / "src"
if src.is_dir():
found.append(ConfigPaths(name=name, root=root, src=src))
return found
def cache_dir(project_root: Path) -> Path:
return project_root / "cache" / "1c_meta"
index1c/db.py
+520
View File
@@ -0,0 +1,520 @@
"""Единый SQLite-индекс проекта (все конфигурации 1С)."""
from __future__ import annotations
import json
import sqlite3
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Iterable
SCHEMA_VERSION = 2
DDL = """
PRAGMA journal_mode=WAL;
PRAGMA synchronous=NORMAL;
CREATE TABLE IF NOT EXISTS meta (
key TEXT PRIMARY KEY,
value TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS configs (
name TEXT PRIMARY KEY,
src TEXT NOT NULL,
objects INTEGER NOT NULL DEFAULT 0,
fields INTEGER NOT NULL DEFAULT 0,
refs INTEGER NOT NULL DEFAULT 0,
modules INTEGER NOT NULL DEFAULT 0,
updated_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS files (
id INTEGER PRIMARY KEY,
config TEXT NOT NULL,
rel_path TEXT NOT NULL,
kind TEXT NOT NULL, -- meta_xml | module_bsl
mtime_ns INTEGER NOT NULL,
size INTEGER NOT NULL,
sha1 TEXT,
object_type TEXT,
UNIQUE(config, rel_path)
);
CREATE TABLE IF NOT EXISTS objects (
id INTEGER PRIMARY KEY,
config TEXT NOT NULL,
object_type TEXT NOT NULL,
name TEXT NOT NULL,
full_name TEXT NOT NULL,
synonym TEXT,
comment TEXT,
tooltip TEXT,
explanation TEXT,
uuid TEXT,
path TEXT,
file_id INTEGER,
json TEXT NOT NULL,
UNIQUE(config, full_name),
FOREIGN KEY(file_id) REFERENCES files(id) ON DELETE CASCADE
);
CREATE TABLE IF NOT EXISTS fields (
id INTEGER PRIMARY KEY,
object_id INTEGER NOT NULL,
config TEXT NOT NULL,
kind TEXT,
name TEXT,
synonym TEXT,
comment TEXT,
types TEXT,
tabular_section TEXT,
refs_json TEXT,
FOREIGN KEY(object_id) REFERENCES objects(id) ON DELETE CASCADE
);
CREATE TABLE IF NOT EXISTS refs (
id INTEGER PRIMARY KEY,
config TEXT NOT NULL,
from_full_name TEXT NOT NULL,
from_field TEXT NOT NULL,
to_kind TEXT NOT NULL,
to_name TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS modules (
id INTEGER PRIMARY KEY,
config TEXT NOT NULL,
rel_path TEXT NOT NULL,
owner TEXT,
module_kind TEXT,
symbols TEXT,
line_count INTEGER,
size INTEGER,
file_id INTEGER,
UNIQUE(config, rel_path),
FOREIGN KEY(file_id) REFERENCES files(id) ON DELETE CASCADE
);
CREATE VIRTUAL TABLE IF NOT EXISTS objects_fts USING fts5(
full_name,
synonym,
comment,
fields_text,
object_id UNINDEXED,
config UNINDEXED,
tokenize='unicode61'
);
CREATE VIRTUAL TABLE IF NOT EXISTS modules_fts USING fts5(
rel_path,
owner,
module_kind,
symbols,
body,
module_id UNINDEXED,
config UNINDEXED,
tokenize='unicode61'
);
CREATE INDEX IF NOT EXISTS idx_objects_config_type ON objects(config, object_type);
CREATE INDEX IF NOT EXISTS idx_objects_name ON objects(name);
CREATE INDEX IF NOT EXISTS idx_fields_object ON fields(object_id);
CREATE INDEX IF NOT EXISTS idx_refs_to ON refs(to_kind, to_name);
CREATE INDEX IF NOT EXISTS idx_refs_from ON refs(from_full_name);
CREATE INDEX IF NOT EXISTS idx_refs_config ON refs(config);
CREATE INDEX IF NOT EXISTS idx_modules_config ON modules(config);
CREATE INDEX IF NOT EXISTS idx_modules_owner ON modules(owner);
CREATE INDEX IF NOT EXISTS idx_files_config ON files(config);
"""
def utcnow() -> str:
return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
def global_db_path(cache_root: Path) -> Path:
return cache_root / "index.sqlite"
def connect(db_path: Path, *, readonly: bool = False) -> sqlite3.Connection:
db_path.parent.mkdir(parents=True, exist_ok=True)
if readonly and db_path.is_file():
uri = f"file:{db_path.resolve()}?mode=ro"
conn = sqlite3.connect(uri, uri=True)
else:
conn = sqlite3.connect(db_path)
conn.row_factory = sqlite3.Row
conn.execute("PRAGMA foreign_keys=ON")
return conn
def init_db(conn: sqlite3.Connection) -> None:
conn.executescript(DDL)
row = conn.execute("SELECT value FROM meta WHERE key='schema_version'").fetchone()
if row is None:
conn.execute(
"INSERT INTO meta(key, value) VALUES('schema_version', ?)",
(str(SCHEMA_VERSION),),
)
conn.execute(
"INSERT OR REPLACE INTO meta(key, value) VALUES('created_at', ?)",
(utcnow(),),
)
elif int(row["value"]) != SCHEMA_VERSION:
raise RuntimeError(
f"Несовместимая схема индекса (есть {row['value']}, нужна {SCHEMA_VERSION}). "
"Запустите: python tools/index/index_1c.py reindex --full"
)
conn.commit()
def reset_db(db_path: Path) -> sqlite3.Connection:
if db_path.exists():
db_path.unlink()
for suffix in ("-wal", "-shm"):
p = Path(str(db_path) + suffix)
if p.exists():
p.unlink()
conn = connect(db_path)
init_db(conn)
return conn
def ensure_db(db_path: Path) -> sqlite3.Connection:
conn = connect(db_path)
try:
init_db(conn)
except RuntimeError:
conn.close()
raise
return conn
def get_file_index(conn: sqlite3.Connection, config: str | None = None) -> dict[str, dict[str, Any]]:
"""rel_path → {mtime_ns, size, id, kind} для инкремента."""
if config:
rows = conn.execute(
"SELECT id, rel_path, kind, mtime_ns, size FROM files WHERE config=?",
(config,),
).fetchall()
else:
rows = conn.execute(
"SELECT id, config, rel_path, kind, mtime_ns, size FROM files"
).fetchall()
out: dict[str, dict[str, Any]] = {}
for r in rows:
key = r["rel_path"] if config else f"{r['config']}:{r['rel_path']}"
out[key] = {
"id": r["id"],
"kind": r["kind"],
"mtime_ns": r["mtime_ns"],
"size": r["size"],
}
return out
def delete_file_cascade(conn: sqlite3.Connection, file_id: int) -> None:
"""Удалить файл и связанные объекты/модули/FTS."""
obj = conn.execute(
"SELECT id FROM objects WHERE file_id=?", (file_id,)
).fetchone()
if obj:
oid = obj["id"]
conn.execute("DELETE FROM objects_fts WHERE object_id=?", (oid,))
conn.execute("DELETE FROM fields WHERE object_id=?", (oid,))
row = conn.execute(
"SELECT config, full_name FROM objects WHERE id=?", (oid,)
).fetchone()
if row:
conn.execute(
"DELETE FROM refs WHERE config=? AND from_full_name=?",
(row["config"], row["full_name"]),
)
conn.execute("DELETE FROM objects WHERE id=?", (oid,))
mod = conn.execute(
"SELECT id FROM modules WHERE file_id=?", (file_id,)
).fetchone()
if mod:
mid = mod["id"]
conn.execute("DELETE FROM modules_fts WHERE module_id=?", (mid,))
conn.execute("DELETE FROM modules WHERE id=?", (mid,))
conn.execute("DELETE FROM files WHERE id=?", (file_id,))
def delete_config(conn: sqlite3.Connection, config: str) -> None:
oids = [
r["id"]
for r in conn.execute("SELECT id FROM objects WHERE config=?", (config,))
]
for oid in oids:
conn.execute("DELETE FROM objects_fts WHERE object_id=?", (oid,))
mids = [
r["id"]
for r in conn.execute("SELECT id FROM modules WHERE config=?", (config,))
]
for mid in mids:
conn.execute("DELETE FROM modules_fts WHERE module_id=?", (mid,))
conn.execute("DELETE FROM fields WHERE config=?", (config,))
conn.execute("DELETE FROM refs WHERE config=?", (config,))
conn.execute("DELETE FROM objects WHERE config=?", (config,))
conn.execute("DELETE FROM modules WHERE config=?", (config,))
conn.execute("DELETE FROM files WHERE config=?", (config,))
conn.execute("DELETE FROM configs WHERE name=?", (config,))
def upsert_file(
conn: sqlite3.Connection,
*,
config: str,
rel_path: str,
kind: str,
mtime_ns: int,
size: int,
object_type: str | None = None,
sha1: str | None = None,
) -> int:
conn.execute(
"""
INSERT INTO files(config, rel_path, kind, mtime_ns, size, sha1, object_type)
VALUES(?,?,?,?,?,?,?)
ON CONFLICT(config, rel_path) DO UPDATE SET
kind=excluded.kind,
mtime_ns=excluded.mtime_ns,
size=excluded.size,
sha1=excluded.sha1,
object_type=excluded.object_type
""",
(config, rel_path, kind, mtime_ns, size, sha1, object_type),
)
row = conn.execute(
"SELECT id FROM files WHERE config=? AND rel_path=?",
(config, rel_path),
).fetchone()
assert row is not None
return int(row["id"])
def upsert_object(
conn: sqlite3.Connection,
obj: dict[str, Any],
*,
file_id: int,
fields_text: str,
) -> int:
config = obj["config"]
full_name = obj["full_name"]
old = conn.execute(
"SELECT id FROM objects WHERE config=? AND full_name=?",
(config, full_name),
).fetchone()
if old:
oid = int(old["id"])
conn.execute("DELETE FROM objects_fts WHERE object_id=?", (oid,))
conn.execute("DELETE FROM fields WHERE object_id=?", (oid,))
conn.execute(
"DELETE FROM refs WHERE config=? AND from_full_name=?",
(config, full_name),
)
conn.execute(
"""
UPDATE objects SET object_type=?, name=?, synonym=?, comment=?, tooltip=?,
explanation=?, uuid=?, path=?, file_id=?, json=?
WHERE id=?
""",
(
obj["object_type"],
obj["name"],
obj.get("synonym"),
obj.get("comment"),
obj.get("tooltip"),
obj.get("explanation"),
obj.get("uuid"),
obj.get("path"),
file_id,
json.dumps(obj, ensure_ascii=False),
oid,
),
)
else:
cur = conn.execute(
"""
INSERT INTO objects(config, object_type, name, full_name, synonym, comment,
tooltip, explanation, uuid, path, file_id, json)
VALUES(?,?,?,?,?,?,?,?,?,?,?,?)
""",
(
config,
obj["object_type"],
obj["name"],
full_name,
obj.get("synonym"),
obj.get("comment"),
obj.get("tooltip"),
obj.get("explanation"),
obj.get("uuid"),
obj.get("path"),
file_id,
json.dumps(obj, ensure_ascii=False),
),
)
oid = int(cur.lastrowid)
conn.execute(
"""
INSERT INTO objects_fts(full_name, synonym, comment, fields_text, object_id, config)
VALUES(?,?,?,?,?,?)
""",
(
full_name,
obj.get("synonym") or "",
obj.get("comment") or "",
fields_text,
oid,
config,
),
)
for f in obj.get("fields") or []:
conn.execute(
"""
INSERT INTO fields(object_id, config, kind, name, synonym, comment, types,
tabular_section, refs_json)
VALUES(?,?,?,?,?,?,?,?,?)
""",
(
oid,
config,
f.get("kind"),
f.get("name"),
f.get("synonym"),
f.get("comment"),
", ".join(f.get("types") or []),
f.get("tabular_section") or "",
json.dumps(f.get("refs") or [], ensure_ascii=False),
),
)
for r in f.get("refs") or []:
loc = f"{f['tabular_section']}." if f.get("tabular_section") else ""
conn.execute(
"""
INSERT INTO refs(config, from_full_name, from_field, to_kind, to_name)
VALUES(?,?,?,?,?)
""",
(
config,
full_name,
f"{loc}{f.get('name')}",
r.get("kind"),
r.get("name"),
),
)
return oid
def upsert_module(
conn: sqlite3.Connection,
*,
config: str,
rel_path: str,
owner: str,
module_kind: str,
symbols: list[str],
body: str,
line_count: int,
size: int,
file_id: int,
) -> int:
sym_text = "\n".join(symbols)
old = conn.execute(
"SELECT id FROM modules WHERE config=? AND rel_path=?",
(config, rel_path),
).fetchone()
if old:
mid = int(old["id"])
conn.execute("DELETE FROM modules_fts WHERE module_id=?", (mid,))
conn.execute(
"""
UPDATE modules SET owner=?, module_kind=?, symbols=?, line_count=?, size=?, file_id=?
WHERE id=?
""",
(owner, module_kind, sym_text, line_count, size, file_id, mid),
)
else:
cur = conn.execute(
"""
INSERT INTO modules(config, rel_path, owner, module_kind, symbols, line_count, size, file_id)
VALUES(?,?,?,?,?,?,?,?)
""",
(config, rel_path, owner, module_kind, sym_text, line_count, size, file_id),
)
mid = int(cur.lastrowid)
conn.execute(
"""
INSERT INTO modules_fts(rel_path, owner, module_kind, symbols, body, module_id, config)
VALUES(?,?,?,?,?,?,?)
""",
(rel_path, owner, module_kind, sym_text, body, mid, config),
)
return mid
def refresh_config_stats(conn: sqlite3.Connection, config: str, src: str) -> dict[str, Any]:
objects = conn.execute(
"SELECT COUNT(*) AS n FROM objects WHERE config=?", (config,)
).fetchone()["n"]
fields = conn.execute(
"SELECT COUNT(*) AS n FROM fields WHERE config=?", (config,)
).fetchone()["n"]
refs = conn.execute(
"SELECT COUNT(*) AS n FROM refs WHERE config=?", (config,)
).fetchone()["n"]
modules = conn.execute(
"SELECT COUNT(*) AS n FROM modules WHERE config=?", (config,)
).fetchone()["n"]
stats = {
"name": config,
"src": src,
"objects": objects,
"fields": fields,
"refs": refs,
"modules": modules,
"updated_at": utcnow(),
}
conn.execute(
"""
INSERT INTO configs(name, src, objects, fields, refs, modules, updated_at)
VALUES(?,?,?,?,?,?,?)
ON CONFLICT(name) DO UPDATE SET
src=excluded.src,
objects=excluded.objects,
fields=excluded.fields,
refs=excluded.refs,
modules=excluded.modules,
updated_at=excluded.updated_at
""",
(
stats["name"],
stats["src"],
stats["objects"],
stats["fields"],
stats["refs"],
stats["modules"],
stats["updated_at"],
),
)
return stats
def list_configs(conn: sqlite3.Connection) -> list[dict[str, Any]]:
return [dict(r) for r in conn.execute("SELECT * FROM configs ORDER BY name")]
def set_meta(conn: sqlite3.Connection, key: str, value: str) -> None:
conn.execute(
"INSERT OR REPLACE INTO meta(key, value) VALUES(?, ?)",
(key, value),
)
index1c/parse_meta.py
+351
View File
@@ -0,0 +1,351 @@
"""Разбор XML метаданных объектов 1С (выгрузка EDT)."""
from __future__ import annotations
import re
import xml.etree.ElementTree as ET
from dataclasses import asdict, dataclass, field
from pathlib import Path
from typing import Any, Iterable
from .config import FIELD_RICH_TYPES, OBJECT_TYPE_FOLDERS
MD = "{http://v8.1c.ru/8.3/MDClasses}"
V8 = "{http://v8.1c.ru/8.1/data/core}"
REF_TYPE_RE = re.compile(
r"^cfg:(CatalogRef|DocumentRef|EnumRef|ChartOfCharacteristicTypesRef|"
r"ChartOfAccountsRef|ChartOfCalculationTypesRef|ExchangePlanRef|"
r"BusinessProcessRef|TaskRef|DefinedType)\.(.+)$"
)
FIELD_TAGS = frozenset(
{
"Attribute",
"Dimension",
"Resource",
"TabularSection",
"Column",
"EnumValue",
"AddressingAttribute",
}
)
def local(tag: str) -> str:
return tag.split("}")[-1] if "}" in tag else tag
def text_of(el: ET.Element | None) -> str:
if el is None:
return ""
return (el.text or "").strip()
def child_text(props: ET.Element, tag: str) -> str:
node = props.find(f"{MD}{tag}")
if node is None:
# без namespace — на всякий случай
for ch in props:
if local(ch.tag) == tag:
return (ch.text or "").strip()
return ""
return (node.text or "").strip()
def collect_localized(props: ET.Element, tag: str) -> dict[str, str]:
"""Синоним / подсказка: lang → content."""
node = props.find(f"{MD}{tag}")
if node is None:
return {}
out: dict[str, str] = {}
for item in node.findall(f"{V8}item"):
lang = text_of(item.find(f"{V8}lang")) or "ru"
content = text_of(item.find(f"{V8}content"))
if content:
out[lang] = content
return out
def synonym_str(loc: dict[str, str]) -> str:
if not loc:
return ""
if "ru" in loc:
parts = [loc["ru"]] + [f"{k}:{v}" for k, v in loc.items() if k != "ru"]
return " | ".join(parts)
return " | ".join(f"{k}:{v}" for k, v in loc.items())
def collect_types(props: ET.Element) -> list[str]:
typ = props.find(f"{MD}Type")
if typ is None:
return []
types: list[str] = []
for t in typ.findall(f".//{V8}Type"):
raw = (t.text or "").strip()
if raw:
types.append(raw)
return types
def parse_ref_targets(types: Iterable[str]) -> list[dict[str, str]]:
refs: list[dict[str, str]] = []
for t in types:
m = REF_TYPE_RE.match(t)
if m:
refs.append({"kind": m.group(1), "name": m.group(2), "raw": t})
elif t.startswith("cfg:") and "Ref." in t:
# прочие cfg:*Ref.*
body = t[4:]
if "." in body:
kind, name = body.split(".", 1)
refs.append({"kind": kind, "name": name, "raw": t})
return refs
@dataclass
class FieldRec:
kind: str # Attribute | Dimension | Resource | Column | EnumValue | ...
name: str
synonym: str = ""
comment: str = ""
tooltip: str = ""
types: list[str] = field(default_factory=list)
refs: list[dict[str, str]] = field(default_factory=list)
tabular_section: str = ""
@dataclass
class ObjectRec:
config: str
object_type: str
name: str
synonym: str = ""
comment: str = ""
tooltip: str = ""
explanation: str = ""
uuid: str = ""
path: str = ""
fields: list[FieldRec] = field(default_factory=list)
owners: list[str] = field(default_factory=list)
register_records: list[str] = field(default_factory=list)
based_on: list[str] = field(default_factory=list)
input_by_string: list[str] = field(default_factory=list)
@property
def full_name(self) -> str:
return f"{self.object_type}.{self.name}"
def search_text(self) -> str:
parts = [
self.object_type,
self.name,
self.synonym,
self.comment,
self.tooltip,
self.explanation,
]
for f in self.fields:
parts.extend([f.kind, f.name, f.synonym, f.comment, f.tooltip, f.tabular_section])
parts.extend(f.types)
parts.extend(self.owners)
parts.extend(self.register_records)
parts.extend(self.based_on)
return "\n".join(p for p in parts if p)
def to_dict(self) -> dict[str, Any]:
d = asdict(self)
d["full_name"] = self.full_name
return d
def _parse_field_props(props: ET.Element, kind: str, tabular: str = "") -> FieldRec | None:
name = child_text(props, "Name")
if not name:
return None
types = collect_types(props)
return FieldRec(
kind=kind,
name=name,
synonym=synonym_str(collect_localized(props, "Synonym")),
comment=child_text(props, "Comment"),
tooltip=synonym_str(collect_localized(props, "ToolTip")),
types=types,
refs=parse_ref_targets(types),
tabular_section=tabular,
)
def _walk_child_objects(co: ET.Element | None, object_type: str) -> list[FieldRec]:
if co is None or object_type not in FIELD_RICH_TYPES:
return []
fields: list[FieldRec] = []
for child in co:
tag = local(child.tag)
if tag == "TabularSection":
ts_props = child.find(f"{MD}Properties")
ts_name = child_text(ts_props, "Name") if ts_props is not None else ""
if ts_props is not None and ts_name:
fields.append(
FieldRec(
kind="TabularSection",
name=ts_name,
synonym=synonym_str(collect_localized(ts_props, "Synonym")),
comment=child_text(ts_props, "Comment"),
)
)
inner = child.find(f"{MD}ChildObjects")
if inner is not None:
for col in inner:
if local(col.tag) != "Attribute":
continue
cprops = col.find(f"{MD}Properties")
if cprops is None:
continue
rec = _parse_field_props(cprops, "Column", tabular=ts_name)
if rec:
fields.append(rec)
continue
if tag not in FIELD_TAGS:
continue
props = child.find(f"{MD}Properties")
if props is None:
continue
rec = _parse_field_props(props, tag)
if rec:
fields.append(rec)
return fields
_META_PREFIXES = (
"Catalog.",
"Document.",
"InformationRegister.",
"AccumulationRegister.",
"Enum.",
"Constant.",
"ChartOfCharacteristicTypes.",
"ExchangePlan.",
"BusinessProcess.",
"Task.",
"DocumentJournal.",
)
def _list_from_props(props: ET.Element, tag: str) -> list[str]:
node = props.find(f"{MD}{tag}")
if node is None:
return []
items: list[str] = []
for el in node.iter():
t = (el.text or "").strip()
if not t or "." not in t:
continue
if any(t.startswith(p) for p in _META_PREFIXES):
items.append(t)
seen: set[str] = set()
out: list[str] = []
for x in items:
if x not in seen:
seen.add(x)
out.append(x)
return out
def parse_metadata_xml(
path: Path,
*,
config: str,
object_type: str,
) -> ObjectRec | None:
try:
tree = ET.parse(path)
except ET.ParseError:
return None
root = tree.getroot()
# Корневой элемент объекта: <Catalog>, <Document>, ...
obj_el: ET.Element | None = None
for ch in root:
if local(ch.tag) == object_type:
obj_el = ch
break
if obj_el is None:
for ch in root:
if ch.find(f"{MD}Properties") is not None or ch.find("Properties") is not None:
obj_el = ch
break
if obj_el is None:
return None
props = obj_el.find(f"{MD}Properties")
if props is None:
props = obj_el.find("Properties")
if props is None:
return None
name = child_text(props, "Name")
if not name:
return None
uuid = obj_el.attrib.get("uuid", "")
fields = _walk_child_objects(obj_el.find(f"{MD}ChildObjects"), object_type)
# Для констант / определяемых типов тип лежит в Properties
if object_type in {"Constant", "DefinedType", "CommonAttribute", "SessionParameter"} and not fields:
types = collect_types(props)
if types:
fields.append(
FieldRec(
kind="ValueType",
name="Type",
types=types,
refs=parse_ref_targets(types),
)
)
return ObjectRec(
config=config,
object_type=object_type,
name=name,
synonym=synonym_str(collect_localized(props, "Synonym")),
comment=child_text(props, "Comment"),
tooltip=synonym_str(collect_localized(props, "ToolTip")),
explanation=synonym_str(collect_localized(props, "Explanation")),
uuid=uuid,
path=str(path),
fields=fields,
owners=_list_from_props(props, "Owners"),
register_records=_list_from_props(props, "RegisterRecords"),
based_on=_list_from_props(props, "BasedOn"),
input_by_string=_list_from_props(props, "InputByString"),
)
def iter_object_xml_files(
src: Path,
types: set[str] | None = None,
skip_types: set[str] | None = None,
) -> list[tuple[str, Path]]:
"""Список (object_type, xml_path) только верхний уровень src/<Folder>/*.xml."""
skip = skip_types or set()
out: list[tuple[str, Path]] = []
if not src.is_dir():
return out
for folder in sorted(src.iterdir()):
if not folder.is_dir():
continue
otype = OBJECT_TYPE_FOLDERS.get(folder.name)
if otype is None:
continue
if otype in skip:
continue
if types is not None and otype not in types:
continue
for xml in sorted(folder.glob("*.xml")):
out.append((otype, xml))
return out
index1c/parse_modules.py
+89
View File
@@ -0,0 +1,89 @@
"""Индексация BSL-модулей конфигурации."""
from __future__ import annotations
import re
from dataclasses import dataclass
from pathlib import Path
from .config import OBJECT_TYPE_FOLDERS
# Процедура/Функция Имя(...)
RE_SYMBOL = re.compile(
r"(?im)^\s*(?:&[А-Яа-яA-Za-z_]+)*\s*"
r"(?:Процедура|Функция|Procedure|Function)\s+"
r"([А-Яа-яA-Za-z_][А-Яа-яA-Za-z0-9_]*)"
)
# Лимит тела для FTS (символы) — полные гигантские модули не кладём целиком.
MAX_BODY_CHARS = 200_000
@dataclass
class ModuleRec:
config: str
rel_path: str
abs_path: str
owner: str
module_kind: str
symbols: list[str]
body: str
line_count: int
size: int
mtime_ns: int
def _module_kind(rel: Path) -> str:
name = rel.name
if name == "Module.bsl":
parent = rel.parent.name
if parent == "Form":
return "FormModule"
if parent == "Ext":
# CommonModules/.../Ext/Module.bsl или HTTPServices/.../Ext/Module.bsl
return "Module"
return "Module"
if name.endswith("Module.bsl"):
return name[: -len(".bsl")] # ObjectModule, ManagerModule, ...
return name
def _owner_from_rel(rel: Path) -> str:
"""Documents/ЗаказКлиента/Ext/ObjectModule.bsl → Document.ЗаказКлиента."""
parts = rel.parts
if not parts:
return ""
folder = parts[0]
otype = OBJECT_TYPE_FOLDERS.get(folder)
if otype and len(parts) >= 2:
return f"{otype}.{parts[1]}"
return folder
def parse_bsl_file(path: Path, *, config: str, src_root: Path) -> ModuleRec | None:
try:
st = path.stat()
text = path.read_text(encoding="utf-8-sig", errors="replace")
except OSError:
return None
rel = path.relative_to(src_root).as_posix()
symbols = sorted(set(RE_SYMBOL.findall(text)))
body = text if len(text) <= MAX_BODY_CHARS else text[:MAX_BODY_CHARS]
return ModuleRec(
config=config,
rel_path=rel,
abs_path=str(path),
owner=_owner_from_rel(Path(rel)),
module_kind=_module_kind(Path(rel)),
symbols=symbols,
body=body,
line_count=text.count("\n") + (1 if text and not text.endswith("\n") else 0),
size=st.st_size,
mtime_ns=getattr(st, "st_mtime_ns", int(st.st_mtime * 1e9)),
)
def iter_bsl_files(src: Path) -> list[Path]:
if not src.is_dir():
return []
return sorted(src.rglob("*.bsl"))
index1c/search.py
+287
View File
@@ -0,0 +1,287 @@
"""Поиск по единому index.sqlite проекта."""
from __future__ import annotations
import json
import re
import sqlite3
from pathlib import Path
from typing import Any
from . import db as dbmod
from .config import cache_dir
def _fts_query(raw: str) -> str:
tokens = re.findall(r"[\wА-Яа-яЁё]+", raw, flags=re.UNICODE)
if not tokens:
return raw
return " ".join(f"{t}*" for t in tokens)
def open_index(project_root: Path, *, readonly: bool = True) -> sqlite3.Connection:
path = dbmod.global_db_path(cache_dir(project_root))
if not path.is_file():
raise FileNotFoundError(
f"Индекс не найден: {path}. Запустите: python tools/index/index_1c.py reindex --full"
)
return dbmod.connect(path, readonly=readonly)
def search_objects(
conn: sqlite3.Connection,
query: str,
*,
configs: list[str] | None = None,
object_type: str | None = None,
limit: int = 30,
) -> list[dict[str, Any]]:
exact_sql = """
SELECT config, full_name, object_type, name, synonym, comment, path, '' AS snip, -100.0 AS score
FROM objects
WHERE (name = ? OR full_name = ? OR full_name LIKE ?)
"""
exact_params: list[Any] = [query, query, f"%.{query}"]
if configs:
exact_sql += f" AND config IN ({','.join('?' * len(configs))})"
exact_params.extend(configs)
if object_type:
exact_sql += " AND object_type = ?"
exact_params.append(object_type)
exact_sql += " LIMIT ?"
exact_params.append(limit)
exact = [dict(r) for r in conn.execute(exact_sql, exact_params)]
seen = {(r["config"], r["full_name"]) for r in exact}
fts = _fts_query(query)
fetch_n = max(limit * 5, 50)
sql = """
SELECT o.config, o.full_name, o.object_type, o.name, o.synonym, o.comment, o.path,
snippet(objects_fts, 0, '[', ']', '…', 12) AS snip,
bm25(objects_fts, 10.0, 5.0, 2.0, 1.0) AS score
FROM objects_fts
JOIN objects o ON o.id = objects_fts.object_id
WHERE objects_fts MATCH ?
"""
params: list[Any] = [fts]
if configs:
sql += f" AND o.config IN ({','.join('?' * len(configs))})"
params.extend(configs)
if object_type:
sql += " AND o.object_type = ?"
params.append(object_type)
sql += " ORDER BY score LIMIT ?"
params.append(fetch_n)
try:
rows = [dict(r) for r in conn.execute(sql, params)]
except sqlite3.OperationalError:
like = f"%{query}%"
sql2 = """
SELECT config, full_name, object_type, name, synonym, comment, path, '' AS snip, 0 AS score
FROM objects
WHERE full_name LIKE ? OR synonym LIKE ? OR comment LIKE ? OR name LIKE ?
"""
params2: list[Any] = [like, like, like, like]
if configs:
sql2 += f" AND config IN ({','.join('?' * len(configs))})"
params2.extend(configs)
if object_type:
sql2 += " AND object_type = ?"
params2.append(object_type)
sql2 += " LIMIT ?"
params2.append(fetch_n)
rows = [dict(r) for r in conn.execute(sql2, params2)]
merged = list(exact)
for r in rows:
key = (r["config"], r["full_name"])
if key in seen:
continue
seen.add(key)
merged.append(r)
q_cf = query.casefold()
for r in merged:
name_cf = (r.get("name") or "").casefold()
full_cf = (r.get("full_name") or "").casefold()
if name_cf == q_cf or full_cf == q_cf or full_cf.endswith("." + q_cf):
r["_boost"] = 0
elif q_cf in name_cf or q_cf in full_cf:
r["_boost"] = 1
else:
r["_boost"] = 2
merged.sort(
key=lambda x: (x.get("_boost", 9), x.get("score") if x.get("score") is not None else 0)
)
return merged[:limit]
def search_modules(
conn: sqlite3.Connection,
query: str,
*,
configs: list[str] | None = None,
limit: int = 30,
) -> list[dict[str, Any]]:
fts = _fts_query(query)
sql = """
SELECT m.config, m.rel_path, m.owner, m.module_kind, m.symbols, m.line_count,
snippet(modules_fts, 4, '[', ']', '…', 16) AS snip,
bm25(modules_fts, 2.0, 5.0, 2.0, 8.0, 1.0) AS score
FROM modules_fts
JOIN modules m ON m.id = modules_fts.module_id
WHERE modules_fts MATCH ?
"""
params: list[Any] = [fts]
if configs:
sql += f" AND m.config IN ({','.join('?' * len(configs))})"
params.extend(configs)
sql += " ORDER BY score LIMIT ?"
params.append(limit)
try:
return [dict(r) for r in conn.execute(sql, params)]
except sqlite3.OperationalError:
like = f"%{query}%"
sql2 = """
SELECT config, rel_path, owner, module_kind, symbols, line_count, '' AS snip, 0 AS score
FROM modules
WHERE rel_path LIKE ? OR owner LIKE ? OR symbols LIKE ?
"""
params2: list[Any] = [like, like, like]
if configs:
sql2 += f" AND config IN ({','.join('?' * len(configs))})"
params2.extend(configs)
sql2 += " LIMIT ?"
params2.append(limit)
return [dict(r) for r in conn.execute(sql2, params2)]
def find_refs_to(
conn: sqlite3.Connection,
target: str,
*,
configs: list[str] | None = None,
limit: int = 100,
) -> list[dict[str, Any]]:
kind = ""
name = target
if "." in target:
kind, name = target.split(".", 1)
kind_ref = kind
if kind and not kind.endswith("Ref"):
if kind in {
"Catalog",
"Document",
"Enum",
"ExchangePlan",
"BusinessProcess",
"Task",
"ChartOfCharacteristicTypes",
"ChartOfAccounts",
"ChartOfCalculationTypes",
}:
kind_ref = kind + "Ref"
if kind_ref:
sql = """
SELECT config, from_full_name, from_field, to_kind, to_name
FROM refs
WHERE to_name = ? AND (to_kind = ? OR to_kind = ?)
"""
params: list[Any] = [name, kind_ref, kind]
else:
sql = """
SELECT config, from_full_name, from_field, to_kind, to_name
FROM refs WHERE to_name = ?
"""
params = [name]
if configs:
sql += f" AND config IN ({','.join('?' * len(configs))})"
params.extend(configs)
sql += " ORDER BY config, from_full_name LIMIT ?"
params.append(limit)
return [dict(r) for r in conn.execute(sql, params)]
def get_object(
conn: sqlite3.Connection,
name: str,
*,
configs: list[str] | None = None,
) -> dict[str, Any] | None:
sql = "SELECT config, json FROM objects WHERE full_name = ? OR name = ?"
params: list[Any] = [name, name]
if configs:
sql += f" AND config IN ({','.join('?' * len(configs))})"
params.extend(configs)
sql += " LIMIT 1"
row = conn.execute(sql, params).fetchone()
if not row:
sql = "SELECT config, json FROM objects WHERE full_name LIKE ?"
params = [f"%.{name}"]
if configs:
sql += f" AND config IN ({','.join('?' * len(configs))})"
params.extend(configs)
sql += " LIMIT 1"
row = conn.execute(sql, params).fetchone()
if not row:
return None
data = json.loads(row["json"])
data["_config"] = row["config"]
return data
def get_module(
conn: sqlite3.Connection,
path_or_owner: str,
*,
configs: list[str] | None = None,
) -> list[dict[str, Any]]:
sql = """
SELECT config, rel_path, owner, module_kind, symbols, line_count, size
FROM modules
WHERE rel_path = ? OR rel_path LIKE ? OR owner = ? OR owner LIKE ?
"""
like = f"%{path_or_owner}%"
params: list[Any] = [path_or_owner, like, path_or_owner, like]
if configs:
sql += f" AND config IN ({','.join('?' * len(configs))})"
params.extend(configs)
sql += " ORDER BY config, rel_path LIMIT 50"
return [dict(r) for r in conn.execute(sql, params)]
def list_indexed_configs(conn_or_cache) -> list[str]:
"""Совместимость: Connection или Path cache_root."""
if isinstance(conn_or_cache, Path):
db_path = dbmod.global_db_path(conn_or_cache)
if not db_path.is_file():
# legacy: подкаталоги с manifest
return [
p.name
for p in sorted(conn_or_cache.iterdir())
if p.is_dir() and (p / "manifest.json").is_file()
]
conn = dbmod.connect(db_path, readonly=True)
try:
return [c["name"] for c in dbmod.list_configs(conn)]
finally:
conn.close()
return [c["name"] for c in dbmod.list_configs(conn_or_cache)]
# --- legacy wrappers (per-config sqlite) ---
def search_sqlite(db_path: Path, query: str, *, limit: int = 30, object_type: str | None = None):
"""Устарело: ищет в per-config sqlite; для нового индекса используйте search_objects."""
if not db_path.is_file():
return []
# если передали index.sqlite — открыть как глобальный
if db_path.name == "index.sqlite":
conn = dbmod.connect(db_path, readonly=True)
try:
return search_objects(conn, query, object_type=object_type, limit=limit)
finally:
conn.close()
return []
index1c/writers.py
+334
View File
@@ -0,0 +1,334 @@
"""Запись индексов: JSONL, Markdown, SQLite FTS5."""
from __future__ import annotations
import json
import sqlite3
from collections import defaultdict
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Iterable
from .parse_meta import ObjectRec
def _utcnow() -> str:
return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
def write_jsonl(path: Path, rows: Iterable[dict[str, Any]]) -> int:
path.parent.mkdir(parents=True, exist_ok=True)
n = 0
with path.open("w", encoding="utf-8") as f:
for row in rows:
f.write(json.dumps(row, ensure_ascii=False) + "\n")
n += 1
return n
def write_objects_markdown(path: Path, objects: list[ObjectRec], *, title: str) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
by_type: dict[str, list[ObjectRec]] = defaultdict(list)
for obj in objects:
by_type[obj.object_type].append(obj)
lines: list[str] = [
f"# {title}",
"",
f"Сгенерировано: {_utcnow()}. Быстрый поиск метаданных без обхода XML.",
"",
]
for otype in sorted(by_type):
lines.append(f"## {otype} ({len(by_type[otype])})")
lines.append("")
for obj in sorted(by_type[otype], key=lambda o: o.name.casefold()):
syn = f" — {obj.synonym}" if obj.synonym else ""
lines.append(f"### `{obj.full_name}`{syn}")
if obj.comment:
lines.append(f"- Комментарий: {obj.comment}")
if obj.tooltip:
lines.append(f"- Подсказка: {obj.tooltip}")
if obj.owners:
lines.append(f"- Владельцы: {', '.join(obj.owners)}")
if obj.register_records:
lines.append(f"- Движения: {', '.join(obj.register_records)}")
if obj.based_on:
lines.append(f"- Вводится на основании: {', '.join(obj.based_on)}")
attrs = [f for f in obj.fields if f.kind != "TabularSection"]
if attrs:
lines.append("- Поля:")
for f in attrs:
loc = f".{f.tabular_section}" if f.tabular_section else ""
types = ", ".join(f.types) if f.types else ""
syn_f = f" ({f.synonym})" if f.synonym else ""
type_s = f" : `{types}`" if types else ""
refs = ""
if f.refs:
refs = " → " + ", ".join(
f"{r['kind']}.{r['name']}" for r in f.refs
)
lines.append(
f" - `{f.kind}{loc}.{f.name}`{syn_f}{type_s}{refs}"
)
lines.append("")
path.write_text("\n".join(lines), encoding="utf-8")
def write_refs_markdown(path: Path, objects: list[ObjectRec], *, title: str) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
inv: dict[str, list[str]] = defaultdict(list)
for obj in objects:
for f in obj.fields:
for r in f.refs:
target = f"{r['kind']}.{r['name']}"
loc = f".{f.tabular_section}" if f.tabular_section else ""
inv[target].append(f"{obj.full_name}{loc}.{f.name}")
lines = [
f"# {title}",
"",
f"Сгенерировано: {_utcnow()}.",
"",
]
for target in sorted(inv, key=lambda s: s.casefold()):
lines.append(f"## `{target}`")
for src in sorted(set(inv[target])):
lines.append(f"- `{src}`")
lines.append("")
path.write_text("\n".join(lines), encoding="utf-8")
def write_sqlite_fts(path: Path, objects: list[ObjectRec], config: str) -> None:
if path.exists():
path.unlink()
path.parent.mkdir(parents=True, exist_ok=True)
conn = sqlite3.connect(path)
try:
conn.execute("PRAGMA journal_mode=WAL")
conn.execute(
"""
CREATE TABLE objects (
id INTEGER PRIMARY KEY,
config TEXT NOT NULL,
object_type TEXT NOT NULL,
name TEXT NOT NULL,
full_name TEXT NOT NULL,
synonym TEXT,
comment TEXT,
path TEXT,
json TEXT NOT NULL
)
"""
)
conn.execute(
"""
CREATE VIRTUAL TABLE objects_fts USING fts5(
full_name,
synonym,
comment,
fields_text,
object_id UNINDEXED,
tokenize='unicode61'
)
"""
)
conn.execute(
"""
CREATE TABLE fields (
object_id INTEGER NOT NULL,
kind TEXT,
name TEXT,
synonym TEXT,
types TEXT,
tabular_section TEXT,
refs_json TEXT,
FOREIGN KEY(object_id) REFERENCES objects(id)
)
"""
)
conn.execute(
"""
CREATE TABLE refs (
from_full_name TEXT NOT NULL,
from_field TEXT NOT NULL,
to_kind TEXT NOT NULL,
to_name TEXT NOT NULL
)
"""
)
for obj in objects:
fields_text = " ".join(
" ".join(
filter(
None,
[
f.kind,
f.name,
f.synonym,
f.comment,
f.tabular_section,
" ".join(f.types),
],
)
)
for f in obj.fields
)
payload = json.dumps(obj.to_dict(), ensure_ascii=False)
cur = conn.execute(
"""
INSERT INTO objects
(config, object_type, name, full_name, synonym, comment, path, json)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
""",
(
config,
obj.object_type,
obj.name,
obj.full_name,
obj.synonym,
obj.comment,
obj.path,
payload,
),
)
oid = cur.lastrowid
conn.execute(
"""
INSERT INTO objects_fts
(full_name, synonym, comment, fields_text, object_id)
VALUES (?, ?, ?, ?, ?)
""",
(obj.full_name, obj.synonym or "", obj.comment or "", fields_text, oid),
)
for f in obj.fields:
conn.execute(
"""
INSERT INTO fields
(object_id, kind, name, synonym, types, tabular_section, refs_json)
VALUES (?, ?, ?, ?, ?, ?, ?)
""",
(
oid,
f.kind,
f.name,
f.synonym,
", ".join(f.types),
f.tabular_section,
json.dumps(f.refs, ensure_ascii=False),
),
)
for r in f.refs:
loc = f"{f.tabular_section}." if f.tabular_section else ""
conn.execute(
"""
INSERT INTO refs
(from_full_name, from_field, to_kind, to_name)
VALUES (?, ?, ?, ?)
""",
(obj.full_name, f"{loc}{f.name}", r["kind"], r["name"]),
)
conn.execute("CREATE INDEX idx_objects_type ON objects(object_type)")
conn.execute("CREATE INDEX idx_objects_name ON objects(name)")
conn.execute("CREATE INDEX idx_refs_to ON refs(to_kind, to_name)")
conn.execute("CREATE INDEX idx_refs_from ON refs(from_full_name)")
conn.commit()
finally:
conn.close()
def write_manifest(path: Path, data: dict[str, Any]) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
data = {**data, "generated_at": _utcnow()}
path.write_text(
json.dumps(data, ensure_ascii=False, indent=2) + "\n", encoding="utf-8"
)
def write_type_summaries(out_dir: Path, objects: list[ObjectRec]) -> None:
by_type: dict[str, list[ObjectRec]] = defaultdict(list)
for obj in objects:
by_type[obj.object_type].append(obj)
types_dir = out_dir / "by_type"
types_dir.mkdir(parents=True, exist_ok=True)
for otype, items in by_type.items():
lines = [f"# {otype}", "", f"Объектов: {len(items)}", ""]
for obj in sorted(items, key=lambda o: o.name.casefold()):
syn = f" — {obj.synonym}" if obj.synonym else ""
n_fields = len([f for f in obj.fields if f.kind != "TabularSection"])
lines.append(f"- `{obj.name}`{syn} (полей: {n_fields})")
(types_dir / f"{otype}.md").write_text("\n".join(lines) + "\n", encoding="utf-8")
def write_global_index_md(cache_root: Path, summary: dict[str, Any]) -> Path:
"""Сводка по единому index.sqlite."""
cache_root.mkdir(parents=True, exist_ok=True)
configs = summary.get("configs") or []
lines = [
"# Индекс метаданных и модулей 1С (проект)",
"",
f"Сгенерировано: {_utcnow()}.",
f"Режим: `{summary.get('mode', '?')}`, БД: `cache/1c_meta/index.sqlite`.",
f"Потоки: {summary.get('workers', '?')}, время: {summary.get('elapsed_sec', '?')} с.",
"",
"Поиск: `python tools/index/query_1c.py search \"…\"`",
"Переиндексация: `python tools/index/index_1c.py reindex` (инкремент) / `--full`.",
"",
"## Конфигурации",
"",
"| Конфигурация | Объектов | Полей | Ссылок | Модулей |",
"|---|---:|---:|---:|---:|",
]
for c in sorted(configs, key=lambda x: x.get("name") or x.get("config") or ""):
name = c.get("name") or c.get("config") or "?"
lines.append(
"| `{name}` | {objects} | {fields} | {refs} | {modules} |".format(
name=name,
objects=c.get("objects", 0),
fields=c.get("fields", 0),
refs=c.get("refs", 0),
modules=c.get("modules", 0),
)
)
lines.extend(
[
"",
"## Статистика прогона",
"",
f"- meta разобрано: {summary.get('meta_parsed', 0)}, пропущено (без изменений): {summary.get('meta_skipped', 0)}",
f"- modules разобрано: {summary.get('modules_parsed', 0)}, пропущено: {summary.get('modules_skipped', 0)}",
f"- удалено файлов: {summary.get('files_removed', 0)}, ошибок: {summary.get('errors', 0)}",
"",
"## Артефакты",
"",
"- `index.sqlite` — единый кэш проекта (объекты, поля, ссылки, модули, FTS)",
"- `INDEX.md` — эта сводка",
"- `md/<config>/` — опциональный Markdown (`reindex --md`)",
"",
]
)
path = cache_root / "INDEX.md"
path.write_text("\n".join(lines) + "\n", encoding="utf-8")
write_manifest(
cache_root / "manifest.json",
{
"db": "index.sqlite",
"mode": summary.get("mode"),
"configs": configs,
"summary": {
k: summary.get(k)
for k in (
"meta_parsed",
"meta_skipped",
"modules_parsed",
"modules_skipped",
"files_removed",
"errors",
"elapsed_sec",
"workers",
)
},
},
)
return path