File size: 8,994 Bytes
4304c6d |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 |
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
@classmethod
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
@classmethod
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))}
@classmethod
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()
@classmethod
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)
@classmethod
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
@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 = 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()
@classmethod
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
}
@classmethod
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)
@classmethod
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
@classmethod
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
|