531 lines
17 KiB
Python
531 lines
17 KiB
Python
import os
|
|
import hashlib
|
|
import base64
|
|
import json
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.generated_resource import GeneratedResource
|
|
from app.models.user_oauth import UserOAuth
|
|
from app.models.resources_material import ResourcesMaterial
|
|
from app.models.pre_test_template import PreTestTemplate
|
|
from app.utils.id_gen import generate_id
|
|
from app.utils.douyinApi import DouyinApi
|
|
|
|
|
|
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": "腾讯平台上传接口暂未开通",
|
|
} |