mz/video-gen
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

修改授权，如果是同一个登录账号，任意授权都会更新全部token

Browse Source
This commit is contained in:
18610128193
2026-07-02 11:38:53 +08:00
parent 0061068ac8
commit 9275556db3
3 changed files with 216 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files
video-gen-api/app/tasks/token_refresh_task.py
+95 -34
View File
@@ -60,12 +60,33 @@ async def refresh_juliang_token(oauth: UserOAuth, app: UserOAuthApp, db: AsyncSe
#如果code=40103或者40107，传入refresh_token已失效，失效原因一般是由于refresh_token已被使用，或授权账号重新授权并生成了新的Token
if data.get("code") in [40103, 40107]:
#清空数据库中的token信息，和Redis缓存中的token
oauth.access_token = None
oauth.access_token_expired = None
oauth.refresh_token = None
oauth.refresh_token_expired = None
from sqlalchemy import update
where_cond = UserOAuth.deleted_at.is_(None)
if oauth.appid:
where_cond = where_cond & (UserOAuth.appid == oauth.appid)
if oauth.account_username:
where_cond = where_cond & (UserOAuth.account_username == oauth.account_username)
if oauth.account_userid:
where_cond = where_cond & (UserOAuth.account_userid == oauth.account_userid)
await db.execute(
update(UserOAuth).where(where_cond).values(
access_token=None,
access_token_expired=None,
refresh_token=None,
refresh_token_expired=None,
)
)
await db.commit()
await _update_redis_token(oauth.id, "", None)
related_oauth_ids = await db.execute(
select(UserOAuth.id).where(where_cond)
)
related_oauth_ids = [row[0] for row in related_oauth_ids.all()]
for related_id in related_oauth_ids:
await _update_redis_token(related_id, "", None)
return
@@ -75,15 +96,34 @@ async def refresh_juliang_token(oauth: UserOAuth, app: UserOAuthApp, db: AsyncSe
expires_in = datetime.now(tz=oauth.access_token_expired.tzinfo) + timedelta(seconds=data.get("expires_in", 0))
refresh_token_expires_in = datetime.now(tz=oauth.refresh_token_expired.tzinfo) + timedelta(seconds=data.get("refresh_token_expires_in", 0))
oauth.access_token = new_access_token
oauth.access_token_expired = expires_in
oauth.refresh_token = new_refresh_token
oauth.refresh_token_expires_in = refresh_token_expires_in
from sqlalchemy import update
where_cond = UserOAuth.deleted_at.is_(None)
if oauth.appid:
where_cond = where_cond & (UserOAuth.appid == oauth.appid)
if oauth.account_username:
where_cond = where_cond & (UserOAuth.account_username == oauth.account_username)
if oauth.account_userid:
where_cond = where_cond & (UserOAuth.account_userid == oauth.account_userid)
await db.execute(
update(UserOAuth).where(where_cond).values(
access_token=new_access_token,
access_token_expired=expires_in,
refresh_token=new_refresh_token,
refresh_token_expired=refresh_token_expires_in,
)
)
await db.commit()
await _update_redis_token(oauth.id, new_access_token, expires_in)
related_oauth_ids = await db.execute(
select(UserOAuth.id).where(where_cond)
)
related_oauth_ids = [row[0] for row in related_oauth_ids.all()]
for related_id in related_oauth_ids:
await _update_redis_token(related_id, new_access_token, expires_in)
logger.info(f"成功刷新巨量引擎token: oauth_id={oauth.id}, account_id={oauth.account_id}")
logger.info(f"成功刷新巨量引擎token: oauth_id={oauth.id}, account_id={oauth.account_id}, 关联账户数={len(related_oauth_ids)}")
except httpx.HTTPError as e:
logger.error(f"HTTP请求失败: oauth_id={oauth.id}, 错误: {str(e)}")
except Exception as e:
@@ -97,37 +137,39 @@ async def check_and_refresh_tokens():
query = select(UserOAuth).where(
UserOAuth.deleted_at.is_(None),
UserOAuth.access_token.is_not(None),
UserOAuth.access_token_expired.is_not(None),
UserOAuth.refresh_token.is_not(None),
UserOAuth.refresh_token_expired.is_not(None),
UserOAuth.refresh_token_expired > now,
)
result = await db.execute(query)
oauth_list = result.scalars().all()
refreshed_keys = set()
for oauth in oauth_list:
try:
#1.检查access_token是否过期，如果未过期，并且大于800秒，直接跳过不处理
if not oauth.access_token_expired:
#检查是否为支持的平台（巨量引擎）
# port_type: 平台端口（1=巨量，2=磁力，3=巨量星图，4=服务单，5=腾讯）
if oauth.port_type not in [1]:
continue
remaining_seconds = (oauth.access_token_expired - now).total_seconds()
#构建登录账号唯一标识，同一登录账号共享token
key_parts = []
if oauth.appid:
key_parts.append(oauth.appid)
if oauth.account_username:
key_parts.append(oauth.account_username)
if oauth.account_userid:
key_parts.append(oauth.account_userid)
login_key = "|".join(key_parts)
# access_token剩余时间大于等于800秒，不需要刷新
if remaining_seconds >= REFRESH_THRESHOLD_SECONDS:
#同一登录账号已刷新过，直接跳过（避免使用旧数据判断）
if login_key in refreshed_keys:
logger.debug(f"跳过重复刷新: oauth_id={oauth.id}, 同一登录账号已刷新")
continue
#2.如果access_token过期，或者剩余时间小于800秒，需要刷新token
#3.如果需要刷新token，检查refresh_token是否过期，如果refresh_token过期，说明不可刷新，需要直接重新授权，直接跳过不处理
if not oauth.refresh_token_expired:
continue
refresh_remaining_seconds = (oauth.refresh_token_expired - now).total_seconds()
if refresh_remaining_seconds <= 0:
continue
#5.获取应用配置
#获取应用配置
app_result = await db.execute(
select(UserOAuthApp).where(UserOAuthApp.app_id == oauth.appid)
)
@@ -135,12 +177,31 @@ async def check_and_refresh_tokens():
if not app:
continue
#检查是否为支持的平台（巨量引擎）
# port_type: 平台端口（1=巨量，2=磁力，3=巨量星图，4=服务单，5=腾讯）
if oauth.port_type in [1]:
#刷新token
await refresh_juliang_token(oauth, app, db)
#检查access_token是否需要刷新
need_refresh = False
# access_token为空，需要刷新
if not oauth.access_token:
need_refresh = True
# access_token_expired为空，需要刷新
elif not oauth.access_token_expired:
need_refresh = True
# access_token即将过期（剩余时间小于800秒），需要刷新
else:
remaining_seconds = (oauth.access_token_expired - now).total_seconds()
if remaining_seconds < REFRESH_THRESHOLD_SECONDS:
need_refresh = True
if not need_refresh:
continue
#refresh_token已在查询条件中过滤，确保有效才能刷新
#刷新token
await refresh_juliang_token(oauth, app, db)
refreshed_keys.add(login_key)
except Exception as e:
#7.增加错误日志