import os import hashlib import base64 import json from sqlalchemy import select, func from sqlalchemy.ext.asyncio import AsyncSession from app.models.generated_resource import GeneratedResource from app.models.user_oauth import UserOAuth from app.models.resources_material import ResourcesMaterial from app.models.pre_test_template import PreTestTemplate from app.models.upload_task import UploadTask from app.utils.id_gen import generate_id from app.utils.douyinApi import DouyinApi douyin_api = DouyinApi() async def upload_material_to_platform( resource_ids: list[str], advertiser_ids: list[str], oauth_id: str, db: AsyncSession, current_user_id: str, pre_test_template_id: str | None = None, ) -> any: oauth = await db.execute( select(UserOAuth).where( UserOAuth.id == oauth_id, UserOAuth.deleted_at.is_(None), UserOAuth.user_id == current_user_id, ) ) oauth = oauth.scalar_one_or_none() if not oauth: return { "success_count": 0, "fail_count": len(resource_ids) * len(advertiser_ids), "total_count": len(resource_ids) * len(advertiser_ids), "results": [{ "resource_id": rid, "advertiser_id": aid, "filename": "", "success": False, "error": "授权记录不存在" } for rid in resource_ids for aid in advertiser_ids], "error": "授权记录不存在" } port_type = oauth.port_type access_token = oauth.access_token if not access_token: return { "success_count": 0, "fail_count": len(resource_ids) * len(advertiser_ids), "total_count": len(resource_ids) * len(advertiser_ids), "results": [{ "resource_id": rid, "advertiser_id": aid, "filename": "", "success": False, "error": "授权token不存在" } for rid in resource_ids for aid in advertiser_ids], "error": "授权token不存在" } resources = await db.execute( select(GeneratedResource).where( GeneratedResource.id.in_(resource_ids), GeneratedResource.user_id == current_user_id, GeneratedResource.deleted_at.is_(None), ) ) resources = resources.scalars().all() resource_dict = {r.id: r for r in resources} all_upload_results = [] success_count = 0 fail_count = 0 for resource_id in resource_ids: if resource_id not in resource_dict: for advertiser_id in advertiser_ids: all_upload_results.append({ "resource_id": resource_id, "advertiser_id": advertiser_id, "filename": "", "success": False, "error": f"资源 {resource_id} 不存在或不属于当前用户" }) fail_count += 1 for advertiser_id in advertiser_ids: # 为每个广告主上传素材 for resource_id in resource_ids: if resource_id not in resource_dict: continue resource = resource_dict[resource_id] resource_type = resource.resource_type storage_path = resource.storage_path if resource_type not in ["image", "video"]: result = { "resource_id": resource_id, "advertiser_id": advertiser_id, "filename": os.path.basename(storage_path) if storage_path else "", "success": False, "error": f"不支持的资源类型: {resource_type}，仅支持image和video", } all_upload_results.append(result) fail_count += 1 continue if not storage_path: result = { "resource_id": resource_id, "advertiser_id": advertiser_id, "filename": "", "success": False, "error": "资源本地存储路径为空", } all_upload_results.append(result) fail_count += 1 continue result = await _upload_to_juliang( oauth_id, storage_path, resource_type, advertiser_id, resource, db, current_user_id ) all_upload_results.append(result) if result.get("success"): success_count += 1 else: fail_count += 1 # 上传完成后提交事务 await db.commit() # 调用前测接口（仅当有模板ID且有视频上传成功时） if pre_test_template_id: success_video_ids = [ r["upload_id"] for r in all_upload_results if r.get("success") and r.get("upload_id") and r.get("advertiser_id") == advertiser_id ] if success_video_ids: await _pre_test_material(oauth_id, advertiser_id, success_video_ids, pre_test_template_id, db) return { "success_count": success_count, "fail_count": fail_count, "total_count": len(resource_ids) * len(advertiser_ids), "results": all_upload_results, } #调用抖音前测接口 async def _pre_test_material( oauth_id: str, advertiser_id: str, video_ids: list[str], pre_test_template_id: str, db: AsyncSession, ) -> any: #1.查询数据库是否存在前测模板 pre_test_template = await db.execute( select(PreTestTemplate).where( PreTestTemplate.id == pre_test_template_id, ) ) pre_test_template = pre_test_template.scalar_one_or_none() if not pre_test_template: note = f"前测模板 {pre_test_template_id} 不存在" for video_id in video_ids: await _update_material_pre_test_status( db, oauth_id, advertiser_id, video_id, task_id=None, status="FAILED", note=note, pre_result=None, ) await db.commit() return {"code": -1, "message": note, "data": {}} diagnose_config = {} if pre_test_template.platform: diagnose_config["platform"] = pre_test_template.platform if pre_test_template.external_action: diagnose_config["external_action"] = pre_test_template.external_action if pre_test_template.cpa_bid: diagnose_config["cpa_bid"] = pre_test_template.cpa_bid if pre_test_template.audience_gender: diagnose_config["audience_gender"] = pre_test_template.audience_gender if pre_test_template.audience_age: diagnose_config["audience_age"] = json.loads(pre_test_template.audience_age) if pre_test_template.audience_region: diagnose_config["audience_region"] = json.loads(pre_test_template.audience_region) if pre_test_template.audience_network: diagnose_config["audience_network"] = json.loads(pre_test_template.audience_network) if pre_test_template.cus_name: diagnose_config["cus_name"] = pre_test_template.cus_name if pre_test_template.pricing_type: diagnose_config["pricing_type"] = pre_test_template.pricing_type if pre_test_template.cost_cap: diagnose_config["cost_cap"] = pre_test_template.cost_cap if pre_test_template.target_cost: diagnose_config["target_cost"] = pre_test_template.target_cost if pre_test_template.nobid: diagnose_config["nobid"] = pre_test_template.nobid if pre_test_template.cpc_bid: diagnose_config["cpc_bid"] = pre_test_template.cpc_bid if pre_test_template.budget: diagnose_config["budget"] = pre_test_template.budget params = { "advertiser_id": int(advertiser_id), "video_ids": video_ids, "diagnose_config": diagnose_config, } response = await douyin_api.pre_test_material(oauth_id, params) # 2. 解析接口返回并更新数据库 code = response.get("code", -1) if code != 0: # 整体失败：更新所有视频状态为FAILED note = response.get("message", "未知错误") for video_id in video_ids: await _update_material_pre_test_status( db, oauth_id, advertiser_id, video_id, task_id=None, status="FAILED", note=note, pre_result=None, ) await db.commit() return response # 成功：解析 data 中的任务信息 data = response.get("data", {}) task_ids = data.get("task_ids", []) fail_video_ids = data.get("fail_video_ids", {}) # 3. 更新成功的视频状态为PENDING（等待审核） success_count = 0 for i, video_id in enumerate(video_ids): if video_id in fail_video_ids: # 失败的视频 fail_info = fail_video_ids[video_id] err_code = fail_info.get("err_code", "") err_message = fail_info.get("err_message", "未知错误") note = f"失败[{err_code}]: {err_message}" await _update_material_pre_test_status( db, oauth_id, advertiser_id, video_id, task_id=None, status="FAILED", note=note, pre_result=None, ) else: # 成功的视频 task_id = str(task_ids[success_count]) if success_count < len(task_ids) else None await _update_material_pre_test_status( db, oauth_id, advertiser_id, video_id, task_id=task_id, status="PENDING", note="", pre_result=None, ) success_count += 1 # 提交事务，保存所有修改 await db.commit() return response async def _update_material_pre_test_status( db: AsyncSession, oauth_id: str, advertiser_id: str, upload_id: str, task_id: str | None, status: str, note: str | None, pre_result: str | None, ): """更新素材的前测状态""" result = await db.execute( select(ResourcesMaterial).where( ResourcesMaterial.oauth_id == oauth_id, ResourcesMaterial.advertiser_id == advertiser_id, ResourcesMaterial.upload_id == upload_id, ResourcesMaterial.deleted_at.is_(None), ) ) material = result.scalar_one_or_none() if material: material.task_id = task_id material.status = status material.note = note material.pre_result = pre_result async def _upload_to_juliang( oauth_id: str, storage_path: str, resource_type: str, advertiser_id: str, resource: GeneratedResource, db: AsyncSession, current_user_id: str, ) -> any: filename = os.path.basename(storage_path) if resource_type == "image": if resource.file_size_bytes > 5 * 1024 * 1024: return { "resource_id": resource.id, "advertiser_id": advertiser_id, "filename": filename, "success": False, "error": "图片大小超过5M，不支持上传", } with open(storage_path, "rb") as f: file_content = f.read() image_signature = hashlib.md5(file_content).hexdigest() data = { "advertiser_id": advertiser_id, "upload_type": "UPLOAD_BY_FILE", "image_signature": image_signature, "filename": filename, } files = { "image_file": (filename, file_content, "image/png"), } response = await douyin_api.upload_image_material(oauth_id, data, files) if response["code"] != 0: return { "resource_id": resource.id, "advertiser_id": advertiser_id, "filename": filename, "success": False, "error": response.get("message", "上传失败"), } image_id = response["data"]["id"] material_id = response["data"]["material_id"] existing = await db.execute( select(ResourcesMaterial).where( ResourcesMaterial.oauth_id == oauth_id, ResourcesMaterial.advertiser_id == advertiser_id, ResourcesMaterial.material_id == str(material_id), ResourcesMaterial.upload_id == str(image_id), ResourcesMaterial.deleted_at.is_(None), ) ) existing = existing.scalar_one_or_none() if existing: return { "resource_id": resource.id, "advertiser_id": advertiser_id, "filename": filename, "success": True, "material_id": str(material_id), "upload_id": str(image_id), "message": "素材已存在，跳过添加", } db.add(ResourcesMaterial( id=generate_id(), oauth_id=oauth_id, advertiser_id=advertiser_id, target_table="generated_resources", target_id=resource.id, material_id=str(material_id), upload_id=str(image_id), resource_type=resource_type, user_id=current_user_id, )) return { "resource_id": resource.id, "advertiser_id": advertiser_id, "filename": filename, "success": True, "material_id": str(material_id), "upload_id": str(image_id), } elif resource_type == "video": if resource.file_size_bytes > 500 * 1024 * 1024: return { "resource_id": resource.id, "advertiser_id": advertiser_id, "filename": filename, "success": False, "error": "视频大小超过500M，不支持上传", } with open(storage_path, "rb") as f: file_content = f.read() video_signature = hashlib.md5(file_content).hexdigest() data = { "advertiser_id": advertiser_id, "upload_type": "UPLOAD_BY_FILE", "video_signature": video_signature, "filename": filename, } files = { "video_file": (filename, file_content, "video/mp4"), } response = await douyin_api.upload_video_material(oauth_id, data, files) if response["code"] != 0: return { "resource_id": resource.id, "advertiser_id": advertiser_id, "filename": filename, "success": False, "error": response.get("message", "上传失败"), } video_id = response["data"]["video_id"] material_id = response["data"]["material_id"] existing = await db.execute( select(ResourcesMaterial).where( ResourcesMaterial.oauth_id == oauth_id, ResourcesMaterial.advertiser_id == advertiser_id, ResourcesMaterial.material_id == str(material_id), ResourcesMaterial.upload_id == str(video_id), ResourcesMaterial.deleted_at.is_(None), ) ) existing = existing.scalar_one_or_none() if existing: return { "resource_id": resource.id, "advertiser_id": advertiser_id, "filename": filename, "success": True, "material_id": str(material_id), "upload_id": str(video_id), "message": "素材已存在，跳过添加", } db.add(ResourcesMaterial( id=generate_id(), oauth_id=oauth_id, advertiser_id=advertiser_id, target_table="generated_resources", target_id=resource.id, material_id=str(material_id), upload_id=str(video_id), resource_type=resource_type, user_id=current_user_id, )) return { "resource_id": resource.id, "advertiser_id": advertiser_id, "filename": filename, "success": True, "material_id": str(material_id), "upload_id": str(video_id), } else: return { "resource_id": resource.id, "advertiser_id": advertiser_id, "filename": filename, "success": False, "error": f"不支持的资源类型: {resource_type}", } async def _upload_to_kuaishou( access_token: str, storage_path: str, resource_type: str, advertiser_id: str, resource: GeneratedResource, db: AsyncSession, current_user_id: str, ) -> dict: filename = os.path.basename(storage_path) return { "resource_id": resource.id, "advertiser_id": advertiser_id, "filename": filename, "success": False, "error": "快手平台上传接口暂未开通", } async def _upload_to_tencent( access_token: str, storage_path: str, resource_type: str, advertiser_id: str, resource: GeneratedResource, db: AsyncSession, current_user_id: str, ) -> dict: filename = os.path.basename(storage_path) return { "resource_id": resource.id, "advertiser_id": advertiser_id, "filename": filename, "success": False, "error": "腾讯平台上传接口暂未开通", } async def get_upload_history( user_id: str, db: AsyncSession, page: int = 1, page_size: int = 20, status: int | None = None, ) -> dict: offset = (page - 1) * page_size query = ( select( UploadTask, GeneratedResource.resource_type, GeneratedResource.resource_url, GeneratedResource.remote_url, GeneratedResource.storage_type, GeneratedResource.storage_path, GeneratedResource.file_size_bytes, GeneratedResource.model_name, ) .outerjoin( GeneratedResource, UploadTask.resource_id == GeneratedResource.id, ) .where( UploadTask.user_id == user_id, UploadTask.deleted_at.is_(None), ) ) if status is not None: query = query.where(UploadTask.status == status) result = await db.execute( query.order_by(UploadTask.created_at.desc()) .offset(offset) .limit(page_size) ) tasks = result.all() count_query = select(func.count(UploadTask.id)).where( UploadTask.user_id == user_id, UploadTask.deleted_at.is_(None), ) if status is not None: count_query = count_query.where(UploadTask.status == status) total_result = await db.execute(count_query) total = total_result.scalar_one() status_map = { 1: "待上传", 2: "上传中", 3: "上传成功", 4: "上传失败", } data = [] for task, resource_type, resource_url, remote_url, storage_type, storage_path, file_size_bytes, model_name in tasks: data.append({ "task_id": task.id, "status": task.status, "status_text": status_map.get(task.status, "未知"), "advertiser_id": task.advertiser_id, "resource_id": task.resource_id, "resource_type": resource_type, "resource_url": resource_url, "remote_url": remote_url, "storage_type": storage_type, "storage_path": storage_path, "file_size_bytes": file_size_bytes, "model_name": model_name, "note": task.note, "created_at": task.created_at, "updated_at": task.updated_at, }) return { "data": data, "pagination": { "page": page, "page_size": page_size, "total": total, }, }