新增更新token,提交素材，前测模板管理

2026-06-18 17:44:21 +08:00
parent c8ad49c736
commit 11f424a333
9 changed files with 648 additions and 62 deletions
@@ -0,0 +1,194 @@
+from datetime import datetime, timedelta, timezone
+import asyncio
+import httpx
+import json
+import os
+import logging
+from logging.handlers import TimedRotatingFileHandler
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.models.user_oauth import UserOAuth
+from app.models.user_oauth_app import UserOAuthApp
+from app.models.base import async_session
+from app.config import settings
+from app.utils.redis import get_redis
+
+REDIS_KEY = "douyin:tokens"
+
+LOG_DIR = os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(__file__))), "logs")
+
+os.makedirs(LOG_DIR, exist_ok=True)
+
+logger = logging.getLogger("token_refresh")
+logger.setLevel(logging.INFO)
+
+def get_log_filename():
+    return os.path.join(LOG_DIR, f"token_refresh-{datetime.now().strftime('%Y-%m-%d')}.log")
+
+class DailyRotatingFileHandler(logging.FileHandler):
+    def __init__(self, directory, encoding=None):
+        self.directory = directory
+        filename = get_log_filename()
+        super().__init__(filename, encoding=encoding)
+    
+    def emit(self, record):
+        current_filename = get_log_filename()
+        if self.baseFilename != current_filename:
+            self.close()
+            self.baseFilename = current_filename
+            self.stream = self._open()
+        super().emit(record)
+
+handler = DailyRotatingFileHandler(LOG_DIR, encoding="utf-8")
+handler.setFormatter(logging.Formatter("%(asctime)s - %(levelname)s - %(message)s", "%Y-%m-%d %H:%M:%S"))
+
+logger.addHandler(handler)
+
+
+
+REFRESH_THRESHOLD_SECONDS = 800
+CHECK_INTERVAL_MINUTES = 5
+
+
+async def _update_redis_token(oauth_id: str, token: str, expired_at: datetime):
+    """更新Redis缓存中的token"""
+    redis = get_redis()
+    if not redis:
+        logger.warning("Redis连接未配置，跳过缓存更新")
+        return
+
+    try:
+        cache = {
+            "token": token,
+            "expired_at": expired_at.isoformat(),
+        }
+        await redis.hset(REDIS_KEY, oauth_id, json.dumps(cache))
+        logger.info(f"Redis缓存已更新: oauth_id={oauth_id}")
+    except Exception as e:
+        logger.error(f"更新Redis缓存失败: {str(e)}")
+
+
+async def refresh_juliang_token(oauth: UserOAuth, app: UserOAuthApp, db: AsyncSession):
+    """刷新巨量引擎token"""
+    try:
+        async with httpx.AsyncClient() as client:
+            url = "https://api.oceanengine.com/open_api/oauth2/refresh_token/"
+            response = await client.post(
+                url,
+                json={
+                    "app_id": app.app_id,
+                    "secret": app.secret,
+                    "refresh_token": oauth.refresh_token,
+                },
+            )
+            response.raise_for_status()
+            data = response.json()
+            if data.get("code") != 0:
+                logger.error(f"刷新巨量引擎token失败: oauth_id={oauth.id}, 错误信息: {data}")
+                #如果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
+                    await db.commit()
+                    await _update_redis_token(oauth.id, "", None)    
+                return
+
+            
+            data = data.get("data", {})
+            new_access_token = data.get("access_token", "")
+            new_refresh_token = data.get("refresh_token", "")
+            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
+            await db.commit()
+            
+            await _update_redis_token(oauth.id, new_access_token, expires_in)
+            
+            logger.info(f"成功刷新巨量引擎token: oauth_id={oauth.id}, account_id={oauth.account_id}")
+    except httpx.HTTPError as e:
+        logger.error(f"HTTP请求失败: oauth_id={oauth.id}, 错误: {str(e)}")
+    except Exception as e:
+        logger.error(f"刷新巨量引擎token发生异常: oauth_id={oauth.id}, 错误: {str(e)}")
+
+
+async def check_and_refresh_tokens():
+    """检查并刷新即将过期的token"""
+    async with async_session() as db:
+        now = datetime.now(timezone.utc)
+        
+        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),
+        )
+        
+        result = await db.execute(query)
+        oauth_list = result.scalars().all()
+        
+        for oauth in oauth_list:
+            try:
+                #1.检查access_token是否过期，如果未过期，并且大于800秒，直接跳过不处理
+                if not oauth.access_token_expired:
+                    continue
+                
+                remaining_seconds = (oauth.access_token_expired - now).total_seconds()
+                
+                # access_token剩余时间大于等于800秒，不需要刷新
+                if remaining_seconds >= REFRESH_THRESHOLD_SECONDS:
+                    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)
+                )
+                app = app_result.scalar_one_or_none()
+                
+                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)
+                
+            except Exception as e:
+                #7.增加错误日志
+                logger.error(f"刷新token失败: {str(e)}")
+
+
+async def token_refresh_scheduler():
+    """定时任务调度器"""
+    while True:
+        try:
+            await check_and_refresh_tokens()
+        except Exception as e:
+            logger.error(f"定时任务token_refresh_scheduler执行失败: {str(e)}")
+        
+        await asyncio.sleep(CHECK_INTERVAL_MINUTES * 60)
+
+
+def start_token_refresh_task():
+    """启动token刷新定时任务"""
+    logger.info("启动token刷新定时任务")
+    asyncio.create_task(token_refresh_scheduler())
@@ -1,5 +1,3 @@
-from asyncio import Condition
-
 import httpx
 from datetime import datetime, timedelta

