tjdify/api/services/app_service.py

import json
import logging
from datetime import UTC, datetime
from typing import Optional, cast

from flask_login import current_user  # type: ignore
from flask_sqlalchemy.pagination import Pagination
from sqlalchemy import select, union
from sqlalchemy.orm import aliased

from configs import dify_config
from constants.model_template import default_app_templates
from core.agent.entities import AgentToolEntity
from core.errors.error import LLMBadRequestError, ProviderTokenNotInitError
from core.model_manager import ModelManager
from core.model_runtime.entities.model_entities import ModelPropertyKey, ModelType
from core.model_runtime.model_providers.__base.large_language_model import LargeLanguageModel
from core.tools.tool_manager import ToolManager
from core.tools.utils.configuration import ToolParameterConfigurationManager
from events.app_event import app_was_created
from extensions.ext_database import db
from models.account import Account
from models.model import App, AppMode, AppModelConfig, AppPermissionAll
from models.tools import ApiToolProvider
from services.tag_service import TagService
from tasks.remove_app_and_related_data_task import remove_app_and_related_data_task


class AppService:

    def get_paginate_apps(self, user_id: str, tenant_id: str, args: dict) -> Pagination | None:
        """
        Get app list with pagination
        :param user_id: user id
        :param tenant_id: tenant id
        :param args: request args
        :return:
        """
        filters = [App.tenant_id == tenant_id, App.is_universal == False]

        if args["mode"] == "workflow":
            filters.append(App.mode == AppMode.WORKFLOW.value)
        elif args["mode"] == "completion":
            filters.append(App.mode == AppMode.COMPLETION.value)
        elif args["mode"] == "chat":
            filters.append(App.mode == AppMode.CHAT.value)
        elif args["mode"] == "advanced-chat":
            filters.append(App.mode == AppMode.ADVANCED_CHAT.value)
        elif args["mode"] == "agent-chat":
            filters.append(App.mode == AppMode.AGENT_CHAT.value)
        elif args["mode"] == "channel":
            filters.append(App.mode == AppMode.CHANNEL.value)

        if args.get("is_created_by_me", False):
            filters.append(App.created_by == user_id)
        if args.get("name"):
            name = args["name"][:30]
            filters.append(App.name.ilike(f"%{name}%"))
        if args.get("tag_ids"):
            target_ids = TagService.get_target_ids_by_tag_ids("app", tenant_id, args["tag_ids"])
            if target_ids:
                filters.append(App.id.in_(target_ids))
            else:
                return None

        # 初始化 main_query
        main_query = db.select(App).where(*filters)
        # 添加 auth_type 参数
        auth_type = args.get("auth_type")
        # 根据 auth_type 构建不同的查询
        if auth_type is None:
            query1 = db.select(App.id).where(
                *filters,
                App.created_by == user_id
            )
            query2 = db.select(App.id).join(
                Account, App.dept_id == Account.dept_id
            ).where(
                *filters,
                App.edit_auth == 2,
                Account.id == user_id
            )
            query3 = db.select(App.id).join(
                AppPermissionAll, App.id == AppPermissionAll.app_id
            ).where(
                *filters,
                AppPermissionAll.has_read_permission == True,
                AppPermissionAll.account_id == user_id
            )
            union_query = union(query1, query2, query3).subquery()
            main_query = db.select(App).where(App.id.in_(select(union_query.c.id)))
        elif auth_type == "1":
            # 只查询部门编辑权限
            main_query = db.select(App).where(
                *filters,
                App.created_by == user_id
            )
        elif auth_type == "2":
            # 只查询部门编辑权限
            main_query = db.select(App).join(
                Account, App.dept_id == Account.dept_id
            ).where(
                *filters,
                App.edit_auth == 2,
                Account.id == user_id
            )
        elif auth_type == "3":
            # 只查询授权编辑权限
            main_query = db.select(App).join(
                AppPermissionAll, App.id == AppPermissionAll.app_id
            ).where(
                *filters,
                AppPermissionAll.has_read_permission == True,
                AppPermissionAll.account_id == user_id
            )
        if args.get("creator") or args.get("creator_dept"):
            # 使用别名避免重复连接
            creator_account = aliased(Account)
            main_query = main_query.join(creator_account, App.created_by == creator_account.id)

            if args.get("creator"):
                main_query = main_query.where(creator_account.name.ilike(f"%{args.get('creator')}%"))
            if args.get("creator_dept"):
                main_query = main_query.where(creator_account.dept_id == args.get("creator_dept"))

        print(str(main_query))

        app_models = db.paginate(
            main_query.order_by(App.created_at.desc()),
            page=args["page"],
            per_page=args["limit"],
            error_out=False,
        )

        return app_models

    def create_app(self, tenant_id: str, args: dict, account: Account) -> App:
        """
        Create app
        :param tenant_id: tenant id
        :param args: request args
        :param account: Account instance
        """
        app_mode = AppMode.value_of(args["mode"])
        app_template = default_app_templates[app_mode]

        # get model config
        default_model_config = app_template.get("model_config")
        default_model_config = default_model_config.copy() if default_model_config else None
        if default_model_config and "model" in default_model_config:
            # get model provider
            model_manager = ModelManager()

            # get default model instance
            try:
                model_instance = model_manager.get_default_model_instance(
                    tenant_id=account.current_tenant_id or "", model_type=ModelType.LLM
                )
            except (ProviderTokenNotInitError, LLMBadRequestError):
                model_instance = None
            except Exception as e:
                logging.exception(f"Get default model instance failed, tenant_id: {tenant_id}")
                model_instance = None

            if model_instance:
                if (
                    model_instance.model == default_model_config["model"]["name"]
                    and model_instance.provider == default_model_config["model"]["provider"]
                ):
                    default_model_dict = default_model_config["model"]
                else:
                    llm_model = cast(LargeLanguageModel, model_instance.model_type_instance)
                    model_schema = llm_model.get_model_schema(model_instance.model, model_instance.credentials)
                    if model_schema is None:
                        raise ValueError(f"model schema not found for model {model_instance.model}")

                    default_model_dict = {
                        "provider": model_instance.provider,
                        "name": model_instance.model,
                        "mode": model_schema.model_properties.get(ModelPropertyKey.MODE),
                        "completion_params": {},
                    }
            else:
                provider, model = model_manager.get_default_provider_model_name(
                    tenant_id=account.current_tenant_id or "", model_type=ModelType.LLM
                )
                default_model_config["model"]["provider"] = provider
                default_model_config["model"]["name"] = model
                default_model_dict = default_model_config["model"]

            default_model_config["model"] = json.dumps(default_model_dict)

        app = App(**app_template["app"])
        app.name = args["name"]
        app.description = args.get("description", "")
        app.mode = args["mode"]
        app.icon_type = args.get("icon_type", "emoji")
        app.icon = args["icon"]
        app.icon_background = args["icon_background"]
        app.tenant_id = tenant_id
        app.api_rph = args.get("api_rph", 0)
        app.api_rpm = args.get("api_rpm", 0)
        app.created_by = account.id
        app.dept_id =account.dept_id
        app.updated_by = account.id

        db.session.add(app)
        db.session.flush()

        if default_model_config:
            app_model_config = AppModelConfig(**default_model_config)
            app_model_config.app_id = app.id
            app_model_config.created_by = account.id
            app_model_config.updated_by = account.id
            db.session.add(app_model_config)
            db.session.flush()

            app.app_model_config_id = app_model_config.id

        db.session.commit()

        app_was_created.send(app, account=account)

        return app

    def get_app_by_id(self, app: App) -> App:

        return app


    def get_app(self, app: App) -> App:
        """
        Get App
        """
        # get original app model config
        if app.mode == AppMode.AGENT_CHAT.value or app.is_agent:
            model_config = app.app_model_config
            agent_mode = model_config.agent_mode_dict
            # decrypt agent tool parameters if it's secret-input
            for tool in agent_mode.get("tools") or []:
                if not isinstance(tool, dict) or len(tool.keys()) <= 3:
                    continue
                agent_tool_entity = AgentToolEntity(**tool)
                # get tool
                try:
                    tool_runtime = ToolManager.get_agent_tool_runtime(
                        tenant_id=current_user.current_tenant_id,
                        app_id=app.id,
                        agent_tool=agent_tool_entity,
                    )
                    manager = ToolParameterConfigurationManager(
                        tenant_id=current_user.current_tenant_id,
                        tool_runtime=tool_runtime,
                        provider_name=agent_tool_entity.provider_id,
                        provider_type=agent_tool_entity.provider_type,
                        identity_id=f"AGENT.{app.id}",
                    )
                    # get decrypted parameters
                    if agent_tool_entity.tool_parameters:
                        parameters = manager.decrypt_tool_parameters(agent_tool_entity.tool_parameters or {})
                        masked_parameter = manager.mask_tool_parameters(parameters or {})
                    else:
                        masked_parameter = {}

                    # override tool parameters
                    tool["tool_parameters"] = masked_parameter
                except Exception as e:
                    pass

            # override agent mode
            model_config.agent_mode = json.dumps(agent_mode)
            class ModifiedApp(App):
                """
                Modified App class
                """
                def __init__(self, app):
                    self.__dict__.update(app.__dict__)

                @property
                def app_model_config(self):
                    return model_config

            app = ModifiedApp(app)
        return app

    def update_app(self, app: App, args: dict) -> App:
        """
        Update app
        :param app: App instance
        :param args: request args
        :return: App instance
        """
        app.name = args.get("name")
        app.description = args.get("description", "")
        app.icon_type = args.get("icon_type", "emoji")
        app.icon = args.get("icon")
        app.icon_background = args.get("icon_background")
        app.use_icon_as_answer_icon = args.get("use_icon_as_answer_icon", False)
        app.updated_by = current_user.id
        app.updated_at = datetime.now(UTC).replace(tzinfo=None)
        db.session.commit()

        return app

    def update_app_name(self, app: App, name: str) -> App:
        """
        Update app name
        :param app: App instance
        :param name: new name
        :return: App instance
        """
        app.name = name
        app.updated_by = current_user.id
        app.updated_at = datetime.now(UTC).replace(tzinfo=None)
        db.session.commit()

        return app

    def update_app_icon(self, app: App, icon: str, icon_background: str) -> App:
        """
        Update app icon
        :param app: App instance
        :param icon: new icon
        :param icon_background: new icon_background
        :return: App instance
        """
        app.icon = icon
        app.icon_background = icon_background
        app.updated_by = current_user.id
        app.updated_at = datetime.now(UTC).replace(tzinfo=None)
        db.session.commit()

        return app

    def update_app_site_status(self, app: App, enable_site: bool) -> App:
        """
        Update app site status
        :param app: App instance
        :param enable_site: enable site status
        :return: App instance
        """
        if enable_site == app.enable_site:
            return app

        app.enable_site = enable_site
        app.updated_by = current_user.id
        app.updated_at = datetime.now(UTC).replace(tzinfo=None)
        db.session.commit()

        return app

    def update_app_api_status(self, app: App, enable_api: bool) -> App:
        """
        Update app api status
        :param app: App instance
        :param enable_api: enable api status
        :return: App instance
        """
        if enable_api == app.enable_api:
            return app

        app.enable_api = enable_api
        app.updated_by = current_user.id
        app.updated_at = datetime.now(UTC).replace(tzinfo=None)
        db.session.commit()

        return app

    def delete_app(self, app: App) -> None:
        """
        Delete app
        :param app: App instance
        """
        db.session.delete(app)
        db.session.commit()

        # Trigger asynchronous deletion of app and related data
        remove_app_and_related_data_task.delay(tenant_id=app.tenant_id, app_id=app.id)

    def get_app_meta(self, app_model: App) -> dict:
        """
        Get app meta info
        :param app_model: app model
        :return:
        """
        app_mode = AppMode.value_of(app_model.mode)

        meta: dict = {"tool_icons": {}}

        if app_mode in {AppMode.ADVANCED_CHAT, AppMode.WORKFLOW}:
            workflow = app_model.workflow
            if workflow is None:
                return meta

            graph = workflow.graph_dict
            nodes = graph.get("nodes", [])
            tools = []
            for node in nodes:
                if node.get("data", {}).get("type") == "tool":
                    node_data = node.get("data", {})
                    tools.append(
                        {
                            "provider_type": node_data.get("provider_type"),
                            "provider_id": node_data.get("provider_id"),
                            "tool_name": node_data.get("tool_name"),
                            "tool_parameters": {},
                        }
                    )
        else:
            app_model_config: Optional[AppModelConfig] = app_model.app_model_config

            if not app_model_config:
                return meta

            agent_config = app_model_config.agent_mode_dict

            # get all tools
            tools = agent_config.get("tools", [])

        url_prefix = dify_config.CONSOLE_API_URL + "/console/api/workspaces/current/tool-provider/builtin/"

        for tool in tools:
            keys = list(tool.keys())
            if len(keys) >= 4:
                # current tool standard
                provider_type = tool.get("provider_type", "")
                provider_id = tool.get("provider_id", "")
                tool_name = tool.get("tool_name", "")
                if provider_type == "builtin":
                    meta["tool_icons"][tool_name] = url_prefix + provider_id + "/icon"
                elif provider_type == "api":
                    try:
                        provider: Optional[ApiToolProvider] = (
                            db.session.query(ApiToolProvider).filter(ApiToolProvider.id == provider_id).first()
                        )
                        if provider is None:
                            raise ValueError(f"provider not found for tool {tool_name}")
                        meta["tool_icons"][tool_name] = json.loads(provider.icon)
                    except:
                        meta["tool_icons"][tool_name] = {"background": "#252525", "content": "\ud83d\ude01"}

        return meta

    def update_app_edit_auth(self, app: App, edit_auth: str) -> App:
        """
        Update app
        :param app: App instance
        :param args: request args
        :return: App instance
        """
        if not app:
            raise ValueError("App not found")
        app.edit_auth = edit_auth
        app.updated_by = current_user.id
        app.updated_at = datetime.now(UTC).replace(tzinfo=None)

        db.session.commit()

        return app

    def get_app_read_permission(self):
        results = (
            db.session.query(AppPermissionAll.account_id, Account.email)
            .join(AppPermissionAll, Account.id == AppPermissionAll.account_id)
            .filter(AppPermissionAll.app_id == self, AppPermissionAll.has_read_permission == True)
            .all()
        )
        read_permission_list = []
        for row in results:
            read_permission_list.append({"id": row.account_id, "email": row.email})
        return read_permission_list

    def update_app_read_permission(self, app, read_permission_list):
        app_id = str(app.id)
        old_read_permission_list = AppService.get_app_read_permission(app_id)
        old_read_permission_set = {item.get("id") for item in old_read_permission_list}
        read_permission_set = {item.get("id") for item in read_permission_list}

        delete_read_permission_set = old_read_permission_set - read_permission_set

        delete_read_permission_list = list(delete_read_permission_set)
        for delete_read in delete_read_permission_list:
            delete_account_id = str(delete_read)
            AppPermissionAll.query.filter_by(app_id=app_id, account_id=delete_account_id).update(
                {"has_read_permission": False, "updated_by": self, "updated_at": datetime.now(UTC).replace(tzinfo=None)}
            )
            db.session.commit()
        for read_permission in read_permission_list:
            account_id = read_permission.get("id")
            email = read_permission.get("email")
            app_permission_all: Optional[AppPermissionAll] = AppPermissionAll.query.filter_by(
                app_id=app_id, account_id=account_id
            ).first()
            if app_permission_all == None:
                # 插入
                new_app_permission_all = AppPermissionAll()
                new_app_permission_all.app_id = app_id
                new_app_permission_all.account_id = account_id
                new_app_permission_all.created_by = self
                new_app_permission_all.updated_by = self
                new_app_permission_all.has_read_permission = True
                new_app_permission_all.email = email
                db.session.add(new_app_permission_all)
                db.session.commit()
                db.session.flush()
            else:
                if app_permission_all.has_read_permission == False:
                    app_permission_all.query.filter_by(app_id=app_id, account_id=account_id).update(
                        {
                            "has_read_permission": True,
                            "updated_by": self,
                            "updated_at": datetime.now(UTC).replace(tzinfo=None),
                        }
                    )
                    db.session.commit()