jingrowlocal/jingrow-framework
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

优化local_app_installer.py

Browse Source
This commit is contained in:
jingrow 2025-11-05 03:20:49 +08:00
parent 73f669d170
commit a9cc8d950f
1 changed files with 20 additions and 38 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

50
apps/jingrow/jingrow/api/local_app_installer.py
View File

@ -25,6 +25,7 @@ from jingrow.utils.auth import get_jingrow_cloud_url, get_jingrow_cloud_api_head
from jingrow.utils.export_app_package import export_app_package_from_local from jingrow.utils.export_app_package import export_app_package_from_local
from jingrow.config import Config from jingrow.config import Config
from jingrow.utils.app_manager import update_apps_txt from jingrow.utils.app_manager import update_apps_txt
import jingrow
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
router = APIRouter() router = APIRouter()
@ -122,8 +123,7 @@ async def install_app_from_upload(
# 对齐扫描安装的执行链 # 对齐扫描安装的执行链
try: try:
# 1. 添加到 Local Installed Apps PageType # 1. 添加到 Local Installed Apps PageType
from jingrow.utils.jingrow_api import get_single_pagetype pagetype_result = jingrow.get_single("Local Installed Apps")
pagetype_result = get_single_pagetype("Local Installed Apps")
if pagetype_result.get('success'): if pagetype_result.get('success'):
config = pagetype_result.get('config', {}) config = pagetype_result.get('config', {})
apps_list = config.get('local_installed_apps', []) apps_list = config.get('local_installed_apps', [])
@ -200,10 +200,8 @@ async def install_app_from_upload(
async def check_app_exists(app_name: str): async def check_app_exists(app_name: str):
"""检查应用或扩展包是否已安装""" """检查应用或扩展包是否已安装"""
try: try:
from jingrow.utils.jingrow_api import get_single_pagetype, get_record_list
# 1. 检查应用是否已安装 # 1. 检查应用是否已安装
result = get_single_pagetype("Local Installed Apps") result = jingrow.get_single("Local Installed Apps")
if result.get('success'): if result.get('success'):
config = result.get('config', {}) config = result.get('config', {})
@ -214,11 +212,10 @@ async def check_app_exists(app_name: str):
return {'exists': True, 'installed': True, 'type': 'app'} return {'exists': True, 'installed': True, 'type': 'app'}
# 2. 检查扩展包是否已安装（通过查询Package PageType） # 2. 检查扩展包是否已安装（通过查询Package PageType）
package_result = get_record_list("Package", filters=[["name", "=", app_name]], limit=1) package_data = jingrow.get_list("Package", filters=[["name", "=", app_name]], limit=1)
if package_result.get('success') and package_result.get('data'): if package_data:
# 如果找到了记录 # 如果找到了记录
if len(package_result.get('data', [])) > 0:
return {'exists': True, 'installed': False, 'type': 'package'} return {'exists': True, 'installed': False, 'type': 'package'}
return {'exists': False, 'installed': False} return {'exists': False, 'installed': False}
@ -230,12 +227,10 @@ async def check_app_exists(app_name: str):
async def get_installed_app_names(): async def get_installed_app_names():
"""获取所有已安装的应用名称列表""" """获取所有已安装的应用名称列表"""
try: try:
from jingrow.utils.jingrow_api import get_single_pagetype, get_record_list
installed = set() installed = set()
# 从 Local Installed Apps 获取应用列表 # 从 Local Installed Apps 获取应用列表
result = get_single_pagetype("Local Installed Apps") result = jingrow.get_single("Local Installed Apps")
if result.get('success'): if result.get('success'):
config = result.get('config', {}) config = result.get('config', {})
local_installed_apps = config.get('local_installed_apps', []) local_installed_apps = config.get('local_installed_apps', [])
@ -245,9 +240,8 @@ async def get_installed_app_names():
installed.add(app_name.lower()) installed.add(app_name.lower())
# 从 Package PageType 获取扩展包列表 # 从 Package PageType 获取扩展包列表
package_result = get_record_list("Package", fields=["name"], limit=1000) package_data = jingrow.get_list("Package", fields=["name"], limit=1000)
if package_result.get('success') and package_result.get('data'): for pkg in package_data:
for pkg in package_result.get('data', []):
pkg_name = pkg.get('name', '') pkg_name = pkg.get('name', '')
if pkg_name: if pkg_name:
installed.add(pkg_name.lower()) installed.add(pkg_name.lower())
@ -266,8 +260,7 @@ async def get_local_installed_apps(request: Request):
apps_dir = root / "apps" apps_dir = root / "apps"
# 获取已安装的App列表 - 通过get_single API获取 # 获取已安装的App列表 - 通过get_single API获取
from jingrow.utils.jingrow_api import get_single_pagetype result = jingrow.get_single("Local Installed Apps")
result = get_single_pagetype("Local Installed Apps")
installed_names = set() installed_names = set()
if result.get('success'): if result.get('success'):
config = result.get('config', {}) config = result.get('config', {})
@ -364,8 +357,7 @@ async def install_local_app(request: Request, app_name: str):
backend_dir = app_dir backend_dir = app_dir
# 检查是否已经安装 - 通过get_single API检查 # 检查是否已经安装 - 通过get_single API检查
from jingrow.utils.jingrow_api import get_single_pagetype result = jingrow.get_single("Local Installed Apps")
result = get_single_pagetype("Local Installed Apps")
if result.get('success'): if result.get('success'):
config = result.get('config', {}) config = result.get('config', {})
local_installed_apps = config.get('local_installed_apps', []) local_installed_apps = config.get('local_installed_apps', [])
@ -375,10 +367,8 @@ async def install_local_app(request: Request, app_name: str):
# 将App信息添加到Local Installed Apps PageType # 将App信息添加到Local Installed Apps PageType
try: try:
from jingrow.utils.jingrow_api import get_single_pagetype
# 获取当前应用列表 # 获取当前应用列表
result = get_single_pagetype("Local Installed Apps") result = jingrow.get_single("Local Installed Apps")
if result.get('success'): if result.get('success'):
config = result.get('config', {}) config = result.get('config', {})
apps_list = config.get('local_installed_apps', []) apps_list = config.get('local_installed_apps', [])
@ -448,9 +438,7 @@ async def ensure_jingrow_registered():
return return
try: try:
from jingrow.utils.jingrow_api import get_single_pagetype result = jingrow.get_single("Local Installed Apps")
result = get_single_pagetype("Local Installed Apps")
if not result.get('success'): if not result.get('success'):
return return
@ -475,9 +463,7 @@ async def get_installed_apps(request: Request):
await ensure_jingrow_registered() await ensure_jingrow_registered()
# 通过get_single API获取Local Installed Apps数据 # 通过get_single API获取Local Installed Apps数据
from jingrow.utils.jingrow_api import get_single_pagetype result = jingrow.get_single("Local Installed Apps")
result = get_single_pagetype("Local Installed Apps")
if not result.get('success'): if not result.get('success'):
return { return {
@ -585,8 +571,7 @@ async def install_from_git(repo_url: str = Form(...)):
if app_dir: if app_dir:
try: try:
from jingrow.utils.jingrow_api import get_single_pagetype pagetype_result = jingrow.get_single("Local Installed Apps")
pagetype_result = get_single_pagetype("Local Installed Apps")
apps_list = pagetype_result.get('config', {}).get('local_installed_apps', []) if pagetype_result.get('success') else [] apps_list = pagetype_result.get('config', {}).get('local_installed_apps', []) if pagetype_result.get('success') else []
app_exists = False app_exists = False
@ -670,8 +655,7 @@ async def install_from_url(url: str = Form(...)):
# 对齐扫描安装的执行链 # 对齐扫描安装的执行链
try: try:
# 1. 添加到 Local Installed Apps PageType # 1. 添加到 Local Installed Apps PageType
from jingrow.utils.jingrow_api import get_single_pagetype pagetype_result = jingrow.get_single("Local Installed Apps")
pagetype_result = get_single_pagetype("Local Installed Apps")
if pagetype_result.get('success'): if pagetype_result.get('success'):
config = pagetype_result.get('config', {}) config = pagetype_result.get('config', {})
apps_list = config.get('local_installed_apps', []) apps_list = config.get('local_installed_apps', [])
@ -901,10 +885,8 @@ async def _remove_from_database(app_name: str) -> Dict[str, Any]:
"""从数据库中删除应用记录""" """从数据库中删除应用记录"""
try: try:
# 使用API方式操作数据库 # 使用API方式操作数据库
from jingrow.utils.jingrow_api import get_single_pagetype
# 获取当前数据 # 获取当前数据
result = get_single_pagetype("Local Installed Apps") result = jingrow.get_single("Local Installed Apps")
if not result.get('success'): if not result.get('success'):
return {'success': True, 'message': '未找到Local Installed Apps记录'} return {'success': True, 'message': '未找到Local Installed Apps记录'}