255 lines
11 KiB
Python
255 lines
11 KiB
Python
#!/usr/bin/env python3
|
|
"""Search within one 1C object context using configurator-visible terms."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import base64
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
|
from get_1c_form_context import attach_structure, select_form # noqa: E402
|
|
from get_1c_object_brief_context import view_forms, view_modules # noqa: E402
|
|
from get_1c_object_metadata import build_object_metadata # noqa: E402
|
|
from resolve_1c_object import load_json, resolve_object # noqa: E402
|
|
|
|
|
|
def decode_arg(value: str | None, encoded: str | None) -> str | None:
|
|
if encoded:
|
|
return base64.b64decode(encoded).decode("utf-8")
|
|
return value
|
|
|
|
|
|
def norm(value: Any) -> str:
|
|
return "".join(ch for ch in str(value or "").casefold() if ch not in " \t\r\n._-")
|
|
|
|
|
|
def text_fields(value: Any) -> list[str]:
|
|
fields = []
|
|
if isinstance(value, dict):
|
|
for item in value.values():
|
|
fields.extend(text_fields(item))
|
|
elif isinstance(value, list):
|
|
for item in value:
|
|
fields.extend(text_fields(item))
|
|
elif value is not None:
|
|
fields.append(str(value))
|
|
return fields
|
|
|
|
|
|
def matches(query: str, payload: dict[str, Any]) -> bool:
|
|
wanted = norm(query)
|
|
return any(wanted in norm(text) for text in text_fields(payload))
|
|
|
|
|
|
def slim(payload: dict[str, Any], keys: list[str]) -> dict[str, Any]:
|
|
return {key: payload.get(key) for key in keys if payload.get(key) not in (None, [], "")}
|
|
|
|
|
|
def add_match(result: list[dict[str, Any]], *, area: str, score: int, payload: dict[str, Any], evidence: dict[str, Any]) -> None:
|
|
row = {"area": area, "score": score, **payload, "evidence": evidence}
|
|
result.append(row)
|
|
|
|
|
|
def search_metadata(query: str, metadata: dict[str, Any], limit: int) -> list[dict[str, Any]]:
|
|
result = []
|
|
for attr in metadata.get("attributes") or []:
|
|
payload = slim(attr, ["name", "synonym", "type", "types", "origin", "effective_action"])
|
|
if matches(query, payload):
|
|
add_match(result, area="metadata.attribute", score=90, payload=payload, evidence={"uuid": attr.get("uuid")})
|
|
for section in metadata.get("tabular_sections") or []:
|
|
payload = slim(section, ["name", "synonym", "origin"])
|
|
if matches(query, payload):
|
|
add_match(result, area="metadata.tabular_section", score=80, payload=payload, evidence={"uuid": section.get("uuid")})
|
|
return result[:limit]
|
|
|
|
|
|
def search_forms(query: str, forms: list[dict[str, Any]], max_form_items: int, limit: int) -> list[dict[str, Any]]:
|
|
result = []
|
|
for form in attach_structure(select_form(forms, None), max_items=max_form_items):
|
|
form_payload = slim(form, ["name", "synonym", "origin", "effective_action", "form_xml_path", "module_path"])
|
|
if matches(query, form_payload):
|
|
add_match(result, area="form", score=75, payload=form_payload, evidence={"uuid": form.get("uuid")})
|
|
structures = []
|
|
if form.get("structure"):
|
|
structures.append((form, form.get("structure")))
|
|
for overlay in form.get("extension_overlays") or []:
|
|
if overlay.get("structure"):
|
|
structures.append((overlay, overlay.get("structure")))
|
|
for owner, structure in structures:
|
|
origin = owner.get("origin") or {}
|
|
for item in structure.get("items") or []:
|
|
payload = slim(item, ["name", "kind", "title", "data_path", "command_name", "parent", "events"])
|
|
if matches(query, payload):
|
|
payload["form"] = owner.get("name")
|
|
payload["origin"] = origin
|
|
add_match(result, area="form.item", score=70, payload=payload, evidence={"id": item.get("id"), "form_xml_path": owner.get("form_xml_path")})
|
|
for attr in structure.get("attributes") or []:
|
|
payload = slim(attr, ["name", "title", "data_path", "type", "saved_data", "view"])
|
|
if matches(query, payload):
|
|
payload["form"] = owner.get("name")
|
|
payload["origin"] = origin
|
|
add_match(result, area="form.attribute", score=72, payload=payload, evidence={"id": attr.get("id"), "form_xml_path": owner.get("form_xml_path")})
|
|
for command in structure.get("commands") or []:
|
|
payload = slim(command, ["name", "title", "action", "group", "representation"])
|
|
if matches(query, payload):
|
|
payload["form"] = owner.get("name")
|
|
payload["origin"] = origin
|
|
add_match(result, area="form.command", score=74, payload=payload, evidence={"id": command.get("id"), "form_xml_path": owner.get("form_xml_path")})
|
|
for event in structure.get("events") or []:
|
|
payload = slim(event, ["name", "handler"])
|
|
if matches(query, payload):
|
|
payload["form"] = owner.get("name")
|
|
payload["origin"] = origin
|
|
add_match(result, area="form.event", score=76, payload=payload, evidence={"form_xml_path": owner.get("form_xml_path")})
|
|
if len(result) >= limit:
|
|
return result[:limit]
|
|
return result[:limit]
|
|
|
|
|
|
def line_snippet(lines: list[str], index: int, radius: int = 1) -> dict[str, Any]:
|
|
start = max(0, index - radius)
|
|
end = min(len(lines), index + radius + 1)
|
|
return {
|
|
"line": index + 1,
|
|
"text": "".join(lines[start:end]).strip(),
|
|
}
|
|
|
|
|
|
def search_modules(query: str, modules: list[dict[str, Any]], *, search_code: bool, limit: int) -> list[dict[str, Any]]:
|
|
result = []
|
|
flat_modules = []
|
|
for module in modules:
|
|
flat_modules.append(module)
|
|
flat_modules.extend(module.get("extension_overlays") or [])
|
|
for module in flat_modules:
|
|
payload = slim(module, ["name", "kind", "origin", "effective_action", "relative_path", "path", "size"])
|
|
if matches(query, payload):
|
|
add_match(result, area="module", score=65, payload=payload, evidence={"path": module.get("path")})
|
|
if not search_code:
|
|
continue
|
|
path = Path(str(module.get("path") or ""))
|
|
if not path.is_file():
|
|
continue
|
|
try:
|
|
lines = path.read_text(encoding="utf-8-sig").splitlines(keepends=True)
|
|
except UnicodeDecodeError:
|
|
lines = path.read_text(encoding="cp1251", errors="replace").splitlines(keepends=True)
|
|
wanted = norm(query)
|
|
for index, line in enumerate(lines):
|
|
if wanted and wanted in norm(line):
|
|
add_match(
|
|
result,
|
|
area="module.code",
|
|
score=60,
|
|
payload=payload,
|
|
evidence={"path": str(path), **line_snippet(lines, index)},
|
|
)
|
|
if len(result) >= limit:
|
|
return result[:limit]
|
|
return result[:limit]
|
|
|
|
|
|
def build_search(
|
|
index: dict[str, Any],
|
|
*,
|
|
kind: str,
|
|
name: str,
|
|
query: str,
|
|
view: str,
|
|
extension: str | None,
|
|
search_code: bool,
|
|
max_form_items: int,
|
|
limit: int,
|
|
) -> dict[str, Any]:
|
|
resolution = resolve_object(index, kind=kind, name=name, limit=100)
|
|
canonical = resolution.get("canonical")
|
|
if not canonical:
|
|
raise SystemExit(f"Object not found: {kind}.{name}")
|
|
metadata = build_object_metadata(index, kind=kind, name=name, view=view, extension=extension, include_storage=False)
|
|
forms, form_counts = view_forms(canonical, resolution, view=view, extension=extension)
|
|
modules, module_counts = view_modules(canonical, resolution, view=view, extension=extension)
|
|
|
|
matches_found = []
|
|
remaining = limit
|
|
for group in (
|
|
search_metadata(query, metadata, remaining),
|
|
search_forms(query, forms, max_form_items=max_form_items, limit=remaining),
|
|
search_modules(query, modules, search_code=search_code, limit=remaining),
|
|
):
|
|
matches_found.extend(group)
|
|
remaining = limit - len(matches_found)
|
|
if remaining <= 0:
|
|
break
|
|
matches_found.sort(key=lambda row: row.get("score", 0), reverse=True)
|
|
return {
|
|
"schema": "onec_object_context_search.v1",
|
|
"view": view,
|
|
"extension": extension if view == "extension" else None,
|
|
"query": {"kind": kind, "name": name, "text": query, "search_code": search_code, "max_form_items": max_form_items, "limit": limit},
|
|
"object": {
|
|
"kind": canonical.get("kind"),
|
|
"name": canonical.get("name"),
|
|
"synonym": canonical.get("synonym"),
|
|
"uuid": canonical.get("guid"),
|
|
},
|
|
"matches": matches_found[:limit],
|
|
"counts": {
|
|
"matches": len(matches_found[:limit]),
|
|
"forms": form_counts,
|
|
"modules": module_counts,
|
|
"metadata_attributes": len(metadata.get("attributes") or []),
|
|
"metadata_tabular_sections": len(metadata.get("tabular_sections") or []),
|
|
},
|
|
}
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description="Search one 1C object context.")
|
|
parser.add_argument("--index", type=Path, required=True)
|
|
parser.add_argument("--kind")
|
|
parser.add_argument("--name")
|
|
parser.add_argument("--text")
|
|
parser.add_argument("--kind-b64")
|
|
parser.add_argument("--name-b64")
|
|
parser.add_argument("--text-b64")
|
|
parser.add_argument("--view", choices=["effective", "base", "extension"], default="effective")
|
|
parser.add_argument("--extension")
|
|
parser.add_argument("--search-code", action="store_true")
|
|
parser.add_argument("--max-form-items", type=int, default=2000)
|
|
parser.add_argument("--limit", type=int, default=50)
|
|
parser.add_argument("--output", type=Path)
|
|
args = parser.parse_args()
|
|
|
|
kind = decode_arg(args.kind, args.kind_b64)
|
|
name = decode_arg(args.name, args.name_b64)
|
|
text = decode_arg(args.text, args.text_b64)
|
|
if not kind or not name or not text:
|
|
raise SystemExit("Use --kind/--name/--text or their --*-b64 variants.")
|
|
result = build_search(
|
|
load_json(args.index),
|
|
kind=kind,
|
|
name=name,
|
|
query=text,
|
|
view=args.view,
|
|
extension=args.extension,
|
|
search_code=args.search_code,
|
|
max_form_items=args.max_form_items,
|
|
limit=args.limit,
|
|
)
|
|
output = json.dumps(result, ensure_ascii=False, indent=2) + "\n"
|
|
if args.output:
|
|
args.output.parent.mkdir(parents=True, exist_ok=True)
|
|
args.output.write_text(output, encoding="utf-8")
|
|
print(json.dumps({"output": str(args.output), "counts": result["counts"], "view": result["view"]}, ensure_ascii=False))
|
|
else:
|
|
print(output)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|