From 9275556db39a16196efd34dc951a45d0d81b542f Mon Sep 17 00:00:00 2001 From: 18610128193 <10574456+chenweiqiang-123@user.noreply.gitee.com> Date: Thu, 2 Jul 2026 11:38:53 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E6=94=B9=E6=8E=88=E6=9D=83=EF=BC=8C?= =?UTF-8?q?=E5=A6=82=E6=9E=9C=E6=98=AF=E5=90=8C=E4=B8=80=E4=B8=AA=E7=99=BB?= =?UTF-8?q?=E5=BD=95=E8=B4=A6=E5=8F=B7=EF=BC=8C=E4=BB=BB=E6=84=8F=E6=8E=88?= =?UTF-8?q?=E6=9D=83=E9=83=BD=E4=BC=9A=E6=9B=B4=E6=96=B0=E5=85=A8=E9=83=A8?= =?UTF-8?q?token?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../app/services/user_oauth_service.py | 59 +++++++- video-gen-api/app/tasks/token_refresh_task.py | 129 +++++++++++++----- video-gen-api/app/utils/douyinRequest.py | 85 +++++++++--- 3 files changed, 216 insertions(+), 57 deletions(-) diff --git a/video-gen-api/app/services/user_oauth_service.py b/video-gen-api/app/services/user_oauth_service.py index 9507c171..de8f7452 100644 --- a/video-gen-api/app/services/user_oauth_service.py +++ b/video-gen-api/app/services/user_oauth_service.py @@ -9,6 +9,7 @@ from app.config import settings from app.models.user_oauth import UserOAuth from app.models.user_oauth_app import UserOAuthApp from app.utils.id_gen import generate_id +from app.tasks.token_refresh_task import _update_redis_token #随机获取一个可用的应用配置 async def get_available_app(open_type: int, db: AsyncSession) -> dict: @@ -172,10 +173,13 @@ async def get_juliang_token(app_id: str, secret: str, code: str, open_type: int, ) oauth = oauth.scalar_one_or_none() if not oauth: + new_oauth_ids = [] for account in account_list: + oauth_id = generate_id() + new_oauth_ids.append(oauth_id) #新增授权记录 db.add(UserOAuth( - id=generate_id(), + id=oauth_id, account_id = str(account.get("account_id", "")), account_name = account.get("account_name", ""), account_role = account.get("account_role", ""), @@ -191,7 +195,11 @@ async def get_juliang_token(app_id: str, secret: str, code: str, open_type: int, refresh_token_expired = refresh_token_expires_in, material_auth_status = material_auth_status, )) - await db.commit() + await db.commit() + + + for oauth_id in new_oauth_ids: + await _update_redis_token(oauth_id, access_token, expires_in) else: # 查询现有授权记录（未删除的） existing_accounts = await db.execute( @@ -209,6 +217,9 @@ async def get_juliang_token(app_id: str, secret: str, code: str, open_type: int, new_account_ids = {str(account.get("account_id")) for account in account_list} old_account_ids = set(existing_accounts.keys()) + # 需要更新Redis的oauth_id列表 + update_redis_ids = [] + # 1. 软删除已消失的账户 for account_id in old_account_ids - new_account_ids: existing_accounts[account_id].deleted_at = datetime.now() @@ -226,10 +237,13 @@ async def get_juliang_token(app_id: str, secret: str, code: str, open_type: int, existing_oauth.refresh_token = refresh_token existing_oauth.refresh_token_expired = refresh_token_expires_in existing_oauth.material_auth_status = material_auth_status + update_redis_ids.append(existing_oauth.id) else: # 新增记录 + oauth_id = generate_id() + update_redis_ids.append(oauth_id) db.add(UserOAuth( - id=generate_id(), + id=oauth_id, account_id=str(account_id), account_name=account.get("account_name", ""), account_role=account.get("account_role", ""), @@ -246,8 +260,45 @@ async def get_juliang_token(app_id: str, secret: str, code: str, open_type: int, material_auth_status=material_auth_status, )) await db.commit() + + for oauth_id in update_redis_ids: + await _update_redis_token(oauth_id, access_token, expires_in) + + #5.本次更新成功以后，判断是否有其他同一个appid，同一个授权登录账号的授权记录,如果有，则更新token信息 + from sqlalchemy import update + + related_oauths = await db.execute( + select(UserOAuth).where( + UserOAuth.account_username == account_username, + UserOAuth.account_userid == account_userid, + UserOAuth.appid == app_id, + UserOAuth.user_id != user_id, + UserOAuth.deleted_at.is_(None), + ) + ) + related_oauths = related_oauths.scalars().all() + + if related_oauths: + await db.execute( + update(UserOAuth).where( + UserOAuth.account_username == account_username, + UserOAuth.account_userid == account_userid, + UserOAuth.appid == app_id, + UserOAuth.user_id != user_id, + UserOAuth.deleted_at.is_(None), + ).values( + access_token=access_token, + access_token_expired=expires_in, + refresh_token=refresh_token, + refresh_token_expired=refresh_token_expires_in, + material_auth_status=material_auth_status, + ) + ) + await db.commit() + + for related_oauth in related_oauths: + await _update_redis_token(related_oauth.id, access_token, expires_in) - #5.返回成功 return {"message": "授权成功"} async def get_kuaishou_token(app_id: str, secret: str, code: str, oauth_type: int, user_id: str, db: AsyncSession) -> dict: diff --git a/video-gen-api/app/tasks/token_refresh_task.py b/video-gen-api/app/tasks/token_refresh_task.py index effe9e54..11e86bc8 100644 --- a/video-gen-api/app/tasks/token_refresh_task.py +++ b/video-gen-api/app/tasks/token_refresh_task.py @@ -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.增加错误日志 diff --git a/video-gen-api/app/utils/douyinRequest.py b/video-gen-api/app/utils/douyinRequest.py index 6943f398..69121af6 100644 --- a/video-gen-api/app/utils/douyinRequest.py +++ b/video-gen-api/app/utils/douyinRequest.py @@ -89,7 +89,6 @@ class DouyinRequest: if token: return token - async with async_session() as db: oauth_data = await db.execute( select(UserOAuth).where( @@ -103,17 +102,30 @@ class DouyinRequest: raise ValueError("无效的oauth_id") if force_refresh: + where_cond = UserOAuth.deleted_at.is_(None) + if oauth_data.appid: + where_cond = where_cond & (UserOAuth.appid == oauth_data.appid) + if oauth_data.account_username: + where_cond = where_cond & (UserOAuth.account_username == oauth_data.account_username) + if oauth_data.account_userid: + where_cond = where_cond & (UserOAuth.account_userid == oauth_data.account_userid) + await db.execute( - update(UserOAuth).where(UserOAuth.id == oauth_id).values( + update(UserOAuth).where(where_cond).values( access_token=None, access_token_expired=None, ) ) await db.commit() - await self._delete_redis_token(oauth_id) + + 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 self._delete_redis_token(related_id) new_token, new_expired_at = await self.refresh_access_token(db, oauth_id, oauth_data.appid, oauth_data.refresh_token) - await self._set_redis_token(oauth_id, new_token, new_expired_at) return new_token if oauth_data.access_token_expired and oauth_data.access_token_expired > datetime.now(timezone.utc): @@ -122,14 +134,46 @@ class DouyinRequest: await self._set_redis_token(oauth_id, token, expired_at) return token + where_cond = UserOAuth.deleted_at.is_(None) + if oauth_data.appid: + where_cond = where_cond & (UserOAuth.appid == oauth_data.appid) + if oauth_data.account_username: + where_cond = where_cond & (UserOAuth.account_username == oauth_data.account_username) + if oauth_data.account_userid: + where_cond = where_cond & (UserOAuth.account_userid == oauth_data.account_userid) + + related_oauths = await db.execute( + select(UserOAuth.access_token, UserOAuth.access_token_expired).where( + where_cond, + UserOAuth.access_token_expired.is_not(None), + UserOAuth.access_token_expired > datetime.now(timezone.utc), + ).limit(1) + ) + related_oauth = related_oauths.first() + if related_oauth: + token, expired_at = related_oauth + await self._set_redis_token(oauth_id, token, expired_at) + return token + if oauth_data.refresh_token_expired and oauth_data.refresh_token_expired < datetime.now(timezone.utc): raise ValueError("授权已过期，请重新授权") new_token, new_expired_at = await self.refresh_access_token(db, oauth_id, oauth_data.appid, oauth_data.refresh_token) - await self._set_redis_token(oauth_id, new_token, new_expired_at) return new_token async def refresh_access_token(self, db: AsyncSession, oauth_id: str, appid: str, refresh_token: str) -> Tuple[str, datetime]: + oauth_info = await db.execute( + select(UserOAuth.account_username, UserOAuth.account_userid).where( + UserOAuth.id == oauth_id, + UserOAuth.deleted_at.is_(None), + ).limit(1) + ) + oauth_info = oauth_info.first() + if not oauth_info: + raise ValueError("无效的oauth_id") + + account_username, account_userid = oauth_info + result = await db.execute( select(UserOAuthApp.secret).where( UserOAuthApp.app_id == appid, @@ -169,9 +213,17 @@ class DouyinRequest: new_expired_at = datetime.now(timezone.utc) + timedelta(seconds=expires_in) + where_cond = UserOAuth.deleted_at.is_(None) + if appid: + where_cond = where_cond & (UserOAuth.appid == appid) + if account_username: + where_cond = where_cond & (UserOAuth.account_username == account_username) + if account_userid: + where_cond = where_cond & (UserOAuth.account_userid == account_userid) + await db.execute( update(UserOAuth).where( - UserOAuth.id == oauth_id, + where_cond, ).values( access_token=new_access_token, refresh_token=new_refresh_token, @@ -181,6 +233,14 @@ class DouyinRequest: ) await db.commit() + 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 self._set_redis_token(related_id, new_access_token, new_expired_at) + return new_access_token, new_expired_at # 有token请求 @@ -267,19 +327,6 @@ class DouyinRequest: else: raise ValueError('网络错误，稍后重试。') - # options_log = {} - # if options: - # for key, value in options.items(): - # if key == 'files': - # options_log[key] = {k: (v[0], 'bytes_content', v[2]) for k, v in value.items()} - # else: - # options_log[key] = value - # raise RuntimeError( - # f'DouYin API request failed after 5 retries. ' - # f'url:{url};oauthId:{oauth_id};options:{json.dumps(options_log)};response:{res}' - # ) - # if code != 0: - # raise ValueError(f'response:{res}') # 无token请求 async def request_with_context(