622 lines
20 KiB
Python
622 lines
20 KiB
Python
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,
|
|
GeneratedResource.file_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, file_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,
|
|
"file_name": file_name,
|
|
})
|
|
|
|
return {
|
|
"data": data,
|
|
"pagination": {
|
|
"page": page,
|
|
"page_size": page_size,
|
|
"total": total,
|
|
},
|
|
} |