.gitea/workflows/api-base-image.yml

@@ -0,0 +1,84 @@
+name: API Base Image Build
+
+on:
+  push:
+    branches:
+      - develop
+      - main
+    paths:
+      - 'requirements-base.txt'
+      - 'requirements.txt'
+      - 'infra/docker/api-base.Dockerfile'
+  workflow_dispatch:
+
+jobs:
+  build-api-base:
+    name: Build API Base Image
+    runs-on: runtime-builder
+    timeout-minutes: 45
+    steps:
+    - name: Checkout code
+      shell: sh
+      env:
+        GITHUB_TOKEN: ${{ github.token }}
+      run: |
+        curl -sH "Authorization: token $GITHUB_TOKEN" \
+          "${GITHUB_API_URL}/repos/${GITHUB_REPOSITORY}/raw/scripts/ci/step_checkout.sh?ref=${GITHUB_SHA}" \
+          | bash
+
+    - name: Docker login to Registry
+      shell: sh
+      env:
+        ACR_USERNAME: ${{ secrets.ACR_USERNAME }}
+        ACR_PASSWORD: ${{ secrets.ACR_PASSWORD }}
+        GITEA_REGISTRY_USER: xiaoxia
+        GITEA_REGISTRY_TOKEN: ${{ secrets.REGISTRY_TOKEN }}
+      run: |
+        set -eu
+        for i in 1 2 3; do
+          echo "=== Docker login 尝试 $i/3 ==="
+          if printf '%s' "${ACR_PASSWORD}" | docker login xiaoxia-registry.cn-hangzhou.cr.aliyuncs.com -u "${ACR_USERNAME}" --password-stdin \
+              && docker login git.xiaoxiajianji.com -u "${GITEA_REGISTRY_USER}" -p "${GITEA_REGISTRY_TOKEN}"; then
+            echo "✅ Docker login successful"
+            break
+          fi
+          echo "❌ Docker login 失败（尝试 $i/3），5s 后重试..."
+          sleep 5
+        done
+
+    - name: Build and push API base image
+      shell: sh
+      run: |
+        set -eu
+        ACR_IMAGE="xiaoxia-registry.cn-hangzhou.cr.aliyuncs.com/xiaoxiakeji/saas-api-base:latest"
+        GITEA_IMAGE="git.xiaoxiajianji.com/xiaoxia-saas/saas-api-base:latest"
+
+        echo "=== Building API base image ==="
+
+        # 使用普通 docker build（单平台不需要 buildx）
+        docker build \
+          -f infra/docker/api-base.Dockerfile \
+          -t "${ACR_IMAGE}" \
+          .
+
+        echo ""
+        echo "✅ Image built successfully"
+
+        # 推送到 ACR
+        echo "=== Pushing to ACR ==="
+        docker push "${ACR_IMAGE}"
+        echo "✅ Pushed to ACR"
+
+        # 打标签并推送到 Gitea Packages 作为备份
+        echo "=== Pushing to Gitea Packages ==="
+        docker tag "${ACR_IMAGE}" "${GITEA_IMAGE}"
+        docker push "${GITEA_IMAGE}" || echo "⚠️ Gitea Packages push failed (non-fatal)"
+        echo "✅ Gitea backup push completed"
+
+    - name: Cleanup
+      if: always()
+      shell: sh
+      run: |
+        ACR_IMAGE="xiaoxia-registry.cn-hangzhou.cr.aliyuncs.com/xiaoxiakeji/saas-api-base:latest"
+        docker rmi "${ACR_IMAGE}" 2>/dev/null || true
+        echo "Cleanup done"

apps/api/app/api/routes/health.py

@@ -1,6 +1,6 @@
 from datetime import datetime, timezone
 
-import psycopg2
+import psycopg
 import redis
 from app.config import settings
 from fastapi import APIRouter, status
@@ -49,7 +49,7 @@ async def _check_database() -> dict:
             "message": "Using in-memory database",
         }
     try:
-        conn = psycopg2.connect(settings.DATABASE_URL, connect_timeout=3)
+        conn = psycopg.connect(settings.DATABASE_URL, connect_timeout=3)
         with conn.cursor() as cur:
             cur.execute("SELECT 1")
             cur.fetchone()