@@ -7,9 +5,13 @@ from sqlalchemy import select

 from app.models.base import async_session
 from app.models.user_oauth import UserOAuth
+from app.models.user_oauth_account import UserOAuthAccount
 from app.models.user_oauth_app import UserOAuthApp
 from app.tasks.async_runner import run_async
 from app.tasks.celery_app import celery_app
+from app.utils.douyinApi import DouyinApi
+from app.utils.douyinRequest import DouyinRequest
+from app.utils.id_gen import generate_id


 async def _update_oauth_accounts(account_id: str, account_userid: str, current_user_id: str, db: async_session):
@@ -27,18 +29,61 @@ async def _update_oauth_accounts(account_id: str, account_userid: str, current_u
    if not oauth_records:
        return True
    for oauth in oauth_records:
-        pass
+        if oauth.port_type == 1:
+            return await update_juliang(oauth, db)

-if celery_app:
-    @celery_app.task(name="user_oauth.update_oauth_accounts", bind=True, max_retries=3, default_retry_delay=60)
-    def update_oauth_accounts(self, account_id: str, account_userid: str, current_user_id: str, db):
-        return run_async(_update_oauth_accounts(account_id, account_userid, current_user_id, db))
-else:
-    class _DisabledTask:
-        def delay(self, *args, **kwargs):
-            pass
+async def update_juliang(oauth: UserOAuth, db: async_session):
+    if oauth.account_role == 'AGENT':
+        #通过代理商获取账户列表
+        cursor : int = 0
+        count : int = 10
+        while True:
+            params = {
+                'advertiser_id': oauth.account_id,
+                'count': count,
+            }
+            if cursor:
+                params['cursor'] = cursor
+            response = await DouyinApi().get_advertiser_by_agent(oauth.id, params)
+            if response.get('code', 0) != 0:
+                #记录错误日志
+                break

-        def apply_async(self, *args, **kwargs):
-            pass
+            data = response['data']['list'] or []
+            if not data:
+                break
+            account_source = response['data']['account_source'] or ''

-    update_oauth_accounts = _DisabledTask()
+            account_list = []
+            for item in data:
+                account_list.append(UserOAuthAccount(
+                    id=generate_id(),
+                    oauth_id=oauth.id,
+                    advertiser_id=str(item),
+                    advertiser_name="",
+                    advertiser_role=account_source,
+                ))
+
+            if account_list:
+                db.add_all(account_list)
+                await db.commit()
+
+            cursor = response['data'].get('cursor_page_info', {}).get('cursor')
+            has_more = response['data'].get('cursor_page_info', {}).get('has_more', False)
+            if not has_more:
+                break
+        
+
+# if celery_app:
+#     @celery_app.task(name="user_oauth.update_oauth_accounts", bind=True, max_retries=3, default_retry_delay=60)
+#     def update_oauth_accounts(self, account_id: str, account_userid: str, current_user_id: str, db):
+#         return run_async(_update_oauth_accounts(account_id, account_userid, current_user_id, db))
+# else:
+#     class _DisabledTask:
+#         def delay(self, *args, **kwargs):
+#             pass
+
+#         def apply_async(self, *args, **kwargs):
+#             pass
+
+#     update_oauth_accounts = _DisabledTask()