fix(scripts): #1894/#1968 抖音 yt-dlp cookies 支持 + 旧档位归一化补测 + CI 覆盖率修复

- scripts_ai.py: 支持通过 DOUYIN_COOKIES_FILE 指定 Netscape cookies 文件；
  cookies 缺失/过期类错误（'fresh cookies' 等）返回 503 + 友好文案
  （'抖音链接解析暂时不可用，请稍后重试或手动输入文案'），
  不再暴露 yt-dlp 原始错误；普通下载失败统一文案，不带内部异常文本
- ci_staging_deploy.sh: API 容器挂载 /var/lib/xiaoxia-saas-staging/configs/douyin_cookies.txt
  到 /app/configs/douyin_cookies.txt；自动 mkdir + 空占位兜底
- infra/docker/compose.yml: API 服务挂载 ../../deploy/configs:/app/configs:ro
- deploy/configs/.env.{staging,production}: 新增 DOUYIN_COOKIES_FILE 环境变量
- deploy/configs/douyin_cookies.txt: Netscape 格式占位文件（真实 cookies 部署时替换）
- 新增 test_helpers_get_user_plan.py 覆盖 _helpers.get_user_plan 旧档位归一化分支
- 新增 test_subscription_routes_legacy.py 覆盖 _build_subscription_info 旧档位归一化/
  cancel_subscription 免费用户分支/change_plan 参数校验/toggle-auto-renew 等
- 扩展 test_extract_from_douyin_errors.py: 新增 5 个 cookies 场景测试（503 友好文案/
  cookiefile 注入/错误信息脱敏等）
- 增量覆盖率从 59% 提升到 60%+，解决 CI Gate 阻塞

apps/api/app/api/routes/scripts_ai.py

@@ -38,6 +38,51 @@ logger = logging.getLogger(__name__)
 
 router = APIRouter()
 
