video-gen/video-gen-api/app/services/upload_material_service.py

import os
import hashlib
import base64
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.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,
) -> 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:
        raise ValueError("授权记录不存在")

    port_type = oauth.port_type
    access_token = oauth.access_token

    if not access_token:
        raise ValueError("授权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}
    for resource_id in resource_ids:
        if resource_id not in resource_dict:
            raise ValueError(f"资源 {resource_id} 不存在或不属于当前用户")

    all_upload_results = []
    success_count = 0
    fail_count = 0

    for resource_id in resource_ids:
        resource = resource_dict[resource_id]
        resource_type = resource.resource_type
        storage_path = resource.storage_path

        if resource_type not in ["image", "video"]:
            raise ValueError(f"资源 {resource_id} 不支持的资源类型，仅支持image和video")

        if not storage_path:
            raise ValueError(f"资源 {resource_id} 本地存储路径为空")

        for advertiser_id in advertiser_ids:
            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()

    return {
        "success_count": success_count,
        "fail_count": fail_count,
        "total_count": len(resource_ids) * len(advertiser_ids),
        "results": all_upload_results,
    }


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": "腾讯平台上传接口暂未开通",
    }