Underground-Digital/Workflow-Engine

1import logging2from typing import Optional3 4import requests5 6from configs import dify_config7from services.recommend_app.buildin.buildin_retrieval import BuildInRecommendAppRetrieval8from services.recommend_app.recommend_app_base import RecommendAppRetrievalBase9from services.recommend_app.recommend_app_type import RecommendAppType10 11logger = logging.getLogger(__name__)12 13 14class RemoteRecommendAppRetrieval(RecommendAppRetrievalBase):15    """16    Retrieval recommended app from dify official17    """18 19    def get_recommend_app_detail(self, app_id: str):20        try:21            result = self.fetch_recommended_app_detail_from_dify_official(app_id)22        except Exception as e:23            logger.warning(f"fetch recommended app detail from dify official failed: {e}, switch to built-in.")24            result = BuildInRecommendAppRetrieval.fetch_recommended_app_detail_from_builtin(app_id)25        return result26 27    def get_recommended_apps_and_categories(self, language: str) -> dict:28        try:29            result = self.fetch_recommended_apps_from_dify_official(language)30        except Exception as e:31            logger.warning(f"fetch recommended apps from dify official failed: {e}, switch to built-in.")32            result = BuildInRecommendAppRetrieval.fetch_recommended_apps_from_builtin(language)33        return result34 35    def get_type(self) -> str:36        return RecommendAppType.REMOTE37 38    @classmethod39    def fetch_recommended_app_detail_from_dify_official(cls, app_id: str) -> Optional[dict]:40        """41        Fetch recommended app detail from dify official.42        :param app_id: App ID43        :return:44        """45        domain = dify_config.HOSTED_FETCH_APP_TEMPLATES_REMOTE_DOMAIN46        url = f"{domain}/apps/{app_id}"47        response = requests.get(url, timeout=(3, 10))48        if response.status_code != 200:49            return None50 51        return response.json()52 53    @classmethod54    def fetch_recommended_apps_from_dify_official(cls, language: str) -> dict:55        """56        Fetch recommended apps from dify official.57        :param language: language58        :return:59        """60        domain = dify_config.HOSTED_FETCH_APP_TEMPLATES_REMOTE_DOMAIN61        url = f"{domain}/apps?language={language}"62        response = requests.get(url, timeout=(3, 10))63        if response.status_code != 200:64            raise ValueError(f"fetch recommended apps failed, status code: {response.status_code}")65 66        result = response.json()67 68        if "categories" in result:69            result["categories"] = sorted(result["categories"])70 71        return result72