小程序、App、H5微应用：企业培训场景下的三端登录态同步方案
·
一、问题全景：三套登录体系，一个用户
先说清楚三端的登录机制差异，这是所有问题的根源。
| 维度 | 微信小程序 | 原生App | H5微应用（钉钉/企微/飞书） |
|---|---|---|---|
| 登录方式 | wx.login获取code → 后端换openid | 账号密码 / 手机号验证码 | OAuth免登（授权码模式） |
| 身份凭证 | openid + unionid | JWT access_token + refresh_token | 平台access_token（钉钉/企微/飞书各自签发） |
| 会话生命周期 | 受微信session管理，可能被系统回收 | App前台/后台切换，可能长时间存活 | 跟随宿主App生命周期，关闭Tab即失效 |
| 多设备 | 同一微信号只能在一个设备登录 | 可以多设备同时登录（手机+平板） | 跟随宿主App的设备 |
| 退出登录 | 用户主动退出 / 长期未使用 | 用户主动退出 / Token过期 | 用户退出宿主App / Token过期 |
这三套完全不同的登录机制，背后对应的是三个不同的身份源：微信体系的用户身份、企学宝自有账户体系的用户身份、以及企业OA平台（钉钉/企微/飞书）的员工身份。
核心挑战是：这三个身份源必须映射到同一个"企学宝用户"上。 一个员工在钉钉里叫"张三（工号10086）“，在微信小程序里叫"微信用户oXXXX”，在App里用手机号13800138000登录——系统必须知道这三个是同一个张三。
二、统一身份层：三源归一
2.1 账号关联模型
我们设计了一个"主账号 + 关联身份"的模型：
-- 主账号：企学宝内部唯一用户标识
CREATE TABLE accounts (
id BIGINT PRIMARY KEY,
account_uid VARCHAR(64) UNIQUE NOT NULL, -- 全局唯一用户标识
display_name VARCHAR(128),
avatar_url VARCHAR(512),
status VARCHAR(20) DEFAULT 'active',
created_at TIMESTAMP DEFAULT NOW(),
updated_at TIMESTAMP DEFAULT NOW()
);
-- 关联身份：每个外部身份源对应一条记录
CREATE TABLE account_identities (
id BIGINT PRIMARY KEY,
account_uid VARCHAR(64) NOT NULL REFERENCES accounts(account_uid),
identity_type VARCHAR(32) NOT NULL, -- 'wechat' / 'mobile' / 'dingtalk' / 'wecom' / 'feishu'
identity_id VARCHAR(256) NOT NULL, -- 各平台的用户唯一标识
identity_name VARCHAR(256), -- 该平台上的用户名
corp_id VARCHAR(128), -- 所属企业ID（OA平台场景）
extra_data JSONB, -- 平台特有的扩展信息
bound_at TIMESTAMP DEFAULT NOW(),
last_login_at TIMESTAMP,
-- 同一identity_type + identity_id + corp_id 唯一
UNIQUE(identity_type, identity_id, corp_id)
);
-- 索引：通过外部身份快速查找主账号
CREATE INDEX idx_identity_lookup ON account_identities(identity_type, identity_id, corp_id);
CREATE INDEX idx_identity_account ON account_identities(account_uid);
这个模型的关键设计点：
第一，account_uid是系统内部的全局唯一标识。 所有业务数据（学习记录、考试成绩、证书）都关联account_uid，而不是关联某个外部平台的用户ID。这样无论用户从哪个端登录，只要映射到同一个account_uid，数据就是通的。
第二，corp_id解决了一个常见难题——同一个员工在不同企业里可能有不同的OA身份。 比如张三同时在A集团（钉钉）和B子公司（企微）任职，他在两个OA平台上的员工ID不同，但可能映射到同一个企学宝账号（如果A集团和B子公司是同一个客户），也可能映射到不同账号（如果是两个独立客户）。corp_id让这个区分成为可能。
2.2 身份绑定流程
用户第一次从某个端登录时，系统需要完成"身份绑定"——把外部身份关联到主账号上。绑定的触发条件因端而异：
class IdentityBinder:
"""身份绑定器：处理首次登录的身份关联"""
async def bind_on_first_login(self, identity_type: str,
identity_id: str,
corp_id: str = None,
profile: dict = None) -> str:
"""
首次登录时的身份绑定逻辑
返回 account_uid
"""
# 1. 先查是否已有绑定
existing = await self.db.find_identity(
identity_type=identity_type,
identity_id=identity_id,
corp_id=corp_id
)
if existing:
# 已有绑定，更新最后登录时间，返回主账号
await self.db.update_last_login(existing.account_uid)
return existing.account_uid
# 2. 没有绑定，尝试自动关联
account_uid = await self._try_auto_bind(
identity_type, identity_id, corp_id, profile
)
return account_uid
async def _try_auto_bind(self, identity_type: str,
identity_id: str,
corp_id: str = None,
profile: dict = None) -> str:
"""
自动绑定策略：按优先级尝试关联到已有主账号
"""
# 策略1：通过手机号关联
# 如果当前身份携带了手机号，且该手机号已注册过主账号
if profile and profile.get("mobile"):
existing_account = await self.db.find_account_by_mobile(
profile["mobile"]
)
if existing_account:
await self._create_identity_link(
existing_account.account_uid,
identity_type, identity_id, corp_id, profile
)
return existing_account.account_uid
# 策略2：通过企业邮箱关联
# OA平台（钉钉/企微/飞书）通常能提供员工的企业邮箱
if profile and profile.get("email"):
existing_account = await self.db.find_account_by_email(
profile["email"]
)
if existing_account:
await self._create_identity_link(
existing_account.account_uid,
identity_type, identity_id, corp_id, profile
)
return existing_account.account_uid
# 策略3：通过企业通讯录匹配
# 如果corp_id对应的企业已有员工列表，通过工号/姓名匹配
if corp_id and profile and profile.get("employee_no"):
existing_account = await self.db.find_account_by_employee_no(
corp_id, profile["employee_no"]
)
if existing_account:
await self._create_identity_link(
existing_account.account_uid,
identity_type, identity_id, corp_id, profile
)
return existing_account.account_uid
# 策略4：无法自动关联，创建新主账号
new_account = await self._create_new_account(profile)
await self._create_identity_link(
new_account.account_uid,
identity_type, identity_id, corp_id, profile
)
return new_account.account_uid
自动绑定的优先级设计很关键——手机号是最强的关联信号（一个人通常只有一个手机号），企业邮箱次之，工号再次之。如果所有策略都匹配不上，才创建新账号。
但自动绑定不是万能的。有一种常见场景：员工先用手机号在App上注册了账号，后来企业在钉钉上开通了企学宝，员工通过钉钉H5免登进入——此时系统可以通过手机号自动关联（如果钉钉通讯录里有手机号）。但如果钉钉通讯录里没有手机号（很多企业不填），自动绑定就失败了，需要走手动绑定流程。
2.3 手动绑定：用户主动关联多端身份
当自动绑定失败时，需要用户在端上主动完成绑定：
class ManualBindFlow:
"""手动绑定流程"""
async def initiate_bind(self, account_uid: str,
target_identity_type: str) -> dict:
"""
发起绑定：生成绑定凭证
场景：用户在App上扫描小程序二维码，或在小程序上输入手机号验证码
"""
bind_token = generate_secure_token(ttl=300) # 5分钟有效
await self.redis.setex(
f"bind:{bind_token}",
300,
json.dumps({
"account_uid": account_uid,
"target_type": target_identity_type,
"created_at": datetime.utcnow().isoformat()
})
)
return {
"bind_token": bind_token,
"expires_in": 300,
"bind_url": f"https://qxb.example.com/bind/{bind_token}"
}
async def confirm_bind(self, bind_token: str,
identity_type: str,
identity_id: str,
verification: dict) -> bool:
"""
确认绑定：验证用户身份后完成关联
"""
# 1. 验证bind_token有效性
bind_data = await self.redis.get(f"bind:{bind_token}")
if not bind_data:
raise BindTokenExpiredError("绑定凭证已过期")
bind_info = json.loads(bind_data)
account_uid = bind_info["account_uid"]
# 2. 验证用户身份（手机号验证码 / 密码验证）
if verification.get("type") == "mobile_sms":
is_valid = await self._verify_sms_code(
verification["mobile"],
verification["code"]
)
if not is_valid:
raise VerificationFailedError("验证码错误")
# 确认该手机号属于目标账号
account = await self.db.get_account(account_uid)
if account.mobile != verification["mobile"]:
raise VerificationFailedError("手机号不匹配")
elif verification.get("type") == "password":
is_valid = await self._verify_password(
account_uid,
verification["password"]
)
if not is_valid:
raise VerificationFailedError("密码错误")
# 3. 检查目标身份是否已被其他账号绑定
existing_link = await self.db.find_identity(
identity_type=identity_type,
identity_id=identity_id
)
if existing_link and existing_link.account_uid != account_uid:
raise IdentityAlreadyBoundError(
"该身份已关联其他账号，请先解绑"
)
# 4. 创建绑定
await self._create_identity_link(
account_uid, identity_type, identity_id
)
# 5. 清理bind_token
await self.redis.delete(f"bind:{bind_token}")
return True
三、Token管理：三端各自的会话策略
登录态同步不只是"识别用户是谁"，还要管理好每一端的Token生命周期。三端的Token策略差异很大。
3.1 统一的Token结构
不管用户从哪个端登录，后端签发的Token结构是统一的：
import jwt
import time
from dataclasses import dataclass
@dataclass
class TokenPayload:
"""统一的Token载荷结构"""
account_uid: str # 主账号标识
device_type: str # 'app_ios' / 'app_android' / 'miniapp' / 'h5_dingtalk' / 'h5_wecom' / 'h5_feishu'
session_id: str # 会话ID（用于单点踢出）
identity_type: str # 本次登录使用的身份类型
corp_id: str # 所属企业
issued_at: int # 签发时间
expires_at: int # 过期时间
def to_dict(self) -> dict:
return {
"uid": self.account_uid,
"dev": self.device_type,
"sid": self.session_id,
"idp": self.identity_type,
"corp": self.corp_id,
"iat": self.issued_at,
"exp": self.expires_at,
}
class TokenService:
"""Token签发与验证服务"""
ACCESS_TOKEN_TTL = 7200 # access_token 2小时
REFRESH_TOKEN_TTL = 2592000 # refresh_token 30天
def __init__(self, secret_key: str, algorithm: str = "HS256"):
self.secret_key = secret_key
self.algorithm = algorithm
def issue_token_pair(self, account_uid: str, device_type: str,
identity_type: str, corp_id: str) -> dict:
"""签发 access_token + refresh_token 对"""
now = int(time.time())
session_id = generate_session_id()
# access_token
access_payload = TokenPayload(
account_uid=account_uid,
device_type=device_type,
session_id=session_id,
identity_type=identity_type,
corp_id=corp_id,
issued_at=now,
expires_at=now + self.ACCESS_TOKEN_TTL,
)
access_token = jwt.encode(
access_payload.to_dict(),
self.secret_key,
algorithm=self.algorithm
)
# refresh_token（更长有效期，存储在Redis中可主动吊销）
refresh_token = generate_secure_token()
await self.redis.setex(
f"refresh:{refresh_token}",
self.REFRESH_TOKEN_TTL,
json.dumps({
"account_uid": account_uid,
"session_id": session_id,
"device_type": device_type,
})
)
return {
"access_token": access_token,
"access_token_expires_in": self.ACCESS_TOKEN_TTL,
"refresh_token": refresh_token,
"refresh_token_expires_in": self.REFRESH_TOKEN_TTL,
"session_id": session_id,
}
def verify_access_token(self, token: str) -> TokenPayload:
"""验证access_token"""
try:
payload = jwt.decode(token, self.secret_key,
algorithms=[self.algorithm])
return TokenPayload(**payload)
except jwt.ExpiredSignatureError:
raise TokenExpiredError()
except jwt.InvalidTokenError:
raise TokenInvalidError()
3.2 各端的Token刷新策略
三端的Token刷新策略必须差异化设计，因为各端的生命周期管理方式完全不同。
App端：后台保活 + 静默刷新
class AppTokenRefresher:
"""
App端Token刷新策略
特点：App可能长时间在后台，恢复前台时Token可能已过期
"""
async def on_app_foreground(self, current_token: str,
refresh_token: str) -> dict:
"""App回到前台时的Token处理"""
# 1. 尝试用当前access_token发一个轻量请求（如获取用户信息）
try:
token_payload = self.token_service.verify_access_token(current_token)
# Token未过期，继续使用
return {"token": current_token, "refreshed": False}
except TokenExpiredError:
pass
# 2. access_token过期，用refresh_token静默刷新
try:
new_tokens = await self._refresh_with_token(refresh_token)
return {
"token": new_tokens["access_token"],
"refresh_token": new_tokens["refresh_token"],
"refreshed": True
}
except RefreshTokenExpiredError:
# 3. refresh_token也过期，需要重新登录
return {"need_relogin": True}
async def _refresh_with_token(self, refresh_token: str) -> dict:
"""用refresh_token换取新的token对"""
# 验证refresh_token有效性
stored = await self.redis.get(f"refresh:{refresh_token}")
if not stored:
raise RefreshTokenExpiredError()
stored_data = json.loads(stored)
# 签发新的token对
new_tokens = self.token_service.issue_token_pair(
account_uid=stored_data["account_uid"],
device_type=stored_data["device_type"],
identity_type="refreshed",
corp_id=stored_data.get("corp_id", "")
)
# 旧的refresh_token失效（rotation策略）
await self.redis.delete(f"refresh:{refresh_token}")
return new_tokens
App端的关键设计是refresh_token rotation——每次用refresh_token换新token时，旧的refresh_token立即失效。这样即使refresh_token泄露，攻击者最多只能用一次。
小程序端：冷启动重建 + 静默登录
class MiniAppSessionManager:
"""
小程序Token管理策略
特点：小程序可能被微信系统回收（冷启动），需要无感重建会话
"""
async def on_launch(self, wx_code: str,
cached_token: str = None) -> dict:
"""
小程序启动时的会话管理
"""
# 1. 如果有缓存的token，先验证是否有效
if cached_token:
try:
payload = self.token_service.verify_access_token(cached_token)
return {
"token": cached_token,
"account_uid": payload.account_uid,
"source": "cache"
}
except (TokenExpiredError, TokenInvalidError):
pass
# 2. 缓存token无效，用wx.code静默登录
# 小程序的wx.login是静默的，用户无感知
wx_session = await self.wechat_api.code2session(wx_code)
openid = wx_session["openid"]
# 3. 通过openid查找关联的主账号
identity = await self.db.find_identity(
identity_type="wechat",
identity_id=openid
)
if identity:
# 已有账号，签发新token
tokens = self.token_service.issue_token_pair(
account_uid=identity.account_uid,
device_type="miniapp",
identity_type="wechat",
corp_id=identity.corp_id or ""
)
return {
"token": tokens["access_token"],
"account_uid": identity.account_uid,
"source": "silent_login"
}
else:
# 新用户，需要走注册/绑定流程
return {"need_register": True, "wx_session": wx_session}
async def on_check_session(self) -> bool:
"""
微信session过期检查
微信的session可能会在以下情况失效：
- 用户长时间未使用小程序
- 微信系统主动回收
- 用户在其他设备登录同一微信号
"""
try:
await self.wechat_api.check_session()
return True
except WxSessionExpiredError:
# 需要重新wx.login
return False
小程序端的关键设计是静默登录——用户打开小程序时，通过wx.login自动获取code，后端通过code换openid，再通过openid找到主账号，整个过程用户完全无感知。只有首次使用（openid未绑定账号）时才需要用户交互。
H5微应用端：跟随宿主 + 免登刷新
class H5MicroAppSessionManager:
"""
H5微应用Token管理策略
特点：生命周期跟随宿主App（钉钉/企微/飞书），关闭Tab即失效
"""
async def on_page_load(self, platform: str,
auth_code: str = None,
cached_token: str = None) -> dict:
"""
H5页面加载时的会话管理
"""
# 1. 检查缓存token
if cached_token:
try:
payload = self.token_service.verify_access_token(cached_token)
return {"token": cached_token, "source": "cache"}
except (TokenExpiredError, TokenInvalidError):
pass
# 2. 使用OA平台的免登授权码
if not auth_code:
# 需要前端跳转到OA平台的授权页面获取auth_code
return {
"need_auth": True,
"auth_url": self._build_auth_url(platform)
}
# 3. 用auth_code换取OA平台的用户信息
user_info = await self._get_oauth_user_info(platform, auth_code)
# 4. 通过OA平台身份查找主账号
identity_type = self._platform_to_identity_type(platform)
identity = await self.db.find_identity(
identity_type=identity_type,
identity_id=user_info["user_id"],
corp_id=user_info["corp_id"]
)
if identity:
tokens = self.token_service.issue_token_pair(
account_uid=identity.account_uid,
device_type=f"h5_{platform}",
identity_type=identity_type,
corp_id=identity.corp_id
)
return {
"token": tokens["access_token"],
"account_uid": identity.account_uid,
"source": "oauth_login"
}
else:
# OA平台用户未绑定企学宝账号
return {
"need_bindaccount": True,
"user_info": user_info
}
def _platform_to_identity_type(self, platform: str) -> str:
mapping = {
"dingtalk": "dingtalk",
"wecom": "wecom",
"feishu": "feishu",
}
return mapping.get(platform, platform)
async def _get_oauth_user_info(self, platform: str,
auth_code: str) -> dict:
"""用授权码换取OA平台用户信息"""
if platform == "dingtalk":
return await self._dingtalk_get_user(auth_code)
elif platform == "wecom":
return await self._wecom_get_user(auth_code)
elif platform == "feishu":
return await self._feishu_get_user(auth_code)
async def _dingtalk_get_user(self, auth_code: str) -> dict:
"""钉钉免登流程"""
# 1. 用auth_code获取钉钉用户access_token
token_resp = await self.dingtalk_api.get_access_token(auth_code)
# 2. 用access_token获取用户信息
user_resp = await self.dingtalk_api.get_user_info(
token_resp["access_token"]
)
return {
"user_id": user_resp["userid"],
"name": user_resp["name"],
"corp_id": user_resp["corp_id"],
"mobile": user_resp.get("mobile"),
"email": user_resp.get("email"),
"avatar": user_resp.get("avatar"),
}
async def _wecom_get_user(self, auth_code: str) -> dict:
"""企业微信免登流程"""
# 企微的免登流程和钉钉类似但细节不同
token_resp = await self.wecom_api.get_access_token()
user_resp = await self.wecom_api.get_user_info(
token_resp["access_token"], auth_code
)
return {
"user_id": user_resp["UserId"],
"name": user_resp.get("name"),
"corp_id": token_resp["corp_id"],
"mobile": user_resp.get("mobile"),
"email": user_resp.get("email"),
}
async def _feishu_get_user(self, auth_code: str) -> dict:
"""飞书免登流程"""
# 飞书的流程又不一样——需要先获取app_access_token再获取user_access_token
app_token = await self.feishu_api.get_app_access_token()
user_token = await self.feishu_api.get_user_access_token(
app_token, auth_code
)
user_info = await self.feishu_api.get_user_info(
user_token["user_access_token"]
)
return {
"user_id": user_info["open_id"],
"name": user_info.get("name"),
"corp_id": user_info.get("tenant_key"),
"mobile": user_info.get("mobile"),
"email": user_info.get("enterprise_email"),
}
H5端的关键特点是生命周期短——用户关闭钉钉里的H5 Tab，会话就没了。所以H5端每次打开都需要重新走免登流程，但因为免登是静默的（用户不需要输入密码），体验上还是"打开即用"。
3.3 跨端Token互不影响
一个重要的设计原则：各端的Token互相独立，一端登出不影响其他端。
class SessionManager:
"""会话管理：支持多端同时在线"""
async def logout(self, account_uid: str, device_type: str = None,
session_id: str = None):
"""
登出策略：
- 指定session_id：只登出该会话
- 指定device_type：登出该类型的所有会话
- 都不指定：登出所有端（全量登出）
"""
if session_id:
# 单会话登出
await self._invalidate_session(session_id)
elif device_type:
# 按设备类型登出
sessions = await self.redis.smembers(
f"sessions:{account_uid}:{device_type}"
)
for sid in sessions:
await self._invalidate_session(sid)
else:
# 全量登出
all_sessions = await self.redis.smembers(
f"sessions:{account_uid}:*"
)
for sid in all_sessions:
await self._invalidate_session(sid)
async def _invalidate_session(self, session_id: str):
"""使一个会话失效"""
# 1. 删除session对应的refresh_token
refresh_tokens = await self.redis.smembers(
f"session:{session_id}:refresh_tokens"
)
for rt in refresh_tokens:
await self.redis.delete(f"refresh:{rt}")
# 2. 将session标记为失效（access_token可能还没过期）
await self.redis.setex(
f"session:{session_id}:revoked",
7200, # 和access_token的TTL对齐
"1"
)
# 3. 从用户的会话集合中移除
session_meta = await self.redis.hget(f"session:{session_id}:meta", "account_uid")
device_type = await self.redis.hget(f"session:{session_id}:meta", "device_type")
if session_meta:
await self.redis.srem(f"sessions:{session_meta}:{device_type}", session_id)
await self.redis.delete(f"session:{session_id}:meta")
await self.redis.delete(f"session:{session_id}:refresh_tokens")
async def is_session_valid(self, session_id: str) -> bool:
"""检查会话是否有效"""
revoked = await self.redis.get(f"session:{session_id}:revoked")
return revoked is None
这个设计让用户可以在App上保持登录的同时，单独退出H5端的会话——比如员工下班后退出钉钉里的H5微应用，但不影响App上的学习进度。
四、学习进度跨端同步：真正的难题
登录态同步只是基础。用户真正关心的是：我在App上看到第3章，打开小程序应该也从第3章继续。 这就是学习进度的跨端同步。
4.1 进度数据模型
-- 学习进度主表
CREATE TABLE learning_progress (
id BIGINT PRIMARY KEY,
account_uid VARCHAR(64) NOT NULL,
course_id VARCHAR(64) NOT NULL,
device_type VARCHAR(32) NOT NULL, -- 最后更新进度的设备类型
current_chapter_id VARCHAR(64),
current_position_sec INT DEFAULT 0, -- 当前播放位置（秒）
completion_ratio DECIMAL(5,4) DEFAULT 0, -- 完成比例（0.0000 - 1.0000）
total_study_duration_sec INT DEFAULT 0, -- 累计学习时长
last_study_at TIMESTAMP,
created_at TIMESTAMP DEFAULT NOW(),
updated_at TIMESTAMP DEFAULT NOW(),
-- 一个用户一门课只有一条进度记录
UNIQUE(account_uid, course_id)
);
-- 章节级进度明细
CREATE TABLE chapter_progress (
id BIGINT PRIMARY KEY,
account_uid VARCHAR(64) NOT NULL,
course_id VARCHAR(64) NOT NULL,
chapter_id VARCHAR(64) NOT NULL,
status VARCHAR(20) DEFAULT 'not_started', -- not_started / in_progress / completed
position_sec INT DEFAULT 0,
duration_sec INT DEFAULT 0, -- 本章累计学习时长
score DECIMAL(5,2), -- 章节测验分数（如有）
completed_at TIMESTAMP,
updated_at TIMESTAMP DEFAULT NOW(),
UNIQUE(account_uid, course_id, chapter_id)
);
-- 进度同步日志（用于冲突检测和审计）
CREATE TABLE progress_sync_log (
id BIGINT PRIMARY KEY,
account_uid VARCHAR(64) NOT NULL,
course_id VARCHAR(64) NOT NULL,
device_type VARCHAR(32) NOT NULL,
sync_type VARCHAR(20) NOT NULL, -- 'push' / 'pull' / 'merge'
progress_snapshot JSONB, -- 同步时的进度快照
server_version INT, -- 服务端版本号（乐观锁）
created_at TIMESTAMP DEFAULT NOW()
);
4.2 冲突场景与合并策略
跨端同步最头疼的是冲突——两端同时学习同一门课，各自推进了进度，然后都要同步到服务端。
场景1：顺序冲突
用户在App上学到了第3章第15分钟，同时在小程序上也学到了第3章第20分钟。两个进度不一致，以哪个为准？
场景2：分支冲突
用户在App上完成了第3章的测验（得分85分），同时在H5上完成了第4章的视频观看。两个进度不矛盾，但需要同步合并。
场景3：回退冲突
用户在App上看到了第5章，然后退出。后来在小程序上重新打开，但小程序的本地缓存还停留在第3章。此时应该用服务端的第5章覆盖小程序的本地缓存，还是保留小程序的状态？
class ProgressSyncService:
"""学习进度同步服务：处理跨端冲突"""
async def sync_progress(self, account_uid: str,
course_id: str,
client_progress: dict,
device_type: str,
client_version: int) -> dict:
"""
同步学习进度
采用"服务端优先 + 乐观合并"策略
"""
# 1. 获取服务端当前进度
server_progress = await self.db.get_progress(account_uid, course_id)
if not server_progress:
# 服务端没有进度记录，直接以客户端为准
await self._save_progress(account_uid, course_id,
client_progress, device_type)
return {
"progress": client_progress,
"conflict": False,
"source": "client"
}
# 2. 检测冲突
conflict = self._detect_conflict(
server_progress, client_progress, client_version
)
if not conflict:
# 无冲突：客户端进度是服务端进度的延续
# 直接更新
await self._update_progress(
account_uid, course_id,
client_progress, device_type,
client_version
)
return {
"progress": client_progress,
"conflict": False,
"source": "merged"
}
# 3. 有冲突：执行合并策略
merged = self._merge_progress(
server_progress, client_progress, device_type
)
await self._save_progress(
account_uid, course_id,
merged, device_type
)
# 4. 记录同步日志
await self._log_sync(
account_uid, course_id,
device_type, "merge",
merged, client_version
)
return {
"progress": merged,
"conflict": True,
"resolution": "server_wins_with_merge",
"source": "server"
}
def _detect_conflict(self, server: dict, client: dict,
client_version: int) -> bool:
"""
检测冲突：
- 如果client_version == server_version，说明客户端拿到的是最新版本，无冲突
- 如果client_version < server_version，说明服务端在客户端拉取后又有更新，可能有冲突
"""
server_version = server.get("version", 0)
return client_version < server_version
def _merge_progress(self, server: dict, client: dict,
device_type: str) -> dict:
"""
合并策略：
1. 章节进度取"最远"——哪个进度走得远就用哪个
2. 视频位置取"最远"——同一章节内，播放位置靠后的为准
3. 测验成绩取"最高"——如果两端都做了测验，取高分
4. 学习时长取"累加"——两端的学习时长相加（去重）
"""
merged = {}
# 章节进度：取最远
server_chapter = server.get("current_chapter_id", "")
client_chapter = client.get("current_chapter_id", "")
if self._chapter_order(client_chapter) >= self._chapter_order(server_chapter):
merged["current_chapter_id"] = client_chapter
merged["current_position_sec"] = client.get("current_position_sec", 0)
else:
merged["current_chapter_id"] = server_chapter
merged["current_position_sec"] = server.get("current_position_sec", 0)
# 完成比例：取较大值
merged["completion_ratio"] = max(
server.get("completion_ratio", 0),
client.get("completion_ratio", 0)
)
# 学习时长：取较大值（不是累加，因为可能重叠）
# 更精确的做法是按时段合并，但工程上取较大值已经够用
merged["total_study_duration_sec"] = max(
server.get("total_study_duration_sec", 0),
client.get("total_study_duration_sec", 0)
)
merged["device_type"] = device_type
merged["last_study_at"] = datetime.utcnow().isoformat()
return merged
def _chapter_order(self, chapter_id: str) -> int:
"""获取章节的顺序号（用于比较先后）"""
# 从课程结构中获取章节顺序
return self.chapter_index.get(chapter_id, 0)
4.3 各端的同步时机
三端的同步时机不同，这直接影响用户体验：
class SyncTimingStrategy:
"""各端的同步时机策略"""
STRATEGIES = {
"app": {
# App端：实时同步 + 后台批量同步
"on_chapter_complete": "immediate", # 完成一章立即同步
"on_video_position_change": "throttle_30s", # 视频位置变更每30秒同步一次
"on_app_background": "immediate", # App进入后台时立即同步
"on_app_foreground": "pull_then_push", # App回到前台时先拉取再推送
"periodic": "interval_60s", # 每60秒定期同步
},
"miniapp": {
# 小程序端：事件驱动同步
"on_chapter_complete": "immediate",
"on_video_position_change": "throttle_60s", # 节流到60秒（小程序request有并发限制）
"on_hide": "immediate", # 小程序切后台时立即同步
"on_show": "pull_then_push", # 小程序切前台时先拉后推
"on_share": "immediate", # 分享前同步（确保进度最新）
},
"h5": {
# H5端：页面生命周期驱动
"on_chapter_complete": "immediate",
"on_video_position_change": "throttle_30s",
"on_page_unload": "immediate", # 页面关闭前同步（用sendBeacon）
"on_page_load": "pull", # 页面加载时拉取最新进度
"on_visibility_change": "throttle_10s", # 页面可见性变化时同步
},
}
App端可以做得最激进——实时同步，因为App有稳定的网络连接和后台运行能力。
小程序端需要节制一些——微信对小程序的wx.request有并发限制（同时最多5-10个），过于频繁的同步请求可能影响正常的业务请求。所以视频播放位置的同步节流到60秒。
H5端最特殊——页面随时可能被关闭（用户切Tab、关闭钉钉），所以必须在beforeunload/visibilitychange事件里用sendBeacon做最后一次同步。sendBeacon的特点是即使页面已经关闭，请求也会发出去，非常适合这个场景。
4.4 离线学习后的同步
最复杂的场景是离线学习——用户在App上离线看了一节课，期间没有网络。恢复网络后，需要把离线期间的进度同步到服务端。
class OfflineProgressSync:
"""离线学习进度同步"""
async def sync_offline_progress(self, account_uid: str,
offline_records: list[dict]) -> dict:
"""
同步离线期间积累的进度记录
offline_records: 离线期间产生的进度变更列表，按时间排序
"""
results = []
for record in offline_records:
course_id = record["course_id"]
# 获取服务端当前进度
server_progress = await self.db.get_progress(
account_uid, course_id
)
if not server_progress:
# 服务端没有进度，直接写入
await self._save_progress(
account_uid, course_id,
record["progress"], "app_offline"
)
results.append({
"course_id": course_id,
"status": "saved",
"conflict": False
})
continue
# 检查是否有冲突
# 离线期间的进度和服务端进度可能不一致
# （比如用户在另一端也学了同一门课）
if self._has_conflict(server_progress, record):
# 有冲突：合并
merged = self._merge_offline_with_server(
server_progress, record
)
await self._save_progress(
account_uid, course_id,
merged, "app_offline_merged"
)
results.append({
"course_id": course_id,
"status": "merged",
"conflict": True,
"merged_progress": merged
})
else:
# 无冲突：直接追加
updated = self._apply_offline_record(
server_progress, record
)
await self._update_progress(
account_uid, course_id,
updated, "app_offline"
)
results.append({
"course_id": course_id,
"status": "updated",
"conflict": False
})
return {
"synced_count": len(results),
"conflict_count": sum(1 for r in results if r.get("conflict")),
"details": results
}
def _merge_offline_with_server(self, server: dict,
offline_record: dict) -> dict:
"""
合并离线进度和服务端进度
策略：离线进度的"完成状态"优先保留，
但"播放位置"以服务端为准（如果服务端更新）
"""
merged = dict(server)
offline_progress = offline_record["progress"]
# 完成比例：取较大值
if offline_progress.get("completion_ratio", 0) > merged.get("completion_ratio", 0):
merged["completion_ratio"] = offline_progress["completion_ratio"]
merged["current_chapter_id"] = offline_progress.get("current_chapter_id")
# 章节完成状态：离线标记为completed的章节，不会被服务端覆盖
offline_chapters = offline_progress.get("chapter_details", {})
for ch_id, ch_progress in offline_chapters.items():
if ch_progress.get("status") == "completed":
merged.setdefault("chapter_details", {})
merged["chapter_details"][ch_id] = ch_progress
# 学习时长：累加（离线期间的时长是真实的增量）
offline_duration = offline_record.get("duration_added_sec", 0)
merged["total_study_duration_sec"] = (
merged.get("total_study_duration_sec", 0) + offline_duration
)
return merged
五、前端SDK：统一的登录与同步接口
三端的前端实现各不相同，但对外暴露的接口应该统一。我们封装了一个跨端的SDK：
// qxb-sdk.ts - 企学宝跨端SDK核心接口
interface QXBAccount {
accountUid: string;
displayName: string;
avatarUrl: string;
corpId?: string;
}
interface QXBLearningProgress {
courseId: string;
chapterId: string;
positionSec: number;
completionRatio: number;
totalDurationSec: number;
}
interface IQXBSDK {
// ── 登录相关 ──
login(): Promise<QXBAccount>;
logout(): Promise<void>;
getAccount(): QXBAccount | null;
onAccountChange(callback: (account: QXBAccount | null) => void): void;
// ── 学习进度 ──
getProgress(courseId: string): Promise<QXBLearningProgress>;
syncProgress(progress: QXBLearningProgress): Promise<void>;
onProgressUpdate(callback: (progress: QXBLearningProgress) => void): void;
// ── 事件 ──
on(event: 'login' | 'logout' | 'progress_sync' | 'conflict',
callback: (...args: any[]) => void): void;
}
各端的实现：
// app-sdk.ts - App端实现
class AppQXBSDK implements IQXBSDK {
private tokenStorage: SecureStorage; // Keychain / Keystore
private syncTimer: number | null = null;
async login(): Promise<QXBAccount> {
// App端：检查本地Token → 静默刷新 → 弹出登录页
const cachedToken = this.tokenStorage.getAccessToken();
if (cachedToken) {
try {
const account = await this.verifyToken(cachedToken);
this.startPeriodicSync();
return account;
} catch (e) {
// Token过期，尝试刷新
const refreshToken = this.tokenStorage.getRefreshToken();
if (refreshToken) {
try {
const newTokens = await this.refreshTokens(refreshToken);
this.tokenStorage.save(newTokens);
const account = await this.verifyToken(newTokens.accessToken);
this.startPeriodicSync();
return account;
} catch (e) {
// refresh也失败，需要重新登录
}
}
}
}
// 弹出登录页
const result = await Navigation.push('LoginPage');
this.tokenStorage.save(result.tokens);
this.startPeriodicSync();
return result.account;
}
async syncProgress(progress: QXBLearningProgress): Promise<void> {
// App端：实时同步，带节流
this.throttledSync(progress, 30000); // 30秒节流
}
private startPeriodicSync() {
// 每60秒定期同步一次进度
this.syncTimer = setInterval(() => {
const currentProgress = this.localProgressManager.getCurrent();
if (currentProgress) {
this.syncProgress(currentProgress);
}
}, 60000);
}
}
// miniapp-sdk.ts - 小程序端实现
class MiniAppQXBSDK implements IQXBSDK {
async login(): Promise<QXBAccount> {
// 小程序端：wx.login静默获取code → 后端换token
const loginRes = await wx.login();
const code = loginRes.code;
const result = await wx.request({
url: 'https://api.qxb.example.com/auth/miniapp/login',
method: 'POST',
data: { code }
});
if (result.data.needRegister) {
// 新用户，需要注册/绑定
wx.navigateTo({ url: '/pages/bindAccount/index' });
throw new Error('Need registration');
}
// 保存Token到小程序Storage
wx.setStorageSync('access_token', result.data.access_token);
wx.setStorageSync('refresh_token', result.data.refresh_token);
return result.data.account;
}
async syncProgress(progress: QXBLearningProgress): Promise<void> {
// 小程序端：节流到60秒，避免触发并发限制
this.throttledSync(progress, 60000);
}
}
// h5-sdk.ts - H5微应用端实现
class H5QXBSDK implements IQXBSDK {
async login(): Promise<QXBAccount> {
// H5端：检测宿主平台 → 获取免登授权码 → 后端换token
const platform = this.detectPlatform(); // dingtalk / wecom / feishu
let authCode: string;
if (platform === 'dingtalk') {
authCode = await dd.runtime.permission.requestAuthCode({
corpId: this.config.corpId
});
} else if (platform === 'wecom') {
// 企微的免登需要OAuth重定向
const url = new URL(window.location.href);
authCode = url.searchParams.get('code');
if (!authCode) {
// 重定向到企微授权页
window.location.href = this.buildWecomAuthUrl();
throw new Error('Redirecting to auth');
}
} else if (platform === 'feishu') {
// 飞书类似
const url = new URL(window.location.href);
authCode = url.searchParams.get('code');
if (!authCode) {
window.location.href = this.buildFeishuAuthUrl();
throw new Error('Redirecting to auth');
}
}
const result = await fetch('/api/auth/h5/login', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ platform, authCode })
}).then(r => r.json());
// Token存在sessionStorage中（关闭Tab即清除）
sessionStorage.setItem('access_token', result.access_token);
return result.account;
}
async syncProgress(progress: QXBLearningProgress): Promise<void> {
// H5端：页面关闭前用sendBeacon做最后同步
this.throttledSync(progress, 30000);
}
private setupUnloadSync() {
// 页面关闭前的最后同步
window.addEventListener('beforeunload', () => {
const currentProgress = this.getCurrentProgress();
if (currentProgress) {
navigator.sendBeacon(
'/api/progress/sync',
JSON.stringify(currentProgress)
);
}
});
// 页面可见性变化时也同步
document.addEventListener('visibilitychange', () => {
if (document.visibilityState === 'hidden') {
const currentProgress = this.getCurrentProgress();
if (currentProgress) {
navigator.sendBeacon(
'/api/progress/sync',
JSON.stringify(currentProgress)
);
}
}
});
}
}
六、效果数据与踩坑实录
6.1 上线效果
三端登录态统一方案上线6个月后的数据：
| 指标 | 上线前 | 上线后 |
|---|---|---|
| 跨端用户占比 | 12%（手动注册多账号） | 68%（自动关联） |
| 登录失败率 | 8.5% | 2.1% |
| 跨端进度丢失投诉 | 每月40+件 | 每月3件 |
| 身份绑定成功率 | - | 91%（自动绑定76% + 手动绑定15%） |
| Token刷新成功率 | - | 99.2% |
6.2 踩过的坑
坑1：微信openid和unionid的混淆。
早期版本用openid作为微信用户的唯一标识。但同一个小程序和同一个公众号（如果有的话）获取到的openid是不同的——只有unionid才是跨应用统一的。我们有一个客户同时运营着企学宝小程序和企学宝公众号，导致同一个用户在两个渠道的openid不同，被识别为两个账号。
修复方案：统一使用unionid作为微信体系的用户标识。但unionid需要用户关注过同主体的公众号或用过同主体的小程序才能获取，所以还需要openid作为兜底。最终方案是同时存储openid和unionid，优先用unionid匹配，匹配不上再用openid。
坑2：钉钉H5的免登code有效期只有5分钟。
钉钉的免登授权码有效期很短（5分钟），如果前端获取code后没有及时调用后端接口（比如网络慢、页面渲染延迟），code就过期了。用户看到的就是"登录失败"。
修复方案：前端获取code后立即调用后端接口，不做任何延迟。同时后端对code过期返回明确的错误码，前端收到后自动重新获取code并重试（用户无感知）。
async function loginWithRetry(maxRetries = 2): Promise<QXBAccount> {
for (let i = 0; i <= maxRetries; i++) {
try {
const authCode = await dd.runtime.permission.requestAuthCode({
corpId: config.corpId
});
return await api.login(authCode);
} catch (e) {
if (e.code === 'AUTH_CODE_EXPIRED' && i < maxRetries) {
console.warn('Auth code expired, retrying...');
continue;
}
throw e;
}
}
}
坑3：小程序Storage被微信清理导致频繁重新登录。
微信小程序的Storage不是永久可靠的——当手机存储空间不足时，微信可能主动清理小程序的Storage。用户再次打开小程序时，Token没了，需要重新登录。
修复方案：不完全依赖Storage。在wx.login的基础上，增加一层"服务端Session"——小程序启动时先尝试用本地的session_key向服务端换取新Token，如果session_key也失效了（说明微信已经回收了session），才走完整的wx.login流程。大多数情况下，session_key是有效的（只要微信没回收session），可以做到无感恢复。
坑4：企微H5的OAuth重定向在iOS和Android上行为不一致。
企业微信的OAuth授权重定向，在Android上能正常跳转回来，但在iOS上有时候会被企微内置浏览器的缓存拦截——用户看到的是上一次的授权页面，而不是最新的。
修复方案：在OAuth重定向URL上添加时间戳参数（&t={timestamp}），破坏缓存。同时在iOS上检测code参数是否存在，如果不存在则强制刷新页面。
坑5：学习进度同步的"惊群效应"。
有一次大促活动（全员必修的合规考试），几千人同时打开考试页面，每端都在做进度同步请求。服务端瞬间收到了数万并发请求，数据库连接池被打满，正常业务请求也受影响了。
修复方案：
- 进度同步请求和业务请求走不同的API路径，方便独立限流
- 进度同步使用异步队列——前端发请求后服务端立即返回"已接收"，实际写入异步处理
- 前端增加随机抖动（jitter），避免所有客户端在同一时刻发起同步
class ProgressSyncAPI:
"""异步进度同步接口"""
async def post(self, request):
# 1. 验证Token
payload = self.verify_token(request)
# 2. 快速校验数据格式
progress_data = self.validate_progress(request.body)
# 3. 放入异步队列，立即返回
await self.sync_queue.put({
"account_uid": payload.account_uid,
"course_id": progress_data["course_id"],
"progress": progress_data,
"device_type": payload.device_type,
"timestamp": datetime.utcnow().isoformat(),
})
# 4. 立即返回（不等待实际写入完成）
return {"status": "accepted", "server_time": datetime.utcnow().isoformat()}
# 异步消费者
async def progress_sync_consumer():
"""后台消费进度同步队列"""
while True:
batch = []
# 批量取出最多100条
for _ in range(100):
try:
item = await sync_queue.get_nowait()
batch.append(item)
except asyncio.QueueEmpty:
break
if batch:
# 批量写入数据库
await db.batch_upsert_progress(batch)
# 等待下一批（100ms间隔）
await asyncio.sleep(0.1)
坑6：飞书的tenant_key和corp_id不是同一个东西。
我们在设计corp_id时，假设所有OA平台都有一个"企业ID"的概念。但飞书的tenant_key和钉钉的corpId、企微的corpId在语义上不完全一样——飞书的tenant_key是应用维度的，不是企业维度的。同一个企业在飞书上安装多个应用，每个应用的tenant_key不同。
这导致同一个企业在飞书上的不同应用中登录，被识别为不同的corp_id，进而映射到不同的主账号。
修复方案：为飞书单独维护一个feishu_corp_mapping表，把tenant_key映射到统一的corp_id。映射关系在企业管理员首次配置飞书集成时建立。
class FeishuCorpMapping:
"""飞书tenant_key到统一corp_id的映射"""
async def resolve_corp_id(self, tenant_key: str) -> str:
"""将飞书tenant_key解析为统一的corp_id"""
mapping = await self.db.get_feishu_mapping(tenant_key)
if mapping:
return mapping.corp_id
# 未找到映射：尝试通过飞书API获取企业信息
tenant_info = await self.feishu_api.get_tenant_info(tenant_key)
if tenant_info:
# 用企业名+管理员信息匹配已有的企业
matched_corp = await self._match_existing_corp(
tenant_info["company_name"],
tenant_info.get("admin_email")
)
if matched_corp:
# 建立映射
await self.db.save_feishu_mapping(tenant_key, matched_corp)
return matched_corp
raise CorpMappingNotFoundError(
f"Cannot resolve feishu tenant_key: {tenant_key}"
)
七、总结
三端登录态同步的本质是三个问题：
“你是谁” — 统一身份层。三套外部身份（微信/OA平台/自有账号）映射到一个主账号。自动绑定（手机号/邮箱/工号匹配）解决80%的场景，手动绑定兜底剩余的20%。
“你的凭证有效吗” — Token管理。三端各自有不同的Token生命周期和刷新策略，但后端的Token结构和验证逻辑是统一的。各端Token互不影响，一端登出不影响其他端。
“你的数据跟得上吗” — 进度同步。采用"服务端优先 + 乐观合并"策略，各端按自己的节奏上报进度，服务端负责冲突检测和合并。离线学习是最复杂的场景，需要本地存储 + 恢复网络后批量同步。
如果只能记住一句话：多端同步的核心不是技术实现，而是冲突策略的设计。 技术实现（Token签发、身份绑定、进度上报）都是确定性的，但冲突策略（两端进度不一致时以谁为准）需要根据业务场景仔细定义。我们在"学习进度取最远"“测验成绩取最高”"学习时长取最大"这三条规则上花了远比代码实现更多的讨论时间——但这些讨论是值得的，因为它们直接决定了用户体验。
作者注：本文涉及的代码为简化版本，生产环境需要补充完整的异常处理、安全校验和监控埋点。各OA平台的API调用方式可能随版本更新变化，请以官方最新文档为准。
一站式 AI 云服务平台
更多推荐
9
0- 0
已为社区贡献1条内容
所有评论(0)