""" 更新服务API 提供客户端版本检测和更新包下载服务 """ from fastapi import APIRouter, HTTPException, Query from fastapi.responses import FileResponse from typing import Dict, Any import os import json from datetime import datetime from app.utils.timezone import beijing_now router = APIRouter() # 模拟版本信息存储(实际应该存储在数据库中) CURRENT_VERSION = "1.0.0" UPDATE_INFO = { "version": "1.0.1", "notes": "修复了扫货功能bug,提升了稳定性", "pub_date": "2026-01-01T10:00:00Z", "platforms": { "windows-x86_64": { "signature": "更新包数字签名", "url": "https://api.youdangdang.com/updates/windows-x86_64/1.0.1/update.exe", "size": 52428800 # 50MB }, "darwin-x86_64": { "signature": "更新包数字签名", "url": "https://api.youdangdang.com/updates/darwin-x86_64/1.0.1/update.dmg", "size": 67108864 # 64MB }, "linux-x86_64": { "signature": "更新包数字签名", "url": "https://api.youdangdang.com/updates/linux-x86_64/1.0.1/update.AppImage", "size": 41943040 # 40MB } } } @router.get("/updates/{target}/{current_version}") async def check_update( target: str, current_version: str, current_app_version: str = Query(None, alias="currentAppVersion"), current_bundle_version: str = Query(None, alias="currentBundleVersion") ) -> Dict[str, Any]: """ 检查更新API - Tauri客户端调用此接口检测更新 Args: target: 目标平台 (windows-x86_64, darwin-x86_64, linux-x86_64) current_version: 当前版本 current_app_version: 当前应用版本 current_bundle_version: 当前包版本 Returns: 更新信息或204 No Content(无更新) """ try: # 检查目标平台是否支持 if target not in UPDATE_INFO["platforms"]: raise HTTPException(status_code=400, detail=f"Unsupported target: {target}") # 版本比较逻辑(简化版,实际需要更复杂的版本比较) if current_version >= UPDATE_INFO["version"]: # 返回204表示无更新 return {"status": "no-update"} # 返回更新信息 platform_info = UPDATE_INFO["platforms"][target] return { "version": UPDATE_INFO["version"], "notes": UPDATE_INFO["notes"], "pub_date": UPDATE_INFO["pub_date"], "platforms": { target: platform_info } } except Exception as e: raise HTTPException(status_code=500, detail=f"Update check failed: {str(e)}") @router.get("/updates/download/{target}/{version}/{filename}") async def download_update(target: str, version: str, filename: str): """ 下载更新包 Args: target: 目标平台 version: 版本号 filename: 文件名 Returns: 更新包文件 """ # 实际实现中,这里应该从文件存储系统(如OSS、S3等)下载文件 # 这里只是示例代码 file_path = f"/path/to/updates/{target}/{version}/{filename}" if not os.path.exists(file_path): raise HTTPException(status_code=404, detail="Update file not found") return FileResponse( path=file_path, filename=filename, media_type="application/octet-stream" ) @router.post("/updates/force/{user_id}") async def force_update_user(user_id: int, version: str = None): """ 强制指定用户更新到指定版本 Args: user_id: 用户ID version: 目标版本(可选,默认最新版本) Returns: 操作结果 """ # 实际实现中,需要在数据库中标记用户需要强制更新 # 客户端在下次启动时会检测到强制更新标志 target_version = version or UPDATE_INFO["version"] # 数据库操作:更新用户的强制更新标志 # user.force_update_version = target_version # user.force_update_required = True return { "success": True, "message": f"User {user_id} will be forced to update to version {target_version}", "target_version": target_version } @router.get("/admin/updates/stats") async def get_update_stats(): """ 获取更新统计信息(管理员接口) Returns: 更新统计数据 """ # 实际实现中,从数据库查询更新统计信息 return { "current_version": CURRENT_VERSION, "latest_version": UPDATE_INFO["version"], "total_users": 1000, "updated_users": 850, "update_rate": "85%", "platforms": { "windows": {"users": 600, "updated": 520}, "macos": {"users": 300, "updated": 280}, "linux": {"users": 100, "updated": 50} } }