""" 微信同步登录/注册 Use Case 供 BFF 层调用的系统级接口： - 优先按 unionid 识别用户（跨应用/跨端识别同一微信用户） - 再按 openid 识别（同一应用内） - openid 命中老账号但 unionid 缺失时补写 unionid（开放平台绑定前的存量账号自动关联） - 都未命中则创建新用户 - 支持 unionid 跨应用关联 """ import secrets from datetime import datetime, timedelta, timezone from typing import Optional from uuid import uuid4 import jwt as pyjwt from packages.adapters.redis import get_session_store from packages.application.auth.jwt_service import jwt_service from packages.domain.entities import User class WechatSyncRequest: """微信同步登录请求""" def __init__( self, openid: str, unionid: str = "", nickname: str = "", avatar_url: str = "", source: str = "miniapp", ): self.openid = openid.strip() self.unionid = unionid.strip() if unionid else "" self.nickname = nickname or "小虾同学" # 微信新规拿不到真实昵称，新用户默认昵称 self.avatar_url = avatar_url or "" self.source = source class WechatSyncResponse: """微信同步登录响应""" def __init__( self, access_token: str, refresh_token: str, user_id: str, nickname: str, avatar_url: str, is_new_user: bool, expires_in: int, ): self.access_token = access_token self.refresh_token = refresh_token self.user_id = user_id self.nickname = nickname self.avatar_url = avatar_url self.is_new_user = is_new_user self.expires_in = expires_in def to_dict(self) -> dict: return { "access_token": self.access_token, "token": self.access_token, # 兼容 BFF 层用 token 字段读取 "refresh_token": self.refresh_token, "user_id": self.user_id, "user": { "id": self.user_id, "nickname": self.nickname, "avatar_url": self.avatar_url, "display_name": self.nickname, }, "user_info": { "id": self.user_id, "nickname": self.nickname, "avatar_url": self.avatar_url, "display_name": self.nickname, }, "is_new_user": self.is_new_user, "expires_in": self.expires_in, } class WechatSyncUseCase: """微信登录/注册同步用例 系统级接口，由 BFF 通过 API Key 调用。 职责：根据 unionid/openid 查找或创建用户，返回 SaaS token。 """ def __init__(self, user_repository, session_store=None, jwt_secret_key: str | None = None): self.user_repository = user_repository self.session_store = session_store or get_session_store() self.jwt_secret_key = jwt_secret_key or jwt_service.config.SECRET_KEY def execute(self, request: WechatSyncRequest) -> tuple[Optional[WechatSyncResponse], Optional[str]]: """ 执行微信同步登录/注册 Returns: (响应对象, 错误信息) - 成功则错误信息为 None """ try: if not request.openid: return None, "openid is required" is_new_user = False user = None openid_user = None unionid_user = None # 1. 先按 unionid 查找（跨应用识别同一微信用户，优先级最高） if request.unionid: unionid_user = self.user_repository.find_by_wechat_unionid(request.unionid) # 2. 再按 openid 查找（同一应用内） openid_user = self.user_repository.find_by_wechat_openid(request.openid) if unionid_user and openid_user: # 3a. 两边都命中 if unionid_user.id == openid_user.id: # 同一个用户，直接登录 user = unionid_user else: # unionid 与 openid 分属两个不同账号：数据异常，拒绝写入， # 交由人工/数据修复合并，避免账号被错误串联 return None, ("wechat account conflict: unionid and openid bound to " "different users") elif unionid_user: # 3b. unionid 命中（跨端老用户），当前 openid 未绑定过： # 确认 openid 没有落在其他账号上后，把新 openid 绑到该用户 if openid_user is not None and openid_user.id != unionid_user.id: return None, ("wechat account conflict: openid bound to another user") if unionid_user.wechat_openid != request.openid: unionid_user.wechat_openid = request.openid self.user_repository.save(unionid_user) user = unionid_user elif openid_user: # 3c. 仅 openid 命中（开放平台绑定前创建的存量账号）： # 本次请求带了 unionid 且该账号还没有 unionid 时补写 if request.unionid and not openid_user.wechat_unionid: # 去重：确认该 unionid 没有关联到其他用户 conflict = self.user_repository.find_by_wechat_unionid(request.unionid) if conflict is not None and conflict.id != openid_user.id: return None, ("wechat account conflict: unionid already bound to " "another user") openid_user.wechat_unionid = request.unionid self.user_repository.save(openid_user) user = openid_user else: # 4. 都没找到，创建新用户 # 额外兜底：若 unionid 已被其他账号占用（理论上上面已查过）， # 不创建带冲突 unionid 的新账号 user = self._create_wechat_user(request) is_new_user = True # 5. 创建 session 并生成 token session_id = secrets.token_urlsafe(16) refresh_token = secrets.token_urlsafe(32) now = datetime.now(timezone.utc) access_token_payload = { "sub": user.id, "sid": session_id, "type": "user_auth", "iat": now, "exp": now + timedelta(minutes=jwt_service.config.ACCESS_TOKEN_EXPIRE_MINUTES), } access_token = pyjwt.encode( access_token_payload, self.jwt_secret_key, algorithm=jwt_service.config.ALGORITHM, ) # 保存 session self.session_store.save_session( session_id=session_id, user_id=user.id, refresh_token=refresh_token, device_info=f"wechat_{request.source}", ip_address="bff_gateway", expires_in_seconds=30 * 24 * 3600, # 30 天 ) # 更新最后登录信息 user.last_login_at = now user.last_login_ip = "bff_gateway" self.user_repository.save(user) response = WechatSyncResponse( access_token=access_token, refresh_token=refresh_token, user_id=user.id, nickname=user.display_name, avatar_url="", # SaaS 用户模型暂存头像，后续可扩展 is_new_user=is_new_user, expires_in=jwt_service.config.ACCESS_TOKEN_EXPIRE_MINUTES * 60, ) return response, None except Exception as e: return None, f"Internal error: {str(e)}" def _create_wechat_user(self, request: WechatSyncRequest) -> User: """创建微信用户""" user_id = uuid4().hex # 生成唯一名和邮箱（微信用户无真实邮箱，用 openid 生成占位） safe_openid = request.openid.replace("-", "_")[:20] username = f"wx_{safe_openid}" email = f"{safe_openid}@wechat.local" # 确保 username 唯一 suffix = 0 while self.user_repository.find_by_username(username): suffix += 1 username = f"wx_{safe_openid}_{suffix}" # 随机密码（微信用户不用密码登录） random_password = secrets.token_urlsafe(32) from packages.application.auth.password_hasher import password_hasher password_hash = password_hasher.hash_password(random_password) user = User( id=user_id, email=email, username=username, display_name=request.nickname or "小虾同学", password_hash=password_hash, email_verified=True, # 微信登录视为已验证 wechat_openid=request.openid, wechat_unionid=request.unionid or None, # 微信新建用户首次登录需引导设置昵称 profile_completed=False, ) self.user_repository.save(user) return user