feat(mini-auth): add WeChat service-account bind APIs

Co-authored-by: Cursor <cursoragent@cursor.com>

backend/app/api/mini_auth.py

@@ -20,6 +20,7 @@ from app.services.wechat_mini_auth import (
     code2session,
     get_phone_number,
 )
+from app.services.wechat_mp_auth import WechatMpAuthError, oauth_code_to_openid
 
 
 router = APIRouter(prefix="/mini/auth", tags=["小程序鉴权"])
@@ -120,3 +121,49 @@ async def wechat_bind(
     db.add(user)
     db.commit()
     return success_response({"bound": True, "message": "绑定成功"})
+
+
+@router.get("/mp-bind-status", response_model=dict)
+async def mp_bind_status(request: Request, db: Session = Depends(get_db)):
+    """当前用户是否已绑定微信服务号 openid。"""
+    try:
+        user_id = await _resolve_user_id(request, db)
+    except HTTPException:
+        return error_response(401, "未授权")
+
+    user = user_crud.get(db, user_id)
+    bound = bool(user and (getattr(user, "wx_mp_openid", None) or "").strip())
+    return success_response({"bound": bound})
+
+
+@router.post("/mp-bind", response_model=dict)
+async def mp_bind(request: Request, body: dict, db: Session = Depends(get_db)):
+    """服务号网页授权 code 绑定当前用户（需 Bearer）。"""
+    try:
+        user_id = await _resolve_user_id(request, db)
+    except HTTPException:
+        return error_response(401, "未授权")
+
+    code = (body.get("code") or "").strip()
+    try:
+        info = oauth_code_to_openid(code)
+    except WechatMpAuthError as exc:
+        return error_response(400, str(exc))
+
+    mp_openid = info["openid"]
+    other = user_crud.get_by_wx_mp_openid(db, mp_openid)
+    if other and other.id != user_id:
+        return error_response(409, "该微信服务号已绑定其他账号")
+
+    user = user_crud.get(db, user_id)
+    if not user or user.del_flag != 0:
+        return error_response(401, "用户不存在")
+
+    user.wx_mp_openid = mp_openid
+    if info.get("unionid"):
+        user.wx_unionid = info["unionid"]
+    user.wx_mp_bind_time = datetime.now()
+    user.update_time = datetime.now()
+    db.add(user)
+    db.commit()
+    return success_response({"bound": True})

backend/app/crud/user.py

@@ -41,6 +41,16 @@ class _UserCRUDCompat(CRUDBase[SysUser]):
             .first()
         )
 
+    def get_by_wx_mp_openid(self, db: Session, wx_mp_openid: str) -> Optional[SysUser]:
+        oid = (wx_mp_openid or "").strip()
+        if not oid:
+            return None
+        return (
+            db.query(SysUser)
+            .filter(SysUser.wx_mp_openid == oid, SysUser.del_flag == 0)
+            .first()
+        )
+
     def list_active_by_phone_no(self, db: Session, phone_no: str) -> List[SysUser]:
         phone = (phone_no or "").strip()
         if not phone:

backend/tests/test_mp_bind.py

@@ -0,0 +1,117 @@
+"""服务号 mp-bind / mp-bind-status：成功与冲突路径（mock oauth）。"""
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+from fastapi import HTTPException
+
+
+@pytest.mark.asyncio
+async def test_mp_bind_success():
+    user = SimpleNamespace(
+        id="u1",
+        del_flag=0,
+        wx_mp_openid=None,
+        wx_unionid=None,
+        wx_mp_bind_time=None,
+        update_time=None,
+    )
+    db = MagicMock()
+    with (
+        patch(
+            "app.api.mini_auth._resolve_user_id",
+            new_callable=AsyncMock,
+            return_value="u1",
+        ),
+        patch(
+            "app.api.mini_auth.oauth_code_to_openid",
+            return_value={"openid": "mp-oid-1", "unionid": "uni-1"},
+        ),
+        patch("app.api.mini_auth.user_crud.get_by_wx_mp_openid", return_value=None),
+        patch("app.api.mini_auth.user_crud.get", return_value=user),
+    ):
+        from app.api.mini_auth import mp_bind
+
+        resp = await mp_bind(
+            request=MagicMock(),
+            body={"code": "oauth-code"},
+            db=db,
+        )
+
+    assert resp["code"] == 200
+    assert resp["data"]["bound"] is True
+    assert user.wx_mp_openid == "mp-oid-1"
+    assert user.wx_unionid == "uni-1"
+    assert user.wx_mp_bind_time is not None
+    db.add.assert_called_once_with(user)
+    db.commit.assert_called_once()
+
+
+@pytest.mark.asyncio
+async def test_mp_bind_conflict_when_openid_bound_to_other():
+    other = SimpleNamespace(id="u-other")
+    db = MagicMock()
+    with (
+        patch(
+            "app.api.mini_auth._resolve_user_id",
+            new_callable=AsyncMock,
+            return_value="u1",
+        ),
+        patch(
+            "app.api.mini_auth.oauth_code_to_openid",
+            return_value={"openid": "mp-oid-taken", "unionid": None},
+        ),
+        patch(
+            "app.api.mini_auth.user_crud.get_by_wx_mp_openid",
+            return_value=other,
+        ),
+    ):
+        from app.api.mini_auth import mp_bind
+
+        resp = await mp_bind(
+            request=MagicMock(),
+            body={"code": "oauth-code"},
+            db=db,
+        )
+
+    assert resp["code"] == 409
+    assert "已绑定其他账号" in resp["message"]
+    db.commit.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_mp_bind_status_bound_true():
+    user = SimpleNamespace(id="u1", wx_mp_openid="mp-oid-1")
+    with (
+        patch(
+            "app.api.mini_auth._resolve_user_id",
+            new_callable=AsyncMock,
+            return_value="u1",
+        ),
+        patch("app.api.mini_auth.user_crud.get", return_value=user),
+    ):
+        from app.api.mini_auth import mp_bind_status
+
+        resp = await mp_bind_status(request=MagicMock(), db=MagicMock())
+
+    assert resp["code"] == 200
+    assert resp["data"]["bound"] is True
+
+
+@pytest.mark.asyncio
+async def test_mp_bind_unauthorized():
+    with patch(
+        "app.api.mini_auth._resolve_user_id",
+        new_callable=AsyncMock,
+        side_effect=HTTPException(status_code=401, detail="未授权"),
+    ):
+        from app.api.mini_auth import mp_bind
+
+        resp = await mp_bind(
+            request=MagicMock(),
+            body={"code": "x"},
+            db=MagicMock(),
+        )
+
+    assert resp["code"] == 401
+    assert resp["message"] == "未授权"