unrhodecode/rhodecode/apps/search/views.py

# Copyright (C) 2011-2024 RhodeCode GmbH
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License, version 3
# (only), as published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.
#
# This program is dual-licensed. If you wish to learn more about the
# RhodeCode Enterprise Edition, including its added features, Support services,
# and proprietary license terms, please see https://rhodecode.com/licenses/

import logging
import urllib.request
import urllib.parse
import urllib.error

from webhelpers2.html.tools import update_params

from rhodecode.apps._base import BaseAppView, RepoAppView, RepoGroupAppView
from rhodecode.lib.auth import LoginRequired, HasRepoPermissionAnyDecorator, HasRepoGroupPermissionAnyDecorator
from rhodecode.lib.helpers import Page
from rhodecode.lib.utils2 import safe_str
from rhodecode.lib.index import searcher_from_config
from rhodecode.model import validation_schema
from rhodecode.model.validation_schema.schemas import search_schema

log = logging.getLogger(__name__)


def perform_search(request, tmpl_context, repo_name=None, repo_group_name=None):
    searcher = searcher_from_config(request.registry.settings)
    formatted_results = []
    execution_time = ""

    schema = search_schema.SearchParamsSchema()
    search_tags = []
    search_params = {}
    errors = []

    try:
        search_params = schema.deserialize(
            dict(
                search_query=request.GET.get("q"),
                search_type=request.GET.get("type"),
                search_sort=request.GET.get("sort"),
                search_max_lines=request.GET.get("max_lines"),
                page_limit=request.GET.get("page_limit"),
                requested_page=request.GET.get("page"),
            )
        )
    except validation_schema.Invalid as e:
        errors = e.children

    def url_generator(page_num):
        query_params = {
            "page": page_num,
            "q": safe_str(search_query),
            "type": safe_str(search_type),
            "max_lines": search_max_lines,
            "sort": search_sort,
        }

        return "?" + urllib.parse.urlencode(query_params)

    c = tmpl_context
    search_query = search_params.get("search_query")
    search_type = search_params.get("search_type")
    search_sort = search_params.get("search_sort")
    search_max_lines = search_params.get("search_max_lines")
    if search_params.get("search_query"):
        page_limit = search_params["page_limit"]
        requested_page = search_params["requested_page"]

        try:
            search_result = searcher.search(
                search_query,
                search_type,
                c.auth_user,
                repo_name,
                repo_group_name,
                requested_page=requested_page,
                page_limit=page_limit,
                sort=search_sort,
            )

            search_result_count = search_result["count"]["value"]

            formatted_results = Page(
                search_result["results"],
                page=requested_page,
                item_count=search_result_count,
                items_per_page=page_limit,
                url_maker=url_generator,
            )
        finally:
            searcher.cleanup()

        search_tags = searcher.extract_search_tags(search_query)

        if not search_result["error"]:
            execution_time = "{} results ({:.4f} seconds)".format(search_result_count, search_result["runtime"])
        elif not errors:
            node = schema["search_query"]
            errors = [validation_schema.Invalid(node, search_result["error"])]

    c.perm_user = c.auth_user
    c.repo_name = repo_name
    c.repo_group_name = repo_group_name
    c.errors = errors
    c.formatted_results = formatted_results
    c.runtime = execution_time
    c.cur_query = search_query
    c.search_type = search_type
    c.searcher = searcher
    c.search_tags = search_tags

    direction, sort_field = searcher.get_sort(search_type, search_sort)
    sort_definition = searcher.sort_def(search_type, direction, sort_field)
    c.sort = ""
    c.sort_tag = None
    c.sort_tag_dir = direction
    if sort_definition:
        c.sort = f"{direction}:{sort_field}"
        c.sort_tag = sort_field


class SearchView(BaseAppView):
    def load_default_context(self):
        c = self._get_local_tmpl_context()
        return c

    @LoginRequired()
    def search(self):
        c = self.load_default_context()
        perform_search(self.request, c)
        return self._get_template_context(c)


class SearchRepoView(RepoAppView):
    def load_default_context(self):
        c = self._get_local_tmpl_context()
        c.active = "search"
        return c

    @LoginRequired()
    @HasRepoPermissionAnyDecorator("repository.read", "repository.write", "repository.admin")
    def search_repo(self):
        c = self.load_default_context()
        perform_search(self.request, c, repo_name=self.db_repo_name)
        return self._get_template_context(c)


class SearchRepoGroupView(RepoGroupAppView):
    def load_default_context(self):
        c = self._get_local_tmpl_context()
        c.active = "search"
        return c

    @LoginRequired()
    @HasRepoGroupPermissionAnyDecorator("group.read", "group.write", "group.admin")
    def search_repo_group(self):
        c = self.load_default_context()
        perform_search(self.request, c, repo_group_name=self.db_repo_group_name)
        return self._get_template_context(c)