diff --git a/video-gen-api/app/api/v1/__init__.py b/video-gen-api/app/api/v1/__init__.py index 72863a45..96dcf50e 100644 --- a/video-gen-api/app/api/v1/__init__.py +++ b/video-gen-api/app/api/v1/__init__.py @@ -30,6 +30,7 @@ from app.api.v1.resources_material import router as resources_material_router from app.api.v1.contact import router as contact_router from app.api.v1.home_materials import router as home_materials_router from app.api.admin import router as admin_module_router +from app.api.v1.material_admin import router as material_admin_router api_router = APIRouter() api_router.include_router(auth_router) @@ -62,3 +63,4 @@ api_router.include_router(resources_material_router) api_router.include_router(contact_router) api_router.include_router(home_materials_router) api_router.include_router(admin_module_router) +api_router.include_router(material_admin_router) \ No newline at end of file diff --git a/video-gen-api/app/api/v1/material_admin.py b/video-gen-api/app/api/v1/material_admin.py new file mode 100644 index 00000000..a6d9eda6 --- /dev/null +++ b/video-gen-api/app/api/v1/material_admin.py @@ -0,0 +1,168 @@ +from typing import Any, Optional, List +from datetime import datetime, timedelta, timezone + +from fastapi import APIRouter, Query, Depends, Body +from sqlalchemy import select, func +from sqlalchemy.ext.asyncio import AsyncSession + +from app.models.user import User +from app.models.resources_material import ResourcesMaterial +from app.models.material_cost import MaterialCost +from app.models.user_oauth import UserOAuth +from app.models.user_oauth_account import UserOAuthAccount +from app.models.pre_test_template import PreTestTemplate +from app.models.upload_task import UploadTask +from app.dependencies import get_current_user, get_db, get_admin_user +from app.services.material_consumption_queue import sync_all_advertisers_consumption, _fetch_and_save_consumption +from app.services.material_consumption_service import get_consumption_list, format_consumption_response +from app.services.material_admin_service import get_oauth_list, get_material_list, get_pre_test_template_list, get_upload_task_list + +router = APIRouter(prefix="/material-admin", tags=["material-admin"]) + +@router.get("/oauth-list", summary="管理员查看所有授权列表") +async def admin_get_oauth_list( + id: Optional[str] = Query(None, description="主键id"), + phone: Optional[str] = Query(None, description="用户登录手机号"), + account_id: Optional[str] = Query(None, description="授权账户id"), + account_userid: Optional[str] = Query(None, description="授权登录账号id"), + created_at: Optional[List[datetime]] = Query(None, description="授权时间"), + appid: Optional[str] = Query(None, description="应用id"), + open_type: Optional[int] = Query(None, description="开户方式"), + page: int = Query(1, ge=1, description="页码"), + page_size: int = Query(10, ge=1, le=3000, description="每页数量"), + db: AsyncSession = Depends(get_db), + admin_user: User = Depends(get_admin_user), +) -> Any | dict: + try: + oauth_list, total = await get_oauth_list( + id=id, + db=db, + phone=phone, + account_id=account_id, + account_userid=account_userid, + created_at=created_at, + appid=appid, + open_type=open_type, + page=page, + page_size=page_size, + ) + + return { + "code": 0, + "data": oauth_list, + "pagination": { + "page": page, + "page_size": page_size, + "total": total, + }, + } + except Exception as e: + return { + "code": 1, + "msg": str(e), + "data": None, + } + +@router.get("/material-list", summary="管理员查看所有素材资源列表") +async def admin_get_material_list( + phone: Optional[str] = Query(None, description="用户登录手机号"), + id: Optional[str] = Query(None, description="主键id"), + resource_type: Optional[str] = Query(None, description="素材资源类型"), + advertiser_id: Optional[str] = Query(None, description="广告主id"), + material_id: Optional[str] = Query(None, description="素材id"), + upload_id: Optional[str] = Query(None, description="平台id【视频id/图片id】"), + created_at: Optional[List[datetime]] = Query(None, description="创建时间"), + status: Optional[str] = Query(None, description="前测状态"), + page: int = Query(1, ge=1, description="页码"), + page_size: int = Query(10, ge=1, le=3000, description="每页数量"), + db: AsyncSession = Depends(get_db), + admin_user: User = Depends(get_admin_user), +) -> Any | dict: + materials, total = await get_material_list( + db=db, + phone=phone, + id=id, + resource_type=resource_type, + advertiser_id=advertiser_id, + material_id=material_id, + upload_id=upload_id, + created_at=created_at, + status=status, + page=page, + page_size=page_size, + ) + + return { + "code": 0, + "data": materials, + "pagination": { + "page": page, + "page_size": page_size, + "total": total, + }, + } + + +@router.get("/pre-test-template-list", summary="管理员查看所有前测模板列表") +async def admin_get_pre_test_template_list( + phone: Optional[str] = Query(None, description="用户登录手机号"), + id: Optional[str] = Query(None, description="主键id"), + created_at: Optional[datetime] = Query(None, description="创建时间"), + page: int = Query(1, ge=1, description="页码"), + page_size: int = Query(10, ge=1, le=3000, description="每页数量"), + db: AsyncSession = Depends(get_db), + admin_user: User = Depends(get_admin_user), +) -> Any | dict: + templates, total = await get_pre_test_template_list( + db=db, + phone=phone, + id=id, + created_at=created_at, + page=page, + page_size=page_size, + ) + + return { + "code": 0, + "data": templates, + "pagination": { + "page": page, + "page_size": page_size, + "total": total, + }, + } + +@router.get("/upload-task-list", summary="管理员查看推送任务列表") +async def admin_get_upload_task_list( + phone: Optional[str] = Query(None, description="用户登录手机号"), + id: Optional[str] = Query(None, description="主键id"), + created_at: Optional[List[str]] = Query(None, description="创建时间范围，示例: ['2023-01-01', '2023-03-01']"), + advertiser_id: Optional[str] = Query(None, description="广告主id"), + status: Optional[int] = Query(None, description="前测状态"), + resource_id: Optional[str] = Query(None, description="资源id"), + page: int = Query(1, ge=1, description="页码"), + page_size: int = Query(10, ge=1, le=3000, description="每页数量"), + db: AsyncSession = Depends(get_db), + admin_user: User = Depends(get_admin_user), +) -> Any | dict: + tasks, total = await get_upload_task_list( + db=db, + phone=phone, + id=id, + created_at=created_at, + advertiser_id=advertiser_id, + status=status, + resource_id=resource_id, + page=page, + page_size=page_size, + ) + + return { + "code": 0, + "data": tasks, + "pagination": { + "page": page, + "page_size": page_size, + "total": total, + }, + } \ No newline at end of file diff --git a/video-gen-api/app/services/material_admin_service.py b/video-gen-api/app/services/material_admin_service.py new file mode 100644 index 00000000..3de246ab --- /dev/null +++ b/video-gen-api/app/services/material_admin_service.py @@ -0,0 +1,379 @@ +from typing import Optional, List, Tuple +from datetime import datetime + +from sqlalchemy import select, func +from sqlalchemy.ext.asyncio import AsyncSession + +from app.models.user import User +from app.models.user_oauth import UserOAuth +from app.models.user_oauth_account import UserOAuthAccount +from app.models.resources_material import ResourcesMaterial +from app.models.pre_test_template import PreTestTemplate +from app.models.upload_task import UploadTask +from app.utils.datetime_util import BEIJING_TZ, datetime_to_db_tz_str, db_tz_str_to_datetime, parse_date_range + + +async def get_oauth_list( + id: Optional[str] = None, + db: AsyncSession = None, + phone: Optional[str] = None, + account_id: Optional[str] = None, + account_userid: Optional[str] = None, + created_at: Optional[datetime] = None, + appid: Optional[str] = None, + open_type: Optional[int] = None, + page: int = 1, + page_size: int = 10, +) -> Tuple[List[UserOAuth], int]: + offset = (page - 1) * page_size + + query = select(UserOAuth).where(UserOAuth.deleted_at.is_(None)) + if id: + query = query.where(UserOAuth.id == id) + if phone: + query = query.join(User, User.id == UserOAuth.user_id).where(User.phone == phone) + if account_id: + query = query.where(UserOAuth.account_id == account_id) + if account_userid: + query = query.where(UserOAuth.account_userid == account_userid) + # 【改动3】适配入参 created_at=["2025-01-01","2026-01-01"] 字符串日期场景 + if created_at and len(created_at) == 2: + try: + start_dt, end_dt = parse_date_range(created_at, is_timezone=True) + # 直接用datetime范围匹配，不走func.date，保留时分秒精度，命中索引 + query = query.where( + UserOAuth.created_at >= start_dt, + UserOAuth.created_at <= end_dt + ) + except ValueError: + # 捕获日期格式错误，非法时间直接不附加该查询条件 + pass + if appid: + query = query.where(UserOAuth.appid == appid) + if open_type: + query = query.where(UserOAuth.open_type == open_type) + + count_query = select(func.count(UserOAuth.id)).where(UserOAuth.deleted_at.is_(None)) + + if phone: + count_query = count_query.join(User, User.id == UserOAuth.user_id).where(User.phone == phone) + if account_id: + count_query = count_query.where(UserOAuth.account_id == account_id) + if account_userid: + count_query = count_query.where(UserOAuth.account_userid == account_userid) + if created_at: + try: + start_dt, end_dt = parse_date_range(created_at, is_timezone=True) + # 直接用datetime范围匹配，不走func.date，保留时分秒精度，命中索引 + count_query = count_query.where( + UserOAuth.created_at >= start_dt, + UserOAuth.created_at <= end_dt + ) + except ValueError: + # 捕获日期格式错误，非法时间直接不附加该查询条件 + pass + if appid: + count_query = count_query.where(UserOAuth.appid == appid) + if open_type: + count_query = count_query.where(UserOAuth.open_type == open_type) + + total_result = await db.execute(count_query) + total = total_result.scalar_one() + + query = query.order_by(UserOAuth.id.desc()).offset(offset).limit(page_size) + result = await db.execute(query) + list = []; + for oauth in result.scalars().all(): + user_query = select(User).where(User.id == oauth.user_id) + user_result = await db.execute(user_query) + user = user_result.scalar_one() + if user: + phone = user.phone + else: + phone = None + item = { + "id": oauth.id, + "account_id": oauth.account_id, + "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, + "user_phone": phone, + "open_type": oauth.open_type, + "port_type": oauth.port_type, + "appid": oauth.appid, + "access_token": oauth.access_token, + "refresh_token": oauth.refresh_token, + "access_token_expired": oauth.access_token_expired, + "refresh_token_expired": oauth.refresh_token_expired, + "created_at": oauth.created_at, + "updated_at": oauth.updated_at, + } + list.append(item) + return list, total + + +async def get_material_list( + db: AsyncSession, + phone: Optional[str] = None, + id: Optional[str] = None, + resource_type: Optional[str] = None, + advertiser_id: Optional[str] = None, + material_id: Optional[str] = None, + upload_id: Optional[str] = None, + created_at: Optional[datetime] = None, + status: Optional[str] = None, + page: int = 1, + page_size: int = 20, +) -> Tuple[List[ResourcesMaterial], int]: + offset = (page - 1) * page_size + + query = select(ResourcesMaterial).where(ResourcesMaterial.deleted_at.is_(None)) + + if phone: + query = query.join(UserOAuth, UserOAuth.id == ResourcesMaterial.oauth_id).join(User, User.id == UserOAuth.user_id).where(User.phone == phone) + if id: + query = query.where(ResourcesMaterial.id == id) + if resource_type: + query = query.where(ResourcesMaterial.resource_type == resource_type) + if advertiser_id: + query = query.where(ResourcesMaterial.advertiser_id == advertiser_id) + if material_id: + query = query.where(ResourcesMaterial.material_id == material_id) + if upload_id: + query = query.where(ResourcesMaterial.upload_id == upload_id) + if created_at: + start_dt, end_dt = parse_date_range(created_at, is_timezone=True) + # 直接用datetime范围匹配，不走func.date，保留时分秒精度，命中索引 + query = query.where( + ResourcesMaterial.created_at >= start_dt, + ResourcesMaterial.created_at <= end_dt + ) + if status: + query = query.where(ResourcesMaterial.status == status) + + count_query = select(func.count(ResourcesMaterial.id)).where(ResourcesMaterial.deleted_at.is_(None)) + + if phone: + count_query = count_query.join(UserOAuth, UserOAuth.id == ResourcesMaterial.oauth_id).join(User, User.id == UserOAuth.user_id).where(User.phone == phone) + if id: + count_query = count_query.where(ResourcesMaterial.id == id) + if resource_type: + count_query = count_query.where(ResourcesMaterial.resource_type == resource_type) + if advertiser_id: + count_query = count_query.where(ResourcesMaterial.advertiser_id == advertiser_id) + if material_id: + count_query = count_query.where(ResourcesMaterial.material_id == material_id) + if upload_id: + count_query = count_query.where(ResourcesMaterial.upload_id == upload_id) + if created_at: + start_dt, end_dt = parse_date_range(created_at, is_timezone=True) + # 直接用datetime范围匹配，不走func.date，保留时分秒精度，命中索引 + count_query = count_query.where( + ResourcesMaterial.created_at >= start_dt, + ResourcesMaterial.created_at <= end_dt + ) + if status: + count_query = count_query.where(ResourcesMaterial.status == status) + + total_result = await db.execute(count_query) + total = total_result.scalar_one() + + query = query.order_by(ResourcesMaterial.id.desc()).offset(offset).limit(page_size) + result = await db.execute(query) + list = [] + for item in result.scalars().all(): + user_query = select(User).where(User.id == item.user_id) + user_result = await db.execute(user_query) + user = user_result.scalar_one() + if user: + phone = user.phone + else: + phone = None + item = { + "id": item.id, + "oauth_id": item.oauth_id, + "user_phone": phone, + "advertiser_id": item.advertiser_id, + "target_table": item.target_table, + "target_id": item.target_id, + "material_id": item.material_id, + "upload_id": item.upload_id, + "resource_type": item.resource_type, + "user_id": item.user_id, + "created_at": item.created_at, + "updated_at": item.updated_at, + "task_id": item.task_id, + "note": item.note, + "status": item.status, + "pre_result": item.pre_result, + "pre_test_template_id": item.pre_test_template_id, + } + list.append(item) + + return list, total + + +async def get_pre_test_template_list( + db: AsyncSession, + phone: Optional[str] = None, + id: Optional[str] = None, + created_at: Optional[datetime] = None, + page: int = 1, + page_size: int = 10, +) -> Tuple[List[PreTestTemplate], int]: + offset = (page - 1) * page_size + + query = select(PreTestTemplate).where(PreTestTemplate.deleted_at.is_(None)) + + if phone: + query = query.join(User, User.id == PreTestTemplate.user_id).where(User.phone == phone) + if id: + query = query.where(PreTestTemplate.id == id) + if created_at: + start_dt, end_dt = parse_date_range(created_at, is_timezone=True) + # 直接用datetime范围匹配，不走func.date，保留时分秒精度，命中索引 + query = query.where( + PreTestTemplate.created_at >= start_dt, + PreTestTemplate.created_at <= end_dt + ) + + count_query = select(func.count(PreTestTemplate.id)).where(PreTestTemplate.deleted_at.is_(None)) + + if phone: + count_query = count_query.join(User, User.id == PreTestTemplate.user_id).where(User.phone == phone) + if id: + count_query = count_query.where(PreTestTemplate.id == id) + if created_at: + # 直接用datetime范围匹配，不走func.date，保留时分秒精度，命中索引 + start_dt, end_dt = parse_date_range(created_at, is_timezone=True) + count_query = count_query.where( + PreTestTemplate.created_at >= start_dt, + PreTestTemplate.created_at <= end_dt + ) + + total_result = await db.execute(count_query) + total = total_result.scalar_one() + + query = query.order_by(PreTestTemplate.id.desc()).offset(offset).limit(page_size) + result = await db.execute(query) + list = [] + for item in result.scalars().all(): + user_query = select(User).where(User.id == item.user_id) + user_result = await db.execute(user_query) + user = user_result.scalar_one() + if user: + phone = user.phone + else: + phone = None + item = { + "id": item.id, + "name": item.name, + "user_phone": phone, + "user_id": item.user_id, + "platform": item.platform, + "external_action": item.external_action, + "cpa_bid": item.cpa_bid, + "audience_gender": item.audience_gender, + "audience_age": item.audience_age, + "audience_region": item.audience_region, + "audience_network": item.audience_network, + "cus_name": item.cus_name, + "pricing_type": item.pricing_type, + "cost_cap": item.cost_cap, + "target_cost": item.target_cost, + "nobid": item.nobid, + "cpc_bid": item.cpc_bid, + "budget": item.budget, + "is_default": item.is_default, + "created_at": item.created_at, + "updated_at": item.updated_at, + "note": item.note, + } + list.append(item) + return list, total + + +async def get_upload_task_list( + db: AsyncSession, + phone: Optional[str] = None, + id: Optional[str] = None, + created_at: Optional[List[str]] = None, + advertiser_id: Optional[str] = None, + status: Optional[int] = None, + resource_id: Optional[str] = None, + page: int = 1, + page_size: int = 10, +) -> Tuple[List[UploadTask], int]: + offset = (page - 1) * page_size + + query = select(UploadTask).where(UploadTask.deleted_at.is_(None)) + + if phone: + query = query.join(User, User.id == UploadTask.user_id).where(User.phone == phone) + if id: + query = query.where(UploadTask.id == id) + if created_at and len(created_at) >= 2: + # 直接用datetime范围匹配，不走func.date，保留时分秒精度，命中索引 + start_dt, end_dt = parse_date_range(created_at, is_timezone=True) + query = query.where( + UploadTask.created_at >= start_dt, + UploadTask.created_at <= end_dt + ) + if advertiser_id: + query = query.where(UploadTask.advertiser_id == advertiser_id) + if status: + query = query.where(UploadTask.status == status) + if resource_id: + query = query.where(UploadTask.resource_id == resource_id) + + count_query = select(func.count(UploadTask.id)).where(UploadTask.deleted_at.is_(None)) + + if phone: + count_query = count_query.join(User, User.id == UploadTask.user_id).where(User.phone == phone) + if id: + count_query = count_query.where(UploadTask.id == id) + if created_at and len(created_at) >= 2: + # 直接用datetime范围匹配，不走func.date，保留时分秒精度，命中索引 + start_dt, end_dt = parse_date_range(created_at, is_timezone=True) + count_query = count_query.where( + UploadTask.created_at >= start_dt, + UploadTask.created_at <= end_dt + ) + if advertiser_id: + count_query = count_query.where(UploadTask.advertiser_id == advertiser_id) + if status: + count_query = count_query.where(UploadTask.status == status) + if resource_id: + count_query = count_query.where(UploadTask.resource_id == resource_id) + + total_result = await db.execute(count_query) + total = total_result.scalar_one() + + query = query.order_by(UploadTask.id.desc()).offset(offset).limit(page_size) + result = await db.execute(query) + list = [] + for item in result.scalars().all(): + user_query = select(User).where(User.id == item.user_id) + user_result = await db.execute(user_query) + user = user_result.scalar_one() + if user: + phone = user.phone + else: + phone = None + item = { + "id": item.id, + "user_phone": phone, + "user_id": item.user_id, + "advertiser_id": item.advertiser_id, + "resource_id": item.resource_id, + "status": item.status, + "note": item.note, + "oauth_id": item.oauth_id, + "created_at": item.created_at, + "updated_at": item.updated_at, + "other_info": item.other_info, + } + list.append(item) + return list, total diff --git a/video-gen-api/app/services/user_oauth_service.py b/video-gen-api/app/services/user_oauth_service.py index b85fcd51..88e87a63 100644 --- a/video-gen-api/app/services/user_oauth_service.py +++ b/video-gen-api/app/services/user_oauth_service.py @@ -1,20 +1,22 @@ import random -from datetime import datetime, timedelta, timezone +from datetime import datetime, timedelta import httpx -from sqlalchemy import select, func +from sqlalchemy import select, func, update from sqlalchemy.ext.asyncio import AsyncSession -from app.config import settings +# 引入项目统一北京时间时区 +from app.utils.datetime_util import BEIJING_TZ 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 +from app.tasks.token_refresh_task import _update_redis_token, _delete_redis_token -#随机获取一个可用的应用配置 +# 随机获取一个可用的应用配置 async def get_available_app(open_type: int, db: AsyncSession) -> dict: - # 从 user_oauth_app 表查询可用应用 - # 过滤条件：open_type 匹配、status=1（正常）、deleted_at is None + if not isinstance(open_type, int): + raise ValueError("open_type必须为整数类型") + result = await db.execute( select(UserOAuthApp).where( UserOAuthApp.open_type == open_type, @@ -30,8 +32,6 @@ async def get_available_app(open_type: int, db: AsyncSession) -> dict: available_apps = [] for app in apps: app_id = app.app_id - - # 查询该应用当前授权数量 count_result = await db.execute( select(func.count(UserOAuth.id)).where( UserOAuth.appid == app_id, @@ -39,8 +39,6 @@ async def get_available_app(open_type: int, db: AsyncSession) -> dict: ) ) count = count_result.scalar() or 0 - - # 检查是否达到最大授权数 max_users = app.max_count if count < max_users: available_apps.append({ @@ -59,24 +57,21 @@ async def get_available_app(open_type: int, db: AsyncSession) -> dict: async def build_oauth_url(open_type: int, user_id: str, db: AsyncSession) -> str: if open_type == 1: - #千川 return await _build_jl_oauth_url(open_type, user_id, db) elif open_type in [2, 3]: - #广告,本地推 return await _build_jl_oauth_url(2, user_id, db) elif open_type == 5: - #快手代理商 return "无配置" elif open_type == 9: - #腾讯营销K2 return "无配置" elif open_type == 10: - #腾讯营销K3 return "无配置" else: - raise ValueError(f"不支持的应用类型: {app_type}") + # 修复BUG：变量名错误 app_type -> open_type + raise ValueError(f"不支持的应用类型: {open_type}") -#千川授权链接构建 + +# 千川授权链接构建 async def _build_jl_oauth_url(open_type: int, user_id: str, db: AsyncSession) -> str: app = await get_available_app(open_type, db) app_id = app.get("app_id") @@ -92,221 +87,232 @@ async def _build_jl_oauth_url(open_type: int, user_id: str, db: AsyncSession) -> async def get_token(code: str, user_id: str, app_id: str, db: AsyncSession) -> dict: - #1.根据app_id查询应用配置 + if not all([code, user_id, app_id]): + raise ValueError("code、user_id、app_id不能为空") + result = await db.execute( - select(UserOAuthApp).where(UserOAuthApp.app_id == app_id) + select(UserOAuthApp).where( + UserOAuthApp.app_id == app_id, + UserOAuthApp.deleted_at.is_(None), + UserOAuthApp.status == 1 + ) ) app = result.scalar_one_or_none() if not app: - raise ValueError("应用配置不存在") - + raise ValueError("应用配置不存在或已禁用") + open_type = app.open_type secret = app.secret - if open_type == 1 or open_type == 2: - #千川或广告 + if open_type in (1, 2): return await get_juliang_token(app_id, secret, code, open_type, user_id, db) elif open_type == 5: - #快手代理商 return await get_kuaishou_token(app_id, secret, code, open_type, user_id, db) - elif open_type == 9 or open_type == 10: - #腾讯营销K2或腾讯营销K3 + elif open_type in (9, 10): return await get_tencent_token(app_id, secret, code, open_type, user_id, db) else: raise ValueError(f"不支持的应用类型: {open_type}") async def get_juliang_token(app_id: str, secret: str, code: str, open_type: int, user_id: str, db: AsyncSession) -> dict: - - async with httpx.AsyncClient() as client: - #1.请求token - url = "https://api.oceanengine.com/open_api/oauth2/access_token/" - response = await client.post( - url, - json={ - "app_id": app_id, - "secret": secret, - "auth_code": code, - }, - ) - response.raise_for_status() - data = response.json() - if data.get("code") != 0: - raise ValueError(data.get("message", "获取token失败")+f"，错误信息：{data.get('message', '')}") - data = data.get("data", {}) - access_token = data.get("access_token", "") - refresh_token = data.get("refresh_token", "") - expires_in = datetime.now(timezone.utc) + timedelta(seconds=data.get("expires_in", 0)) - refresh_token_expires_in = datetime.now(timezone.utc) + timedelta(seconds=data.get("refresh_token_expires_in", 0)) - #2.获取已授权角色账户，一个授权可能有多个角色账户 - url = "https://api.oceanengine.com/open_api/oauth2/advertiser/get/" - response = await client.get( - url, - headers={"Access-Token": access_token}, - ) - response.raise_for_status() - data = response.json() - if data.get("code") != 0: - raise ValueError(data.get("message", "获取已授权账户失败")+f"，错误信息：{data.get('message', '')}") - data = data.get("data", {}) - account_list = data.get("list", []) - #3.获取已授权登录信息 - url = "https://api.oceanengine.com/open_api/2/user/info/" - response = await client.get( - url, - headers={"Access-Token": access_token}, - ) - response.raise_for_status() - data = response.json() - if data.get("code") != 0: - raise ValueError(data.get("message", "获取已授权登录信息失败")+f"，错误信息：{data.get('message', '')}") - data = data.get("data", {}) - account_username = data.get("email", "") - account_userid = str(data.get("id", "")) - material_auth_status = data.get("material_auth_status", False) - #4.根据登录信息判断是否新增token或更新token，不同的登录信息对应不同的token，然后更新数据库 - #查询email，appid，user_id是否存在已授权记录 - oauth = 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) - .limit(1) - ) - 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=oauth_id, - account_id = str(account.get("account_id", "")), - account_name = account.get("account_name", ""), - account_role = account.get("account_role", ""), - account_username = account_username, - account_userid = account_userid, - user_id = user_id, - appid = app_id, - open_type = open_type, - port_type = 1, - 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 oauth_id in new_oauth_ids: - await _update_redis_token(oauth_id, access_token, expires_in) - else: - # 查询现有授权记录（未删除的） - existing_accounts = await db.execute( - select(UserOAuth).where( + timeout = httpx.Timeout(30.0) + try: + async with httpx.AsyncClient(timeout=timeout) as client: + # 1. 获取token + url = "https://api.oceanengine.com/open_api/oauth2/access_token/" + resp = await client.post( + url, + json={ + "app_id": app_id, + "secret": secret, + "auth_code": code, + }, + ) + resp.raise_for_status() + data = resp.json() + if data.get("code") != 0: + msg = data.get("message", "获取token失败") + raise ValueError(f"获取token失败：{msg}") + + resp_data = data.get("data", {}) + access_token = resp_data.get("access_token", "") + refresh_token = resp_data.get("refresh_token", "") + expires_sec = resp_data.get("expires_in", 7200) + refresh_expires_sec = resp_data.get("refresh_token_expires_in", 30 * 24 * 3600) + + # 【修复：统一使用北京时间计算过期时间】 + now_beijing = datetime.now(tz=BEIJING_TZ) + access_expired = now_beijing + timedelta(seconds=expires_sec) + refresh_expired = now_beijing + timedelta(seconds=refresh_expires_sec) + + # 2. 获取授权广告账户列表 + resp = await client.get( + "https://api.oceanengine.com/open_api/oauth2/advertiser/get/", + headers={"Access-Token": access_token}, + ) + resp.raise_for_status() + data = resp.json() + if data.get("code") != 0: + raise ValueError(f"获取已授权账户失败：{data.get('message')}") + account_list = data.get("data", {}).get("list", []) + + # 3. 获取登录用户信息 + resp = await client.get( + "https://api.oceanengine.com/open_api/2/user/info/", + headers={"Access-Token": access_token}, + ) + resp.raise_for_status() + data = resp.json() + if data.get("code") != 0: + raise ValueError(f"获取登录信息失败：{data.get('message')}") + + user_info = data.get("data", {}) + account_username = user_info.get("email", "") + account_userid = str(user_info.get("id", "")) + material_auth_status = user_info.get("material_auth_status", False) + + # 4. 查询当前用户+应用+登录账号下的授权记录 + oauth_query = 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) + ) + oauth_exist = await db.execute(oauth_query.limit(1)) + oauth_exist = oauth_exist.scalar_one_or_none() + + if not oauth_exist: + # 新增授权 + new_oauth_ids = [] + for account in account_list: + oauth_id = generate_id() + new_oauth_ids.append(oauth_id) + db.add(UserOAuth( + id=oauth_id, + account_id=str(account.get("account_id", "")), + account_name=account.get("account_name", ""), + account_role=account.get("account_role", ""), + account_username=account_username, + account_userid=account_userid, + user_id=user_id, + appid=app_id, + open_type=open_type, + port_type=1, + access_token=access_token, + access_token_expired=access_expired, + refresh_token=refresh_token, + refresh_token_expired=refresh_expired, + material_auth_status=material_auth_status, + )) + await db.commit() + # 批量写入Redis + for oauth_id in new_oauth_ids: + await _update_redis_token(oauth_id, access_token, access_expired) + else: + # 查询当前所有有效授权账户 + exist_query = 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), ) - ) - existing_accounts = {acc.account_id: acc for acc in existing_accounts.scalars().all()} + exist_res = await db.execute(exist_query) + exist_map = {item.account_id: item for item in exist_res.scalars().all()} - # 新的账户列表 - new_account_ids = {str(account.get("account_id")) for account in account_list} - old_account_ids = set(existing_accounts.keys()) + new_account_ids = {str(acc.get("account_id")) for acc in account_list} + old_account_ids = set(exist_map.keys()) + update_redis_ids = [] - # 需要更新Redis的oauth_id列表 - update_redis_ids = [] + # 下线的账户：软删除 + 清理Redis脏缓存 + for del_account_id in old_account_ids - new_account_ids: + del_oauth = exist_map[del_account_id] + del_oauth.deleted_at = now_beijing + await _delete_redis_token(del_oauth.id) - # 1. 软删除已消失的账户 - for account_id in old_account_ids - new_account_ids: - existing_accounts[account_id].deleted_at = datetime.now(timezone.utc) - - # 2. 更新或新增账户 - for account in account_list: - account_id = str(account.get("account_id", "")) - if account_id in existing_accounts: - # 更新现有记录 - existing_oauth = existing_accounts[account_id] - existing_oauth.account_name = account.get("account_name", "") - existing_oauth.account_role = account.get("account_role", "") - existing_oauth.access_token = access_token - existing_oauth.access_token_expired = expires_in - 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( + # 更新/新增当前授权账户 + for account in account_list: + aid = str(account.get("account_id", "")) + if aid in exist_map: + item = exist_map[aid] + item.account_name = account.get("account_name", "") + item.account_role = account.get("account_role", "") + item.access_token = access_token + item.access_token_expired = access_expired + item.refresh_token = refresh_token + item.refresh_token_expired = refresh_expired + item.material_auth_status = material_auth_status + update_redis_ids.append(item.id) + else: + oauth_id = generate_id() + update_redis_ids.append(oauth_id) + db.add(UserOAuth( id=oauth_id, - account_id=str(account_id), + account_id=aid, account_name=account.get("account_name", ""), account_role=account.get("account_role", ""), account_username=account_username, - account_userid = account_userid, + account_userid=account_userid, user_id=user_id, appid=app_id, open_type=open_type, port_type=1, - access_token = access_token, - access_token_expired=expires_in, + access_token=access_token, + access_token_expired=access_expired, refresh_token=refresh_token, - refresh_token_expired=refresh_token_expires_in, + refresh_token_expired=refresh_expired, 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) + await db.commit() + # 更新有效账号缓存 + for oid in update_redis_ids: + await _update_redis_token(oid, access_token, access_expired) - #5.本次更新成功以后，判断是否有其他同一个appid，同一个授权登录账号的授权记录,如果有，则更新token信息 - from sqlalchemy import update - - related_oauths = await db.execute( - select(UserOAuth).where( + # 5. 同登录账号、同应用、其他用户下的授权批量同步最新token + related_query = 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, + related_res = await db.execute(related_query) + related_list = related_res.scalars().all() + if related_list: + 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=access_expired, + refresh_token=refresh_token, + refresh_token_expired=refresh_expired, + 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) + await db.commit() + for item in related_list: + await _update_redis_token(item.id, access_token, access_expired) + + return {"message": "授权成功"} + + except httpx.HTTPError as e: + raise ValueError(f"第三方接口请求异常：{str(e)}") + except Exception as e: + raise ValueError(f"授权处理异常：{str(e)}") - return {"message": "授权成功"} async def get_kuaishou_token(app_id: str, secret: str, code: str, oauth_type: int, user_id: str, db: AsyncSession) -> dict: - return "未配置" + return {"message": "快手渠道暂未实现授权逻辑"} async def get_tencent_token(app_id: str, secret: str, code: str, oauth_type: int, user_id: str, db: AsyncSession) -> str: - return "未配置" + return "腾讯营销渠道暂未实现授权逻辑" async def get_oauth_list( @@ -318,36 +324,35 @@ async def get_oauth_list( page: int = 1, page_size: int = 10, ) -> dict: - if page < 1: - page = 1 - if page_size < 1: - page_size = 10 + # 分页参数容错 + page = max(page, 1) + page_size = max(min(page_size, 100), 1) - query = select(UserOAuth).where( + base_where = [ UserOAuth.user_id == user_id, UserOAuth.deleted_at.is_(None), - ) - + ] if account_userid: - query = query.where(UserOAuth.account_userid == account_userid) + base_where.append(UserOAuth.account_userid == account_userid) if open_type: - query = query.where(UserOAuth.open_type == open_type) + base_where.append(UserOAuth.open_type == open_type) if account_id: - query = query.where(UserOAuth.account_id == account_id) + base_where.append(UserOAuth.account_id == account_id) - query = query.order_by(UserOAuth.created_at.desc()) + # 统计总条数（优化：使用count，避免全量查询） + count_stmt = select(func.count(UserOAuth.id)).where(*base_where) + total = await db.scalar(count_stmt) or 0 - total_result = await db.execute(query.with_only_columns(UserOAuth.id)) - total = len(total_result.scalars().all()) - - offset = (page - 1) * page_size - query = query.offset(offset).limit(page_size) - - result = await db.execute(query) - oauth_list = result.scalars().all() + # 分页查询数据 + data_stmt = select(UserOAuth).where(*base_where)\ + .order_by(UserOAuth.created_at.desc())\ + .offset((page - 1) * page_size)\ + .limit(page_size) + result = await db.execute(data_stmt) + data_list = result.scalars().all() return { - "data": oauth_list, + "data": data_list, "total": total, "page": page, "page_size": page_size, diff --git a/video-gen-api/app/tasks/token_refresh_task.py b/video-gen-api/app/tasks/token_refresh_task.py index 37a9830e..6f6a4279 100644 --- a/video-gen-api/app/tasks/token_refresh_task.py +++ b/video-gen-api/app/tasks/token_refresh_task.py @@ -1,49 +1,64 @@ -from datetime import datetime, timedelta, timezone +from datetime import datetime, timedelta import asyncio import httpx import json +from typing import Optional -from sqlalchemy import select +from sqlalchemy import select, update from sqlalchemy.ext.asyncio import AsyncSession +# 引入自定义北京时间工具类 +from app.utils.datetime_util import BEIJING_TZ 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 from app.utils.logger import get_logger +from app.utils.datetime_util import datetime_to_db_tz_str, db_tz_str_to_datetime REDIS_KEY = "douyin:tokens" logger = get_logger("token_refresh", "token_refresh") - - +# 配置常量 REFRESH_THRESHOLD_SECONDS = 800 CHECK_INTERVAL_MINUTES = 5 +HTTP_TIMEOUT = httpx.Timeout(30.0) + + +async def _delete_redis_token(oauth_id: str): + """【改动1：废弃空值覆盖，直接删除Hash脏缓存】refresh失效则移除字段""" + redis = get_redis() + if not redis: + logger.warning("Redis连接未配置，跳过缓存清理") + return + try: + await redis.hdel(REDIS_KEY, oauth_id) + logger.info(f"清理失效授权Redis缓存: oauth_id={oauth_id}") + except Exception as e: + logger.error(f"删除Redis缓存失败 oauth_id:{oauth_id}, err:{str(e)}") async def _update_redis_token(oauth_id: str, token: str, expired_at: datetime): - """更新Redis缓存中的token""" + """更新Redis缓存中的token（统一北京时间序列化）""" redis = get_redis() if not redis: logger.warning("Redis连接未配置，跳过缓存更新") return - try: cache = { - "token": token, - "expired_at": expired_at.isoformat(), - } + "token": token, + "expired_at": datetime_to_db_tz_str(expired_at), + } 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)}") + logger.error(f"更新Redis缓存失败 oauth_id:{oauth_id}, err:{str(e)}") async def refresh_juliang_token(oauth: UserOAuth, app: UserOAuthApp, db: AsyncSession): """刷新巨量引擎token""" try: - async with httpx.AsyncClient() as client: + async with httpx.AsyncClient(timeout=HTTP_TIMEOUT) as client: url = "https://api.oceanengine.com/open_api/oauth2/refresh_token/" response = await client.post( url, @@ -55,13 +70,11 @@ async def refresh_juliang_token(oauth: UserOAuth, app: UserOAuthApp, db: AsyncSe ) 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 + # refresh_token已失效 if data.get("code") in [40103, 40107]: - #清空数据库中的token信息，和Redis缓存中的token - from sqlalchemy import update - where_cond = UserOAuth.deleted_at.is_(None) if oauth.appid: where_cond = where_cond & (UserOAuth.appid == oauth.appid) @@ -69,7 +82,7 @@ async def refresh_juliang_token(oauth: UserOAuth, app: UserOAuthApp, db: AsyncSe 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, @@ -79,25 +92,23 @@ async def refresh_juliang_token(oauth: UserOAuth, app: UserOAuthApp, db: AsyncSe ) ) await db.commit() - - related_oauth_ids = await db.execute( - select(UserOAuth.id).where(where_cond) - ) + + related_oauth_ids = await db.execute(select(UserOAuth.id).where(where_cond)) related_oauth_ids = [row[0] for row in related_oauth_ids.all()] + # 【改动2：失效直接删除Redis缓存，不再写入空值】 for related_id in related_oauth_ids: - await _update_redis_token(related_id, "", None) - + await _delete_redis_token(related_id) return - - data = data.get("data", {}) - new_access_token = data.get("access_token", "") - new_refresh_token = data.get("refresh_token", "") - expires_in = datetime.now(timezone.utc) + timedelta(seconds=data.get("expires_in", 0)) - refresh_token_expires_in = datetime.now(timezone.utc) + timedelta(seconds=data.get("refresh_token_expires_in", 0)) + resp_data = data.get("data", {}) + new_access_token = resp_data.get("access_token", "") + new_refresh_token = resp_data.get("refresh_token", "") + + # 【改动3：全部使用北京时间计算过期时间，统一时区】 + now_beijing = datetime.now(tz=BEIJING_TZ) + expires_in = now_beijing + timedelta(seconds=resp_data.get("expires_in", 0)) + refresh_expires = now_beijing + timedelta(seconds=resp_data.get("refresh_token_expires_in", 0)) - from sqlalchemy import update - where_cond = UserOAuth.deleted_at.is_(None) if oauth.appid: where_cond = where_cond & (UserOAuth.appid == oauth.appid) @@ -105,25 +116,26 @@ async def refresh_juliang_token(oauth: UserOAuth, app: UserOAuthApp, db: AsyncSe 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, + refresh_token_expired=refresh_expires, ) ) await db.commit() - - related_oauth_ids = await db.execute( - select(UserOAuth.id).where(where_cond) - ) + + 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}, 关联账户数={len(related_oauth_ids)}") + + logger.info( + f"成功刷新巨量引擎token: oauth_id={oauth.id}, " + f"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: @@ -133,93 +145,88 @@ async def refresh_juliang_token(oauth: UserOAuth, app: UserOAuthApp, db: AsyncSe async def check_and_refresh_tokens(): """检查并刷新即将过期的token""" async with async_session() as db: - now = datetime.now(timezone.utc) - + # 【改动4：统一使用北京时间当前时间，避免时区运算异常】 + now_beijing = datetime.now(tz=BEIJING_TZ) + query = select(UserOAuth).where( UserOAuth.deleted_at.is_(None), UserOAuth.refresh_token.is_not(None), UserOAuth.refresh_token_expired.is_not(None), - UserOAuth.refresh_token_expired > now, + UserOAuth.refresh_token_expired > now_beijing, ) - + result = await db.execute(query) oauth_list = result.scalars().all() - + refreshed_keys = set() - + for oauth in oauth_list: try: - #检查是否为支持的平台（巨量引擎） - # port_type: 平台端口（1=巨量，2=磁力，3=巨量星图，4=服务单，5=腾讯） if oauth.port_type not in [1]: continue - - #构建登录账号唯一标识，同一登录账号共享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) + key_parts.append(str(oauth.account_userid)) login_key = "|".join(key_parts) - - #同一登录账号已刷新过，直接跳过（避免使用旧数据判断） + if login_key in refreshed_keys: logger.debug(f"跳过重复刷新: oauth_id={oauth.id}, 同一登录账号已刷新") continue - - #获取应用配置 + + # 【改动5：过滤已禁用、已删除应用】 app_result = await db.execute( - select(UserOAuthApp).where(UserOAuthApp.app_id == oauth.appid) + select(UserOAuthApp).where( + UserOAuthApp.app_id == oauth.appid, + UserOAuthApp.status == 1, + UserOAuthApp.deleted_at.is_(None) + ) ) app = app_result.scalar_one_or_none() - if not app: + logger.warning(f"oauth_id:{oauth.id} 对应应用不存在/已禁用，跳过刷新") continue - - #检查access_token是否需要刷新 + need_refresh = False - - # access_token为空，需要刷新 - if not oauth.access_token: + if not oauth.access_token or not oauth.access_token_expired: 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: + # 同时区时间运算，不会抛异常 + remain_sec = (oauth.access_token_expired - now_beijing).total_seconds() + if remain_sec < 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) - + + # 简单限流，防止瞬间大量请求 + await asyncio.sleep(0.2) + except Exception as e: - #7.增加错误日志 - logger.error(f"刷新token失败: {str(e)}") + logger.error(f"oauth_id:{oauth.id} 刷新token异常: {str(e)}", exc_info=True) async def token_refresh_scheduler(): """定时任务调度器""" + logger.info("开始执行定时Token刷新任务") while True: try: await check_and_refresh_tokens() except Exception as e: - logger.error(f"定时任务token_refresh_scheduler执行失败: {str(e)}") - + logger.error(f"定时任务token_refresh_scheduler执行失败: {str(e)}", exc_info=True) + 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 + logger.info("启动token刷新定时后台任务") + asyncio.create_task(token_refresh_scheduler()) diff --git a/video-gen-api/app/utils/datetime_util.py b/video-gen-api/app/utils/datetime_util.py new file mode 100644 index 00000000..f17fa7e3 --- /dev/null +++ b/video-gen-api/app/utils/datetime_util.py @@ -0,0 +1,270 @@ +from datetime import datetime, timedelta, timezone, date +from typing import Optional, Union, Tuple +from dateutil import parser + +# ===================== 时区与格式化常量（PHP风格映射） ===================== +# 北京时间 东八区 UTC+8 +BEIJING_TZ = timezone(timedelta(hours=8)) +UTC_TZ = timezone.utc + +# Python strftime 标准格式 +FORMAT_DATETIME = "%Y-%m-%d %H:%M:%S" +FORMAT_DATE = "%Y-%m-%d" +FORMAT_TIME = "%H:%M:%S" + +# PHP格式 -> Python格式 映射，贴近PHP使用习惯 +PHP_FMT_MAP = { + "Y-m-d": FORMAT_DATE, + "Y-m-d H:i:s": FORMAT_DATETIME, + "H:i:s": FORMAT_TIME +} + +# ===================== 私有公共工具函数（内部复用） ===================== +def _normalize_date_str(date_str: str) -> str: + """ + 预处理中文格式日期字符串，统一转为横杠分隔标准格式 + 支持：2025年01月01日 12时30分00秒 / 2025：01：01 等格式 + """ + if not isinstance(date_str, str): + return "" + s = date_str.strip() + # 中文、全角符号替换清洗 + s = s.replace("：", ":") \ + .replace("年", "-") \ + .replace("月", "-") \ + .replace("日", " ") \ + .replace("时", ":") \ + .replace("分", ":") \ + .replace("秒", "") + return s + + +def _get_python_fmt(fmt: str) -> str: + """兼容PHP格式字符串，自动转为Python strftime格式""" + return PHP_FMT_MAP.get(fmt, fmt) + + +def _convert_ts_to_datetime(timestamp: Union[int, float], tz: timezone = BEIJING_TZ) -> datetime: + """ + 统一处理 10位秒 / 13位毫秒 时间戳 -> 带时区datetime + """ + ts = float(timestamp) + # 毫秒时间戳兼容 + if ts > 10 ** 12: + ts /= 1000 + return datetime.fromtimestamp(ts, tz=tz) + + +def _parse_datetime_with_tz(dt_str: str, fmt: str, tz: timezone = BEIJING_TZ) -> Optional[datetime]: + """ + 通用：固定格式日期字符串解析+绑定时区，支持中文日期预处理 + 仅支持指定fmt格式，不支持带时区后缀字符串 + """ + try: + fmt = _get_python_fmt(fmt) + normalize_str = _normalize_date_str(dt_str) + dt = datetime.strptime(normalize_str, fmt) + return dt.replace(tzinfo=tz) + except (ValueError, TypeError): + return None + + +def _smart_parse_datetime(dt_str: str, base_dt: datetime, is_timezone: bool) -> Optional[datetime]: + """ + 私有：dateutil智能解析日期（用于带时区、相对时间、中文复杂日期） + """ + clean_str = _normalize_date_str(dt_str) + try: + parsed_dt = parser.parse(clean_str, default=base_dt, tzinfos=None) + if is_timezone: + # 保留原始时区，无时区则兜底北京时间 + if parsed_dt.tzinfo is None: + parsed_dt = parsed_dt.replace(tzinfo=BEIJING_TZ) + else: + # 普通日期强制绑定北京时间 + parsed_dt = parsed_dt.replace(tzinfo=BEIJING_TZ) + return parsed_dt + except (parser.ParserError, ValueError, TypeError): + return None + +# ===================== 对外工具方法（对标PHP + dateutil增强） ===================== +def time() -> int: + """ + 对标 PHP time() + 获取当前北京时间 10位 秒级时间戳 + """ + return int(datetime.now(tz=BEIJING_TZ).timestamp()) + + +def microtime(get_as_float: bool = False) -> Union[float, str]: + """ + 对标 PHP microtime() + :param get_as_float: True 返回浮点时间戳，False 返回 "微秒 秒" 字符串 + """ + now = datetime.now(tz=BEIJING_TZ) + ts = now.timestamp() + if get_as_float: + return ts + sec = int(ts) + usec = int((ts - sec) * 1000000) + return f"{usec:06d} {sec}" + + +def strtotime( + time_str: str, + now: Optional[int] = None, + is_timezone: bool = False +) -> Optional[int]: + """ + 【增强版 对标 PHP strtotime】依托 dateutil 支持相对时间、带时区时间解析 + :param time_str: 待解析日期/相对时间字符串 + :param now: 基准时间戳，默认当前北京时间 + :param is_timezone: 是否为带时区格式的时间字符串（如 2026-07-03 10:01:29.621351+08） + - True：优先使用字符串自带时区，无时区则兜底北京时间 + - False：默认按北京时间解析普通日期字符串 + :return: 秒级时间戳，解析失败返回 None + """ + if not isinstance(time_str, str) or not time_str.strip(): + return None + + # 基准时间：默认当前北京时间 + base_dt = datetime.fromtimestamp(now, tz=BEIJING_TZ) if now else datetime.now(tz=BEIJING_TZ) + parsed_dt = _smart_parse_datetime(time_str, base_dt, is_timezone) + if not parsed_dt: + return None + return int(parsed_dt.timestamp()) + + +def date(fmt: str, timestamp: Optional[int] = None) -> str: + """ + 对标 PHP date() + 时间戳格式化，默认北京时间 + :param fmt: 支持 Y-m-d / Y-m-d H:i:s 或 %Y-%m-%d 原生格式 + :param timestamp: 秒时间戳，None则取当前时间 + """ + if timestamp is None: + dt = datetime.now(tz=BEIJING_TZ) + else: + dt = _convert_ts_to_datetime(timestamp) + + python_fmt = _get_python_fmt(fmt) + return dt.strftime(python_fmt) + + +def timestamp_to_datetime(timestamp: Union[int, float], fmt: str = "Y-m-d H:i:s") -> str: + """ + 时间戳（秒/毫秒）→ 北京时间格式化字符串 + :param timestamp: 10位秒 /13位毫秒 + :param fmt: PHP风格格式 Y-m-d / Y-m-d H:i:s + """ + dt = _convert_ts_to_datetime(timestamp) + python_fmt = _get_python_fmt(fmt) + return dt.strftime(python_fmt) + + +def datetime_to_timestamp(dt_str: str, fmt: str = "Y-m-d H:i:s") -> Optional[int]: + """ + 日期字符串(含中文日期) → 北京时间秒时间戳 + :param dt_str: 日期字符串 + :param fmt: PHP风格格式化模板 + :return: 秒时间戳，解析失败返回None + """ + dt = _parse_datetime_with_tz(dt_str, fmt) + if not dt: + return None + return int(dt.timestamp()) + + +def get_today_start_timestamp() -> int: + """获取今日 00:00:00 北京时间 秒时间戳""" + today: date = datetime.now(tz=BEIJING_TZ).date() + start_dt = datetime.combine(today, datetime.min.time(), tzinfo=BEIJING_TZ) + return int(start_dt.timestamp()) + + +def get_today_end_timestamp() -> int: + """获取今日 23:59:59.999999 北京时间 秒时间戳""" + today: date = datetime.now(tz=BEIJING_TZ).date() + end_dt = datetime.combine(today, datetime.max.time(), tzinfo=BEIJING_TZ) + return int(end_dt.timestamp()) + + +def parse_date_range(date_list: list, fmt: str = "Y-m-d", is_timezone: bool = False) -> Optional[Tuple[datetime, datetime]]: + """ + 时间范围解析：["2025-01-01","2026-01-01"] → (开始0点,结束23:59:59.999999) 带北京时间 + :param date_list: 长度为2的日期字符串数组 + :param fmt: 日期格式，默认Y-m-d + :param is_timezone: 是否为带时区格式的时间字符串（如 2026-07-03 10:01:29.621351+08） + - True：使用dateutil智能解析，优先保留字符串自带时区，无时区兜底北京时间 + - False：按指定fmt格式精准解析，强制北京时间 + :return: (start_dt, end_dt) 格式非法返回None + """ + # 强参数校验 + if not isinstance(date_list, list) or len(date_list) != 2: + return None + start_str, end_str = date_list[0].strip(), date_list[1].strip() + if not start_str or not end_str: + return None + + base_now = datetime.now(tz=BEIJING_TZ) + + if is_timezone: + # 带时区场景：智能解析，支持 2026-07-03 08:16:26.88303+08 + start_dt = _smart_parse_datetime(start_str, base_now, is_timezone=True) + end_dt = _smart_parse_datetime(end_str, base_now, is_timezone=True) + else: + # 普通前端日期：固定格式解析 + start_dt = _parse_datetime_with_tz(start_str, fmt, BEIJING_TZ) + end_dt = _parse_datetime_with_tz(end_str, fmt, BEIJING_TZ) + + if not start_dt or not end_dt: + return None + + # 结束时间补全到当日最后一毫秒 + end_dt = end_dt.replace(hour=23, minute=59, second=59, microsecond=999999) + return start_dt, end_dt + + +def db_tz_str_to_datetime(db_datetime_str: str) -> Optional[datetime]: + """ + 数据库timestamptz格式字符串（2026-07-03 08:16:26.88303+08）转为带时区datetime + """ + try: + dt = datetime.fromisoformat(db_datetime_str) + if dt.tzinfo is None: + dt = dt.replace(tzinfo=BEIJING_TZ) + return dt + except (ValueError, TypeError): + return None + + +def datetime_to_db_tz_str(dt: datetime) -> str: + """ + 带时区datetime转为数据库timestamptz标准字符串，统一北京时间存储 + """ + if dt.tzinfo is None: + dt = dt.replace(tzinfo=BEIJING_TZ) + else: + dt = dt.astimezone(BEIJING_TZ) + return dt.isoformat() + + +# 导出列表 +__all__ = [ + "BEIJING_TZ", + "UTC_TZ", + "FORMAT_DATETIME", + "FORMAT_DATE", + "FORMAT_TIME", + "time", + "microtime", + "strtotime", + "date", + "timestamp_to_datetime", + "datetime_to_timestamp", + "get_today_start_timestamp", + "get_today_end_timestamp", + "parse_date_range", + "db_tz_str_to_datetime", + "datetime_to_db_tz_str" +] \ 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 a817d513..551d912b 100644 --- a/video-gen-api/app/utils/douyinRequest.py +++ b/video-gen-api/app/utils/douyinRequest.py @@ -5,8 +5,14 @@ from typing import Any, Dict, Optional, Tuple import httpx from sqlalchemy import select, update from sqlalchemy.ext.asyncio import AsyncSession -from datetime import datetime, timezone, timedelta +from datetime import datetime, timedelta +# 【改动1：引入自定义北京时间工具类】 +from app.utils.datetime_util import ( + BEIJING_TZ, + datetime_to_db_tz_str, + db_tz_str_to_datetime +) from app.models.user_oauth import UserOAuth from app.models.user_oauth_app import UserOAuthApp from app.models.base import async_session @@ -40,50 +46,75 @@ class DouyinRequest: self._client = None async def _get_redis_token(self, oauth_id: str) -> Optional[str]: + """ + 【核心改动：北京时间解析+脏缓存自动清理】 + 1. 缓存格式异常/字段缺失：直接删除脏缓存 + 2. Token过期则不返回，交由上层判断refresh是否过期决定是否删除缓存 + """ redis = get_redis() if not redis: raise ValueError("Redis连接未配置") try: cache_str = await redis.hget(self._redis_key, oauth_id) - if cache_str: - cache = json.loads(cache_str) - token = cache.get("token") - expired_at_str = cache.get("expired_at") - if token and expired_at_str: - expired_at = datetime.fromisoformat(expired_at_str).replace(tzinfo=timezone.utc) - if expired_at > datetime.now(timezone.utc): - return token - except Exception as e: - raise ValueError(f"获取Redis缓存失败: {e}") + if not cache_str: + return None - return None + cache = json.loads(cache_str) + token = cache.get("token") + expired_at_str = cache.get("expired_at") + + # 脏缓存：关键字段缺失，直接删除 + if not (token and expired_at_str): + await redis.hdel(self._redis_key, oauth_id) + logger.warning(f"oauth_id:{oauth_id} Redis缓存字段缺失，已清理脏数据") + return None + + # 【改动3：使用工具类解析带时区时间，不再强制覆盖UTC】 + expired_at = db_tz_str_to_datetime(expired_at_str) + now_beijing = datetime.now(tz=BEIJING_TZ) + + if expired_at > now_beijing: + return token + + # AccessToken已过期，返回None，上层会校验refresh_token状态决定是否清理缓存 + return None + + except json.JSONDecodeError: + # JSON格式损坏，清理脏缓存 + await redis.hdel(self._redis_key, oauth_id) + logger.error(f"oauth_id:{oauth_id} Redis缓存JSON格式异常，已清理脏数据") + return None + except Exception as e: + raise ValueError(f"获取Redis缓存失败: {str(e)}") async def _set_redis_token(self, oauth_id: str, token: str, expired_at: datetime): redis = get_redis() if not redis: raise ValueError("Redis连接未配置") - + # 【改动4：统一转为北京时间序列化存入Redis，和数据库时区对齐】 + cache = { + "token": token, + "expired_at": datetime_to_db_tz_str(expired_at), + } try: - cache = { - "token": token, - "expired_at": expired_at.isoformat(), - } - await redis.hset(self._redis_key, oauth_id, json.dumps(cache)) + await redis.hset(self._redis_key, oauth_id, json.dumps(cache, ensure_ascii=False)) except Exception as e: - raise ValueError(f"设置Redis缓存失败: {e}") + raise ValueError(f"设置Redis缓存失败: {str(e)}") async def _delete_redis_token(self, oauth_id: str): + """删除单个oauth_id缓存（授权彻底失效时调用）""" redis = get_redis() if not redis: raise ValueError("Redis连接未配置") - try: await redis.hdel(self._redis_key, oauth_id) + logger.info(f"oauth_id:{oauth_id} 授权失效，已清理Redis缓存脏数据") except Exception as e: - raise ValueError(f"删除Redis缓存失败: {e}") + raise ValueError(f"删除Redis缓存失败: {str(e)}") async def get_access_token(self, oauth_id: str, force_refresh: bool = False) -> str: + # 优先读取缓存 if not force_refresh: token = await self._get_redis_token(oauth_id) if token: @@ -101,7 +132,11 @@ class DouyinRequest: if not oauth_data: raise ValueError("无效的oauth_id") + # 【改动5：统一北京时间当前时间】 + now_beijing = datetime.now(tz=BEIJING_TZ) + if force_refresh: + # 强制刷新：清空同条件下所有账号缓存与数据库token where_cond = UserOAuth.deleted_at.is_(None) if oauth_data.appid: where_cond = where_cond & (UserOAuth.appid == oauth_data.appid) @@ -118,9 +153,7 @@ class DouyinRequest: ) await db.commit() - related_oauth_ids = await db.execute( - select(UserOAuth.id).where(where_cond) - ) + 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) @@ -128,12 +161,14 @@ class DouyinRequest: new_token, new_expired_at = await self.refresh_access_token(db, oauth_id, oauth_data.appid, oauth_data.refresh_token) return new_token - if oauth_data.access_token_expired and oauth_data.access_token_expired > datetime.now(timezone.utc): + # 数据库AccessToken未过期，写入缓存直接返回 + if oauth_data.access_token_expired and oauth_data.access_token_expired > now_beijing: token = oauth_data.access_token expired_at = oauth_data.access_token_expired await self._set_redis_token(oauth_id, token, expired_at) return token + # 查找同账号下未过期有效token复用 where_cond = UserOAuth.deleted_at.is_(None) if oauth_data.appid: where_cond = where_cond & (UserOAuth.appid == oauth_data.appid) @@ -146,7 +181,7 @@ class DouyinRequest: 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), + UserOAuth.access_token_expired > now_beijing, ).limit(1) ) related_oauth = related_oauths.first() @@ -155,9 +190,13 @@ class DouyinRequest: 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("授权已过期，请重新授权") + # =========【核心业务规则实现：判断RefreshToken是否过期】========= + if oauth_data.refresh_token_expired and oauth_data.refresh_token_expired < now_beijing: + # RefreshToken过期 → 授权彻底失效，清理Redis脏缓存 + await self._delete_redis_token(oauth_id) + raise ValueError("授权已过期，请重新授权登录") + # RefreshToken有效，执行刷新并更新缓存 new_token, new_expired_at = await self.refresh_access_token(db, oauth_id, oauth_data.appid, oauth_data.refresh_token) return new_token @@ -171,7 +210,7 @@ class DouyinRequest: oauth_info = oauth_info.first() if not oauth_info: raise ValueError("无效的oauth_id") - + account_username, account_userid = oauth_info result = await db.execute( @@ -211,7 +250,10 @@ class DouyinRequest: expires_in = data.get('expires_in', 0) refresh_token_expires_in = data.get('refresh_token_expires_in', 0) - new_expired_at = datetime.now(timezone.utc) + timedelta(seconds=expires_in) + # 【改动6：北京时间计算过期时间】 + now_beijing = datetime.now(tz=BEIJING_TZ) + new_expired_at = now_beijing + timedelta(seconds=expires_in) + new_refresh_expired = now_beijing + timedelta(seconds=refresh_token_expires_in) where_cond = UserOAuth.deleted_at.is_(None) if appid: @@ -222,20 +264,16 @@ class DouyinRequest: where_cond = where_cond & (UserOAuth.account_userid == account_userid) await db.execute( - update(UserOAuth).where( - where_cond, - ).values( + update(UserOAuth).where(where_cond).values( access_token=new_access_token, refresh_token=new_refresh_token, access_token_expired=new_expired_at, - refresh_token_expired=datetime.now(timezone.utc) + timedelta(seconds=refresh_token_expires_in), + refresh_token_expired=new_refresh_expired, ) ) await db.commit() - related_oauth_ids = await db.execute( - select(UserOAuth.id).where(where_cond) - ) + 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: @@ -243,92 +281,79 @@ class DouyinRequest: return new_access_token, new_expired_at - # 有token请求 async def request_with_token_with_context( self, oauth_id: str, url: str, method: str = 'GET', - options: any = None, + options: Optional[Dict[str, Any]] = None, request_count: int = 1, ) -> Any: options = options or {} token = await self.get_access_token(oauth_id) + resp_data: Optional[Dict[str, Any]] = None - for i in range(1, request_count+1): - # 有一些错误是触发频次管理的，需要重试 + for i in range(1, request_count + 1): try: headers = options.get('headers', {}).copy() headers['Access-Token'] = token - + has_files = 'files' in options if has_files: - # 移除可能错误设置的 Content-Type，让库自动生成 multipart 头 headers.pop('Content-Type', None) else: headers.setdefault('Content-Type', 'application/json') - + options['headers'] = headers - response = await self.client.request(method, url, **options) response.raise_for_status() - - try: - data = response.json() - except json.JSONDecodeError: - data = {'code': 0, 'data': response.text, 'msg':'JSON解析失败'} - if 'code' not in data: + try: + resp_data = response.json() + except json.JSONDecodeError: + resp_data = {'code': 0, 'data': response.text, 'msg': 'JSON解析失败'} + + if 'code' not in resp_data: await asyncio.sleep(i * 5) continue - code = data.get('code', 0) - + code = resp_data.get('code', 0) if code in [40102, 40104]: await asyncio.sleep(i * 5) token = await self.get_access_token(oauth_id, force_refresh=True) continue - if code in [40100, 40110]: wait_time = min(2 * (2 ** (i - 1)), 10) await asyncio.sleep(wait_time) continue - if code == 50000: await asyncio.sleep(i * 10) continue - return data + return resp_data - except httpx.HTTPStatusError as e: - await asyncio.sleep(i * 10) - continue - except httpx.RequestError as e: + except (httpx.HTTPStatusError, httpx.RequestError): await asyncio.sleep(i * 10) continue + # 重试耗尽，日志记录 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 - - res = json.dumps(data, ensure_ascii=False) if 'data' in locals() else '' - + 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 + + res = json.dumps(resp_data, ensure_ascii=False) if resp_data else '' logger.error( f'DouYin API request failed after {request_count} retries. ' f'url:{url};method:{method};oauth_id:{oauth_id};options:{json.dumps(options_log, ensure_ascii=False)};response:{res}' ) - - if 'data' in locals() and data.get('code', 0) != 0: - raise ValueError(f'接口返回错误[code:{data.get("code", "接口编码")}]{data.get("message", "接口返回错误")}') - else: - raise ValueError('网络错误，稍后重试。') + if resp_data and resp_data.get("code", 0) != 0: + raise ValueError(f'接口返回错误[code:{resp_data.get("code")}]{resp_data.get("message", "接口异常")}') + raise ValueError("网络请求失败，请稍后重试") - # 无token请求 async def request_with_context( self, url: str, @@ -336,8 +361,9 @@ class DouyinRequest: options: Optional[Dict[str, Any]] = None, ) -> Dict[str, Any]: options = options or {} + resp_data: Optional[Dict[str, Any]] = None - for i in range(1, 2): + for i in range(1, 3): try: headers = options.get('headers', {}).copy() headers.setdefault('Content-Type', 'application/json') @@ -347,30 +373,24 @@ class DouyinRequest: response.raise_for_status() try: - data = response.json() + resp_data = response.json() except json.JSONDecodeError: - data = {'code': 0, 'data': response.text} + resp_data = {'code': 0, 'data': response.text} - if 'code' not in data: + if 'code' not in resp_data: await asyncio.sleep(i * 5) continue - - code = data.get('code', 0) - if code >= 50000: + if resp_data.get("code", 0) >= 50000: await asyncio.sleep(i * 10) continue + return resp_data - return data - - except httpx.HTTPStatusError as e: - await asyncio.sleep(i * 10) - continue - except httpx.RequestError as e: + except (httpx.HTTPStatusError, httpx.RequestError): await asyncio.sleep(i * 10) continue - res = json.dumps(data) if 'data' in locals() else '' + res = json.dumps(resp_data, ensure_ascii=False) if resp_data else '' raise RuntimeError( - f'DouYin API request failed after 5 retries. ' - f'url:{url};options:{json.dumps(options)};response:{res}' + f'DouYin API request failed after 2 retries. ' + f'url:{url};options:{json.dumps(options, ensure_ascii=False)};response:{res}' ) \ No newline at end of file