feat: OTA 版本下发端点(检查更新)

- app_config 加 get/set_app_version(独立 key,复用表不进 CONFIG_DEFS)
- GET /api/v1/platform/app-version 公开读 + POST /internal/app-version 内部写(X-Internal-Secret)
- tests/test_app_version.py 5 端到端测试

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

app/api/internal/app_version.py

@@ -0,0 +1,58 @@
+"""最新 App 版本写入端点(发布流程 → app-server)。
+
+发车流程出 APK 后,把版本号 / 下载链接 / sha256 等 POST 到这里,落 app_config(key=latest_app_version)。
+客户端再 GET /api/v1/platform/app-version 读取做 OTA 检查更新。**不是给客户端的接口**:
+不走用户 JWT,靠 server 间共享密钥头 `X-Internal-Secret` 校验(== settings.INTERNAL_API_SECRET)。
+密钥未配置(默认空)时直接 503,避免裸奔的写端点。也是应急改版本信息(紧急下线/改 apk_url)的入口。
+"""
+from __future__ import annotations
+
+import hmac
+import logging
+from typing import Annotated
+
+from fastapi import APIRouter, Header, HTTPException, status
+
+from app.api.deps import DbSession
+from app.core.config import settings
+from app.repositories import app_config
+from app.schemas.platform import AppVersionOut, AppVersionWriteIn
+
+logger = logging.getLogger("shagua.internal.app_version")
+
+router = APIRouter(prefix="/internal", tags=["internal"])
+
+
+def _check_secret(x_internal_secret: str | None) -> None:
+    """共享密钥校验。未配置 → 503(挡住裸奔写端点);不匹配 → 401(常量时间比较)。"""
+    configured = settings.INTERNAL_API_SECRET
+    if not configured:
+        raise HTTPException(
+            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+            detail="internal api not configured",
+        )
+    if not x_internal_secret or not hmac.compare_digest(x_internal_secret, configured):
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="invalid internal secret",
+        )
+
+
+@router.post(
+    "/app-version",
+    response_model=AppVersionOut,
+    summary="写最新 App 版本(发布流程→app-server,落 app_config)",
+)
+def write_app_version(
+    payload: AppVersionWriteIn,
+    db: DbSession,
+    x_internal_secret: Annotated[str | None, Header()] = None,
+) -> AppVersionOut:
+    _check_secret(x_internal_secret)
+    data = payload.model_dump()
+    app_config.set_app_version(db, data)
+    logger.info(
+        "app_version set code=%d name=%s url=%s",
+        payload.latest_version_code, payload.latest_version_name, payload.apk_url,
+    )
+    return AppVersionOut(**data)

app/api/v1/platform.py

@@ -18,6 +18,7 @@ from app.repositories import ops_marquee as marquee_crud
 from app.repositories import ops_stat as crud
 from app.schemas.platform import (
     AppFlagsOut,
+    AppVersionOut,
     PlatformStatsOut,
     SavingsFeedItem,
     SavingsFeedOut,
@@ -51,3 +52,13 @@ def flags(db: DbSession) -> AppFlagsOut:
     return AppFlagsOut(
         comparing_ad_enabled=bool(app_config.get_value(db, "comparing_ad_enabled")),
     )
+
+
+@router.get("/app-version", response_model=AppVersionOut, summary="最新 App 版本(OTA 检查更新,不鉴权)")
+def app_version(db: DbSession) -> AppVersionOut:
+    """客户端启动 / 手动检查更新时拉取。不鉴权:版本信息非敏感,且检查更新可能在登录前。
+    用 latest_version_code 与本机 versionCode 比;未配置(返回默认 0)时客户端视为已是最新。"""
+    data = app_config.get_app_version(db)
+    if not data:
+        return AppVersionOut()
+    return AppVersionOut(**data)

app/main.py

@@ -21,6 +21,7 @@ from app.api.v1.compare_milestone import router as compare_milestone_router
 from app.api.v1.compare_record import router as compare_record_router
 from app.api.v1.coupon import router as coupon_router
 from app.api.v1.cps_redirect import router as cps_redirect_router
+from app.api.internal.app_version import router as internal_app_version_router
 from app.api.internal.price import router as internal_price_router
 from app.api.internal.store import router as internal_store_router
 from app.api.v1.feedback import router as feedback_router
@@ -107,6 +108,7 @@ app.include_router(report_router)
 # 内部(server→server)端点:pricebot 上报价格观测 / 店铺映射,靠共享密钥头校验,不对客户端开放。
 app.include_router(internal_price_router)
 app.include_router(internal_store_router)
+app.include_router(internal_app_version_router)
 app.include_router(platform_router)
 # CPS 群发短链跳转 /c/{code}(公网无鉴权:记点击 → 302 跳美团)
 app.include_router(cps_redirect_router)

app/repositories/app_config.py

@@ -63,3 +63,31 @@ def list_all(db: Session) -> list[dict]:
             }
         )
     return out
