Merge branch 'main' of https://gitee.com/wg123/video-gen

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:

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
+                    from sqlalchemy import update
-                    oauth.access_token_expired = None
+                    
-                    oauth.refresh_token = None
+                    where_cond = UserOAuth.deleted_at.is_(None)
-                    oauth.refresh_token_expired = 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
+            from sqlalchemy import update
-            oauth.access_token_expired = expires_in
+            
-            oauth.refresh_token = new_refresh_token
+            where_cond = UserOAuth.deleted_at.is_(None)
-            oauth.refresh_token_expires_in = refresh_token_expires_in
+            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
-
+                
-                #检查是否为支持的平台（巨量引擎）
+                #检查access_token是否需要刷新
-                # port_type: 平台端口（1=巨量，2=磁力，3=巨量星图，4=服务单，5=腾讯）
+                need_refresh = False
-                if oauth.port_type in [1]:  
+                
-                    #刷新token
+                # access_token为空，需要刷新
-                    await refresh_juliang_token(oauth, app, db)
+                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.增加错误日志

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(