"""安卓 OTA:最新版本清单 + APK 下载。 为什么不用一张表:一个安装包就是一条"发布记录",而 APK 是构建产物、不能进仓库。 所以发布目录里放「latest.json + 同名 apk」,换版本只换文件,运维不需要动数据库。 清单由 `backend/scripts/publish_apk.py` 生成(算好大小与 sha256),本模块只读。 """ from __future__ import annotations import json from pathlib import Path from typing import Any from fastapi import APIRouter from fastapi.responses import FileResponse from ..config import get_settings from ..errors import AppError, ErrorCode from ..logging_setup import get_logger from ..schemas import ok router = APIRouter(prefix="/app", tags=["版本升级"]) settings = get_settings() logger = get_logger("app.release") MANIFEST_NAME = "latest.json" def apk_dir() -> Path: """发布目录:优先用配置里的 APK_DIR,默认在 backend/static/apk(部署时保留、不入库)。""" if settings.apk_dir: return Path(settings.apk_dir) return Path(__file__).resolve().parents[2] / "static" / "apk" def _manifest() -> dict[str, Any]: path = apk_dir() / MANIFEST_NAME if not path.is_file(): raise AppError(ErrorCode.APP_RELEASE_NOT_FOUND, "还没有发布过安装包") try: data = json.loads(path.read_text(encoding="utf-8")) except (OSError, ValueError) as exc: # noqa: BLE001 - 清单坏了也要说人话 logger.exception("读取 OTA 清单失败") raise AppError(ErrorCode.APP_RELEASE_NOT_FOUND, "安装包清单损坏,请重新发布") from exc if not isinstance(data, dict) or not data.get("file"): raise AppError(ErrorCode.APP_RELEASE_NOT_FOUND, "安装包清单缺少文件名") return data def _payload(data: dict[str, Any], base: str, current: int | None) -> dict[str, Any]: version_code = int(data.get("version_code") or 0) return { "platform": data.get("platform", "android"), "version_code": version_code, "version_name": data.get("version_name", ""), "notes": data.get("notes", ""), "file": data["file"], "size": int(data.get("size") or 0), "sha256": data.get("sha256", ""), "min_sdk": int(data.get("min_sdk") or 26), "published_at": data.get("published_at"), # 客户端拿相对路径拼自己的 API 前缀,换域名不用改包 "download_url": f"{base}/app/release/download/{data['file']}", # 带了当前版本号就顺手算好要不要升级,客户端不用再比一次 "has_update": current is not None and version_code > current, } @router.get("/release/latest", summary="安卓最新版本") async def latest_release(version_code: int | None = None) -> dict[str, Any]: """公开接口:装完 App 之前就要能查,不该要登录。 `version_code` 传客户端当前版本(`BuildConfig.VERSION_CODE`),返回里的 `has_update` 直接告诉它要不要提示升级。 """ data = _manifest() return ok(_payload(data, settings.api_prefix, version_code)) @router.api_route("/release/download/{name}", methods=["GET", "HEAD"], summary="下载安装包") async def download_release(name: str) -> FileResponse: # 只认清单里那一个文件:不做目录列举,也不接受 ../ 之类 data = _manifest() if name != data["file"]: raise AppError(ErrorCode.APP_RELEASE_NOT_FOUND, "没有这个安装包") path = apk_dir() / name if not path.is_file(): raise AppError(ErrorCode.APP_RELEASE_NOT_FOUND, "安装包文件不见了,请重新发布") return FileResponse( path, media_type="application/vnd.android.package-archive", filename=name, )