+
+
+# ── 最新 App 版本(OTA 检查更新)──────────────────────────────────────────────
+# 版本信息不是"运营可配置项",不进 CONFIG_DEFS:它由发布流程(发车)/应急整组写入、客户端读取做
+# OTA,语义是"发布系统状态"而非"运营调参"。故走下面专用读写,直接操作 AppConfig 表(复用表不复用 repo)。
+APP_VERSION_KEY = "latest_app_version"
+
+
+def get_app_version(db: Session) -> dict | None:
+    """读最新 App 版本信息(OTA)。无记录返回 None(客户端视为无更新)。"""
+    row = db.get(AppConfig, APP_VERSION_KEY)
+    return row.value if row is not None else None
+
+
+def set_app_version(db: Session, data: dict, *, commit: bool = True) -> AppConfig:
+    """整组原子覆盖最新版本信息(发布流程/应急写,非 admin,故 updated_by_admin_id 留空)。"""
+    row = db.get(AppConfig, APP_VERSION_KEY)
+    if row is None:
+        row = AppConfig(key=APP_VERSION_KEY, value=data, updated_by_admin_id=None)
+        db.add(row)
+    else:
+        row.value = data
+    if commit:
+        db.commit()
+        db.refresh(row)
+    else:
+        db.flush()
+    return row

app/schemas/platform.py

@@ -1,7 +1,7 @@
 """首页平台级展示数据 schemas(客户端首页门面数字)。"""
 from __future__ import annotations
 
-from pydantic import BaseModel
+from pydantic import BaseModel, Field
 
 
 class PlatformStatsOut(BaseModel):
@@ -28,3 +28,31 @@ class AppFlagsOut(BaseModel):
     """客户端拉取的运营 feature flag(不鉴权,登录前也能拉)。客户端缓存后按需读。"""
 
     comparing_ad_enabled: bool  # 比价/领券期是否展示信息流广告(远程 kill-switch)
+
+
+class AppVersionOut(BaseModel):
+    """最新 App 版本信息(OTA 检查更新,不鉴权)。
+
+    客户端用 latest_version_code(整数,单调递增)与本机 versionCode 比较,**不要**用 version_name 字符串比。
+    latest_version_code=0 表示后台尚未配置任何版本 → 客户端一律视为"已是最新"。
+    """
+
+    latest_version_code: int = 0  # 最新版 versionCode;0=未配置(无更新)
+    latest_version_name: str = ""  # 展示用,如 "0.1.4"
+    apk_url: str = ""  # 下载链接(发车产出的永久版本化链接)
+    update_note: str = ""  # 更新说明,弹窗展示
+    min_supported_version_code: int = 0  # 本机低于此版本=强制更新;0=不强更(全可选)
+    apk_size_bytes: int = 0  # 包大小(字节),展示"约 xMB"
+    apk_sha256: str = ""  # APK 完整性校验(客户端下载后比对,防损坏/劫持)
+
+
+class AppVersionWriteIn(BaseModel):
+    """内部写入最新版本(发布流程/应急,X-Internal-Secret 校验)。整组原子覆盖。"""
+
+    latest_version_code: int = Field(gt=0)
+    latest_version_name: str = Field(min_length=1)
+    apk_url: str = Field(min_length=1)
+    update_note: str = ""
+    min_supported_version_code: int = 0
+    apk_size_bytes: int = 0
+    apk_sha256: str = ""

tests/test_app_version.py

