From 242030c50b5faed978eae6a7ceeda886834f2d25 Mon Sep 17 00:00:00 2001 From: 18610128193 <10574456+chenweiqiang-123@user.noreply.gitee.com> Date: Tue, 23 Jun 2026 14:16:31 +0800 Subject: [PATCH] =?UTF-8?q?=E6=96=B0=E5=A2=9E=E7=B4=A0=E6=9D=90=E6=9B=B4?= =?UTF-8?q?=E6=96=B0=E5=8A=9F=E8=83=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- video-gen-api/app/api/v1/__init__.py | 4 + .../app/api/v1/material_consumption.py | 301 ++++++++++++++++ video-gen-api/app/api/v1/open_type.py | 262 ++++++++++++++ video-gen-api/app/api/v1/upload_material.py | 8 + video-gen-api/app/main.py | 16 +- video-gen-api/app/schemas/open_type.py | 48 +++ .../services/material_consumption_queue.py | 334 ++++++++++++++++++ video-gen-api/app/services/upload_queue.py | 47 +++ .../app/tasks/material_consumption_task.py | 68 ++++ video-gen-api/app/utils/douyinApi.py | 24 ++ video-gen-api/app/utils/douyinRequest.py | 4 +- 11 files changed, 1109 insertions(+), 7 deletions(-) create mode 100644 video-gen-api/app/api/v1/material_consumption.py create mode 100644 video-gen-api/app/api/v1/open_type.py create mode 100644 video-gen-api/app/schemas/open_type.py create mode 100644 video-gen-api/app/services/material_consumption_queue.py create mode 100644 video-gen-api/app/tasks/material_consumption_task.py diff --git a/video-gen-api/app/api/v1/__init__.py b/video-gen-api/app/api/v1/__init__.py index 6833e4a7..06f60775 100644 --- a/video-gen-api/app/api/v1/__init__.py +++ b/video-gen-api/app/api/v1/__init__.py @@ -22,6 +22,8 @@ from app.api.v1.user_oauth import router as user_oauth_router from app.api.v1.user_oauth_app import router as user_oauth_app_router from app.api.v1.upload_material import router as upload_material_router from app.api.v1.pre_test_template import router as pre_test_template_router +from app.api.v1.material_consumption import router as material_consumption_router +from app.api.v1.open_type import router as open_type_router from app.api.admin import router as admin_module_router api_router = APIRouter() @@ -47,4 +49,6 @@ api_router.include_router(user_oauth_router) api_router.include_router(user_oauth_app_router) api_router.include_router(upload_material_router) api_router.include_router(pre_test_template_router) +api_router.include_router(material_consumption_router) +api_router.include_router(open_type_router) api_router.include_router(admin_module_router) diff --git a/video-gen-api/app/api/v1/material_consumption.py b/video-gen-api/app/api/v1/material_consumption.py new file mode 100644 index 00000000..3db0c85c --- /dev/null +++ b/video-gen-api/app/api/v1/material_consumption.py @@ -0,0 +1,301 @@ +from typing import Any, Optional +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.material_cost import MaterialCost +from app.models.user_oauth import UserOAuth +from app.models.user_oauth_account import UserOAuthAccount +from app.dependencies import get_current_user, get_db +from app.services.material_consumption_queue import sync_all_advertisers_consumption, _fetch_and_save_consumption + +router = APIRouter(prefix="/material-consumption", tags=["material-consumption"]) + + +@router.get( + "/list", + summary="查询素材消耗列表", + description="查询当前用户的素材消耗列表", +) +async def get_consumption_list( + page: int = Query(1, ge=1, description="页码"), + page_size: int = Query(20, ge=1, le=100, description="每页数量"), + advertiser_id: Optional[str] = Query(None, description="广告主ID筛选"), + start_date: Optional[str] = Query(None, description="开始日期"), + end_date: Optional[str] = Query(None, description="结束日期"), + current_user: User = Depends(get_current_user), + db: AsyncSession = Depends(get_db), +) -> Any | dict: + offset = (page - 1) * page_size + + query = ( + select(MaterialCost) + .join( + UserOAuth, + MaterialCost.oauth_id == UserOAuth.id, + ) + .where( + UserOAuth.user_id == current_user.id, + MaterialCost.deleted_at.is_(None), + UserOAuth.deleted_at.is_(None), + ) + ) + + if advertiser_id: + query = query.where(MaterialCost.advertiser_id == advertiser_id) + + if start_date: + start_date_obj = datetime.strptime(start_date, "%Y-%m-%d").date() + query = query.where(MaterialCost.consume_date >= start_date_obj) + + if end_date: + end_date_obj = datetime.strptime(end_date, "%Y-%m-%d").date() + query = query.where(MaterialCost.consume_date <= end_date_obj) + + result = await db.execute( + query.order_by(MaterialCost.consume_date.desc()) + .offset(offset) + .limit(page_size) + ) + consumptions = result.scalars().all() + + count_query = ( + select(func.count(MaterialCost.id)) + .join( + UserOAuth, + MaterialCost.oauth_id == UserOAuth.id, + ) + .where( + UserOAuth.user_id == current_user.id, + MaterialCost.deleted_at.is_(None), + UserOAuth.deleted_at.is_(None), + ) + ) + + if advertiser_id: + count_query = count_query.where(MaterialCost.advertiser_id == advertiser_id) + + if start_date: + start_date_obj = datetime.strptime(start_date, "%Y-%m-%d").date() + count_query = count_query.where(MaterialCost.consume_date >= start_date_obj) + + if end_date: + end_date_obj = datetime.strptime(end_date, "%Y-%m-%d").date() + count_query = count_query.where(MaterialCost.consume_date <= end_date_obj) + + total_result = await db.execute(count_query) + total = total_result.scalar_one() + + return { + "code": 0, + "data": [ + { + "id": consumption.id, + "advertiser_id": consumption.advertiser_id, + "material_id": consumption.material_id, + "consume_date": consumption.consume_date.isoformat() if consumption.consume_date else None, + "stat_cost": consumption.stat_cost, + "show_cnt": consumption.show_cnt, + "cpm_platform": consumption.cpm_platform, + "click_cnt": consumption.click_cnt, + "ctr": consumption.ctr, + "cpc_platform": consumption.cpc_platform, + "convert_cnt": consumption.convert_cnt, + "conversion_cost": consumption.conversion_cost, + "conversion_rate": consumption.conversion_rate, + "deep_convert_cnt": consumption.deep_convert_cnt, + "deep_convert_cost": consumption.deep_convert_cost, + "deep_convert_rate": consumption.deep_convert_rate, + "active": consumption.active, + "active_cost": consumption.active_cost, + "active_rate": consumption.active_rate, + "active_register": consumption.active_register, + "active_register_cost": consumption.active_register_cost, + "active_register_rate": consumption.active_register_rate, + "attribution_next_day_open_cnt": consumption.attribution_next_day_open_cnt, + "attribution_next_day_open_cost": consumption.attribution_next_day_open_cost, + "attribution_next_day_open_rate": consumption.attribution_next_day_open_rate, + "active_pay": consumption.active_pay, + "active_pay_cost": consumption.active_pay_cost, + "active_pay_rate": consumption.active_pay_rate, + "phone": consumption.phone, + "form": consumption.form, + "download_start": consumption.download_start, + "form_submit": consumption.form_submit, + "button": consumption.button, + "view": consumption.view, + "message": consumption.message, + "consult": consumption.consult, + "consult_effective": consumption.consult_effective, + "shopping": consumption.shopping, + "customer_effective": consumption.customer_effective, + "attribution_game_in_app_ltv_1day": consumption.attribution_game_in_app_ltv_1day, + "attribution_game_in_app_roi_1day": consumption.attribution_game_in_app_roi_1day, + "loan_completion": consumption.loan_completion, + "loan_completion_cost": consumption.loan_completion_cost, + "loan_completion_rate": consumption.loan_completion_rate, + "loan_credit": consumption.loan_credit, + "loan_credit_cost": consumption.loan_credit_cost, + "loan_credit_rate": consumption.loan_credit_rate, + "in_app_order_gmv": consumption.in_app_order_gmv, + "in_app_order_roi": consumption.in_app_order_roi, + "in_app_pay_gmv": consumption.in_app_pay_gmv, + "in_app_pay_roi": consumption.in_app_pay_roi, + "total_play": consumption.total_play, + "valid_play": consumption.valid_play, + "valid_play_cost": consumption.valid_play_cost, + "valid_play_rate": consumption.valid_play_rate, + "valid_play_of_mille": consumption.valid_play_of_mille, + "valid_play_cost_of_mille": consumption.valid_play_cost_of_mille, + "average_play_time_per_play": consumption.average_play_time_per_play, + "play_over_rate": consumption.play_over_rate, + "dy_like": consumption.dy_like, + "dy_comment": consumption.dy_comment, + "dy_share": consumption.dy_share, + "report_cnt": consumption.report_cnt, + "created_at": consumption.created_at, + } + for consumption in consumptions + ], + "pagination": { + "page": page, + "page_size": page_size, + "total": total, + }, + } + + +@router.get( + "/sync", + summary="手动同步素材消耗", + description="手动触发素材消耗同步任务，加入队列顺序执行", +) +async def sync_consumption( + date: str = Query(None, description="同步日期，默认为昨天"), + current_user: User = Depends(get_current_user), + db: AsyncSession = Depends(get_db), +) -> Any | dict: + if date is None: + date = (datetime.now(timezone.utc) - timedelta(days=1)).strftime("%Y-%m-%d") + + # 测试用，手动触发同步素材消耗任务 + result = await _fetch_and_save_consumption("0019eb9f130027c05b8", "1863675913228435", date) + return { + "code": 0, + "message": result, + } + result = await sync_all_advertisers_consumption(date) + + return { + "code": 0, + "message": result["message"], + } + + +@router.get( + "/fields", + summary="获取素材消耗字段描述", + description="返回素材消耗表每个字段对应的中文描述，帮助前端理解字段含义", +) +async def get_consumption_fields() -> Any | dict: + fields = [ + {"field": "id", "description": "主键"}, + {"field": "advertiser_id", "description": "广告主id"}, + {"field": "material_id", "description": "素材id"}, + {"field": "consume_date", "description": "消耗日期"}, + {"field": "stat_cost", "description": "消耗金额"}, + {"field": "show_cnt", "description": "展示数"}, + {"field": "cpm_platform", "description": "平均千次展现费用(元)"}, + {"field": "click_cnt", "description": "点击数"}, + {"field": "ctr", "description": "点击率"}, + {"field": "cpc_platform", "description": "平均点击单价(元)"}, + {"field": "convert_cnt", "description": "转化数"}, + {"field": "conversion_cost", "description": "平均转化成本(元)"}, + {"field": "conversion_rate", "description": "转化率"}, + {"field": "deep_convert_cnt", "description": "深度转化数"}, + {"field": "deep_convert_cost", "description": "深度转化成本(元)"}, + {"field": "deep_convert_rate", "description": "深度转化率"}, + {"field": "active", "description": "激活数"}, + {"field": "active_cost", "description": "激活成本(元)"}, + {"field": "active_rate", "description": "激活率"}, + {"field": "active_register", "description": "注册数"}, + {"field": "active_register_cost", "description": "注册成本(元)"}, + {"field": "active_register_rate", "description": "注册率"}, + {"field": "attribution_next_day_open_cnt", "description": "次留数"}, + {"field": "attribution_next_day_open_cost", "description": "次留成本"}, + {"field": "attribution_next_day_open_rate", "description": "次留率"}, + {"field": "active_pay", "description": "首次付费数"}, + {"field": "active_pay_cost", "description": "首次付费成本(元)"}, + {"field": "active_pay_rate", "description": "首次付费率"}, + {"field": "phone", "description": "点击电话按钮"}, + {"field": "form", "description": "用户在门店落地页多线沟通提交表单的次数"}, + {"field": "download_start", "description": "用户点击下载开始的次数"}, + {"field": "form_submit", "description": "用户查看附加创意后，提交表单的次数"}, + {"field": "button", "description": "用户点击按钮button的次数"}, + {"field": "view", "description": "用户在关键页面的浏览次数"}, + {"field": "message", "description": "用户点击短信咨询的次数"}, + {"field": "consult", "description": "用户点击在线咨询按钮的次数"}, + {"field": "consult_effective", "description": "用户在门店落地页多线沟通的在线咨询中有效咨询的次数"}, + {"field": "shopping", "description": "用户购买商品的次数"}, + {"field": "customer_effective", "description": "有效获客"}, + {"field": "attribution_game_in_app_ltv_1day", "description": "当日付费金额"}, + {"field": "attribution_game_in_app_roi_1day", "description": "当日付费ROI"}, + {"field": "loan_completion", "description": "完件数(互联网金融-贷款行业中，用户成功提交贷款额度申请的行为)"}, + {"field": "loan_completion_cost", "description": "完件成本(元)"}, + {"field": "loan_completion_rate", "description": "完件率"}, + {"field": "loan_credit", "description": "授信数(互联网金融-贷款行业中，用户提交贷款额度申请后，客户审批通过，给予用户可贷款的额度)"}, + {"field": "loan_credit_cost", "description": "授信成本(元)"}, + {"field": "loan_credit_rate", "description": "授信率"}, + {"field": "in_app_order_gmv", "description": "引流电商订单GMV(当您使用\"in_app_order\"事件回传订单金额时，对应的GMV金额)"}, + {"field": "in_app_order_roi", "description": "引流电商订单ROI"}, + {"field": "in_app_pay_gmv", "description": "引流电商支付GMV"}, + {"field": "in_app_pay_roi", "description": "引流电商支付ROI"}, + {"field": "total_play", "description": "播放量(播放时间大于0S的数量，在某些蜂窝网络环境下，需要您手动点击开始才会开始播放，因此有时播放数小于展示数)"}, + {"field": "valid_play", "description": "有效播放数"}, + {"field": "valid_play_cost", "description": "有效播放成本(元)"}, + {"field": "valid_play_rate", "description": "有效播放率"}, + {"field": "valid_play_of_mille", "description": "千次有效播放数"}, + {"field": "valid_play_cost_of_mille", "description": "千次有效播放成本(元)"}, + {"field": "average_play_time_per_play", "description": "平均单次播放时长"}, + {"field": "play_over_rate", "description": "完播率"}, + {"field": "dy_like", "description": "点赞数"}, + {"field": "dy_comment", "description": "评论量"}, + {"field": "dy_share", "description": "分享量"}, + {"field": "report_cnt", "description": "举报数"}, + {"field": "created_at", "description": "创建时间"}, + ] + + return { + "code": 0, + "data": fields, + } + + +@router.get( + "/advertisers", + summary="获取用户的广告主列表", + description="获取当前用户有权限的广告主列表，用于筛选", +) +async def get_advertisers( + current_user: User = Depends(get_current_user), + db: AsyncSession = Depends(get_db), +) -> Any | dict: + result = await db.execute( + select(UserOAuthAccount.advertiser_id) + .join(UserOAuth, UserOAuthAccount.oauth_id == UserOAuth.id) + .where( + UserOAuth.user_id == current_user.id, + UserOAuth.deleted_at.is_(None), + UserOAuthAccount.deleted_at.is_(None), + ) + .distinct() + ) + advertisers = result.scalars().all() + + return { + "code": 0, + "data": [{"advertiser_id": str(aid)} for aid in advertisers], + } \ No newline at end of file diff --git a/video-gen-api/app/api/v1/open_type.py b/video-gen-api/app/api/v1/open_type.py new file mode 100644 index 00000000..7d134843 --- /dev/null +++ b/video-gen-api/app/api/v1/open_type.py @@ -0,0 +1,262 @@ +from typing import Any, Optional + +from fastapi import APIRouter, Body, Depends, HTTPException, Query, status +from sqlalchemy import select, func +from sqlalchemy.ext.asyncio import AsyncSession + +from app.dependencies import get_admin_user, get_db +from app.models.user import User +from app.models.open_type import OpenType +from app.schemas.open_type import ( + OpenTypeCreate, + OpenTypeUpdate, + OpenTypeListResponse, + OpenTypeResponse, +) +from app.utils.id_gen import generate_id + +router = APIRouter(prefix="/open-type", tags=["open-type"]) + + +@router.get( + "/list", + summary="获取开户方式列表", + description="获取开户方式列表，支持分页和筛选", + response_model=OpenTypeListResponse, +) +async def get_open_type_list( + page: int = Query(1, ge=1, description="页码"), + page_size: int = Query(20, ge=1, le=100, description="每页数量"), + type_name: Optional[str] = Query(None, description="标题名称筛选"), + open_type: Optional[int] = Query(None, description="开户方式id筛选"), + admin: User = Depends(get_admin_user), + db: AsyncSession = Depends(get_db), +) -> Any | dict: + offset = (page - 1) * page_size + + query = select(OpenType).where(OpenType.deleted_at.is_(None)) + + if type_name: + query = query.where(OpenType.type_name.like(f"%{type_name}%")) + + if open_type: + query = query.where(OpenType.open_type == open_type) + + result = await db.execute( + query.order_by(OpenType.open_type.asc()) + .offset(offset) + .limit(page_size) + ) + open_types = result.scalars().all() + + count_query = select(func.count(OpenType.id)).where(OpenType.deleted_at.is_(None)) + if type_name: + count_query = count_query.where(OpenType.type_name.like(f"%{type_name}%")) + if open_type: + count_query = count_query.where(OpenType.open_type == open_type) + + total_result = await db.execute(count_query) + total = total_result.scalar_one() + + return { + "code": 0, + "message": "查询成功", + "data": [ + { + "id": item.id, + "type_name": item.type_name, + "open_type": item.open_type, + "description": item.description, + "thumb": item.thumb, + "created_at": item.created_at, + "updated_at": item.updated_at, + } + for item in open_types + ], + "pagination": { + "page": page, + "page_size": page_size, + "total": total, + "total_pages": (total + page_size - 1) // page_size, + }, + } + + +@router.get( + "/{id}", + summary="获取开户方式详情", + description="根据ID获取开户方式详情", + response_model=OpenTypeResponse, +) +async def get_open_type_detail( + id: str, + admin: User = Depends(get_admin_user), + db: AsyncSession = Depends(get_db), +) -> Any | dict: + result = await db.execute( + select(OpenType).where(OpenType.id == id, OpenType.deleted_at.is_(None)) + ) + open_type = result.scalar_one_or_none() + + if not open_type: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail="开户方式不存在", + ) + + return { + "code": 0, + "message": "查询成功", + "data": { + "id": open_type.id, + "type_name": open_type.type_name, + "open_type": open_type.open_type, + "description": open_type.description, + "thumb": open_type.thumb, + "created_at": open_type.created_at, + "updated_at": open_type.updated_at, + }, + } + + +@router.post( + "/", + summary="创建开户方式", + description="创建新的开户方式", + response_model=OpenTypeResponse, +) +async def create_open_type( + req: OpenTypeCreate = Body(...), + admin: User = Depends(get_admin_user), + db: AsyncSession = Depends(get_db), +) -> Any | dict: + existing = await db.execute( + select(OpenType).where(OpenType.open_type == req.open_type, OpenType.deleted_at.is_(None)) + ) + if existing.scalar_one_or_none(): + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail=f"开户方式ID {req.open_type} 已存在", + ) + + open_type = OpenType( + id=generate_id(), + type_name=req.type_name, + open_type=req.open_type, + description=req.description, + thumb=req.thumb, + ) + + db.add(open_type) + await db.commit() + await db.refresh(open_type) + + return { + "code": 0, + "message": "创建成功", + "data": { + "id": open_type.id, + "type_name": open_type.type_name, + "open_type": open_type.open_type, + "description": open_type.description, + "thumb": open_type.thumb, + "created_at": open_type.created_at, + "updated_at": open_type.updated_at, + }, + } + + +@router.put( + "/{id}", + summary="更新开户方式", + description="更新指定的开户方式", + response_model=OpenTypeResponse, +) +async def update_open_type( + id: str, + req: OpenTypeUpdate = Body(...), + admin: User = Depends(get_admin_user), + db: AsyncSession = Depends(get_db), +) -> Any | dict: + result = await db.execute( + select(OpenType).where(OpenType.id == id, OpenType.deleted_at.is_(None)) + ) + open_type = result.scalar_one_or_none() + + if not open_type: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail="开户方式不存在", + ) + + if req.open_type is not None and req.open_type != open_type.open_type: + existing = await db.execute( + select(OpenType).where( + OpenType.open_type == req.open_type, + OpenType.id != id, + OpenType.deleted_at.is_(None), + ) + ) + if existing.scalar_one_or_none(): + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail=f"开户方式ID {req.open_type} 已存在", + ) + + if req.type_name is not None: + open_type.type_name = req.type_name + if req.open_type is not None: + open_type.open_type = req.open_type + if req.description is not None: + open_type.description = req.description + if req.thumb is not None: + open_type.thumb = req.thumb + + await db.commit() + await db.refresh(open_type) + + return { + "code": 0, + "message": "更新成功", + "data": { + "id": open_type.id, + "type_name": open_type.type_name, + "open_type": open_type.open_type, + "description": open_type.description, + "thumb": open_type.thumb, + "created_at": open_type.created_at, + "updated_at": open_type.updated_at, + }, + } + + +@router.delete( + "/{id}", + summary="删除开户方式", + description="软删除指定的开户方式", + response_model=OpenTypeResponse, +) +async def delete_open_type( + id: str, + admin: User = Depends(get_admin_user), + db: AsyncSession = Depends(get_db), +) -> Any | dict: + result = await db.execute( + select(OpenType).where(OpenType.id == id, OpenType.deleted_at.is_(None)) + ) + open_type = result.scalar_one_or_none() + + if not open_type: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail="开户方式不存在", + ) + + open_type.deleted_at = func.now() + await db.commit() + + return { + "code": 0, + "message": "删除成功", + "data": None, + } \ No newline at end of file diff --git a/video-gen-api/app/api/v1/upload_material.py b/video-gen-api/app/api/v1/upload_material.py index 509eb757..30116a8d 100644 --- a/video-gen-api/app/api/v1/upload_material.py +++ b/video-gen-api/app/api/v1/upload_material.py @@ -222,6 +222,14 @@ async def async_batch_upload_material( current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ) -> Any | dict: + + # param = { + # "account_ids" : json.dumps([1836693172153543]), + # } + + # account_info = await DouyinApi().get_account_info("0019eb9f130027c05b8", param) + # return account_info + if not req.tasks: return { "code": 0, diff --git a/video-gen-api/app/main.py b/video-gen-api/app/main.py index 9fd46e74..c486f832 100644 --- a/video-gen-api/app/main.py +++ b/video-gen-api/app/main.py @@ -68,10 +68,13 @@ async def lifespan(app: FastAPI): await upload_queue.recover() upload_queue_task = asyncio.create_task(upload_queue.run()) - # 启动上传队列（异步处理素材上传） - from app.services.upload_queue import upload_queue - await upload_queue.recover() - upload_queue_task = asyncio.create_task(upload_queue.run()) + # 启动素材消耗队列 + from app.services.material_consumption_queue import material_consumption_queue + consumption_queue_task = asyncio.create_task(material_consumption_queue.run()) + + # 启动素材消耗计划任务（每天9点自动同步） + from app.tasks.material_consumption_task import schedule_daily_sync + consumption_schedule_task = asyncio.create_task(schedule_daily_sync()) # 启动时立即同步一次未支付订单 asyncio.create_task(asyncio.sleep(5)) # 等待5秒后再同步，让系统完全启动 @@ -97,10 +100,11 @@ async def lifespan(app: FastAPI): await queue_task upload_queue.stop() await upload_queue_task + material_consumption_queue.stop() + await consumption_queue_task + consumption_schedule_task.cancel() expiry_task.cancel() token_refresh_task.cancel() - upload_queue.stop() - await upload_queue_task await close_database() await close_redis() diff --git a/video-gen-api/app/schemas/open_type.py b/video-gen-api/app/schemas/open_type.py new file mode 100644 index 00000000..0553e432 --- /dev/null +++ b/video-gen-api/app/schemas/open_type.py @@ -0,0 +1,48 @@ +from datetime import datetime +from typing import List, Optional + +from pydantic import BaseModel, Field + + +class OpenTypeCreate(BaseModel): + type_name: str = Field(..., description="标题名称") + open_type: int = Field(..., description="开户方式id") + description: Optional[str] = Field(None, description="开户方式描述") + thumb: Optional[str] = Field(None, description="缩略图") + + +class OpenTypeUpdate(BaseModel): + type_name: Optional[str] = Field(None, description="标题名称") + open_type: Optional[int] = Field(None, description="开户方式id") + description: Optional[str] = Field(None, description="开户方式描述") + thumb: Optional[str] = Field(None, description="缩略图") + + +class OpenTypeOut(BaseModel): + id: str = Field(..., description="主键") + type_name: str = Field(..., description="标题名称") + open_type: int = Field(..., description="开户方式id") + description: Optional[str] = Field(None, description="开户方式描述") + thumb: Optional[str] = Field(None, description="缩略图") + created_at: datetime = Field(..., description="创建时间") + updated_at: datetime = Field(..., description="更新时间") + + +class PaginationInfo(BaseModel): + page: int = Field(..., description="当前页码") + page_size: int = Field(..., description="每页数量") + total: int = Field(..., description="总记录数") + total_pages: int = Field(..., description="总页数") + + +class OpenTypeListResponse(BaseModel): + code: int = Field(0, description="返回码，0表示成功") + message: str = Field("查询成功", description="返回消息") + data: List[OpenTypeOut] = Field(..., description="开户方式列表") + pagination: PaginationInfo = Field(..., description="分页信息") + + +class OpenTypeResponse(BaseModel): + code: int = Field(0, description="返回码，0表示成功") + message: str = Field("操作成功", description="返回消息") + data: Optional[OpenTypeOut] = Field(None, description="开户方式信息") \ No newline at end of file diff --git a/video-gen-api/app/services/material_consumption_queue.py b/video-gen-api/app/services/material_consumption_queue.py new file mode 100644 index 00000000..1028d7ad --- /dev/null +++ b/video-gen-api/app/services/material_consumption_queue.py @@ -0,0 +1,334 @@ +import asyncio +import logging +from datetime import datetime, timezone, timedelta + +from sqlalchemy import select, delete +from sqlalchemy.ext.asyncio import AsyncSession + +from app.models.base import async_session +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.resources_material import ResourcesMaterial +from app.utils.id_gen import generate_id +from app.utils.douyinApi import DouyinApi + +import os +import json + +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("material_consumption_task") +logger.setLevel(logging.INFO) + + +class DailyRotatingFileHandler(logging.FileHandler): + def __init__(self, directory, encoding=None): + self.directory = directory + filename = self._get_log_filename() + super().__init__(filename, encoding=encoding) + + def _get_log_filename(self): + return os.path.join(self.directory, f"material_consumption-{datetime.now(timezone.utc).strftime('%Y-%m-%d')}.log") + + def emit(self, record): + current_filename = self._get_log_filename() + if self.baseFilename != current_filename: + self.close() + self.baseFilename = current_filename + self.stream = self._open() + super().emit(record) + + +if not logger.handlers: + 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) + +douyin_api = DouyinApi() + + +def _convert_numeric(value): + """Convert string numeric values to appropriate Python types.""" + if value is None or value == "": + return None + try: + # Try int first + return int(value) + except ValueError: + try: + # Then try float + return float(value) + except ValueError: + # Return original value if conversion fails + return None + + +class MaterialConsumptionQueue: + def __init__(self): + self.queue: asyncio.Queue[dict] = asyncio.Queue() + self.running = False + + async def enqueue(self, task_data: dict): + """Add a task to the queue.""" + await self.queue.put(task_data) + + async def run(self): + """Main processing loop.""" + self.running = True + logger.info("Material consumption queue started") + + while self.running: + try: + task_data = await asyncio.wait_for(self.queue.get(), timeout=5.0) + except asyncio.TimeoutError: + continue + + try: + await self._process(task_data) + except Exception as e: + logger.error(f"Error processing material consumption task: {e}") + finally: + self.queue.task_done() + + logger.info("Material consumption queue stopped") + + async def _process(self, task_data: dict): + """Process a single consumption update task.""" + oauth_id = task_data.get("oauth_id") + advertiser_id = task_data.get("advertiser_id") + date = task_data.get("date") + + logger.info(f"Processing consumption update for oauth_id={oauth_id}, advertiser_id={advertiser_id}, date={date}") + + try: + result = await _fetch_and_save_consumption(oauth_id, advertiser_id, date) + if result.get("success"): + logger.info(f"Successfully updated consumption for advertiser {advertiser_id} on {date}") + else: + logger.error(f"Failed to update consumption for advertiser {advertiser_id} on {date}: {result.get('error')}") + except Exception as e: + logger.error(f"Exception processing consumption for advertiser {advertiser_id} on {date}: {e}") + + def stop(self): + """Stop the queue.""" + self.running = False + + +async def _fetch_and_save_consumption(oauth_id: str, advertiser_id: str, date: str) -> dict: + async with async_session() as db: + try: + consume_date_obj = datetime.strptime(date, "%Y-%m-%d").date() + + # 查询当前授权下系统上传的素材ID列表 + material_result = await db.execute( + select(ResourcesMaterial.upload_id) + .where( + ResourcesMaterial.oauth_id == oauth_id, + ResourcesMaterial.advertiser_id == advertiser_id, + ResourcesMaterial.deleted_at.is_(None), + ResourcesMaterial.upload_id.isnot(None), + ResourcesMaterial.upload_id != "", + ) + ) + system_material_ids = [str(row[0]) for row in material_result.all()] + + if not system_material_ids: + return { + "success": True, + "message": "没有系统上传的素材" + } + + #强制删除旧数据,真实删除，不要软删除 + await db.execute( + delete(MaterialCost).where( + MaterialCost.oauth_id == oauth_id, + MaterialCost.advertiser_id == advertiser_id, + MaterialCost.consume_date == consume_date_obj, + MaterialCost.deleted_at.is_(None), + ) + ) + + BATCH_SIZE = 50 + total_saved = 0 + total_requests = 0 + + # 分批请求，每次请求50个素材 + for i in range(0, len(system_material_ids), BATCH_SIZE): + batch_materials = system_material_ids[i:i+BATCH_SIZE] + total_requests += 1 + + params = { + "dimensions": json.dumps(["ad_platform_material_name","image_mode","material_id","stat_time_day"]), + "advertiser_id": int(advertiser_id), + "metrics": json.dumps([ + "stat_cost","show_cnt","cpm_platform","click_cnt","ctr","cpc_platform","convert_cnt","conversion_cost","conversion_rate","deep_convert_cnt","deep_convert_cost","deep_convert_rate","click_start_cnt","click_start_cost","download_finish_rate", + "install_finish_cnt","install_finish_cost","install_finish_rate","active","active_cost", + "active_rate","active_register","active_register_cost","active_register_rate","game_addiction", + "game_addiction_cost","game_addiction_rate","attribution_next_day_open_cnt","attribution_next_day_open_cost","attribution_next_day_open_rate","next_day_open","active_pay","active_pay_cost","active_pay_rate", + "game_pay_count","game_pay_cost","in_app_uv","in_app_detail_uv","in_app_cart","in_app_pay", + "in_app_order","attribution_billing_game_in_app_ltv_1day","attribution_billing_game_in_app_roi_1day","phone","form", + "form_submit","map","button","view","download_start","qq","vote","lottery","message", + "redirect","shopping","consult","consult_effective","phone_confirm","phone_connect","phone_effective", + "redirect_to_shop","coupon_single_page","poi_address_click","poi_collect","customer_effective", + "attribution_customer_effective","attribution_customer_effective_cost","attribution_clue_pay_succeed", + "attribution_clue_pay_succeed_cost","attribution_clue_interflow","attribution_clue_interflow_cost","attribution_clue_high_intention","attribution_clue_high_intention_cost", + "consult_clue","clue_message_count","attribution_work_wechat_unfriend_count","attribution_clue_connected_count","attribution_clue_connected_cost","attribution_clue_connected_rate","attribution_form","form_and_submit_count","form_and_submit_cost","intention_form_and_submit_count","intention_form_and_submit_cost","attribution_micro_game_0d_ltv","attribution_micro_game_3d_ltv","attribution_micro_game_7d_ltv","attribution_micro_game_0d_roi","attribution_micro_game_3d_roi","attribution_micro_game_7d_roi","attribution_game_in_app_ltv_1day","attribution_game_in_app_roi_1day","active_pay_intra_day_count","active_pay_intra_day_cost","active_pay_intra_day_rate","first_pay_intra_24hour_amount","loan_completion","loan_completion_cost","loan_completion_rate","pre_loan_credit","pre_loan_credit_cost","loan_credit_cost","loan_credit","loan_credit_rate","in_wechat_pay_count","unfollow_in_wechat_count","loan","loan_cost","loan_rate","open_account_count","withdraw_m2_count","in_app_order_gmv","in_app_order_roi","in_app_pay_gmv","in_app_pay_roi","first_rental_order_count","commute_first_pay_count","first_order_count","submit_certification_count","approval_count","total_play","play_duration_3s","valid_play","valid_play_cost","valid_play_rate","valid_play_of_mille","valid_play_cost_of_mille","average_play_time_per_play","play_over_rate","dy_like","dy_comment","dy_share","report_cnt","location_click","dislike_cnt","dy_home_visited","dy_follow","message_action","click_landing_page","click_shopwindow","click_website","click_call_dy","click_download","luban_live_enter_cnt","luban_live_follow_cnt","luban_live_share_cnt","luban_live_comment_cnt","luban_live_gift_cnt","luban_live_gift_amount","click_call_cnt","click_counsel","a3_ask_count","a3_ask_cost" + ]), + "filters": json.dumps([ + { + "field": "stat_cost", + "type": 3, + "operator": 4, + "values": ["0"] + }, + { + "field": "material_id", + "type": 2, + "operator": 7, + "values": batch_materials + } + ]), + "start_time": date, + "end_time": date, + "order_by": json.dumps([ + { + "field": "stat_cost", + "type": "ASC" + } + ]) + } + + response = await douyin_api.get_material_cost(oauth_id, params, request_count=3) + + if response.get("code") != 0: + logger.warning(f"获取素材消耗失败(批次 {i//BATCH_SIZE + 1}): {response.get('message', '未知错误')}") + continue + + data = response.get("data", {}) + list_data = data.get("rows", []) + + for item in list_data: + dimensions = item.get("dimensions", {}) + metrics = item.get("metrics", {}) + material_id = dimensions.get("material_id", "") + + cost = MaterialCost( + id=generate_id(), + oauth_id=oauth_id, + advertiser_id=str(advertiser_id), + material_id=material_id, + consume_date=consume_date_obj, + stat_cost=_convert_numeric(metrics.get("stat_cost")), + show_cnt=_convert_numeric(metrics.get("show_cnt")), + cpm_platform=_convert_numeric(metrics.get("cpm_platform")), + click_cnt=_convert_numeric(metrics.get("click_cnt")), + ctr=_convert_numeric(metrics.get("ctr")), + cpc_platform=_convert_numeric(metrics.get("cpc_platform")), + convert_cnt=_convert_numeric(metrics.get("convert_cnt")), + conversion_cost=_convert_numeric(metrics.get("conversion_cost")), + conversion_rate=_convert_numeric(metrics.get("conversion_rate")), + deep_convert_cnt=_convert_numeric(metrics.get("deep_convert_cnt")), + deep_convert_cost=_convert_numeric(metrics.get("deep_convert_cost")), + deep_convert_rate=_convert_numeric(metrics.get("deep_convert_rate")), + active=_convert_numeric(metrics.get("active")), + active_cost=_convert_numeric(metrics.get("active_cost")), + active_rate=_convert_numeric(metrics.get("active_rate")), + active_register=_convert_numeric(metrics.get("active_register")), + active_register_cost=_convert_numeric(metrics.get("active_register_cost")), + active_register_rate=_convert_numeric(metrics.get("active_register_rate")), + attribution_next_day_open_cnt=_convert_numeric(metrics.get("attribution_next_day_open_cnt")), + attribution_next_day_open_cost=_convert_numeric(metrics.get("attribution_next_day_open_cost")), + attribution_next_day_open_rate=_convert_numeric(metrics.get("attribution_next_day_open_rate")), + active_pay=_convert_numeric(metrics.get("active_pay")), + active_pay_cost=_convert_numeric(metrics.get("active_pay_cost")), + active_pay_rate=_convert_numeric(metrics.get("active_pay_rate")), + phone=_convert_numeric(metrics.get("phone")), + form=_convert_numeric(metrics.get("form")), + download_start=_convert_numeric(metrics.get("download_start")), + form_submit=_convert_numeric(metrics.get("form_submit")), + button=_convert_numeric(metrics.get("button")), + view=_convert_numeric(metrics.get("view")), + message=_convert_numeric(metrics.get("message")), + consult=_convert_numeric(metrics.get("consult")), + consult_effective=_convert_numeric(metrics.get("consult_effective")), + shopping=_convert_numeric(metrics.get("shopping")), + customer_effective=_convert_numeric(metrics.get("customer_effective")), + attribution_game_in_app_ltv_1day=_convert_numeric(metrics.get("attribution_game_in_app_ltv_1day")), + attribution_game_in_app_roi_1day=_convert_numeric(metrics.get("attribution_game_in_app_roi_1day")), + loan_completion=_convert_numeric(metrics.get("loan_completion")), + loan_completion_cost=_convert_numeric(metrics.get("loan_completion_cost")), + loan_completion_rate=_convert_numeric(metrics.get("loan_completion_rate")), + loan_credit=_convert_numeric(metrics.get("loan_credit")), + loan_credit_cost=_convert_numeric(metrics.get("loan_credit_cost")), + loan_credit_rate=_convert_numeric(metrics.get("loan_credit_rate")), + in_app_order_gmv=_convert_numeric(metrics.get("in_app_order_gmv")), + in_app_order_roi=_convert_numeric(metrics.get("in_app_order_roi")), + in_app_pay_gmv=_convert_numeric(metrics.get("in_app_pay_gmv")), + in_app_pay_roi=_convert_numeric(metrics.get("in_app_pay_roi")), + total_play=_convert_numeric(metrics.get("total_play")), + valid_play=_convert_numeric(metrics.get("valid_play")), + valid_play_cost=_convert_numeric(metrics.get("valid_play_cost")), + valid_play_rate=_convert_numeric(metrics.get("valid_play_rate")), + valid_play_of_mille=_convert_numeric(metrics.get("valid_play_of_mille")), + valid_play_cost_of_mille=_convert_numeric(metrics.get("valid_play_cost_of_mille")), + average_play_time_per_play=_convert_numeric(metrics.get("average_play_time_per_play")), + play_over_rate=_convert_numeric(metrics.get("play_over_rate")), + dy_like=_convert_numeric(metrics.get("dy_like")), + dy_comment=_convert_numeric(metrics.get("dy_comment")), + dy_share=_convert_numeric(metrics.get("dy_share")), + report_cnt=_convert_numeric(metrics.get("report_cnt")), + ) + db.add(cost) + total_saved += 1 + + await db.commit() + + return { + "success": True, + "message": f"成功更新 {total_saved} 条记录，共请求 {total_requests} 批次，系统素材总数 {len(system_material_ids)}", + "count": total_saved + } + + except Exception as e: + await db.rollback() + return { + "success": False, + "error": str(e) + } + +#异步获取所有要更新的广告主，素材消耗 +async def sync_all_advertisers_consumption(date: str = None): + """Sync consumption data for all advertisers.""" + if date is None: + date = (datetime.now(timezone.utc) - timedelta(days=1)).strftime("%Y-%m-%d") + + async with async_session() as db: + result = await db.execute( + select(UserOAuth.id, UserOAuthAccount.advertiser_id) + .join(UserOAuthAccount, UserOAuth.id == UserOAuthAccount.oauth_id) + .join(ResourcesMaterial, UserOAuthAccount.oauth_id == ResourcesMaterial.oauth_id) + .where( + UserOAuth.deleted_at.is_(None), + UserOAuthAccount.deleted_at.is_(None), + ResourcesMaterial.deleted_at.is_(None), + ResourcesMaterial.material_id.isnot(None), + ResourcesMaterial.material_id != "", + ) + .distinct() + ) + oauth_advertiser_pairs = result.all() + + for oauth_id, advertiser_id in oauth_advertiser_pairs: + await material_consumption_queue.enqueue({ + "oauth_id": oauth_id, + "advertiser_id": advertiser_id, + "date": date, + }) + logger.info(f"Enqueued consumption sync for oauth_id={oauth_id}, advertiser_id={advertiser_id}, date={date}") + + return {"message": f"已将 {len(oauth_advertiser_pairs)} 个广告主的消耗更新任务加入队列"} + + +material_consumption_queue = MaterialConsumptionQueue() \ No newline at end of file diff --git a/video-gen-api/app/services/upload_queue.py b/video-gen-api/app/services/upload_queue.py index 145bd5f2..267b1ebf 100644 --- a/video-gen-api/app/services/upload_queue.py +++ b/video-gen-api/app/services/upload_queue.py @@ -10,11 +10,13 @@ from app.models.upload_task import UploadTask from app.models.generated_resource import GeneratedResource from app.models.user_oauth import UserOAuth from app.models.resources_material import ResourcesMaterial +from app.models.user_oauth_account import UserOAuthAccount from app.utils.id_gen import generate_id from app.utils.douyinApi import DouyinApi import os import hashlib +import json LOG_DIR = os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(__file__))), "logs") @@ -133,6 +135,51 @@ class UploadQueue: note=result.get("message", "上传成功"), ) ) + + #请求巨量接口获取账户信息，如果存在则更新，否则插入新记录 + param = { + "account_ids" : json.dumps([int(task.advertiser_id)]), + } + try: + account_info = await douyin_api.get_account_info(task.oauth_id, param) + + code = account_info.get("code", 0) + try: + code = int(code) + except (ValueError, TypeError): + code = -1 + + if code != 0: + logger.error(f"Error getting account info: {json.dumps(account_info)}") + else: + existing_account = await db.execute( + select(UserOAuthAccount).where( + UserOAuthAccount.oauth_id == task.oauth_id, + UserOAuthAccount.advertiser_id == task.advertiser_id, + ) + ) + existing_account = existing_account.scalar_one_or_none() + + if existing_account: + await db.execute( + update(UserOAuthAccount).where( + UserOAuthAccount.id == existing_account.id, + ).values( + deleted_at=None, + advertiser_name= account_info.get("data", {}).get("account_detail_list", [{}])[0].get("advertiser_name", ""), + advertiser_role= "", + ) + ) + else: + db.add(UserOAuthAccount( + id=generate_id(), + oauth_id=task.oauth_id, + advertiser_id=task.advertiser_id, + advertiser_name= account_info.get("data", {}).get("account_detail_list", [{}])[0].get("advertiser_name", ""), + advertiser_role= "", + )) + except Exception as e: + logger.error(f"Exception getting account info: {e}") else: await db.execute( update(UploadTask).where(UploadTask.id == task_id).values( diff --git a/video-gen-api/app/tasks/material_consumption_task.py b/video-gen-api/app/tasks/material_consumption_task.py new file mode 100644 index 00000000..28dd1580 --- /dev/null +++ b/video-gen-api/app/tasks/material_consumption_task.py @@ -0,0 +1,68 @@ +from datetime import datetime, timedelta, timezone +import asyncio +import os +import logging + +from app.services.material_consumption_queue import sync_all_advertisers_consumption + +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("material_consumption_task") +logger.setLevel(logging.INFO) + + +class DailyRotatingFileHandler(logging.FileHandler): + def __init__(self, directory, encoding=None): + self.directory = directory + filename = self._get_log_filename() + super().__init__(filename, encoding=encoding) + + def _get_log_filename(self): + return os.path.join(self.directory, f"material_consumption_task-{datetime.now(timezone.utc).strftime('%Y-%m-%d')}.log") + + def emit(self, record): + current_filename = self._get_log_filename() + if self.baseFilename != current_filename: + self.close() + self.baseFilename = current_filename + self.stream = self._open() + super().emit(record) + + +if not logger.handlers: + 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) + + +async def schedule_daily_sync(): + """每天9点自动同步素材消耗数据""" + logger.info("Daily material consumption sync task started") + + while True: + try: + now = datetime.now(timezone.utc) + target_time = now.replace(hour=9, minute=0, second=0, microsecond=0) + + if now >= target_time: + target_time += timedelta(days=1) + + wait_seconds = (target_time - now).total_seconds() + logger.info(f"Waiting {wait_seconds/3600:.1f} hours until next sync at {target_time}") + + await asyncio.sleep(wait_seconds) + + logger.info("Starting daily material consumption sync...") + try: + result = await sync_all_advertisers_consumption() + logger.info(f"Daily sync completed: {result['message']}") + except Exception as e: + logger.error(f"Daily sync failed: {e}") + + except asyncio.CancelledError: + logger.info("Daily sync task cancelled") + break + except Exception as e: + logger.error(f"Error in schedule_daily_sync: {e}") + await asyncio.sleep(60) \ 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 e7bbbaca..87d2b7fa 100644 --- a/video-gen-api/app/utils/douyinApi.py +++ b/video-gen-api/app/utils/douyinApi.py @@ -78,4 +78,28 @@ class DouyinApi: url, 'POST', {"json": params or {}} + ) + + #获取素材消耗 + async def get_material_cost(self, oauth_id: str, params: any, request_count: int = 3) -> Dict[str, Any]: + if not oauth_id: + raise RuntimeError('OAuth ID is not set.') + url = "https://api.oceanengine.com/open_api/v3.0/report/custom/get/" + return await self.request.request_with_token_with_context( + oauth_id, + url, + 'GET', + {'params': params or {}}, + request_count=request_count + ) + #获取账户信息 + async def get_account_info(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/agent/advertiser_info/query/" + return await self.request.request_with_token_with_context( + oauth_id, + url, + 'GET', + {'params': 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 5a066b5a..5136ffca 100644 --- a/video-gen-api/app/utils/douyinRequest.py +++ b/video-gen-api/app/utils/douyinRequest.py @@ -187,11 +187,13 @@ class DouyinRequest: url: str, method: str = 'GET', options: any = None, + request_count: int = 1, ) -> Any: options = options or {} token = await self.get_access_token(oauth_id) - for i in range(1, 2): + for i in range(1, request_count+1): + # 有一些错误是触发频次管理的，需要重试 try: headers = options.get('headers', {}).copy() headers['Access-Token'] = token