Function _keyword_search_nodes

skills/wiki-store/scripts/wiki_db.py:522–578

(
    user_id: str,
    query_text: str,
    node_type: Optional[str],
    limit: int,
)

def _keyword_search_nodes(
    user_id: str,
    query_text: str,
    node_type: Optional[str],
    limit: int,
) -> List[Dict[str, Any]]:
    conn = db_ops.get_connection()
    cursor = conn.cursor()
    params: List[Any] = [_fts_query(query_text), user_id]
    type_filter = ""
    if node_type:
        type_filter = " AND n.node_type = ?"
        params.append(node_type)
    params.append(max(1, int(limit)))
    try:
        rows = cursor.execute(
            f"""
            SELECT n.*, bm25(wiki_nodes_fts) AS search_rank
            FROM wiki_nodes_fts
            JOIN wiki_nodes n ON n.id = wiki_nodes_fts.rowid
            WHERE wiki_nodes_fts MATCH ?
              AND n.user_id = ?
              {type_filter}
            ORDER BY search_rank, n.updated_at DESC
            LIMIT ?
            """,
            params,
        ).fetchall()
        conn.close()
        nodes = _rows_to_nodes(rows)
        for index, node in enumerate(nodes):
            node["keyword_score"] = 1.0 - (index / max(1, len(nodes)))
        return nodes
    except sqlite3.OperationalError:
        like = f"%{query_text}%"
        params = [user_id, like, like, like]
        type_filter = ""
        if node_type:
            type_filter = " AND node_type = ?"
            params.append(node_type)
        params.append(max(1, int(limit)))
        rows = cursor.execute(
            f"""
            SELECT * FROM wiki_nodes
            WHERE user_id = ?
              AND (title LIKE ? OR body LIKE ? OR keywords LIKE ?)
              {type_filter}
            ORDER BY updated_at DESC, id DESC
            LIMIT ?
            """,
            params,
        ).fetchall()
        conn.close()
        nodes = _rows_to_nodes(rows)
        for index, node in enumerate(nodes):
            node["keyword_score"] = 1.0 - (index / max(1, len(nodes)))
        return nodes