@@ -0,0 +1,95 @@
+"""OTA 检查更新端点测试。
+
+覆盖:公开读端点未配置时返回默认(latest_version_code=0=无更新)、内部写端点鉴权
+(503 未配 / 401 头错 / 200)、写后能读到同一组值、写入参数校验(422)。
+"""
+from __future__ import annotations
+
+import pytest
+from fastapi.testclient import TestClient
+
+from app.core.config import settings
+from app.db.session import SessionLocal
+from app.models.app_config import AppConfig
+from app.repositories.app_config import APP_VERSION_KEY
+
+_SECRET = "test-internal-secret-only-for-pytest"
+
+# 合法 body(够过 pydantic 校验,用于鉴权用例)
+_VALID = {
+    "latest_version_code": 31,
+    "latest_version_name": "0.1.6",
+    "apk_url": "https://app-api.shaguabijia.com/media/releases/shaguabijia-v0.1.6-build31.apk",
+}
+
+
+@pytest.fixture()
+def clean_version():
+    """每个用例前后清掉 latest_app_version 行(表由 conftest 建好,session 级共享)。"""
+    def _clear():
+        s = SessionLocal()
+        s.query(AppConfig).filter(AppConfig.key == APP_VERSION_KEY).delete()
+        s.commit()
+        s.close()
+    _clear()
+    yield
+    _clear()
+
+
+def test_read_default_when_unset(client: TestClient, clean_version) -> None:
+    """未配置任何版本时,公开读端点返回 latest_version_code=0(客户端视为已是最新);不鉴权。"""
+    r = client.get("/api/v1/platform/app-version")
+    assert r.status_code == 200, r.text
+    assert r.json()["latest_version_code"] == 0
+
+
+def test_write_secret_unset_503(client: TestClient, clean_version, monkeypatch) -> None:
+    monkeypatch.setattr(settings, "INTERNAL_API_SECRET", "")  # 未配 = 写端点关闭
+    r = client.post("/internal/app-version", json=_VALID)
+    assert r.status_code == 503
+
+
+def test_write_auth(client: TestClient, clean_version, monkeypatch) -> None:
+    monkeypatch.setattr(settings, "INTERNAL_API_SECRET", _SECRET)
+    assert client.post("/internal/app-version", json=_VALID).status_code == 401
+    assert client.post(
+        "/internal/app-version", json=_VALID, headers={"X-Internal-Secret": "wrong"}
+    ).status_code == 401
+    r = client.post("/internal/app-version", json=_VALID, headers={"X-Internal-Secret": _SECRET})
+    assert r.status_code == 200, r.text
+
+
+def test_write_then_read(client: TestClient, clean_version, monkeypatch) -> None:
+    monkeypatch.setattr(settings, "INTERNAL_API_SECRET", _SECRET)
+    payload = {
+        "latest_version_code": 99,
+        "latest_version_name": "9.9.9",
+        "apk_url": "https://app-api.shaguabijia.com/media/releases/shaguabijia-v9.9.9-build99.apk",
+        "update_note": "测试更新说明",
+        "min_supported_version_code": 50,
+        "apk_size_bytes": 12345678,
+        "apk_sha256": "abc123",
+    }
+    w = client.post("/internal/app-version", json=payload, headers={"X-Internal-Secret": _SECRET})
+    assert w.status_code == 200, w.text
+    assert w.json()["latest_version_code"] == 99
+
+    body = client.get("/api/v1/platform/app-version").json()
+    assert body["latest_version_code"] == 99
+    assert body["latest_version_name"] == "9.9.9"
+    assert body["apk_url"] == payload["apk_url"]
+    assert body["update_note"] == "测试更新说明"
+    assert body["min_supported_version_code"] == 50
+    assert body["apk_size_bytes"] == 12345678
+    assert body["apk_sha256"] == "abc123"
+
+
+def test_write_rejects_invalid(client: TestClient, clean_version, monkeypatch) -> None:
+    monkeypatch.setattr(settings, "INTERNAL_API_SECRET", _SECRET)
+    h = {"X-Internal-Secret": _SECRET}
+    # latest_version_code 必须 > 0
+    bad_code = {"latest_version_code": 0, "latest_version_name": "x", "apk_url": "u"}
+    assert client.post("/internal/app-version", json=bad_code, headers=h).status_code == 422
+    # apk_url 不能空
+    bad_url = {"latest_version_code": 5, "latest_version_name": "x", "apk_url": ""}
+    assert client.post("/internal/app-version", json=bad_url, headers=h).status_code == 422