@@ -124,7 +124,7 @@ async def _check_migrations() -> dict:
             "message": "Using in-memory database, no migrations needed",
         }
     try:
-        conn = psycopg2.connect(settings.DATABASE_URL, connect_timeout=3)
+        conn = psycopg.connect(settings.DATABASE_URL, connect_timeout=3)
         with conn.cursor() as cur:
             cur.execute("""
                 SELECT COUNT(*) FROM information_schema.tables

infra/docker/api-base.Dockerfile

@@ -0,0 +1,43 @@
+# ============================================================
+# API 基础镜像（预构建）
+# 预装系统依赖 + Python 依赖，业务构建从此镜像开始
+# 当 requirements-base.txt 或 requirements.txt 变更时重新构建
+# 目标：将 API Image 构建时间从 15-20 分钟降至 3-5 分钟
+# ============================================================
+
+FROM git.xiaoxiajianji.com/xiaoxia/base/python:3.12-slim
+
+# 使用阿里云镜像加速
+RUN sed -i 's|deb.debian.org|mirrors.aliyun.com|g' /etc/apt/sources.list.d/debian.sources 2>/dev/null || \
+    sed -i 's|deb.debian.org|mirrors.aliyun.com|g' /etc/apt/sources.list 2>/dev/null || true
+
+# 预装系统依赖（gcc 编译 psycopg/pg 扩展，libpq-dev 编译期，libpq5 运行期，ffmpeg 封面取帧）
+RUN apt-get update && apt-get install -y --no-install-recommends \
+    gcc \
+    libpq-dev \
+    libpq5 \
+    ffmpeg \
+    && rm -rf /var/lib/apt/lists/*
+
+# 创建虚拟环境
+RUN python -m venv /opt/venv
+ENV PATH="/opt/venv/bin:$PATH"
+
+WORKDIR /tmp
+
+# 预装 Python 基础依赖
+COPY requirements-base.txt requirements.txt ./
+RUN pip install --no-cache-dir \
+    -i https://mirrors.aliyun.com/pypi/simple/ \
+    --trusted-host mirrors.aliyun.com \
+    -r requirements-base.txt -r requirements.txt
+
+# 虚拟环境瘦身
+RUN find /opt/venv -name "*.so" -type f -exec strip --strip-all {} \; 2>/dev/null || true
+RUN find /opt/venv -type d -name "__pycache__" -exec rm -rf {} + 2>/dev/null; \
+    find /opt/venv -name "*.pyc" -delete 2>/dev/null || true
+
+# 清理临时文件
+RUN rm -f /tmp/requirements-base.txt /tmp/requirements.txt
+
+ENV PYTHONPATH=/app

infra/docker/api.Dockerfile

@@ -1,77 +1,24 @@
 # ============================================================
 # API Dockerfile - FastAPI 应用
-# 优化：多阶段构建 + pip cache mount + 依赖分层缓存
+# 优化：从预构建基础镜像开始，仅叠加业务代码
+# 基础镜像包含所有系统依赖和 Python 依赖，构建时间 < 5 分钟
 # ============================================================
 
-# ==================== Builder 阶段 ====================
-FROM git.xiaoxiajianji.com/xiaoxia/base/python:3.12-slim AS builder
-
-# 使用阿里云镜像加速
-RUN sed -i 's|deb.debian.org|mirrors.aliyun.com|g' /etc/apt/sources.list.d/debian.sources 2>/dev/null || \
-    sed -i 's|deb.debian.org|mirrors.aliyun.com|g' /etc/apt/sources.list 2>/dev/null || true
-
-# 安装编译依赖（仅 builder 需要）
-RUN apt-get update && apt-get install -y --no-install-recommends \
-    gcc \
-    libpq-dev \
-    && rm -rf /var/lib/apt/lists/*
-
-# 创建虚拟环境
-RUN python -m venv /opt/venv
-ENV PATH="/opt/venv/bin:$PATH"
-
-WORKDIR /tmp
-
-# ---- 依赖分层：基础依赖（变化少，缓存命中率高）----
-COPY requirements-base.txt /tmp/requirements-base.txt
-
-RUN --mount=type=cache,target=/root/.cache/pip,sharing=locked \
-    pip install --no-cache-dir -i https://mirrors.aliyun.com/pypi/simple/ --trusted-host mirrors.aliyun.com \
-    -r /tmp/requirements-base.txt \
-    && rm /tmp/requirements-base.txt
-
-# ---- 依赖分层：业务依赖（变化频繁）----
-COPY requirements.txt /tmp/requirements.txt
-
-RUN --mount=type=cache,target=/root/.cache/pip,sharing=locked \
-    pip install --no-cache-dir -i https://mirrors.aliyun.com/pypi/simple/ --trusted-host mirrors.aliyun.com \
-    -r /tmp/requirements.txt \
-    && rm /tmp/requirements.txt
-
-# ---- Python 依赖瘦身 ----
-RUN find /opt/venv -name "*.so" -type f -exec strip --strip-all {} \; 2>/dev/null || true
-RUN find /opt/venv -type d -name "__pycache__" -exec rm -rf {} + 2>/dev/null; \
-    find /opt/venv -name "*.pyc" -delete 2>/dev/null || true
-
-# ==================== Runtime 阶段 ====================
-FROM git.xiaoxiajianji.com/xiaoxia/base/python:3.12-slim AS runtime
+FROM xiaoxia-registry.cn-hangzhou.cr.aliyuncs.com/xiaoxiakeji/saas-api-base:latest
 
 # 构建参数：版本号（CI 传入 commit hash）
 ARG APP_VERSION=dev
 
-# 使用阿里云镜像加速
-RUN sed -i 's|deb.debian.org|mirrors.aliyun.com|g' /etc/apt/sources.list.d/debian.sources 2>/dev/null || \
-    sed -i 's|deb.debian.org|mirrors.aliyun.com|g' /etc/apt/sources.list 2>/dev/null || true
-
-# 只装运行时需要的库（libpq5 是 psycopg2 运行时依赖，ffmpeg 用于封面兜底取帧）
-RUN apt-get update && apt-get install -y --no-install-recommends \
-    libpq5 \
-    ffmpeg \
-    && rm -rf /var/lib/apt/lists/*
-
-# 从 builder 复制虚拟环境
-COPY --from=builder /opt/venv /opt/venv
-
 # 设置工作目录
 WORKDIR /app
 
-# 复制应用代码
-COPY apps/api/ /app/apps/api/
-COPY packages/ /app/packages/
-COPY alembic.ini /app/alembic.ini
-COPY migrations/ /app/migrations/
-COPY alembic/ /app/alembic/
-COPY scripts/ /app/scripts/
+# 复制应用代码（按变化频率从低到高排序，最大化层缓存命中）
+COPY alembic.ini ./alembic.ini
+COPY migrations/ ./migrations/
+COPY alembic/ ./alembic/
+COPY scripts/ ./scripts/
+COPY packages/ ./packages/
+COPY apps/api/ ./apps/api/
 
 # 设置环境变量
 ENV PATH="/opt/venv/bin:/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin"
@@ -84,5 +31,4 @@ HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \
     CMD python -c "import urllib.request; urllib.request.urlopen('http://localhost:8000/health', timeout=5)"
 
 # API 入口点
-WORKDIR /app/apps/api
-CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]
+CMD ["uvicorn", "apps.api.main:app", "--host", "0.0.0.0", "--port", "8000"]

requirements-base.txt

@@ -2,7 +2,6 @@
 # 修改此文件会触发完整重新构建，请谨慎修改
 
 # 数据库（基础层）
-psycopg2-binary==2.9.9
 psycopg[binary]==3.2.2
 sqlalchemy==2.0.35
 alembic==1.13.3

scripts/ci/rebuild-api-base-image.sh

@@ -0,0 +1,40 @@
+#!/bin/bash
+# ============================================================
+# 重建 API 基础镜像脚本
+# 用途：当 requirements-base.txt 或 requirements.txt 变更时手动触发
+# 前提：需要在已登录 ACR 的构建服务器上执行
+# ============================================================
+
+set -euo pipefail
+
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+REPO_ROOT="$(cd "$SCRIPT_DIR/../.." && pwd)"
+
+REGISTRY="xiaoxia-registry.cn-hangzhou.cr.aliyuncs.com/xiaoxiakeji"
+IMAGE_NAME="saas-api-base"
+TAG="latest"
+FULL_TAG="${REGISTRY}/${IMAGE_NAME}:${TAG}"
+
+echo "========================================="
+echo "🔨 Rebuilding API base image"
+echo "   Registry: ${REGISTRY}"
+echo "   Image:    ${FULL_TAG}"
+echo "   Context:  ${REPO_ROOT}"
+echo "========================================="
+
+cd "$REPO_ROOT"
+
+# 构建并推送
+docker buildx build \
+    --platform linux/amd64 \
+    --tag "${FULL_TAG}" \
+    --push \
+    -f infra/docker/api-base.Dockerfile \
+    .
+
+echo ""
+echo "✅ API base image pushed: ${FULL_TAG}"
+
+# 显示镜像大小
+docker pull "${FULL_TAG}" > /dev/null 2>&1
+docker images "${FULL_TAG}" --format "table {{.Repository}}:{{.Tag}}\t{{.Size}}"

tests/unit/test_health_routes.py

@@ -0,0 +1,186 @@
+"""Unit tests for apps/api/app/api/routes/health.py
+
+覆盖 _check_database() 和 _check_migrations() 中 psycopg3 连接逻辑。
+确保增量覆盖率 ≥ 60%（目标覆盖 lines 52, 127）。
+"""
+
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+
+@pytest.mark.asyncio
+class TestCheckDatabase:
+    """Tests for _check_database() health check function."""
+
+    @patch("apps.api.app.api.routes.health.settings")
+    @patch("apps.api.app.api.routes.health.psycopg.connect")
+    async def test_check_database_success(self, mock_connect, mock_settings):
+        """PostgreSQL 连接成功时返回 healthy。"""
+        mock_settings.USE_IN_MEMORY_DB = False
+        mock_settings.DATABASE_URL = "postgresql+psycopg://test:test@localhost/test"
+
+        # Mock connection and cursor
+        mock_conn = MagicMock()
+        mock_cursor = MagicMock()
+        mock_cursor.__enter__ = MagicMock(return_value=mock_cursor)
+        mock_cursor.__exit__ = MagicMock(return_value=False)
+        mock_cursor.fetchone.return_value = (1,)
+        mock_conn.cursor.return_value = mock_cursor
+        mock_connect.return_value = mock_conn
+
+        from apps.api.app.api.routes.health import _check_database
+
+        result = await _check_database()
+
+        assert result["status"] == "healthy"
+        assert result["type"] == "postgresql"
+        assert result["message"] == "Database connection successful"
+        mock_connect.assert_called_once_with(
+            "postgresql+psycopg://test:test@localhost/test", connect_timeout=3
+        )
+        mock_cursor.execute.assert_called_once_with("SELECT 1")
+        mock_conn.close.assert_called_once()
+
+    @patch("apps.api.app.api.routes.health.settings")
+    @patch("apps.api.app.api.routes.health.psycopg.connect")
+    async def test_check_database_connection_failure(self, mock_connect, mock_settings):
+        """PostgreSQL 连接失败时返回 unhealthy。"""
+        mock_settings.USE_IN_MEMORY_DB = False
+        mock_settings.DATABASE_URL = "postgresql+psycopg://test:test@localhost/test"
+        mock_connect.side_effect = Exception("connection refused")
+
+        from apps.api.app.api.routes.health import _check_database
+
+        result = await _check_database()
+
+        assert result["status"] == "unhealthy"
+        assert result["type"] == "postgresql"
+        assert "connection refused" in result["message"]
+
+    @patch("apps.api.app.api.routes.health.settings")
+    async def test_check_database_in_memory(self, mock_settings):
+        """使用内存数据库时跳过 PostgreSQL 检查。"""
+        mock_settings.USE_IN_MEMORY_DB = True
+
+        from apps.api.app.api.routes.health import _check_database
+
+        result = await _check_database()
+
+        assert result["status"] == "healthy"
+        assert result["type"] == "in_memory"
+
+
+@pytest.mark.asyncio
+class TestCheckMigrations:
+    """Tests for _check_migrations() health check function."""
+
+    @patch("apps.api.app.api.routes.health.settings")
+    @patch("apps.api.app.api.routes.health.psycopg.connect")
+    async def test_check_migrations_success(self, mock_connect, mock_settings):
+        """所有迁移表存在时返回 healthy。"""
+        mock_settings.USE_IN_MEMORY_DB = False
+        mock_settings.DATABASE_URL = "postgresql+psycopg://test:test@localhost/test"
+
+        mock_conn = MagicMock()
+        mock_cursor = MagicMock()
+        mock_cursor.__enter__ = MagicMock(return_value=mock_cursor)
+        mock_cursor.__exit__ = MagicMock(return_value=False)
+        mock_cursor.fetchone.return_value = (5,)  # 5 tables found
+        mock_conn.cursor.return_value = mock_cursor
+        mock_connect.return_value = mock_conn
+
+        from apps.api.app.api.routes.health import _check_migrations
+
+        result = await _check_migrations()
+
+        assert result["status"] == "healthy"
+        assert result["message"] == "Database migrations applied"
+        mock_connect.assert_called_once_with(
+            "postgresql+psycopg://test:test@localhost/test", connect_timeout=3
+        )
+        mock_conn.close.assert_called_once()
+
+    @patch("apps.api.app.api.routes.health.settings")
+    @patch("apps.api.app.api.routes.health.psycopg.connect")
+    async def test_check_migrations_missing_tables(self, mock_connect, mock_settings):
+        """迁移表不完整时返回 unhealthy。"""
+        mock_settings.USE_IN_MEMORY_DB = False
+        mock_settings.DATABASE_URL = "postgresql+psycopg://test:test@localhost/test"
+
+        mock_conn = MagicMock()
+        mock_cursor = MagicMock()
+        mock_cursor.__enter__ = MagicMock(return_value=mock_cursor)
+        mock_cursor.__exit__ = MagicMock(return_value=False)
+        mock_cursor.fetchone.return_value = (2,)  # Only 2 of 5 tables
+        mock_conn.cursor.return_value = mock_cursor
+        mock_connect.return_value = mock_conn
+
+        from apps.api.app.api.routes.health import _check_migrations
+
+        result = await _check_migrations()
+
+        assert result["status"] == "unhealthy"
+        assert "Missing tables" in result["message"]
+        assert "2/5" in result["message"]
+
+    @patch("apps.api.app.api.routes.health.settings")
+    @patch("apps.api.app.api.routes.health.psycopg.connect")
+    async def test_check_migrations_connection_failure(self, mock_connect, mock_settings):
+        """数据库连接失败时返回 unhealthy。"""
+        mock_settings.USE_IN_MEMORY_DB = False
+        mock_settings.DATABASE_URL = "postgresql+psycopg://test:test@localhost/test"
+        mock_connect.side_effect = Exception("connection refused")
+
+        from apps.api.app.api.routes.health import _check_migrations
+
+        result = await _check_migrations()
+
+        assert result["status"] == "unhealthy"
+        assert "Migration check failed" in result["message"]
+
+    @patch("apps.api.app.api.routes.health.settings")
+    async def test_check_migrations_in_memory(self, mock_settings):
+        """使用内存数据库时跳过迁移检查。"""
+        mock_settings.USE_IN_MEMORY_DB = True
+
+        from apps.api.app.api.routes.health import _check_migrations
+
+        result = await _check_migrations()
+
+        assert result["status"] == "healthy"
+        assert "no migrations needed" in result["message"]
+
+
+@pytest.mark.asyncio
+class TestStartupCheck:
+    """Tests for startup_check() endpoint."""
+
+    @patch("apps.api.app.api.routes.health._check_migrations")
+    @patch("apps.api.app.api.routes.health._check_database")
+    async def test_startup_all_healthy(self, mock_db, mock_mig):
+        """所有检查通过时返回 started。"""
+        mock_db.return_value = {"status": "healthy"}
+        mock_mig.return_value = {"status": "healthy"}
+
+        from apps.api.app.api.routes.health import startup_check
+
+        result = await startup_check()
+
+        assert result["status"] == "started"
+
+    @patch("apps.api.app.api.routes.health._check_migrations")
+    @patch("apps.api.app.api.routes.health._check_database")
+    async def test_startup_db_unhealthy(self, mock_db, mock_mig):
+        """数据库不健康时返回 starting + 503。"""
+        mock_db.return_value = {"status": "unhealthy", "message": "fail"}
+        mock_mig.return_value = {"status": "healthy"}
+
+        from apps.api.app.api.routes.health import startup_check
+
+        result = await startup_check()
+
+        assert result.status_code == 503
+        import json
+        body = json.loads(result.body)
+        assert body["status"] == "starting"