+# 抖音 cookies 文件路径（Netscape 格式），由环境变量 DOUYIN_COOKIES_FILE 覆盖
+# 默认路径与 deploy/configs/douyin_cookies.txt 对应（容器内挂载到 /app/configs/）
+DOUYIN_COOKIES_FILE = os.environ.get(
+    "DOUYIN_COOKIES_FILE",
+    "/app/configs/douyin_cookies.txt",
+)
+
+# cookies 失效/需要刷新的错误关键词
+_COOKIES_ERROR_KEYWORDS = (
+    "fresh cookies",
+    "cookies (not necessarily logged in)",
+    "cookies are needed",
+    "need cookies",
+    "cookie is expired",
+    "login required",
+    "sign in to continue",
+    "未登录",
+    "需要登录",
+    "cookies过期",
+)
+
+
+def _cookies_file_exists() -> bool:
+    """检查抖音 cookies 文件是否存在且非空."""
+    try:
+        return bool(DOUYIN_COOKIES_FILE) and os.path.isfile(DOUYIN_COOKIES_FILE) and os.path.getsize(DOUYIN_COOKIES_FILE) > 100
+    except OSError:
+        return False
+
+
+def _is_cookies_related_error(msg: str) -> bool:
+    """判断 yt-dlp 的错误是否与 cookies 缺失/过期有关."""
+    low = msg.lower()
+    return any(kw in low for kw in _COOKIES_ERROR_KEYWORDS)
+
+
+# 启动时记录 cookies 状态，便于排查
+if _cookies_file_exists():
+    logger.info("抖音 cookies 文件已加载: %s (%d bytes)", DOUYIN_COOKIES_FILE, os.path.getsize(DOUYIN_COOKIES_FILE))
+else:
+    logger.warning(
+        "抖音 cookies 文件未找到或无效: path=%s 抖音提取功能可能因 cookies 缺失失败",
+        DOUYIN_COOKIES_FILE,
+    )
+
 # 抖音 URL 校验：支持短链 v.douyin.com 和长链 www.douyin.com/video/
 _DOUYIN_URL_RE = re.compile(
     r"^(https?://)?(v\.douyin\.com/\S+|www\.douyin\.com/video/\S+)$",
@@ -104,26 +149,45 @@ def extract_from_douyin(
                 "noplaylist": True,
             }
 
+            # 如果存在抖音 cookies 文件，传给 yt-dlp 绕过反爬
+            if _cookies_file_exists():
+                ydl_opts["cookiefile"] = DOUYIN_COOKIES_FILE
+                logger.debug("使用抖音 cookies 文件: %s", DOUYIN_COOKIES_FILE)
+
             try:
                 ydl = yt_dlp.YoutubeDL(ydl_opts)
                 info = ydl.extract_info(url_for_download, download=True)
             except yt_dlp.utils.DownloadError as exc:
-                # yt-dlp 官方异常类型：HTTP 错误、短链失效、视频下架等
+                # yt-dlp 官方异常类型：HTTP 错误、短链失效、视频下架、cookies 过期等
                 msg = str(exc)
                 logger.warning("抖音下载失败: url=%s error=%s", source_url, msg)
-                # 404/视频不存在/不可下载 → 400；网络问题/上游异常 → 502
+                # 404/视频不存在/不可下载 → 400
                 is_bad_url = any(
-                    kw in msg.lower() for kw in ("404", "not found", "unable to download webpage", "unsupported url", "no video formats")
+                    kw in msg.lower() for kw in ("404", "not found", "unable to download webpage", "unsupported url", "no video formats", "video unavailable", "this video isn't available")
                 )
+                # cookies 缺失/过期 → 返回友好提示，不暴露 yt-dlp 原始错误
+                if _is_cookies_related_error(msg):
+                    logger.error("抖音 cookies 失效或缺失，需要刷新: %s", msg[:300])
+                    raise HTTPException(
+                        status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+                        detail="抖音链接解析暂时不可用，请稍后重试或手动输入文案",
+                    ) from exc
                 raise HTTPException(
                     status_code=status.HTTP_400_BAD_REQUEST if is_bad_url else status.HTTP_502_BAD_GATEWAY,
-                    detail=("无法解析该抖音链接，请确认链接有效且视频未被下架" if is_bad_url else f"视频下载失败: {msg[:200]}"),
+                    detail=("无法解析该抖音链接，请确认链接有效且视频未被下架" if is_bad_url else "视频下载失败，请稍后重试"),
                 ) from exc
             except Exception as exc:
-                logger.exception("抖音视频下载异常: url=%s", source_url)
+                msg = str(exc)
+                logger.exception("抖音视频下载异常: url=%s error=%s", source_url, msg)
+                # cookies 相关的未知异常也走友好提示
+                if _is_cookies_related_error(msg):
+                    raise HTTPException(
+                        status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+                        detail="抖音链接解析暂时不可用，请稍后重试或手动输入文案",
+                    ) from exc
                 raise HTTPException(
                     status_code=status.HTTP_502_BAD_GATEWAY,
-                    detail=f"视频下载失败: {str(exc)[:200]}",
+                    detail="视频下载失败，请稍后重试",
                 ) from exc
 
             if info is None:

deploy/configs/.env.production

@@ -240,3 +240,6 @@ DOUBAO_MAX_RETRIES=2
 WECHAT_OPEN_APP_ID=${WECHAT_APP_ID}
 WECHAT_OPEN_APP_SECRET=${WECHAT_APP_SECRET}
 WECHAT_OPEN_REDIRECT_URI=https://saas.xiaoxiajianji.com/auth/wechat/callback
+
+# 抖音 cookies 文件路径（yt-dlp 解析抖音视频需要）
+DOUYIN_COOKIES_FILE=/app/configs/douyin_cookies.txt

deploy/configs/.env.staging

@@ -257,3 +257,6 @@ DOUBAO_MAX_RETRIES=2
 WECHAT_OPEN_APP_ID=${WECHAT_APP_ID}
 WECHAT_OPEN_APP_SECRET=${WECHAT_APP_SECRET}
 WECHAT_OPEN_REDIRECT_URI=https://staging.xiaoxiajianji.com/auth/wechat/callback
+
+# 抖音 cookies 文件路径（yt-dlp 解析抖音视频需要）
+DOUYIN_COOKIES_FILE=/app/configs/douyin_cookies.txt

deploy/configs/douyin_cookies.txt

@@ -0,0 +1,5 @@
+# Netscape HTTP Cookie File
+# https://docs.yt-dlp.org/en/latest/howto/cookies.html
+# 此文件为抖音 cookies，由浏览器导出为 Netscape 格式
+# 若出现 "抖音链接解析暂时不可用"，说明 cookies 已过期，需重新导出
+# Placeholder - 等待浏览器任务导出真实 cookies 后覆盖

infra/docker/compose.yml

@@ -67,9 +67,10 @@ services:
     ports:
       - "127.0.0.1:${API_PORT:-8000}:8000"
     
-    # 共享生成文件目录
+    # 共享生成文件目录 + 抖音 cookies 等运行时配置
     volumes:
       - generated-files:/app/generated
+      - ../../deploy/configs:/app/configs:ro
     
     networks:
       - xiaoxia-net

scripts/ci_staging_deploy.sh

@@ -47,6 +47,7 @@ ENV_FILE="${ENV_FILE:-/var/lib/xiaoxia-saas-staging/.env}"
 GENERATED_DIR="${GENERATED_DIR:-/var/lib/xiaoxia-saas-staging/generated}"
 LEGACY_ASSETS_DIR="${LEGACY_ASSETS_DIR:-/var/lib/xiaoxia-saas-staging/legacy-assets}"
 NGINX_CONF_FILE="${NGINX_CONF_FILE:-/var/lib/xiaoxia-saas-staging/nginx-staging.conf}"
+COOKIES_FILE="${COOKIES_FILE:-/var/lib/xiaoxia-saas-staging/configs/douyin_cookies.txt}"
 
 SKIP_MIGRATION="${SKIP_MIGRATION:-false}"
 SKIP_ROLLBACK="${SKIP_ROLLBACK:-false}"
@@ -65,6 +66,12 @@ fi
 echo "✅ .env file found: $ENV_FILE ($(wc -l < "$ENV_FILE") lines)"
 mkdir -p "$GENERATED_DIR"
 mkdir -p "$LEGACY_ASSETS_DIR"
+mkdir -p "$(dirname "$COOKIES_FILE")"
+# 确保 cookies 文件存在（即使是空占位），避免容器启动失败
+if [ ! -f "$COOKIES_FILE" ]; then
+  printf '# Netscape HTTP Cookie File\n# 抖音 cookies 文件，如抖音提取功能报 503 请刷新此文件\n' > "$COOKIES_FILE"
+  echo "Created empty cookies placeholder at $COOKIES_FILE"
+fi
 
 # ── 写入 Staging Nginx 配置 ──
 # 运行时覆盖 nginx 配置，确保 upstream 指向正确的 staging 网络
@@ -480,7 +487,9 @@ docker run -d \
   -e GENERATED_FILES_DIR=/app/generated \
   -e GENERATED_FILES_URL_PREFIX=/generated-files \
   -e PUBLIC_API_BASE_URL=https://staging-api.xiaoxiajianji.com \
+  -e DOUYIN_COOKIES_FILE=/app/configs/douyin_cookies.txt \
   -v "$GENERATED_DIR:/app/generated" \
+  -v "$COOKIES_FILE:/app/configs/douyin_cookies.txt:ro" \
   --restart unless-stopped \
   --health-cmd "python -c \"import urllib.request; urllib.request.urlopen('http://localhost:8000/health', timeout=5)\"" \
   --health-interval 30s \

tests/unit/test_extract_from_douyin_errors.py

@@ -220,3 +220,126 @@ def test_any_unexpected_error_does_not_return_500_raw(fake_user):
             scripts_ai.extract_from_douyin(request=body, current_user=fake_user, db=mock.MagicMock())
         # 只要不是被全局 INTERNAL_ERROR 吞掉就行（带 detail 的 500 也比通用 500 强）
         assert "抖音" in exc.value.detail or "失败" in exc.value.detail or exc.value.status_code != 500
+
+
+# ── cookies 相关测试 ──────────────────────────────────────────────
+
+
+def test_cookies_error_returns_503_friendly_message(fake_user):
+    """cookies 缺失/过期（yt-dlp 报 'Fresh cookies ... are needed'）→ 返回 503 + 友好文案，不暴露原始错误"""
+    scripts_ai = _import_target()
+    body = scripts_ai.ExtractFromDouyinRequest(url="https://v.douyin.com/test123/")
+
+    class DownloadError(Exception):
+        pass
+
+    class CookiesYDL(_FakeYDLBase):
+        extract_info_raises = DownloadError(
+            "ERROR: [Douyin] 7623712911260650802: Fresh cookies (not necessarily logged in) are needed"
+        )
+
+    _install_fake_ytdlp(CookiesYDL, download_error_cls=DownloadError)
+
+    with mock.patch.object(scripts_ai, "get_doubao_client", return_value=mock.MagicMock(is_available=True)):
+        with pytest.raises(HTTPException) as exc:
+            scripts_ai.extract_from_douyin(request=body, current_user=fake_user, db=mock.MagicMock())
+        assert exc.value.status_code == status.HTTP_503_SERVICE_UNAVAILABLE, f"应为503，实际 {exc.value.status_code}"
+        assert "暂时不可用" in exc.value.detail or "稍后重试" in exc.value.detail, f"应有友好提示，实际: {exc.value.detail}"
+        # 确认不暴露 yt-dlp 原始错误
+        assert "Fresh cookies" not in exc.value.detail
+
+
+def test_cookies_error_in_generic_except_also_returns_503(fake_user):
+    """cookies 错误绕过 DownloadError（例如被其他异常包装）时，兜底异常分支也应识别并返回 503"""
+    scripts_ai = _import_target()
+    body = scripts_ai.ExtractFromDouyinRequest(url="https://v.douyin.com/abc/")
+
+    class CookieBugYDL(_FakeYDLBase):
+        def extract_info(self, url, download=True):
+            # 抛非 DownloadError 的普通异常，但 message 含 cookies 关键词
+            raise RuntimeError("Fresh cookies are needed to access this video")
+
+    _install_fake_ytdlp(CookieBugYDL)
+
+    with mock.patch.object(scripts_ai, "get_doubao_client", return_value=mock.MagicMock(is_available=True)):
+        with pytest.raises(HTTPException) as exc:
+            scripts_ai.extract_from_douyin(request=body, current_user=fake_user, db=mock.MagicMock())
+        assert exc.value.status_code == status.HTTP_503_SERVICE_UNAVAILABLE
+
+
+def test_ydl_opts_includes_cookiefile_when_file_exists(fake_user):
+    """cookies 文件存在时，ydl_opts 应包含 cookiefile 指向该路径"""
+    scripts_ai = _import_target()
+    body = scripts_ai.ExtractFromDouyinRequest(url="https://v.douyin.com/abc/")
+
+    captured_opts = {}
+
+    class CaptureOptsYDL(_FakeYDLBase):
+        def __init__(self, opts):
+            captured_opts.update(opts)
+            super().__init__()
+
+        extract_info_result = {"id": "x", "duration": 5, "title": "t"}
+
+    _install_fake_ytdlp(CaptureOptsYDL)
+
+    with (
+        mock.patch.object(scripts_ai, "get_doubao_client", return_value=mock.MagicMock(is_available=True)),
+        mock.patch.object(scripts_ai, "_cookies_file_exists", return_value=True),
+        mock.patch.object(scripts_ai, "DOUYIN_COOKIES_FILE", "/tmp/fake_cookies.txt"),
+        mock.patch.object(scripts_ai.os.path, "isfile", return_value=True),
+        mock.patch.object(scripts_ai.os.path, "getsize", return_value=1024),
+        mock.patch.object(scripts_ai, "transcribe_to_text", return_value="ok"),
+    ):
+        scripts_ai.extract_from_douyin(request=body, current_user=fake_user, db=mock.MagicMock())
+
+    assert captured_opts.get("cookiefile") == "/tmp/fake_cookies.txt", f"cookiefile 应被设置，opts={captured_opts}"
+
+
+def test_ydl_opts_no_cookiefile_when_file_missing(fake_user):
+    """cookies 文件不存在时，ydl_opts 不应包含 cookiefile 键"""
+    scripts_ai = _import_target()
+    body = scripts_ai.ExtractFromDouyinRequest(url="https://v.douyin.com/abc/")
+
+    captured_opts = {}
+
+    class CaptureOptsYDL(_FakeYDLBase):
+        def __init__(self, opts):
+            captured_opts.update(opts)
+            super().__init__()
+
+        extract_info_result = {"id": "x", "duration": 5, "title": "t"}
+
+    _install_fake_ytdlp(CaptureOptsYDL)
+
+    with (
+        mock.patch.object(scripts_ai, "get_doubao_client", return_value=mock.MagicMock(is_available=True)),
+        mock.patch.object(scripts_ai, "_cookies_file_exists", return_value=False),
+        mock.patch.object(scripts_ai.os.path, "isfile", return_value=True),
+        mock.patch.object(scripts_ai.os.path, "getsize", return_value=1024),
+        mock.patch.object(scripts_ai, "transcribe_to_text", return_value="ok"),
+    ):
+        scripts_ai.extract_from_douyin(request=body, current_user=fake_user, db=mock.MagicMock())
+
+    assert "cookiefile" not in captured_opts, f"cookies 文件缺失时不应设置 cookiefile，opts={captured_opts}"
+
+
+def test_generic_download_error_hides_raw_message(fake_user):
+    """非 cookies 非 404 的通用下载错误 → 502，且不暴露 yt-dlp 原始错误文本"""
+    scripts_ai = _import_target()
+    body = scripts_ai.ExtractFromDouyinRequest(url="https://v.douyin.com/abc/")
+
+    class DownloadError(Exception):
+        pass
+
+    class GenErrYDL(_FakeYDLBase):
+        extract_info_raises = DownloadError("ERROR: some internal yt-dlp weird failure with trace")
+
+    _install_fake_ytdlp(GenErrYDL, download_error_cls=DownloadError)
+
+    with mock.patch.object(scripts_ai, "get_doubao_client", return_value=mock.MagicMock(is_available=True)):
+        with pytest.raises(HTTPException) as exc:
+            scripts_ai.extract_from_douyin(request=body, current_user=fake_user, db=mock.MagicMock())
+        assert exc.value.status_code == status.HTTP_502_BAD_GATEWAY
+        assert "下载失败" in exc.value.detail
+        assert "weird failure" not in exc.value.detail, "不应暴露 yt-dlp 内部错误文本"

tests/unit/test_helpers_get_user_plan.py

@@ -0,0 +1,72 @@
+"""验证 _helpers.get_user_plan 档位归一化逻辑（#1894 旧档位兼容）"""
+
+from __future__ import annotations
+
+from unittest import mock
+
+import pytest
+from app.api.routes import _helpers
+
+
+class _FakeUser:
+    def __init__(self, plan):
+        self.subscription_plan = plan
+
+
+class _FakeUserNoPlan:
+    pass
+
+
+class _FakeRepo:
+    def __init__(self, user=None):
+        self._user = user
+
+    def find_by_id(self, uid):
+        return self._user
+
+
+def test_user_not_found_returns_free():
+    """用户不存在时返回 free（覆盖 _helpers.py 第 41 行 user is None 分支）"""
+    repo = _FakeRepo(user=None)
+    assert _helpers.get_user_plan("u-missing", repo) == "free"
+
+
+def test_user_plan_none_returns_free():
+    """用户 plan 属性为 None 时返回 free"""
+    repo = _FakeRepo(user=_FakeUser(None))
+    assert _helpers.get_user_plan("u1", repo) == "free"
+
+
+@pytest.mark.parametrize(
+    "legacy,expected",
+    [
+        ("standard", "monthly"),
+        ("basic", "monthly"),
+        ("pro", "quarterly"),
+        ("premium", "quarterly"),
+        ("enterprise", "quarterly"),
+    ],
+)
+def test_legacy_plans_normalized(legacy, expected):
+    """旧档位值正确归一化到新体系"""
+    repo = _FakeRepo(user=_FakeUser(legacy))
+    assert _helpers.get_user_plan("u1", repo) == expected
+
+
+def test_unknown_plan_returns_free():
+    """未知 plan 值（非新旧任一档位）→ 回落到 free（覆盖第 47 行）"""
+    repo = _FakeRepo(user=_FakeUser("totally_unknown_plan_xyz"))
+    assert _helpers.get_user_plan("u1", repo) == "free"
+
+
+def test_user_without_subscription_plan_attr_returns_free():
+    """user 对象没有 subscription_plan 属性时返回 free（getattr 默认值分支）"""
+    repo = _FakeRepo(user=_FakeUserNoPlan())
+    assert _helpers.get_user_plan("u1", repo) == "free"
+
+
+@pytest.mark.parametrize("plan", ["free", "monthly", "quarterly", "yearly"])
+def test_valid_new_plans_passthrough(plan):
+    """新档位直接透传"""
+    repo = _FakeRepo(user=_FakeUser(plan))
+    assert _helpers.get_user_plan("u1", repo) == plan

tests/unit/test_subscription_routes_legacy.py

@@ -0,0 +1,185 @@
+"""#1894 旧档位归一化逻辑测试（覆盖 _build_subscription_info / change_plan / cancel 等分支）"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, replace
+from datetime import UTC, datetime
+from unittest import mock
+
+import pytest
+from app.api.routes import subscription
+from app.auth import AuthenticatedUser
+from app.schemas.subscription import (
+    BillingCycle,
+    ChangePlanRequest,
+    MembershipType,
+    ToggleAutoRenewRequest,
+)
+
+
+@dataclass
+class _FakeUserModel:
+    id: str = "u-1234567890"
+    subscription_plan: str | None = MembershipType.FREE
+    subscription_status: str | None = "active"
+    subscription_expires_at: datetime | None = None
+    created_at: datetime | None = None
+
+
+@pytest.fixture
+def auth_user():
+    return AuthenticatedUser(user=_FakeUserModel())
+
+
+class TestBuildSubscriptionInfoLegacy:
+    """覆盖 _build_subscription_info 旧档位归一化（subscription.py 57-58 行）"""
+
+    def test_legacy_standard_plan_normalized_to_monthly(self, auth_user):
+        new_user = replace(auth_user.user, subscription_plan="standard")
+        auth_user = replace(auth_user, user=new_user)
+
+        info = subscription._build_subscription_info(auth_user)
+        assert info.plan_id == MembershipType.MONTHLY
+        assert info.plan_name == "月卡会员"
+
+    def test_legacy_pro_plan_normalized_to_monthly(self, auth_user):
+        new_user = replace(auth_user.user, subscription_plan="pro")
+        auth_user = replace(auth_user, user=new_user)
+
+        info = subscription._build_subscription_info(auth_user)
+        # 旧 pro/standard/enterprise 都归一化到 monthly（按代码逻辑 {standard,pro,enterprise} → monthly）
+        assert info.plan_id == MembershipType.MONTHLY
+
+    def test_legacy_enterprise_plan_normalized(self, auth_user):
+        new_user = replace(auth_user.user, subscription_plan="enterprise")
+        auth_user = replace(auth_user, user=new_user)
+
+        info = subscription._build_subscription_info(auth_user)
+        assert info.plan_id == MembershipType.MONTHLY
+
+    def test_no_expiry_gives_now_period(self, auth_user):
+        """无过期时间时 period_start 和 period_end 都为 now（覆盖 else 分支 55-56 行）"""
+        new_user = replace(auth_user.user, subscription_expires_at=None)
+        auth_user = replace(auth_user, user=new_user)
+
+        info = subscription._build_subscription_info(auth_user)
+        # 两个时间都应非空且接近当前时间
+        assert info.current_period_start
+        assert info.current_period_end
+
+    def test_free_user_billing_cycle_defaults_to_monthly(self, auth_user):
+        """免费用户 billing_cycle 回落到 monthly（覆盖第 64 行 !=FREE 判定 else 分支）"""
+        new_user = replace(auth_user.user, subscription_plan=MembershipType.FREE)
+        auth_user = replace(auth_user, user=new_user)
+
+        info = subscription._build_subscription_info(auth_user)
+        assert info.billing_cycle == BillingCycle.MONTHLY
+        assert info.amount == 0
+
+    def test_yearly_user_passthrough(self, auth_user):
+        """yearly 用户档位直接透传"""
+        new_user = replace(auth_user.user, subscription_plan=MembershipType.YEARLY)
+        auth_user = replace(auth_user, user=new_user)
+
+        info = subscription._build_subscription_info(auth_user)
+        assert info.plan_id == MembershipType.YEARLY
+        assert info.plan_name == "年卡会员"
+
+
+class TestChangePlanValidation:
+    """覆盖 change_plan 入参校验 / 同档位提示 / 旧档位归一化"""
+
+    @pytest.mark.asyncio
+    async def test_invalid_plan_returns_400(self, auth_user):
+        """无效 plan_id → 400（覆盖 169 行）"""
+        from fastapi import HTTPException
+        req = ChangePlanRequest(target_plan_id="totally_bogus_plan", billing_cycle=BillingCycle.MONTHLY)
+        with pytest.raises(HTTPException) as exc:
+            await subscription.change_plan(request=req, current_user=auth_user, user_repository=mock.MagicMock())
+        assert exc.value.status_code == 400
+        assert "无效" in exc.value.detail
+
+    @pytest.mark.asyncio
+    async def test_invalid_billing_cycle_returns_400(self, auth_user):
+        """无效 billing_cycle → 400（覆盖 176/178-179 行）"""
+        from fastapi import HTTPException
+        req = ChangePlanRequest(target_plan_id=MembershipType.MONTHLY, billing_cycle="bogus_cycle")
+        with pytest.raises(HTTPException) as exc:
+            await subscription.change_plan(request=req, current_user=auth_user, user_repository=mock.MagicMock())
+        assert exc.value.status_code == 400
+
+    @pytest.mark.asyncio
+    async def test_same_plan_returns_message(self, auth_user):
+        """同档位变更 → 返回提示（覆盖 185 行分支）"""
+        new_user = replace(auth_user.user, subscription_plan=MembershipType.MONTHLY)
+        auth_user = replace(auth_user, user=new_user)
+        req = ChangePlanRequest(target_plan_id=MembershipType.MONTHLY, billing_cycle=BillingCycle.MONTHLY)
+        resp = await subscription.change_plan(request=req, current_user=auth_user, user_repository=mock.MagicMock())
+        assert resp.success is False
+        assert "已经是" in resp.message
+
+    @pytest.mark.asyncio
+    async def test_legacy_plan_normalized_for_same_plan_check(self, auth_user):
+        """旧档位用户升级到 monthly → 应先归一化 current_plan 到 monthly，再判定为'同档位'"""
+        new_user = replace(auth_user.user, subscription_plan="standard")
+        auth_user = replace(auth_user, user=new_user)
+  # legacy → monthly
+        req = ChangePlanRequest(target_plan_id=MembershipType.MONTHLY, billing_cycle=BillingCycle.MONTHLY)
+        resp = await subscription.change_plan(request=req, current_user=auth_user, user_repository=mock.MagicMock())
+        # standard 归一化到 monthly，所以 target monthly == current monthly → same plan
+        assert resp.success is False
+
+
+class TestCancelSubscription:
+    """覆盖 cancel_subscription 分支"""
+
+    @pytest.mark.asyncio
+    async def test_cancel_free_user_returns_400(self, auth_user):
+        """免费用户取消订阅 → 400（覆盖 252 行）"""
+        from fastapi import HTTPException
+        new_user = replace(auth_user.user, subscription_plan=MembershipType.FREE)
+        auth_user = replace(auth_user, user=new_user)
+
+        with pytest.raises(HTTPException) as exc:
+            await subscription.cancel_subscription(current_user=auth_user, user_repository=mock.MagicMock())
+        assert exc.value.status_code == 400
+        assert "免费" in exc.value.detail
+
+    @pytest.mark.asyncio
+    async def test_cancel_paid_user_marks_cancelled(self, auth_user):
+        """付费用户取消订阅 → save 被调用且 subscription_status='cancelled'"""
+        new_user = replace(auth_user.user, subscription_plan=MembershipType.MONTHLY)
+        auth_user = replace(auth_user, user=new_user)
+        repo = mock.MagicMock()
+        resp = await subscription.cancel_subscription(current_user=auth_user, user_repository=repo)
+        assert resp.success is True
+        repo.save.assert_called_once()
+        saved_user = repo.save.call_args[0][0]
+        assert saved_user.subscription_status == "cancelled"
+
+
+class TestToggleAutoRenew:
+    """覆盖 toggle_auto_renew"""
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize("enabled,msg", [(True, "已开启"), (False, "已关闭")])
+    async def test_toggle_returns_message(self, auth_user, enabled, msg):
+        req = ToggleAutoRenewRequest(enabled=enabled)
+        resp = await subscription.toggle_auto_renew(request=req, current_user=auth_user)
+        assert resp.success is True
+        assert msg in resp.message
+
+
+class TestBuildSubscriptionInfoEdgeCases:
+    """覆盖 _build_subscription_info 的边缘分支"""
+
+    def test_created_at_none_uses_now(self, auth_user):
+        """user.created_at 为 None 时，created_at 字段回落到 now.isoformat（覆盖 69 行）"""
+        new_user = replace(auth_user.user, created_at=None, subscription_plan=MembershipType.MONTHLY)
+        auth_user2 = replace(auth_user, user=new_user)
+        info = subscription._build_subscription_info(auth_user2)
+        assert info.created_at  # 非空
+        # 应为 ISO 格式字符串
+        from datetime import datetime
+        # 能解析即通过
+        datetime.fromisoformat(info.created_at)