Spaces:
Running
Running
| import json | |
| import logging | |
| from os import path | |
| from typing import Optional | |
| import requests | |
| from flask import current_app | |
| from constants.languages import languages | |
| from extensions.ext_database import db | |
| from models.model import App, RecommendedApp | |
| from services.app_service import AppService | |
| logger = logging.getLogger(__name__) | |
| class RecommendedAppService: | |
| builtin_data: Optional[dict] = None | |
| def get_recommended_apps_and_categories(cls, language: str) -> dict: | |
| """ | |
| Get recommended apps and categories. | |
| :param language: language | |
| :return: | |
| """ | |
| mode = current_app.config.get('HOSTED_FETCH_APP_TEMPLATES_MODE', 'remote') | |
| if mode == 'remote': | |
| try: | |
| result = cls._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 = cls._fetch_recommended_apps_from_builtin(language) | |
| elif mode == 'db': | |
| result = cls._fetch_recommended_apps_from_db(language) | |
| elif mode == 'builtin': | |
| result = cls._fetch_recommended_apps_from_builtin(language) | |
| else: | |
| raise ValueError(f'invalid fetch recommended apps mode: {mode}') | |
| if not result.get('recommended_apps') and language != 'en-US': | |
| result = cls._fetch_recommended_apps_from_builtin('en-US') | |
| return result | |
| def _fetch_recommended_apps_from_db(cls, language: str) -> dict: | |
| """ | |
| Fetch recommended apps from db. | |
| :param language: language | |
| :return: | |
| """ | |
| recommended_apps = db.session.query(RecommendedApp).filter( | |
| RecommendedApp.is_listed == True, | |
| RecommendedApp.language == language | |
| ).all() | |
| if len(recommended_apps) == 0: | |
| recommended_apps = db.session.query(RecommendedApp).filter( | |
| RecommendedApp.is_listed == True, | |
| RecommendedApp.language == languages[0] | |
| ).all() | |
| categories = set() | |
| recommended_apps_result = [] | |
| for recommended_app in recommended_apps: | |
| app = recommended_app.app | |
| if not app or not app.is_public: | |
| continue | |
| site = app.site | |
| if not site: | |
| continue | |
| recommended_app_result = { | |
| 'id': recommended_app.id, | |
| 'app': { | |
| 'id': app.id, | |
| 'name': app.name, | |
| 'mode': app.mode, | |
| 'icon': app.icon, | |
| 'icon_background': app.icon_background | |
| }, | |
| 'app_id': recommended_app.app_id, | |
| 'description': site.description, | |
| 'copyright': site.copyright, | |
| 'privacy_policy': site.privacy_policy, | |
| 'custom_disclaimer': site.custom_disclaimer, | |
| 'category': recommended_app.category, | |
| 'position': recommended_app.position, | |
| 'is_listed': recommended_app.is_listed | |
| } | |
| recommended_apps_result.append(recommended_app_result) | |
| categories.add(recommended_app.category) # add category to categories | |
| return {'recommended_apps': recommended_apps_result, 'categories': sorted(list(categories))} | |
| def _fetch_recommended_apps_from_dify_official(cls, language: str) -> dict: | |
| """ | |
| Fetch recommended apps from dify official. | |
| :param language: language | |
| :return: | |
| """ | |
| domain = current_app.config.get('HOSTED_FETCH_APP_TEMPLATES_REMOTE_DOMAIN', 'https://tmpl.dify.ai') | |
| 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}') | |
| return response.json() | |
| def _fetch_recommended_apps_from_builtin(cls, language: str) -> dict: | |
| """ | |
| Fetch recommended apps from builtin. | |
| :param language: language | |
| :return: | |
| """ | |
| builtin_data = cls._get_builtin_data() | |
| return builtin_data.get('recommended_apps', {}).get(language) | |
| def get_recommend_app_detail(cls, app_id: str) -> Optional[dict]: | |
| """ | |
| Get recommend app detail. | |
| :param app_id: app id | |
| :return: | |
| """ | |
| mode = current_app.config.get('HOSTED_FETCH_APP_TEMPLATES_MODE', 'remote') | |
| if mode == 'remote': | |
| try: | |
| result = cls._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 = cls._fetch_recommended_app_detail_from_builtin(app_id) | |
| elif mode == 'db': | |
| result = cls._fetch_recommended_app_detail_from_db(app_id) | |
| elif mode == 'builtin': | |
| result = cls._fetch_recommended_app_detail_from_builtin(app_id) | |
| else: | |
| raise ValueError(f'invalid fetch recommended app detail mode: {mode}') | |
| return result | |
| 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 = current_app.config.get('HOSTED_FETCH_APP_TEMPLATES_REMOTE_DOMAIN', 'https://tmpl.dify.ai') | |
| url = f'{domain}/apps/{app_id}' | |
| response = requests.get(url, timeout=(3, 10)) | |
| if response.status_code != 200: | |
| return None | |
| return response.json() | |
| def _fetch_recommended_app_detail_from_db(cls, app_id: str) -> Optional[dict]: | |
| """ | |
| Fetch recommended app detail from db. | |
| :param app_id: App ID | |
| :return: | |
| """ | |
| # is in public recommended list | |
| recommended_app = db.session.query(RecommendedApp).filter( | |
| RecommendedApp.is_listed == True, | |
| RecommendedApp.app_id == app_id | |
| ).first() | |
| if not recommended_app: | |
| return None | |
| # get app detail | |
| app_model = db.session.query(App).filter(App.id == app_id).first() | |
| if not app_model or not app_model.is_public: | |
| return None | |
| app_service = AppService() | |
| export_str = app_service.export_app(app_model) | |
| return { | |
| 'id': app_model.id, | |
| 'name': app_model.name, | |
| 'icon': app_model.icon, | |
| 'icon_background': app_model.icon_background, | |
| 'mode': app_model.mode, | |
| 'export_data': export_str | |
| } | |
| def _fetch_recommended_app_detail_from_builtin(cls, app_id: str) -> Optional[dict]: | |
| """ | |
| Fetch recommended app detail from builtin. | |
| :param app_id: App ID | |
| :return: | |
| """ | |
| builtin_data = cls._get_builtin_data() | |
| return builtin_data.get('app_details', {}).get(app_id) | |
| def _get_builtin_data(cls) -> dict: | |
| """ | |
| Get builtin data. | |
| :return: | |
| """ | |
| if cls.builtin_data: | |
| return cls.builtin_data | |
| root_path = current_app.root_path | |
| with open(path.join(root_path, 'constants', 'recommended_apps.json'), encoding='utf-8') as f: | |
| json_data = f.read() | |
| data = json.loads(json_data) | |
| cls.builtin_data = data | |
| return cls.builtin_data | |
| def fetch_all_recommended_apps_and_export_datas(cls): | |
| """ | |
| Fetch all recommended apps and export datas | |
| :return: | |
| """ | |
| templates = { | |
| "recommended_apps": {}, | |
| "app_details": {} | |
| } | |
| for language in languages: | |
| try: | |
| result = cls._fetch_recommended_apps_from_dify_official(language) | |
| except Exception as e: | |
| logger.warning(f'fetch recommended apps from dify official failed: {e}, skip.') | |
| continue | |
| templates['recommended_apps'][language] = result | |
| for recommended_app in result.get('recommended_apps'): | |
| app_id = recommended_app.get('app_id') | |
| # get app detail | |
| app_detail = cls._fetch_recommended_app_detail_from_dify_official(app_id) | |
| if not app_detail: | |
| continue | |
| templates['app_details'][app_id] = app_detail | |
| return templates | |