mihailkudravcev
543 lines
19 KiB
Python
543 lines
19 KiB
Python
"""Поиск объекта/реквизита и вывод дерева структуры метаданных."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import re
|
||
import sqlite3
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
from .parse_meta import FieldRec, ObjectRec, parse_metadata_xml
|
||
|
||
_QUERY_TOKEN_RE = re.compile(r"[\wА-Яа-яЁё]+", flags=re.UNICODE)
|
||
|
||
|
||
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 _query_tokens(query: str) -> list[str]:
|
||
return _QUERY_TOKEN_RE.findall(query or "")
|
||
|
||
|
||
def _words_cf(text: str) -> list[str]:
|
||
return [w.casefold() for w in _QUERY_TOKEN_RE.findall(text or "")]
|
||
|
||
|
||
def _tokens_match_text(tokens: list[str], text: str) -> bool:
|
||
"""Как FTS search: каждый токен — префикс какого-либо слова в тексте."""
|
||
if not tokens:
|
||
return False
|
||
words = _words_cf(text)
|
||
if not words:
|
||
return False
|
||
for tok in tokens:
|
||
t = tok.casefold()
|
||
if not any(w.startswith(t) for w in words):
|
||
return False
|
||
return True
|
||
|
||
|
||
def object_header_matches_search(row: dict[str, Any], query: str) -> bool:
|
||
"""Совпадение по имени/синониму/комментарию объекта (не по реквизитам)."""
|
||
tokens = _query_tokens(query)
|
||
if not tokens:
|
||
return False
|
||
for part in (
|
||
row.get("name") or "",
|
||
row.get("full_name") or "",
|
||
row.get("synonym") or "",
|
||
row.get("comment") or "",
|
||
):
|
||
if _tokens_match_text(tokens, part):
|
||
return True
|
||
return False
|
||
|
||
|
||
def field_matches_search_query(f: dict[str, Any], query: str) -> bool:
|
||
"""Совпадение реквизита с поисковым запросом (префикс по словам, как FTS)."""
|
||
tokens = _query_tokens(query)
|
||
if not tokens:
|
||
return False
|
||
text = " ".join(
|
||
filter(
|
||
None,
|
||
[
|
||
f.get("kind") or "",
|
||
f.get("name") or "",
|
||
f.get("synonym") or "",
|
||
f.get("comment") or "",
|
||
f.get("tabular_section") or "",
|
||
" ".join(f.get("types") or []),
|
||
],
|
||
)
|
||
)
|
||
return _tokens_match_text(tokens, text)
|
||
|
||
|
||
def _field_dict_from_db_row(row: sqlite3.Row) -> dict[str, Any]:
|
||
return {
|
||
"kind": row["kind"],
|
||
"name": row["name"],
|
||
"synonym": row["synonym"] or "",
|
||
"comment": row["comment"] or "" if "comment" in row.keys() else "",
|
||
"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": "",
|
||
}
|
||
|
||
|
||
def find_fields_for_search_query(
|
||
conn: sqlite3.Connection,
|
||
object_id: int,
|
||
query: str,
|
||
) -> list[dict[str, Any]]:
|
||
"""Реквизиты объекта, совпавшие с FTS-запросом search."""
|
||
rows = conn.execute(
|
||
"""
|
||
SELECT kind, name, synonym, comment, types, tabular_section, refs_json
|
||
FROM fields
|
||
WHERE object_id = ?
|
||
ORDER BY tabular_section, kind, name
|
||
""",
|
||
(object_id,),
|
||
)
|
||
matched: list[dict[str, Any]] = []
|
||
seen: set[tuple[str, str, str]] = set()
|
||
for row in rows:
|
||
fdict = _field_dict_from_db_row(row)
|
||
if not field_matches_search_query(fdict, query):
|
||
continue
|
||
sig = (fdict["tabular_section"], fdict["name"], fdict["kind"])
|
||
if sig in seen:
|
||
continue
|
||
seen.add(sig)
|
||
matched.append(fdict)
|
||
return matched
|
||
|
||
|
||
def enrich_matched_fields(obj: dict[str, Any], matched: list[dict[str, Any]]) -> None:
|
||
"""Подтянуть FillChecking и типы из актуальной структуры объекта."""
|
||
if not matched or not obj.get("fields"):
|
||
return
|
||
by_f = {
|
||
(f.get("tabular_section") or "", f.get("name") or "", f.get("kind") or ""): f
|
||
for f in obj["fields"]
|
||
}
|
||
for mf in matched:
|
||
key = (mf.get("tabular_section") or "", mf.get("name") or "", mf.get("kind") or "")
|
||
src = by_f.get(key)
|
||
if not src:
|
||
continue
|
||
mf["fill_checking"] = src.get("fill_checking") or ""
|
||
mf["types"] = src.get("types") or mf.get("types")
|
||
mf["synonym"] = src.get("synonym") or mf.get("synonym")
|
||
mf["refs"] = src.get("refs") or mf.get("refs")
|
||
|
||
|
||
def load_object_structure_for_row(
|
||
conn: sqlite3.Connection,
|
||
row: dict[str, Any],
|
||
) -> dict[str, Any]:
|
||
match = {
|
||
"config": row["config"],
|
||
"full_name": row["full_name"],
|
||
"path": row.get("path"),
|
||
"object_type": row.get("object_type"),
|
||
}
|
||
if row.get("json"):
|
||
match["json"] = row["json"]
|
||
elif row.get("object_id"):
|
||
jrow = conn.execute(
|
||
"SELECT json FROM objects WHERE id=?", (row["object_id"],)
|
||
).fetchone()
|
||
match["json"] = jrow["json"] if jrow else "{}"
|
||
else:
|
||
jrow = conn.execute(
|
||
"SELECT json FROM objects WHERE config=? AND full_name=?",
|
||
(row["config"], row["full_name"]),
|
||
).fetchone()
|
||
match["json"] = jrow["json"] if jrow else "{}"
|
||
return load_object_structure(match)
|
||
|
||
|
||
def annotate_search_hits(
|
||
conn: sqlite3.Connection,
|
||
rows: list[dict[str, Any]],
|
||
query: str,
|
||
) -> list[dict[str, Any]]:
|
||
"""Уточнить search: object — по шапке; field — по реквизитам в fields_text."""
|
||
for row in rows:
|
||
if object_header_matches_search(row, query):
|
||
row["match_kind"] = "object"
|
||
row["matched_fields"] = []
|
||
continue
|
||
oid = row.get("object_id")
|
||
matched = find_fields_for_search_query(conn, int(oid), query) if oid else []
|
||
if matched:
|
||
row["match_kind"] = "field"
|
||
row["matched_fields"] = matched
|
||
else:
|
||
row["match_kind"] = "object"
|
||
row["matched_fields"] = []
|
||
return rows
|
||
|
||
|
||
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 _select_matched_field_nodes(
|
||
fields: list[dict[str, Any]],
|
||
matched_fields: list[dict[str, Any]],
|
||
) -> tuple[list[dict[str, Any]], set[tuple[str, str, str]], set[str]]:
|
||
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)
|
||
return selected, want, need_ts
|
||
|
||
|
||
def format_object_tree(
|
||
obj: dict[str, Any],
|
||
*,
|
||
clear: bool = False,
|
||
matched_fields: list[dict[str, Any]] | None = None,
|
||
match_kind: str = "object",
|
||
omit_header: bool = False,
|
||
) -> 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}"
|
||
if not omit_header:
|
||
lines.append(header)
|
||
|
||
fields: list[dict[str, Any]] = list(obj.get("fields") or [])
|
||
|
||
if clear:
|
||
if match_kind == "field" and matched_fields:
|
||
selected, want, need_ts = _select_matched_field_nodes(fields, matched_fields)
|
||
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
|