|
|
import logging |
|
|
from typing import Optional |
|
|
|
|
|
import requests |
|
|
|
|
|
from configs import dify_config |
|
|
from services.recommend_app.buildin.buildin_retrieval import BuildInRecommendAppRetrieval |
|
|
from services.recommend_app.recommend_app_base import RecommendAppRetrievalBase |
|
|
from services.recommend_app.recommend_app_type import RecommendAppType |
|
|
|
|
|
logger = logging.getLogger(__name__) |
|
|
|
|
|
|
|
|
class RemoteRecommendAppRetrieval(RecommendAppRetrievalBase): |
|
|
""" |
|
|
Retrieval recommended app from dify official |
|
|
""" |
|
|
|
|
|
def get_recommend_app_detail(self, app_id: str): |
|
|
try: |
|
|
result = self.fetch_recommended_app_detail_from_dify_official(app_id) |
|
|
except Exception as e: |
|
|
logger.warning(f"fetch recommended app detail from dify official failed: {e}, switch to built-in.") |
|
|
result = BuildInRecommendAppRetrieval.fetch_recommended_app_detail_from_builtin(app_id) |
|
|
return result |
|
|
|
|
|
def get_recommended_apps_and_categories(self, language: str) -> dict: |
|
|
try: |
|
|
result = self.fetch_recommended_apps_from_dify_official(language) |
|
|
except Exception as e: |
|
|
logger.warning(f"fetch recommended apps from dify official failed: {e}, switch to built-in.") |
|
|
result = BuildInRecommendAppRetrieval.fetch_recommended_apps_from_builtin(language) |
|
|
return result |
|
|
|
|
|
def get_type(self) -> str: |
|
|
return RecommendAppType.REMOTE |
|
|
|
|
|
@classmethod |
|
|
def fetch_recommended_app_detail_from_dify_official(cls, app_id: str) -> Optional[dict]: |
|
|
""" |
|
|
Fetch recommended app detail from dify official. |
|
|
:param app_id: App ID |
|
|
:return: |
|
|
""" |
|
|
domain = dify_config.HOSTED_FETCH_APP_TEMPLATES_REMOTE_DOMAIN |
|
|
url = f"{domain}/apps/{app_id}" |
|
|
response = requests.get(url, timeout=(3, 10)) |
|
|
if response.status_code != 200: |
|
|
return None |
|
|
|
|
|
return response.json() |
|
|
|
|
|
@classmethod |
|
|
def fetch_recommended_apps_from_dify_official(cls, language: str) -> dict: |
|
|
""" |
|
|
Fetch recommended apps from dify official. |
|
|
:param language: language |
|
|
:return: |
|
|
""" |
|
|
domain = dify_config.HOSTED_FETCH_APP_TEMPLATES_REMOTE_DOMAIN |
|
|
url = f"{domain}/apps?language={language}" |
|
|
response = requests.get(url, timeout=(3, 10)) |
|
|
if response.status_code != 200: |
|
|
raise ValueError(f"fetch recommended apps failed, status code: {response.status_code}") |
|
|
|
|
|
result = response.json() |
|
|
|
|
|
if "categories" in result: |
|
|
result["categories"] = sorted(result["categories"]) |
|
|
|
|
|
return result |
|
|
|
