mihailkudravcev
288 lines
9.4 KiB
Python
288 lines
9.4 KiB
Python
"""Поиск по единому 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 []
|