From 11f424a333c231265329433f764e64c0a8afb8e9 Mon Sep 17 00:00:00 2001 From: 18610128193 <10574456+chenweiqiang-123@user.noreply.gitee.com> Date: Thu, 18 Jun 2026 17:44:21 +0800 Subject: [PATCH] =?UTF-8?q?=E6=96=B0=E5=A2=9E=E6=9B=B4=E6=96=B0token,?= =?UTF-8?q?=E6=8F=90=E4=BA=A4=E7=B4=A0=E6=9D=90=EF=BC=8C=E5=89=8D=E6=B5=8B?= =?UTF-8?q?=E6=A8=A1=E6=9D=BF=E7=AE=A1=E7=90=86?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- video-gen-api/app/api/v1/pre_test_template.py | 22 +- video-gen-api/app/api/v1/upload_material.py | 139 ++++++++-- video-gen-api/app/api/v1/user_oauth.py | 3 +- video-gen-api/app/main.py | 5 + .../app/services/upload_material_service.py | 257 ++++++++++++++++-- video-gen-api/app/tasks/token_refresh_task.py | 194 +++++++++++++ video-gen-api/app/tasks/user_oauth_tasks.py | 73 ++++- video-gen-api/app/utils/douyinApi.py | 13 + video-gen-api/app/utils/douyinRequest.py | 4 +- 9 files changed, 648 insertions(+), 62 deletions(-) create mode 100644 video-gen-api/app/tasks/token_refresh_task.py diff --git a/video-gen-api/app/api/v1/pre_test_template.py b/video-gen-api/app/api/v1/pre_test_template.py index d19403bf..fefb5063 100644 --- a/video-gen-api/app/api/v1/pre_test_template.py +++ b/video-gen-api/app/api/v1/pre_test_template.py @@ -66,7 +66,7 @@ async def create_template( db: AsyncSession = Depends(get_db), ) -> Any: try: - #新增一个判断，如果模板名称重复，提示用户修改 + #新增一个判断,如果模板名称重复,提示用户修改 result = await db.execute( select(PreTestTemplate).where( PreTestTemplate.name == req.name, @@ -119,13 +119,13 @@ async def create_template( @router.get( "/list", summary="获取前测模板列表", - description="获取当前用户的前测模板列表，支持按平台筛选和分页", + description="获取当前用户的前测模板列表,支持按平台筛选和分页", response_model=PreTestTemplateListResponse, ) async def list_templates( platform: Optional[str] = Query(None, description="投放平台筛选（AD/QIANCHUAN/LOCAL）"), - page: int = Query(1, description="页码，默认1"), - page_size: int = Query(10, description="每页数量，默认10，最大100"), + page: int = Query(1, description="页码,默认1"), + page_size: int = Query(10, description="每页数量,默认10,最大100"), current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ) -> Any: @@ -300,20 +300,20 @@ async def delete_template( @router.get( "/getArea", summary="获取行政区域信息", - description="获取指定级别的行政区域信息，支持一级、二级、三级区域,如果需要更新地区，执行：/api/pre-test-template/getArea?oauth_id=0019ecab9b8bc57d964&advertiser_id=1836693172153543", + description="获取指定级别的行政区域信息,支持一级、二级、三级区域,如果需要更新地区,执行：/api/pre-test-template/getArea?oauth_id=0019ecab9b8bc57d964&advertiser_id=1836693172153543", ) async def get_template_area( - oauth_id: str = Query(None, description="授权ID选填，更新地区必填"), - advertiser_id: str = Query(default="1836693172153543", description="授权ID选填，更新地区必填"), - code: Optional[str] = Query("CN", description="行政区域编码，默认中国CN，选填"), - level: Optional[str] = Query("ONE_LEVEL", description="行政区域层级，可选值：ONE_LEVEL(获取省份)、TWO_LEVEL（市级）、THREE_LEVEL（区级）"), - parent_code: Optional[str] = Query(None, description="父级区域编码，获取二级时传一级编码，获取三级时传二级编码"), + oauth_id: str = Query(None, description="授权ID选填,更新地区必填"), + advertiser_id: str = Query(default="1836693172153543", description="授权ID选填,更新地区必填"), + code: Optional[str] = Query("CN", description="行政区域编码,默认中国CN,选填"), + level: Optional[str] = Query("ONE_LEVEL", description="行政区域层级,可选值：ONE_LEVEL(获取省份)、TWO_LEVEL（市级）、THREE_LEVEL（区级）"), + parent_code: Optional[str] = Query(None, description="父级区域编码,获取二级时传一级编码,获取三级时传二级编码"), ) -> Any: try: # 1. 先检查缓存是否存在 area_list = get_cached_area_data() - # 2. 如果缓存不存在，调用接口获取数据并保存到缓存 + # 2. 如果缓存不存在,调用接口获取数据并保存到缓存 if not area_list: area_list = await fetch_and_cache_area_data(oauth_id, advertiser_id, code) diff --git a/video-gen-api/app/api/v1/upload_material.py b/video-gen-api/app/api/v1/upload_material.py index d3580019..bc8e052f 100644 --- a/video-gen-api/app/api/v1/upload_material.py +++ b/video-gen-api/app/api/v1/upload_material.py @@ -1,12 +1,16 @@ import os -from typing import Any, Optional +from typing import Any, Optional, Dict from fastapi import APIRouter, Depends, HTTPException, status from pydantic import BaseModel, Field from sqlalchemy.ext.asyncio import AsyncSession from app.dependencies import get_current_user, get_db from app.models.user import User +from app.models.pre_test_template import PreTestTemplate from app.services.upload_material_service import upload_material_to_platform +from sqlalchemy import select +from app.utils.douyinApi import DouyinApi + router = APIRouter(prefix="/upload-material", tags=["上传素材"]) @@ -15,8 +19,8 @@ class UploadTask(BaseModel): advertiser_ids: list[str] = Field(..., description="广告主id数组，支持多条") resource_ids: list[str] = Field(..., description="资源id数组（generated_resources表主键）") oauth_id: str = Field(..., description="授权表id") - is_pre_test: Optional[str] = Field(None, description="是否开启前测：是/否/0/1，预留字段") - pre_test_template: Optional[str] = Field(None, description="前测模板id，预留字段") + is_pre_test: Optional[str] = Field(None, description="是否开启前测：1=是/2=否") + pre_test_template: Optional[str] = Field(None, description="前测模板id") class BatchUploadRequest(BaseModel): @@ -33,11 +37,35 @@ async def batch_upload_material( current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ) -> Any | dict: + # try: + # params = { + # "advertiser_id": 1836693172153543, + # "video_ids": ["tos-cn-i-sd07hgqsbj/2db9a53ee3ff4b6f8a2763b3bb463047"], + # "diagnose_config": {"platform": "AD", "external_action": "AD_APP_ACTIVATE"}, + # } + # response = await DouyinApi().pre_test_material("0019ecab9b8bc57d964", params) + # return response + # except Exception as e: + # import traceback + # return { + # "code": 500, + # "message": "请求失败", + # "error": str(e), + # "traceback": traceback.format_exc() + # } if not req.tasks: - raise HTTPException( - status_code=status.HTTP_400_BAD_REQUEST, - detail="上传任务列表不能为空", - ) + return { + "code": 0, + "message": "批量上传完成", + "summary": { + "total_tasks": 0, + "total_success_count": 0, + "total_fail_count": 0, + "total_requested": 0, + }, + "details": [], + "error": "上传任务列表不能为空" + } all_results = [] total_success = 0 @@ -56,9 +84,73 @@ async def batch_upload_material( try: if not task.advertiser_ids: - raise ValueError("广告主id数组不能为空") + task_result["result"] = { + "success": False, + "error": "广告主id数组不能为空", + "success_count": 0, + "fail_count": 0, + "total_count": 0, + "results": [], + } + all_results.append(task_result) + continue + if not task.resource_ids: - raise ValueError("资源id数组不能为空") + task_result["result"] = { + "success": False, + "error": "资源id数组不能为空", + "success_count": 0, + "fail_count": 0, + "total_count": 0, + "results": [], + } + all_results.append(task_result) + continue + + if (task.is_pre_test == "1") and (not task.pre_test_template): + task_result["result"] = { + "success": False, + "error": "开启前测功能时，必须指定前测模板id", + "success_count": 0, + "fail_count": len(task.resource_ids) * len(task.advertiser_ids), + "total_count": len(task.resource_ids) * len(task.advertiser_ids), + "results": [{ + "resource_id": rid, + "advertiser_id": aid, + "filename": "", + "success": False, + "error": "开启前测功能时，必须指定前测模板id" + } for rid in task.resource_ids for aid in task.advertiser_ids], + } + total_fail += len(task.resource_ids) * len(task.advertiser_ids) + all_results.append(task_result) + continue + + if task.is_pre_test == "1": + template = await db.execute( + select(PreTestTemplate).where(PreTestTemplate.id == task.pre_test_template). + where(PreTestTemplate.deleted_at.is_(None)). + where(PreTestTemplate.user_id == current_user.id) + ) + template = template.scalar_one_or_none() + if not template: + task_result["result"] = { + "success": False, + "error": "前测模板id不存在", + "success_count": 0, + "fail_count": len(task.resource_ids) * len(task.advertiser_ids), + "total_count": len(task.resource_ids) * len(task.advertiser_ids), + "results": [{ + "resource_id": rid, + "advertiser_id": aid, + "filename": "", + "success": False, + "error": "前测模板id不存在" + } for rid in task.resource_ids for aid in task.advertiser_ids], + } + total_fail += len(task.resource_ids) * len(task.advertiser_ids) + all_results.append(task_result) + continue result = await upload_material_to_platform( task.resource_ids, @@ -66,6 +158,7 @@ async def batch_upload_material( task.oauth_id, db, current_user.id, + task.pre_test_template if task.is_pre_test == "1" else None, ) task_result["result"] = { @@ -75,23 +168,37 @@ async def batch_upload_material( total_success += result["success_count"] total_fail += result["fail_count"] - if task.is_pre_test and task.is_pre_test in ["是", "1", "true", True]: - task_result["pre_test_reserved"] = { - "status": "reserved", - "message": "前测功能已预留，待后续开通", - "template": task.pre_test_template, - } - + except ValueError as e: task_result["result"] = { "success": False, "error": str(e), + "success_count": 0, + "fail_count": len(task.resource_ids) * len(task.advertiser_ids), + "total_count": len(task.resource_ids) * len(task.advertiser_ids), + "results": [{ + "resource_id": rid, + "advertiser_id": aid, + "filename": "", + "success": False, + "error": str(e) + } for rid in task.resource_ids for aid in task.advertiser_ids], } total_fail += len(task.resource_ids) * len(task.advertiser_ids) except Exception as e: task_result["result"] = { "success": False, "error": f"上传失败: {str(e)}", + "success_count": 0, + "fail_count": len(task.resource_ids) * len(task.advertiser_ids), + "total_count": len(task.resource_ids) * len(task.advertiser_ids), + "results": [{ + "resource_id": rid, + "advertiser_id": aid, + "filename": "", + "success": False, + "error": f"上传失败: {str(e)}" + } for rid in task.resource_ids for aid in task.advertiser_ids], } total_fail += len(task.resource_ids) * len(task.advertiser_ids) diff --git a/video-gen-api/app/api/v1/user_oauth.py b/video-gen-api/app/api/v1/user_oauth.py index 1bcca525..aaa4cd72 100644 --- a/video-gen-api/app/api/v1/user_oauth.py +++ b/video-gen-api/app/api/v1/user_oauth.py @@ -86,7 +86,7 @@ async def juliang_callback( await get_token(auth_code, user_id, app_id, db) return { - "message": "授权成功", + "message": "授权成功，这里需要跳转页面路径到 /user-oauth/oauth_list", "code": 0, } @@ -139,6 +139,7 @@ async def oauth_list( "account_name": oauth.account_name, "account_role": oauth.account_role, "account_username": oauth.account_username, + "account_userid": oauth.account_userid, "user_id": oauth.user_id, "open_type": oauth.open_type, "port_type": oauth.port_type, diff --git a/video-gen-api/app/main.py b/video-gen-api/app/main.py index 8cc52a03..7acc1321 100644 --- a/video-gen-api/app/main.py +++ b/video-gen-api/app/main.py @@ -59,6 +59,10 @@ async def lifespan(app: FastAPI): expiry_task = asyncio.create_task(_order_expiry_loop()) + # 启动token刷新定时任务（每5分钟检查一次，小于800秒有效期的token进行刷新） + from app.tasks.token_refresh_task import token_refresh_scheduler + token_refresh_task = asyncio.create_task(token_refresh_scheduler()) + # 启动时立即同步一次未支付订单 asyncio.create_task(asyncio.sleep(5)) # 等待5秒后再同步，让系统完全启动 async def startup_sync(): @@ -82,6 +86,7 @@ async def lifespan(app: FastAPI): task_queue.stop() await queue_task expiry_task.cancel() + token_refresh_task.cancel() await close_database() await close_redis() diff --git a/video-gen-api/app/services/upload_material_service.py b/video-gen-api/app/services/upload_material_service.py index df6a8a2d..8a264afa 100644 --- a/video-gen-api/app/services/upload_material_service.py +++ b/video-gen-api/app/services/upload_material_service.py @@ -1,12 +1,14 @@ import os import hashlib import base64 +import json from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.models.generated_resource import GeneratedResource from app.models.user_oauth import UserOAuth from app.models.resources_material import ResourcesMaterial +from app.models.pre_test_template import PreTestTemplate from app.utils.id_gen import generate_id from app.utils.douyinApi import DouyinApi @@ -20,6 +22,7 @@ async def upload_material_to_platform( oauth_id: str, db: AsyncSession, current_user_id: str, + pre_test_template_id: str | None = None, ) -> any: oauth = await db.execute( select(UserOAuth).where( @@ -30,13 +33,37 @@ async def upload_material_to_platform( ) oauth = oauth.scalar_one_or_none() if not oauth: - raise ValueError("授权记录不存在") + return { + "success_count": 0, + "fail_count": len(resource_ids) * len(advertiser_ids), + "total_count": len(resource_ids) * len(advertiser_ids), + "results": [{ + "resource_id": rid, + "advertiser_id": aid, + "filename": "", + "success": False, + "error": "授权记录不存在" + } for rid in resource_ids for aid in advertiser_ids], + "error": "授权记录不存在" + } port_type = oauth.port_type access_token = oauth.access_token if not access_token: - raise ValueError("授权token不存在") + return { + "success_count": 0, + "fail_count": len(resource_ids) * len(advertiser_ids), + "total_count": len(resource_ids) * len(advertiser_ids), + "results": [{ + "resource_id": rid, + "advertiser_id": aid, + "filename": "", + "success": False, + "error": "授权token不存在" + } for rid in resource_ids for aid in advertiser_ids], + "error": "授权token不存在" + } resources = await db.execute( select(GeneratedResource).where( @@ -48,26 +75,57 @@ async def upload_material_to_platform( resources = resources.scalars().all() resource_dict = {r.id: r for r in resources} - for resource_id in resource_ids: - if resource_id not in resource_dict: - raise ValueError(f"资源 {resource_id} 不存在或不属于当前用户") - + all_upload_results = [] success_count = 0 fail_count = 0 - + for resource_id in resource_ids: - resource = resource_dict[resource_id] - resource_type = resource.resource_type - storage_path = resource.storage_path + if resource_id not in resource_dict: + for advertiser_id in advertiser_ids: + all_upload_results.append({ + "resource_id": resource_id, + "advertiser_id": advertiser_id, + "filename": "", + "success": False, + "error": f"资源 {resource_id} 不存在或不属于当前用户" + }) + fail_count += 1 - if resource_type not in ["image", "video"]: - raise ValueError(f"资源 {resource_id} 不支持的资源类型，仅支持image和video") + for advertiser_id in advertiser_ids: + # 为每个广告主上传素材 + for resource_id in resource_ids: + if resource_id not in resource_dict: + continue + + resource = resource_dict[resource_id] + resource_type = resource.resource_type + storage_path = resource.storage_path - if not storage_path: - raise ValueError(f"资源 {resource_id} 本地存储路径为空") + if resource_type not in ["image", "video"]: + result = { + "resource_id": resource_id, + "advertiser_id": advertiser_id, + "filename": os.path.basename(storage_path) if storage_path else "", + "success": False, + "error": f"不支持的资源类型: {resource_type}，仅支持image和video", + } + all_upload_results.append(result) + fail_count += 1 + continue + + if not storage_path: + result = { + "resource_id": resource_id, + "advertiser_id": advertiser_id, + "filename": "", + "success": False, + "error": "资源本地存储路径为空", + } + all_upload_results.append(result) + fail_count += 1 + continue - for advertiser_id in advertiser_ids: result = await _upload_to_juliang( oauth_id, storage_path, resource_type, advertiser_id, resource, db, current_user_id ) @@ -76,9 +134,19 @@ async def upload_material_to_platform( success_count += 1 else: fail_count += 1 - - await db.commit() - + + # 上传完成后提交事务 + await db.commit() + + # 调用前测接口（仅当有模板ID且有视频上传成功时） + if pre_test_template_id: + success_video_ids = [ + r["upload_id"] for r in all_upload_results + if r.get("success") and r.get("upload_id") and r.get("advertiser_id") == advertiser_id + ] + if success_video_ids: + await _pre_test_material(oauth_id, advertiser_id, success_video_ids, pre_test_template_id, db) + return { "success_count": success_count, "fail_count": fail_count, @@ -87,6 +155,159 @@ async def upload_material_to_platform( } +#调用抖音前测接口 +async def _pre_test_material( + oauth_id: str, + advertiser_id: str, + video_ids: list[str], + pre_test_template_id: str, + db: AsyncSession, +) -> any: + #1.查询数据库是否存在前测模板 + pre_test_template = await db.execute( + select(PreTestTemplate).where( + PreTestTemplate.id == pre_test_template_id, + ) + ) + pre_test_template = pre_test_template.scalar_one_or_none() + if not pre_test_template: + note = f"前测模板 {pre_test_template_id} 不存在" + for video_id in video_ids: + await _update_material_pre_test_status( + db, oauth_id, advertiser_id, video_id, + task_id=None, + status="FAILED", + note=note, + pre_result=None, + ) + await db.commit() + return {"code": -1, "message": note, "data": {}} + + diagnose_config = {} + if pre_test_template.platform: + diagnose_config["platform"] = pre_test_template.platform + if pre_test_template.external_action: + diagnose_config["external_action"] = pre_test_template.external_action + if pre_test_template.cpa_bid: + diagnose_config["cpa_bid"] = pre_test_template.cpa_bid + if pre_test_template.audience_gender: + diagnose_config["audience_gender"] = pre_test_template.audience_gender + if pre_test_template.audience_age: + diagnose_config["audience_age"] = json.loads(pre_test_template.audience_age) + if pre_test_template.audience_region: + diagnose_config["audience_region"] = json.loads(pre_test_template.audience_region) + if pre_test_template.audience_network: + diagnose_config["audience_network"] = json.loads(pre_test_template.audience_network) + if pre_test_template.cus_name: + diagnose_config["cus_name"] = pre_test_template.cus_name + if pre_test_template.pricing_type: + diagnose_config["pricing_type"] = pre_test_template.pricing_type + if pre_test_template.cost_cap: + diagnose_config["cost_cap"] = pre_test_template.cost_cap + if pre_test_template.target_cost: + diagnose_config["target_cost"] = pre_test_template.target_cost + if pre_test_template.nobid: + diagnose_config["nobid"] = pre_test_template.nobid + if pre_test_template.cpc_bid: + diagnose_config["cpc_bid"] = pre_test_template.cpc_bid + if pre_test_template.budget: + diagnose_config["budget"] = pre_test_template.budget + + params = { + "advertiser_id": int(advertiser_id), + "video_ids": video_ids, + "diagnose_config": diagnose_config, + } + response = await douyin_api.pre_test_material(oauth_id, params) + + # 2. 解析接口返回并更新数据库 + code = response.get("code", -1) + + if code != 0: + # 整体失败：更新所有视频状态为FAILED + note = response.get("message", "未知错误") + for video_id in video_ids: + await _update_material_pre_test_status( + db, oauth_id, advertiser_id, video_id, + task_id=None, + status="FAILED", + note=note, + pre_result=None, + ) + await db.commit() + return response + + # 成功：解析 data 中的任务信息 + data = response.get("data", {}) + task_ids = data.get("task_ids", []) + fail_video_ids = data.get("fail_video_ids", {}) + + # 3. 更新成功的视频状态为PENDING（等待审核） + success_count = 0 + for i, video_id in enumerate(video_ids): + if video_id in fail_video_ids: + # 失败的视频 + fail_info = fail_video_ids[video_id] + err_code = fail_info.get("err_code", "") + err_message = fail_info.get("err_message", "未知错误") + note = f"失败[{err_code}]: {err_message}" + + await _update_material_pre_test_status( + db, oauth_id, advertiser_id, video_id, + task_id=None, + status="FAILED", + note=note, + pre_result=None, + ) + else: + # 成功的视频 + task_id = str(task_ids[success_count]) if success_count < len(task_ids) else None + + await _update_material_pre_test_status( + db, oauth_id, advertiser_id, video_id, + task_id=task_id, + status="PENDING", + note="", + pre_result=None, + ) + success_count += 1 + + # 提交事务，保存所有修改 + await db.commit() + + return response + + +async def _update_material_pre_test_status( + db: AsyncSession, + oauth_id: str, + advertiser_id: str, + upload_id: str, + task_id: str | None, + status: str, + note: str | None, + pre_result: str | None, +): + """更新素材的前测状态""" + result = await db.execute( + select(ResourcesMaterial).where( + ResourcesMaterial.oauth_id == oauth_id, + ResourcesMaterial.advertiser_id == advertiser_id, + ResourcesMaterial.upload_id == upload_id, + ResourcesMaterial.deleted_at.is_(None), + ) + ) + material = result.scalar_one_or_none() + + if material: + material.task_id = task_id + material.status = status + material.note = note + material.pre_result = pre_result + + + + async def _upload_to_juliang( oauth_id: str, storage_path: str, diff --git a/video-gen-api/app/tasks/token_refresh_task.py b/video-gen-api/app/tasks/token_refresh_task.py new file mode 100644 index 00000000..3ea9d4fe --- /dev/null +++ b/video-gen-api/app/tasks/token_refresh_task.py @@ -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()) \ No newline at end of file diff --git a/video-gen-api/app/tasks/user_oauth_tasks.py b/video-gen-api/app/tasks/user_oauth_tasks.py index 74f3c8b5..d43a19a3 100644 --- a/video-gen-api/app/tasks/user_oauth_tasks.py +++ b/video-gen-api/app/tasks/user_oauth_tasks.py @@ -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() \ No newline at end of file + 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() \ No newline at end of file diff --git a/video-gen-api/app/utils/douyinApi.py b/video-gen-api/app/utils/douyinApi.py index 32960f84..e7bbbaca 100644 --- a/video-gen-api/app/utils/douyinApi.py +++ b/video-gen-api/app/utils/douyinApi.py @@ -37,6 +37,7 @@ class DouyinApi: options ) + #上传视频素材 async def upload_video_material(self, oauth_id: str, data: Optional[Dict[str, Any]] = None, files: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: if not oauth_id: raise RuntimeError('OAuth ID is not set.') @@ -65,4 +66,16 @@ class DouyinApi: url, 'GET', {'params': params or {}} + ) + + #前测素材,Adv创建前测任务 + async def pre_test_material(self, oauth_id: str, params: any) -> Dict[str, Any]: + if not oauth_id: + raise RuntimeError('OAuth ID is not set.') + url = "https://api.oceanengine.com/open_api/2/diagnosis_task/adv/create/" + return await self.request.request_with_token_with_context( + oauth_id, + url, + 'POST', + {"json": params or {}} ) \ No newline at end of file diff --git a/video-gen-api/app/utils/douyinRequest.py b/video-gen-api/app/utils/douyinRequest.py index da45bc5d..5a066b5a 100644 --- a/video-gen-api/app/utils/douyinRequest.py +++ b/video-gen-api/app/utils/douyinRequest.py @@ -186,7 +186,7 @@ class DouyinRequest: oauth_id: str, url: str, method: str = 'GET', - options: Optional[Dict[str, Any]] = None, + options: any = None, ) -> Any: options = options or {} token = await self.get_access_token(oauth_id) @@ -229,7 +229,7 @@ class DouyinRequest: await asyncio.sleep(wait_time) continue - if code >= 50000: + if code == 50000: await asyncio.sleep(i * 10) continue