mihailkudravcev
a9d1215820
Add query object (--clear), modules -n, composite refs labels, FillChecking/TypeSet parsing, VERSION/CHANGELOG and --version. Co-authored-by: Cursor <cursoragent@cursor.com>
362 lines
13 KiB
Python
362 lines
13 KiB
Python
"""Поиск объекта/реквизита и вывод дерева структуры метаданных."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import sqlite3
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from .parse_meta import FieldRec, ObjectRec, parse_metadata_xml
|
|
|
|
|
|
FILL_LABEL = {
|
|
"ShowError": "обяз.",
|
|
"DontCheck": "необяз.",
|
|
}
|
|
|
|
|
|
def _norm(s: str) -> str:
|
|
return " ".join((s or "").casefold().split())
|
|
|
|
|
|
def _type_label(raw: str) -> str:
|
|
if raw.startswith("cfg:"):
|
|
return raw[4:]
|
|
if raw.startswith("xs:"):
|
|
return raw[3:]
|
|
return raw
|
|
|
|
|
|
def _fill_label(fill: str) -> str:
|
|
if not fill:
|
|
return "?"
|
|
return FILL_LABEL.get(fill, fill)
|
|
|
|
|
|
def _field_matches(f: dict[str, Any], q: str, qn: str) -> bool:
|
|
"""Только точное совпадение имени или синонима реквизита."""
|
|
name_cf = _norm(f.get("name") or "")
|
|
syn_cf = _norm(f.get("synonym") or "")
|
|
# синоним может быть «Клиент | en:Client» — сравниваем по частям
|
|
syn_parts = [_norm(p) for p in (f.get("synonym") or "").split("|")]
|
|
return name_cf == qn or syn_cf == qn or qn in syn_parts
|
|
|
|
|
|
def find_structure_matches(
|
|
conn: sqlite3.Connection,
|
|
query: str,
|
|
*,
|
|
configs: list[str] | None = None,
|
|
object_type: str | None = None,
|
|
limit: int = 20,
|
|
) -> list[dict[str, Any]]:
|
|
"""Найти объекты по имени/синониму объекта или реквизита.
|
|
|
|
Каждый элемент:
|
|
config, full_name, path, json, match_kind: object|field,
|
|
matched_fields: list[field dict keys],
|
|
score: int (меньше — лучше)
|
|
"""
|
|
q = query.strip()
|
|
qn = _norm(q)
|
|
if not qn:
|
|
return []
|
|
|
|
sql_obj = """
|
|
SELECT id, config, full_name, name, synonym, path, json, object_type
|
|
FROM objects
|
|
WHERE (
|
|
name = ? OR full_name = ? OR full_name LIKE ?
|
|
OR synonym LIKE ? OR name LIKE ?
|
|
)
|
|
"""
|
|
params: list[Any] = [q, q, f"%.{q}", f"%{q}%", f"%{q}%"]
|
|
if configs:
|
|
sql_obj += f" AND config IN ({','.join('?' * len(configs))})"
|
|
params.extend(configs)
|
|
if object_type:
|
|
sql_obj += " AND object_type = ?"
|
|
params.append(object_type)
|
|
sql_obj += " ORDER BY config, full_name LIMIT ?"
|
|
params.append(max(limit * 5, 50))
|
|
|
|
by_key: dict[tuple[str, str], dict[str, Any]] = {}
|
|
|
|
def _ensure(row) -> dict[str, Any]:
|
|
key = (row["config"], row["full_name"])
|
|
if key not in by_key:
|
|
oid = row["object_id"] if "object_id" in row.keys() else row["id"]
|
|
by_key[key] = {
|
|
"config": row["config"],
|
|
"full_name": row["full_name"],
|
|
"path": row["path"],
|
|
"json": row["json"],
|
|
"object_type": row["object_type"],
|
|
"match_kind": "none",
|
|
"matched_fields": [],
|
|
"object_id": oid,
|
|
"score": 100,
|
|
}
|
|
return by_key[key]
|
|
|
|
for row in conn.execute(sql_obj, params):
|
|
entry = _ensure(row)
|
|
name_cf = _norm(row["name"])
|
|
full_cf = _norm(row["full_name"])
|
|
syn_cf = _norm(row["synonym"] or "")
|
|
if name_cf == qn or full_cf == qn or full_cf.endswith("." + qn):
|
|
entry["match_kind"] = "object"
|
|
entry["score"] = min(entry["score"], 0)
|
|
elif syn_cf == qn:
|
|
entry["match_kind"] = "object"
|
|
entry["score"] = min(entry["score"], 2)
|
|
elif qn in name_cf or qn in full_cf:
|
|
if entry["match_kind"] == "none":
|
|
entry["match_kind"] = "object"
|
|
entry["score"] = min(entry["score"], 20)
|
|
elif qn in syn_cf:
|
|
if entry["match_kind"] == "none":
|
|
entry["match_kind"] = "object"
|
|
entry["score"] = min(entry["score"], 25)
|
|
|
|
sql_f = """
|
|
SELECT o.id AS object_id, o.config, o.full_name, o.path, o.json, o.object_type,
|
|
o.name AS object_name, o.synonym AS object_synonym,
|
|
f.kind, f.name, f.synonym, f.types, f.tabular_section, f.refs_json
|
|
FROM fields f
|
|
JOIN objects o ON o.id = f.object_id
|
|
WHERE (
|
|
f.name = ? OR f.synonym LIKE ? OR f.name LIKE ?
|
|
)
|
|
"""
|
|
params_f: list[Any] = [q, f"%{q}%", f"%{q}%"]
|
|
if configs:
|
|
sql_f += f" AND o.config IN ({','.join('?' * len(configs))})"
|
|
params_f.extend(configs)
|
|
if object_type:
|
|
sql_f += " AND o.object_type = ?"
|
|
params_f.append(object_type)
|
|
sql_f += " ORDER BY o.config, o.full_name LIMIT ?"
|
|
params_f.append(max(limit * 15, 100))
|
|
|
|
for row in conn.execute(sql_f, params_f):
|
|
fdict = {
|
|
"kind": row["kind"],
|
|
"name": row["name"],
|
|
"synonym": row["synonym"] or "",
|
|
"types": [t.strip() for t in (row["types"] or "").split(",") if t.strip()],
|
|
"tabular_section": row["tabular_section"] or "",
|
|
"refs": json.loads(row["refs_json"] or "[]"),
|
|
"fill_checking": "",
|
|
}
|
|
if not _field_matches(fdict, q, qn):
|
|
continue
|
|
entry = _ensure(row)
|
|
fname_cf = _norm(fdict["name"])
|
|
fsyn_cf = _norm(fdict["synonym"])
|
|
if fname_cf == qn:
|
|
field_score = 5
|
|
elif fsyn_cf == qn:
|
|
field_score = 6
|
|
else:
|
|
field_score = 15
|
|
# если объект уже найден точно — оставляем object, но добавляем поля
|
|
if entry["score"] > field_score or entry["match_kind"] in {"none", "field"}:
|
|
if entry["match_kind"] != "object" or entry["score"] > 0:
|
|
entry["match_kind"] = "field"
|
|
entry["score"] = min(entry["score"], field_score)
|
|
elif entry["match_kind"] == "object" and entry["score"] == 0:
|
|
# точный объект + ещё реквизит с тем же именем (редко) — остаётся object
|
|
pass
|
|
else:
|
|
entry["score"] = min(entry["score"], field_score)
|
|
|
|
sig = (fdict["tabular_section"], fdict["name"], fdict["kind"])
|
|
existing = {
|
|
(x.get("tabular_section"), x.get("name"), x.get("kind"))
|
|
for x in entry["matched_fields"]
|
|
}
|
|
if sig not in existing:
|
|
entry["matched_fields"].append(fdict)
|
|
|
|
# убрать слабые «просто LIKE по имени объекта», если есть точные хиты
|
|
items = [v for v in by_key.values() if v["match_kind"] != "none"]
|
|
has_exact = any(v["score"] <= 6 for v in items)
|
|
if has_exact:
|
|
items = [v for v in items if v["score"] <= 15]
|
|
|
|
items.sort(key=lambda x: (x["score"], x["config"], x["full_name"]))
|
|
return items[:limit]
|
|
|
|
|
|
def load_object_structure(match: dict[str, Any]) -> dict[str, Any]:
|
|
"""Загрузить структуру: предпочесть свежий разбор XML (FillChecking)."""
|
|
path_s = match.get("path") or ""
|
|
path = Path(path_s) if path_s else None
|
|
config = match["config"]
|
|
otype = match.get("object_type") or match["full_name"].split(".", 1)[0]
|
|
|
|
if path and path.is_file():
|
|
rec = parse_metadata_xml(path, config=config, object_type=otype)
|
|
if rec:
|
|
return rec.to_dict()
|
|
|
|
data = json.loads(match["json"])
|
|
# старые индексы без fill_checking
|
|
for f in data.get("fields") or []:
|
|
f.setdefault("fill_checking", "")
|
|
return data
|
|
|
|
|
|
def format_types(types: list[str], refs: list[dict[str, Any]] | None = None) -> str:
|
|
if not types:
|
|
return "—"
|
|
labels = [_type_label(t) for t in types]
|
|
if len(labels) == 1:
|
|
return labels[0]
|
|
return " | ".join(labels)
|
|
|
|
|
|
def format_field_line(
|
|
f: dict[str, Any],
|
|
*,
|
|
indent: str = "",
|
|
last: bool = True,
|
|
under_ts: bool = False,
|
|
) -> str:
|
|
branch = "└── " if last else "├── "
|
|
name = f.get("name") or ""
|
|
kind = f.get("kind") or "Attribute"
|
|
syn = f.get("synonym") or ""
|
|
syn_s = f" — {syn}" if syn else ""
|
|
if kind == "TabularSection":
|
|
return f"{indent}{branch}`{kind}.{name}`{syn_s}"
|
|
fill = _fill_label(f.get("fill_checking") or "")
|
|
types_s = format_types(f.get("types") or [], f.get("refs"))
|
|
if kind == "Column" and not under_ts and f.get("tabular_section"):
|
|
name = f"{f['tabular_section']}.{name}"
|
|
return f"{indent}{branch}`{kind}.{name}`{syn_s} [{fill}] : {types_s}"
|
|
|
|
|
|
def format_object_tree(
|
|
obj: dict[str, Any],
|
|
*,
|
|
clear: bool = False,
|
|
matched_fields: list[dict[str, Any]] | None = None,
|
|
match_kind: str = "object",
|
|
) -> str:
|
|
"""Дерево реквизитов объекта. clear — только совпадения."""
|
|
lines: list[str] = []
|
|
syn = obj.get("synonym") or ""
|
|
syn_s = f" — {syn}" if syn else ""
|
|
cfg = obj.get("config") or obj.get("_config") or ""
|
|
header = f"`{obj.get('full_name') or obj.get('object_type') + '.' + obj.get('name')}`{syn_s}"
|
|
if cfg:
|
|
header = f"[{cfg}] {header}"
|
|
lines.append(header)
|
|
|
|
fields: list[dict[str, Any]] = list(obj.get("fields") or [])
|
|
|
|
if clear:
|
|
if match_kind == "field" and matched_fields:
|
|
want: set[tuple[str, str, str]] = set()
|
|
need_ts: set[str] = set()
|
|
for mf in matched_fields:
|
|
want.add(
|
|
(
|
|
mf.get("kind") or "",
|
|
mf.get("tabular_section") or "",
|
|
mf.get("name") or "",
|
|
)
|
|
)
|
|
if mf.get("tabular_section") and mf.get("kind") == "Column":
|
|
need_ts.add(mf["tabular_section"])
|
|
selected: list[dict[str, Any]] = []
|
|
for f in fields:
|
|
key = (
|
|
f.get("kind") or "",
|
|
f.get("tabular_section") or "",
|
|
f.get("name") or "",
|
|
)
|
|
if key in want:
|
|
selected.append(f)
|
|
elif f.get("kind") == "TabularSection" and f.get("name") in need_ts:
|
|
selected.append(f)
|
|
if not selected:
|
|
lines.append(" (совпавшие реквизиты не найдены в структуре)")
|
|
return "\n".join(lines)
|
|
lines.extend(_render_field_tree(selected, only_keys=want, need_ts=need_ts))
|
|
return "\n".join(lines)
|
|
return "\n".join(lines)
|
|
|
|
if not fields:
|
|
lines.append(" (нет реквизитов в индексе)")
|
|
return "\n".join(lines)
|
|
|
|
lines.extend(_render_field_tree(fields))
|
|
return "\n".join(lines)
|
|
|
|
|
|
def _render_field_tree(
|
|
fields: list[dict[str, Any]],
|
|
*,
|
|
only_keys: set[tuple[str, str, str]] | None = None,
|
|
need_ts: set[str] | None = None,
|
|
) -> list[str]:
|
|
"""Рендер: сначала шапка объекта уже напечатана; здесь реквизиты и ТЧ."""
|
|
lines: list[str] = []
|
|
# группировка: header fields, then TS with columns
|
|
headers = [
|
|
f
|
|
for f in fields
|
|
if f.get("kind") not in {"TabularSection", "Column"}
|
|
and not f.get("tabular_section")
|
|
]
|
|
tab_sections = [f for f in fields if f.get("kind") == "TabularSection"]
|
|
columns_by_ts: dict[str, list[dict[str, Any]]] = {}
|
|
for f in fields:
|
|
if f.get("kind") == "Column" and f.get("tabular_section"):
|
|
columns_by_ts.setdefault(f["tabular_section"], []).append(f)
|
|
|
|
if only_keys is not None:
|
|
headers = [
|
|
f
|
|
for f in headers
|
|
if (f.get("kind") or "", f.get("tabular_section") or "", f.get("name") or "")
|
|
in only_keys
|
|
]
|
|
tab_sections = [
|
|
f
|
|
for f in tab_sections
|
|
if f.get("name") in (need_ts or set())
|
|
or (f.get("kind") or "", "", f.get("name") or "") in only_keys
|
|
]
|
|
|
|
# плоский список узлов верхнего уровня для корректных └──/├──
|
|
top: list[tuple[str, Any]] = [("h", f) for f in headers] + [("ts", f) for f in tab_sections]
|
|
for i, (kind, node) in enumerate(top):
|
|
last = i == len(top) - 1
|
|
if kind == "h":
|
|
lines.append(format_field_line(node, indent="", last=last))
|
|
continue
|
|
# tabular section
|
|
lines.append(format_field_line(node, indent="", last=last))
|
|
ts_name = node.get("name") or ""
|
|
cols = columns_by_ts.get(ts_name, [])
|
|
if only_keys is not None:
|
|
cols = [
|
|
c
|
|
for c in cols
|
|
if (c.get("kind") or "", c.get("tabular_section") or "", c.get("name") or "")
|
|
in only_keys
|
|
]
|
|
child_indent = " " if last else "│ "
|
|
for j, col in enumerate(cols):
|
|
lines.append(
|
|
format_field_line(
|
|
col, indent=child_indent, last=(j == len(cols) - 1), under_ts=True
|
|
)
|
|
)
|
|
return lines
|