Release 0.2.0: object tree, refs/modules options, docs
Add query object (--clear), modules -n, composite refs labels, FillChecking/TypeSet parsing, VERSION/CHANGELOG and --version. Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
mihailkudravcev
@@ -0,0 +1,361 @@
|
||||
"""Поиск объекта/реквизита и вывод дерева структуры метаданных."""
|
||||
|
||||
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
|
||||
Reference in New Issue
Block a user