feat(backend): resolve wechat login user by phone and bind openid.

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

backend/app/services/wechat_login_service.py

@@ -0,0 +1,58 @@
+"""微信小程序登录业务：openid + 手机号一致性校验（仅快捷登录绑号）。"""
+from __future__ import annotations
+
+from datetime import datetime
+from typing import Optional, Tuple
+
+from sqlalchemy.orm import Session
+
+from app.crud.user import user_crud
+from app.models.models import SysUser
+from app.services.wechat_mini_auth import normalize_cn_mobile, phones_match
+
+
+def resolve_wechat_login_user(
+    db: Session, *, openid: str, phone: str
+) -> Tuple[Optional[SysUser], Optional[str], Optional[int]]:
+    oid = (openid or "").strip()
+    if not oid:
+        return None, "微信未返回 openid", 400
+    if not normalize_cn_mobile(phone):
+        return None, "请授权手机号登录，或使用账号密码登录", 400
+
+    bound = user_crud.get_by_wx_openid(db, oid)
+    if bound:
+        if getattr(bound, "status", "1") != "1":
+            return None, "账号已被停用，请联系管理员", 403
+        if not phones_match(getattr(bound, "phone_no", None), phone):
+            return (
+                None,
+                "微信与账号不匹配，请联系管理员",
+                401,
+            )
+        return bound, None, None
+
+    users = user_crud.list_active_by_phone_no(db, phone)
+    if not users:
+        return (
+            None,
+            "未找到对应员工账号（请确认手机号已在系统维护）",
+            401,
+        )
+    if len(users) > 1:
+        return None, "手机号对应多个账号，请联系管理员", 400
+
+    user = users[0]
+    other = user_crud.get_by_wx_openid(db, oid)
+    if other and other.id != user.id:
+        return None, "微信与账号不匹配，请联系管理员", 409
+    existing_oid = (getattr(user, "wx_openid", None) or "").strip()
+    if existing_oid and existing_oid != oid:
+        return None, "微信与账号不匹配，请联系管理员", 409
+
+    user.wx_openid = oid
+    user.update_time = datetime.now()
+    db.add(user)
+    db.commit()
+    db.refresh(user)
+    return user, None, None

backend/tests/test_wechat_login_resolve.py

@@ -0,0 +1,88 @@
+from types import SimpleNamespace
+from unittest.mock import MagicMock, patch
+
+from app.services.wechat_login_service import resolve_wechat_login_user
+
+
+def test_openid_bound_phone_mismatch_rejects():
+    bound = SimpleNamespace(id="u1", phone_no="13800138000", wx_openid="oid-a")
+    db = MagicMock()
+    with patch(
+        "app.services.wechat_login_service.user_crud.get_by_wx_openid",
+        return_value=bound,
+    ):
+        user, err, code = resolve_wechat_login_user(
+            db, openid="oid-a", phone="13900139000"
+        )
+    assert user is None
+    assert code == 401
+    assert "不一致" in err or "不匹配" in err
+
+
+def test_openid_bound_phone_match_ok():
+    bound = SimpleNamespace(id="u1", phone_no="13800138000", wx_openid="oid-a")
+    db = MagicMock()
+    with patch(
+        "app.services.wechat_login_service.user_crud.get_by_wx_openid",
+        return_value=bound,
+    ):
+        user, err, code = resolve_wechat_login_user(
+            db, openid="oid-a", phone="+8613800138000"
+        )
+    assert user is bound
+    assert err is None
+    assert code is None
+
+
+def test_unbound_no_user_rejects():
+    db = MagicMock()
+    with patch(
+        "app.services.wechat_login_service.user_crud.get_by_wx_openid",
+        return_value=None,
+    ), patch(
+        "app.services.wechat_login_service.user_crud.list_active_by_phone_no",
+        return_value=[],
+    ):
+        user, err, code = resolve_wechat_login_user(
+            db, openid="oid-new", phone="13800138000"
+        )
+    assert user is None
+    assert code == 401
+
+
+def test_unbound_multi_user_rejects():
+    db = MagicMock()
+    u1 = SimpleNamespace(id="a", phone_no="13800138000", wx_openid=None)
+    u2 = SimpleNamespace(id="b", phone_no="13800138000", wx_openid=None)
+    with patch(
+        "app.services.wechat_login_service.user_crud.get_by_wx_openid",
+        return_value=None,
+    ), patch(
+        "app.services.wechat_login_service.user_crud.list_active_by_phone_no",
+        return_value=[u1, u2],
+    ):
+        user, err, code = resolve_wechat_login_user(
+            db, openid="oid-new", phone="13800138000"
+        )
+    assert user is None
+    assert code == 400
+    assert "多个" in err
+
+
+def test_unbound_unique_binds_openid():
+    db = MagicMock()
+    u = SimpleNamespace(id="a", phone_no="13800138000", wx_openid=None, update_time=None)
+    with patch(
+        "app.services.wechat_login_service.user_crud.get_by_wx_openid",
+        side_effect=[None, None],
+    ), patch(
+        "app.services.wechat_login_service.user_crud.list_active_by_phone_no",
+        return_value=[u],
+    ):
+        user, err, code = resolve_wechat_login_user(
+            db, openid="oid-new", phone="13800138000"
+        )
+    assert err is None
+    assert user is u
+    assert u.wx_openid == "oid-new"
+    db.commit.assert_called()