TERES_fastapi_backend/api/apps/search_app.py

#
#  Copyright 2025 The InfiniFlow Authors. All Rights Reserved.
#
#  Licensed under the Apache License, Version 2.0 (the "License");
#  you may not use this file except in compliance with the License.
#  You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
#  Unless required by applicable law or agreed to in writing, software
#  distributed under the License is distributed on an "AS IS" BASIS,
#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#  See the License for the specific language governing permissions and
#  limitations under the License.
#

from typing import Optional

from fastapi import APIRouter, Depends, Query

from api.apps.models.auth_dependencies import get_current_user
from api.apps.models.search_models import (
    CreateSearchRequest,
    UpdateSearchRequest,
    DeleteSearchRequest,
    ListSearchAppsQuery,
    ListSearchAppsBody,
)

from api import settings
from api.constants import DATASET_NAME_LIMIT
from api.db import StatusEnum
from api.db.db_models import DB
from api.db.services import duplicate_name
from api.db.services.search_service import SearchService
from api.db.services.user_service import TenantService, UserTenantService
from api.utils import get_uuid
from api.utils.api_utils import (
    get_data_error_result,
    get_json_result,
    server_error_response,
)

# 创建路由器
router = APIRouter()


@router.post('/create')
async def create(
    request: CreateSearchRequest,
    current_user = Depends(get_current_user)
):
    """创建搜索应用"""
    req = request.model_dump(exclude_unset=True)
    search_name = req["name"]
    description = req.get("description", "")
    if not isinstance(search_name, str):
        return get_data_error_result(message="Search name must be string.")
    if search_name.strip() == "":
        return get_data_error_result(message="Search name can't be empty.")
    if len(search_name.encode("utf-8")) > 255:
        return get_data_error_result(message=f"Search name length is {len(search_name)} which is large than 255.")
    e, _ = TenantService.get_by_id(current_user.id)
    if not e:
        return get_data_error_result(message="Authorized identity.")

    search_name = search_name.strip()
    search_name = duplicate_name(SearchService.query, name=search_name, tenant_id=current_user.id, status=StatusEnum.VALID.value)

    req["id"] = get_uuid()
    req["name"] = search_name
    req["description"] = description
    req["tenant_id"] = current_user.id
    req["created_by"] = current_user.id
    with DB.atomic():
        try:
            if not SearchService.save(**req):
                return get_data_error_result()
            return get_json_result(data={"search_id": req["id"]})
        except Exception as e:
            return server_error_response(e)


@router.post('/update')
async def update(
    request: UpdateSearchRequest,
    current_user = Depends(get_current_user)
):
    """更新搜索应用"""
    req = request.model_dump(exclude_unset=True)
    if not isinstance(req["name"], str):
        return get_data_error_result(message="Search name must be string.")
    if req["name"].strip() == "":
        return get_data_error_result(message="Search name can't be empty.")
    if len(req["name"].encode("utf-8")) > DATASET_NAME_LIMIT:
        return get_data_error_result(message=f"Search name length is {len(req['name'])} which is large than {DATASET_NAME_LIMIT}")
    req["name"] = req["name"].strip()
    tenant_id = req["tenant_id"]
    e, _ = TenantService.get_by_id(tenant_id)
    if not e:
        return get_data_error_result(message="Authorized identity.")

    search_id = req["search_id"]
    if not SearchService.accessible4deletion(search_id, current_user.id):
        return get_json_result(data=False, message="No authorization.", code=settings.RetCode.AUTHENTICATION_ERROR)

    # 验证不允许的参数
    not_allowed = ["id", "created_by", "create_time", "update_time", "create_date", "update_date"]
    for key in not_allowed:
        if key in req:
            del req[key]

    try:
        search_app = SearchService.query(tenant_id=tenant_id, id=search_id)[0]
        if not search_app:
            return get_json_result(data=False, message=f"Cannot find search {search_id}", code=settings.RetCode.DATA_ERROR)

        if req["name"].lower() != search_app.name.lower() and len(SearchService.query(name=req["name"], tenant_id=tenant_id, status=StatusEnum.VALID.value)) >= 1:
            return get_data_error_result(message="Duplicated search name.")

        if "search_config" in req:
            current_config = search_app.search_config or {}
            new_config = req["search_config"]

            if not isinstance(new_config, dict):
                return get_data_error_result(message="search_config must be a JSON object")

            updated_config = {**current_config, **new_config}
            req["search_config"] = updated_config

        req.pop("search_id", None)
        req.pop("tenant_id", None)

        updated = SearchService.update_by_id(search_id, req)
        if not updated:
            return get_data_error_result(message="Failed to update search")

        e, updated_search = SearchService.get_by_id(search_id)
        if not e:
            return get_data_error_result(message="Failed to fetch updated search")

        return get_json_result(data=updated_search.to_dict())

    except Exception as e:
        return server_error_response(e)


@router.get('/detail')
async def detail(
    search_id: str = Query(..., description="搜索应用ID"),
    current_user = Depends(get_current_user)
):
    """获取搜索应用详情"""
    try:
        tenants = UserTenantService.query(user_id=current_user.id)
        for tenant in tenants:
            if SearchService.query(tenant_id=tenant.tenant_id, id=search_id):
                break
        else:
            return get_json_result(data=False, message="Has no permission for this operation.", code=settings.RetCode.OPERATING_ERROR)

        search = SearchService.get_detail(search_id)
        if not search:
            return get_data_error_result(message="Can't find this Search App!")
        return get_json_result(data=search)
    except Exception as e:
        return server_error_response(e)


@router.post('/list')
async def list_search_app(
    query: ListSearchAppsQuery = Depends(),
    body: Optional[ListSearchAppsBody] = None,
    current_user = Depends(get_current_user)
):
    """列出搜索应用"""
    if body is None:
        body = ListSearchAppsBody()
    
    keywords = query.keywords or ""
    page_number = int(query.page or 0)
    items_per_page = int(query.page_size or 0)
    orderby = query.orderby or "create_time"
    desc = query.desc.lower() == "true" if query.desc else True

    owner_ids = body.owner_ids or [] if body else []
    try:
        if not owner_ids:
            # tenants = TenantService.get_joined_tenants_by_user_id(current_user.id)
            # tenants = [m["tenant_id"] for m in tenants]
            tenants = []
            search_apps, total = SearchService.get_by_tenant_ids(tenants, current_user.id, page_number, items_per_page, orderby, desc, keywords)
        else:
            tenants = owner_ids
            search_apps, total = SearchService.get_by_tenant_ids(tenants, current_user.id, 0, 0, orderby, desc, keywords)
            search_apps = [search_app for search_app in search_apps if search_app["tenant_id"] in tenants]
            total = len(search_apps)
            if page_number and items_per_page:
                search_apps = search_apps[(page_number - 1) * items_per_page : page_number * items_per_page]
        return get_json_result(data={"search_apps": search_apps, "total": total})
    except Exception as e:
        return server_error_response(e)


@router.post('/rm')
async def rm(
    request: DeleteSearchRequest,
    current_user = Depends(get_current_user)
):
    """删除搜索应用"""
    search_id = request.search_id
    if not SearchService.accessible4deletion(search_id, current_user.id):
        return get_json_result(data=False, message="No authorization.", code=settings.RetCode.AUTHENTICATION_ERROR)

    try:
        if not SearchService.delete_by_id(search_id):
            return get_data_error_result(message=f"Failed to delete search App {search_id}")
        return get_json_result(data=True)
    except Exception as e:
        return server_error_response(e)