56 lines
1.7 KiB
Python
56 lines
1.7 KiB
Python
"""App 版本更新 API — OTA 热更新检测"""
|
||
from fastapi import APIRouter, Depends, Query
|
||
from sqlalchemy import select, desc
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from app.core.database import get_db
|
||
from app.models.app_version import AppVersion
|
||
from app.schemas.app_version import AppVersionResponse
|
||
|
||
router = APIRouter(prefix="/app", tags=["App版本"])
|
||
|
||
|
||
@router.get("/version", response_model=AppVersionResponse)
|
||
async def check_version(
|
||
current: str = Query(..., description="当前 App 版本号，如 T1.0.0"),
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
"""检测是否有新版本可用"""
|
||
# 查询最新的激活版本
|
||
result = await db.execute(
|
||
select(AppVersion)
|
||
.where(AppVersion.is_active.is_(True))
|
||
.order_by(desc(AppVersion.version_code))
|
||
.limit(1)
|
||
)
|
||
latest = result.scalar_one_or_none()
|
||
|
||
if not latest:
|
||
# 无版本记录 → 无更新
|
||
return AppVersionResponse(
|
||
version=current,
|
||
version_code=0,
|
||
has_update=False,
|
||
)
|
||
|
||
# 比较版本号
|
||
has_update = latest.version_code > _parse_version_code(current)
|
||
|
||
return AppVersionResponse(
|
||
version=latest.version,
|
||
version_code=latest.version_code,
|
||
has_update=has_update,
|
||
wgt_url=latest.wgt_url if has_update else None,
|
||
description=latest.description if has_update else None,
|
||
force_update=False,
|
||
)
|
||
|
||
|
||
def _parse_version_code(version_str: str) -> int:
|
||
"""从版本字符串提取数字版本号（兜底用最后3位）"""
|
||
import re
|
||
nums = re.findall(r"\d+", version_str)
|
||
if nums:
|
||
return int("".join(nums[-3:]).ljust(3, "0")[:3])
|
||
return 0
|