index/index1c/search.py

"""Поиск по единому 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,
    names_only: bool = False,
) -> list[dict[str, Any]]:
    fts = _fts_query(query)
    snip_col = "'' AS snip" if names_only else "snippet(modules_fts, 4, '[', ']', '…', 16) AS snip"
    sql = f"""
        SELECT m.config, m.rel_path, m.owner, m.module_kind, m.symbols, m.line_count,
               o.synonym AS owner_synonym,
               {snip_col},
               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
        LEFT JOIN objects o ON o.config = m.config AND o.full_name = m.owner
        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 m.config, m.rel_path, m.owner, m.module_kind, m.symbols, m.line_count,
                   o.synonym AS owner_synonym, '' AS snip, 0 AS score
            FROM modules m
            LEFT JOIN objects o ON o.config = m.config AND o.full_name = m.owner
            WHERE m.rel_path LIKE ? OR m.owner LIKE ? OR m.symbols LIKE ?
        """
        params2: list[Any] = [like, like, like]
        if configs:
            sql2 += f" AND m.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,
) -> tuple[list[dict[str, Any]], dict[str, int]]:
    """Обратные ссылки на объект.

    Возвращает (строки, totals_by_config).
    При поиске по нескольким конфигурациям ``limit`` — максимум **на каждую**
    конфигурацию (чтобы crm3-26 не вытеснял crm3-dev из-за ORDER BY + LIMIT).
    """
    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"

    where = "r.to_name = ?"
    params: list[Any] = [name]
    if kind_ref:
        where += " AND (r.to_kind = ? OR r.to_kind = ?)"
        params.extend([kind_ref, kind])
    if configs:
        where += f" AND r.config IN ({','.join('?' * len(configs))})"
        params.extend(configs)

    # Итоги по конфигурациям (без LIMIT)
    totals: dict[str, int] = {
        r["config"]: int(r["n"])
        for r in conn.execute(
            f"SELECT r.config, COUNT(*) AS n FROM refs r WHERE {where} GROUP BY r.config",
            params,
        )
    }

    # limit на каждую конфигурацию — иначе алфавитный ORDER BY отрезает остальные
    sql = f"""
        WITH matched AS (
            SELECT
                r.config,
                r.from_full_name,
                r.from_field,
                r.to_kind,
                r.to_name,
                f.types AS field_types,
                f.refs_json AS field_refs_json,
                ROW_NUMBER() OVER (
                    PARTITION BY r.config
                    ORDER BY r.from_full_name, r.from_field
                ) AS rn
            FROM refs r
            LEFT JOIN objects o
                ON o.config = r.config AND o.full_name = r.from_full_name
            LEFT JOIN fields f
                ON f.object_id = o.id
                AND (
                    (COALESCE(f.tabular_section, '') = '' AND f.name = r.from_field)
                    OR (f.tabular_section || '.' || f.name = r.from_field)
                )
            WHERE {where}
        )
        SELECT *
        FROM matched
        WHERE rn <= ?
        ORDER BY config, from_full_name, from_field
    """
    rows = [dict(r) for r in conn.execute(sql, [*params, limit])]

    for row in rows:
        types_s = row.get("field_types") or ""
        refs_raw = row.get("field_refs_json") or "[]"
        try:
            refs_list = json.loads(refs_raw) if refs_raw else []
        except json.JSONDecodeError:
            refs_list = []
        type_parts = [t.strip() for t in types_s.split(",") if t.strip()]
        # составной: несколько cfg:* или несколько ссылок в типе
        composite = len(refs_list) > 1 or len(type_parts) > 1
        row["is_composite"] = composite
        others: list[str] = []
        target_full = f"{row['to_kind']}.{row['to_name']}"
        for ref in refs_list:
            label = f"{ref.get('kind')}.{ref.get('name')}"
            if label != target_full and label not in others:
                others.append(label)
        for t in type_parts:
            # cfg:DocumentRef.X → DocumentRef.X
            short = t[4:] if t.startswith("cfg:") else t
            if short != target_full and short not in others and "Ref." in short:
                others.append(short)
        row["other_types"] = others

    return rows, totals


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 []