from fastapi import APIRouter, Depends, UploadFile, File, Form, HTTPException from app.core.database import get_db_connection from app.core.auth import get_current_admin_user from app.core.response import create_api_response from app.core.config import BASE_DIR, EXTERNAL_APPS_DIR, ALLOWED_IMAGE_EXTENSIONS, MAX_IMAGE_SIZE from app.utils.apk_parser import parse_apk_with_androguard from typing import Optional from pathlib import Path from pydantic import BaseModel import os import shutil import json import uuid import hashlib router = APIRouter() # APK上传配置 ALLOWED_APK_EXTENSIONS = {'.apk'} MAX_APK_SIZE = 200 * 1024 * 1024 # 200MB class CreateExternalAppRequest(BaseModel): app_name: str app_type: str # 'native' or 'web' app_info: str # JSON string icon_url: Optional[str] = None description: Optional[str] = None sort_order: int = 0 is_active: bool = True class UpdateExternalAppRequest(BaseModel): app_name: Optional[str] = None app_type: Optional[str] = None app_info: Optional[str] = None icon_url: Optional[str] = None description: Optional[str] = None sort_order: Optional[int] = None is_active: Optional[bool] = None @router.get("/external-apps", response_model=dict) async def get_external_apps( app_type: Optional[str] = None, is_active: Optional[bool] = None, page: int = 1, size: int = 50, current_user: dict = Depends(get_current_admin_user) ): """ 获取外部应用列表（管理后台接口） """ try: with get_db_connection() as conn: cursor = conn.cursor(dictionary=True) # 构建查询条件 where_clauses = [] params = [] if app_type: where_clauses.append("app_type = %s") params.append(app_type) if is_active is not None: where_clauses.append("is_active = %s") params.append(is_active) where_clause = " AND ".join(where_clauses) if where_clauses else "1=1" # 获取总数 count_query = f"SELECT COUNT(*) as total FROM external_apps WHERE {where_clause}" cursor.execute(count_query, params) total = cursor.fetchone()['total'] # 获取列表数据 offset = (page - 1) * size list_query = f""" SELECT ea.*, u.username as creator_username FROM external_apps ea LEFT JOIN users u ON ea.created_by = u.user_id WHERE {where_clause} ORDER BY ea.sort_order ASC, ea.created_at DESC LIMIT %s OFFSET %s """ cursor.execute(list_query, params + [size, offset]) apps = cursor.fetchall() # 解析 app_info JSON for app in apps: if app.get('app_info'): try: app['app_info'] = json.loads(app['app_info']) except: app['app_info'] = {} cursor.close() return create_api_response( code="200", message="获取成功", data={ "apps": apps, "total": total, "page": page, "size": size } ) except Exception as e: return create_api_response( code="500", message=f"获取外部应用列表失败: {str(e)}" ) @router.get("/external-apps/active", response_model=dict) async def get_active_external_apps(): """ 获取所有启用的外部应用（公开接口，供客户端调用） """ try: with get_db_connection() as conn: cursor = conn.cursor(dictionary=True) query = """ SELECT id, app_name, app_type, app_info, icon_url, description, sort_order FROM external_apps WHERE is_active = TRUE ORDER BY sort_order ASC, created_at DESC """ cursor.execute(query) apps = cursor.fetchall() # 解析 app_info JSON for app in apps: if app.get('app_info'): try: app['app_info'] = json.loads(app['app_info']) except: app['app_info'] = {} cursor.close() # 按类型分组 native_apps = [app for app in apps if app['app_type'] == 'native'] web_apps = [app for app in apps if app['app_type'] == 'web'] return create_api_response( code="200", message="获取成功", data={ "native": native_apps, "web": web_apps, "all": apps } ) except Exception as e: return create_api_response( code="500", message=f"获取外部应用失败: {str(e)}" ) @router.post("/external-apps", response_model=dict) async def create_external_app( request: CreateExternalAppRequest, current_user: dict = Depends(get_current_admin_user) ): """ 创建外部应用 """ try: # 验证 app_info 是否为有效JSON try: json.loads(request.app_info) except: return create_api_response( code="400", message="app_info 必须是有效的JSON格式" ) with get_db_connection() as conn: cursor = conn.cursor() query = """ INSERT INTO external_apps (app_name, app_type, app_info, icon_url, description, sort_order, is_active, created_by) VALUES (%s, %s, %s, %s, %s, %s, %s, %s) """ cursor.execute(query, ( request.app_name, request.app_type, request.app_info, request.icon_url, request.description, request.sort_order, request.is_active, current_user['user_id'] )) app_id = cursor.lastrowid conn.commit() cursor.close() return create_api_response( code="200", message="创建成功", data={"id": app_id} ) except Exception as e: return create_api_response( code="500", message=f"创建外部应用失败: {str(e)}" ) @router.put("/external-apps/{app_id}", response_model=dict) async def update_external_app( app_id: int, request: UpdateExternalAppRequest, current_user: dict = Depends(get_current_admin_user) ): """ 更新外部应用 """ try: # 验证 app_info 是否为有效JSON if request.app_info: try: json.loads(request.app_info) except: return create_api_response( code="400", message="app_info 必须是有效的JSON格式" ) with get_db_connection() as conn: cursor = conn.cursor() # 构建更新字段 update_fields = [] params = [] if request.app_name is not None: update_fields.append("app_name = %s") params.append(request.app_name) if request.app_type is not None: update_fields.append("app_type = %s") params.append(request.app_type) if request.app_info is not None: update_fields.append("app_info = %s") params.append(request.app_info) if request.icon_url is not None: update_fields.append("icon_url = %s") params.append(request.icon_url) if request.description is not None: update_fields.append("description = %s") params.append(request.description) if request.sort_order is not None: update_fields.append("sort_order = %s") params.append(request.sort_order) if request.is_active is not None: update_fields.append("is_active = %s") params.append(request.is_active) if not update_fields: return create_api_response( code="400", message="没有需要更新的字段" ) params.append(app_id) query = f""" UPDATE external_apps SET {', '.join(update_fields)} WHERE id = %s """ cursor.execute(query, params) conn.commit() cursor.close() return create_api_response( code="200", message="更新成功" ) except Exception as e: return create_api_response( code="500", message=f"更新外部应用失败: {str(e)}" ) @router.delete("/external-apps/{app_id}", response_model=dict) async def delete_external_app( app_id: int, current_user: dict = Depends(get_current_admin_user) ): """ 删除外部应用 """ try: with get_db_connection() as conn: cursor = conn.cursor(dictionary=True) # 获取应用信息（用于删除APK文件） cursor.execute("SELECT * FROM external_apps WHERE id = %s", (app_id,)) app = cursor.fetchone() if not app: cursor.close() return create_api_response( code="404", message="应用不存在" ) # 删除数据库记录 cursor.execute("DELETE FROM external_apps WHERE id = %s", (app_id,)) conn.commit() # 删除相关文件 files_to_delete = [] # 如果是原生应用，添加APK文件到删除列表 if app['app_type'] == 'native' and app.get('app_info'): try: app_info = json.loads(app['app_info']) if isinstance(app['app_info'], str) else app['app_info'] apk_url = app_info.get('apk_url', '') if apk_url and apk_url.startswith('/uploads/'): apk_path = BASE_DIR / apk_url.lstrip('/') files_to_delete.append(('APK', apk_path)) except Exception as e: print(f"Failed to parse app_info for APK deletion: {e}") # 添加图标文件到删除列表 if app.get('icon_url'): icon_url = app['icon_url'] if icon_url and icon_url.startswith('/uploads/'): icon_path = BASE_DIR / icon_url.lstrip('/') files_to_delete.append(('Icon', icon_path)) # 执行文件删除 for file_type, file_path in files_to_delete: try: if file_path.exists(): os.remove(file_path) print(f"Deleted {file_type} file: {file_path}") except Exception as e: print(f"Failed to delete {file_type} file: {e}") cursor.close() return create_api_response( code="200", message="删除成功" ) except Exception as e: return create_api_response( code="500", message=f"删除外部应用失败: {str(e)}" ) @router.post("/external-apps/upload-apk", response_model=dict) async def upload_apk( apk_file: UploadFile = File(...), current_user: dict = Depends(get_current_admin_user) ): """ 上传APK文件并解析信息 """ try: # 验证文件类型 file_extension = os.path.splitext(apk_file.filename)[1].lower() if file_extension not in ALLOWED_APK_EXTENSIONS: return create_api_response( code="400", message=f"不支持的文件类型。仅支持: {', '.join(ALLOWED_APK_EXTENSIONS)}" ) # 验证文件大小 apk_file.file.seek(0, 2) # 移动到文件末尾 file_size = apk_file.file.tell() apk_file.file.seek(0) # 重置到文件开头 if file_size > MAX_APK_SIZE: return create_api_response( code="400", message=f"文件大小超过 {MAX_APK_SIZE // (1024 * 1024)}MB 限制" ) # 使用原始文件名 original_filename = apk_file.filename file_path = EXTERNAL_APPS_DIR / original_filename # 如果同名文件已存在，先删除 if file_path.exists(): try: os.remove(file_path) print(f"删除已存在的同名文件: {file_path}") except Exception as e: print(f"删除同名文件失败: {e}") # 保存文件 with open(file_path, "wb") as buffer: shutil.copyfileobj(apk_file.file, buffer) # 计算MD5 md5_hash = hashlib.md5() with open(file_path, "rb") as f: for chunk in iter(lambda: f.read(4096), b""): md5_hash.update(chunk) apk_md5 = md5_hash.hexdigest() # 解析APK try: apk_info = parse_apk_with_androguard(str(file_path)) print(f"APK解析成功: {apk_info}") except Exception as e: # 删除已上传的文件 if file_path.exists(): os.remove(file_path) return create_api_response( code="400", message=f"APK解析失败: {str(e)}" ) # 计算相对路径 relative_path = file_path.relative_to(EXTERNAL_APPS_DIR.parent.parent) # 返回解析结果 return create_api_response( code="200", message="APK上传并解析成功", data={ "apk_url": "/" + str(relative_path).replace("\\", "/"), "apk_size": file_size, "apk_md5": apk_md5, "package_name": apk_info.get('package_name'), "version_name": apk_info.get('version_name'), "version_code": apk_info.get('version_code'), "app_name": apk_info.get('app_name'), "min_sdk_version": apk_info.get('min_sdk_version'), "target_sdk_version": apk_info.get('target_sdk_version') } ) except Exception as e: return create_api_response( code="500", message=f"上传APK失败: {str(e)}" ) @router.post("/external-apps/upload-icon", response_model=dict) async def upload_icon( icon_file: UploadFile = File(...), current_user: dict = Depends(get_current_admin_user) ): """ 上传应用图标 """ try: # 验证文件类型 file_extension = os.path.splitext(icon_file.filename)[1].lower() if file_extension not in ALLOWED_IMAGE_EXTENSIONS: return create_api_response( code="400", message=f"不支持的文件类型。仅支持: {', '.join(ALLOWED_IMAGE_EXTENSIONS)}" ) # 验证文件大小 icon_file.file.seek(0, 2) file_size = icon_file.file.tell() icon_file.file.seek(0) if file_size > MAX_IMAGE_SIZE: return create_api_response( code="400", message=f"文件大小超过 {MAX_IMAGE_SIZE // (1024 * 1024)}MB 限制" ) # 生成唯一文件名（图标使用UUID避免冲突） unique_filename = f"icon_{uuid.uuid4()}{file_extension}" file_path = EXTERNAL_APPS_DIR / unique_filename # 保存文件 with open(file_path, "wb") as buffer: shutil.copyfileobj(icon_file.file, buffer) # 计算相对路径 relative_path = file_path.relative_to(EXTERNAL_APPS_DIR.parent.parent) return create_api_response( code="200", message="图标上传成功", data={ "icon_url": "/" + str(relative_path).replace("\\", "/"), "file_size": file_size } ) except Exception as e: return create_api_response( code="500", message=f"上传图标失败: {str(e)}" )