diff --git a/.gitignore b/.gitignore index 4c3b48dc..411858d3 100644 --- a/.gitignore +++ b/.gitignore @@ -22,4 +22,10 @@ video-gen-api/dist/ # 忽略特定类型文件但保留目录 # *.pyc # !dir/*.pycnode_modules/ -*.tmp.* \ No newline at end of file +*.tmp.* +video-gen-api/.env +video-gen-api/videogen_api.egg-info/requires.txt +video-gen-api/videogen_api.egg-info/SOURCES.txt +video-gen-api/.env +video-gen-api/app/utils/redis.py +video-gen-api/.env diff --git a/video-gen-api/.env b/video-gen-api/.env index 783ade76..c78d9786 100644 --- a/video-gen-api/.env +++ b/video-gen-api/.env @@ -1,64 +1,108 @@ -# App APP_NAME=VideoGen API APP_VERSION=1.0.0 -DEBUG=false -SECRET_KEY=local-dev-secret-key-not-for-production +DEBUG=true +SECRET_KEY=your-secret-key-change-in-production -# Database (PostgreSQL) DATABASE_URL=postgresql+asyncpg://videogen:7k33pnXdPL62Yyb4@180.184.42.66:5432/videogen -# Redis (leave empty to disable - rate limiting and captcha will use in-memory fallback) -REDIS_URL=redis://localhost:6379/3 -CELERY_BROKER_URL=redis://localhost:6379/5 -CELERY_RESULT_BACKEND=redis://localhost:6379/6 +REDIS_URL=redis://127.0.0.1:6379/0 -# JWT JWT_ALGORITHM=HS256 JWT_EXPIRE_MINUTES=1440 +JWT_EXPIRE_REMEMBER_MINUTES=10080 -# Seedance SEEDANCE_API_KEY= SEEDANCE_API_BASE=https://ark.cn-beijing.volces.com/api/v3 SEEDANCE_CALLBACK_URL= -# LLM LLM_API_BASE=https://api.openai.com/v1 LLM_API_KEY= LLM_MODEL=gpt-4o LLM_MOCK=true -# Encryption -ENCRYPTION_KEY=dGhpc19pc18zMl9ieXRlX2tleV9mb3JfYWVzXzI1NiE= +ENCRYPTION_KEY=changeme-32bytes-base64-key-here!! + +SMS_API_URL= +SMS_API_KEY= +SMS_SIGN_NAME=VideoGen +SMS_TEMPLATE_CODE=SMS_001 +SMS_MOCK=true + +VOLC_SMS_ACCESS_KEY_ID= +VOLC_SMS_SECRET_ACCESS_KEY= +VOLC_SMS_ACCOUNT= +VOLC_SMS_TEMPLATE_ID= +VOLC_SMS_SIGN= +SMS_CODE_LENGTH=4 +SMS_CODE_TTL_SECONDS=300 +SMS_SEND_INTERVAL_SECONDS=60 +SMS_DAILY_LIMIT=20 + +WECHAT_MCH_ID= +WECHAT_API_KEY= +WECHAT_CERT_PATH= +ALIPAY_APP_ID= +ALIPAY_PRIVATE_KEY= +ALIPAY_PUBLIC_KEY= +PAYMENT_MOCK=true -# Storage STORAGE_TYPE=local STORAGE_LOCAL_PATH=./storage/generate/videos STORAGE_IMAGE_LOCAL_PATH=./storage/generate/images STORAGE_VIDEO_COVER_LOCAL_PATH=./storage/generate/covers +UPLOAD_LOCAL_PATH=./storage/uploads -# Captcha -CAPTCHA_ENABLED=true - -# CORS -CORS_ORIGINS=["*"] - -# RESOURCE -RESOURCE_SIGN_SECRET=EOTpDZsEgkaYWPxgtIedOO0lDlH1moTS2rnSIemjzmO3 -RESOURCE_SIGN_EXPIRE_SECONDS=86400 -RESOURCE_SIGN_ARG_EXPIRE=exp -RESOURCE_SIGN_ARG_SIGNATURE=sign - -# FFMPEG + COVER -FFMPEG_BIN=/usr/bin/ffmpeg +FFMPEG_BIN= VIDEO_COVER_SEEK_TIME=00:00:01 VIDEO_COVER_FALLBACK_SEEK_TIME=00:00:00 -VIDEO_COVER_WIDTH=600 +VIDEO_COVER_WIDTH=720 VIDEO_COVER_TIMEOUT_SECONDS=15 -VIDEO_COVER_FORMAT=png +VIDEO_COVER_FORMAT=jpg -# VOLC_SMS -VOLC_SMS_ACCESS_KEY_ID=AKLTYWY5Yjc5YjM3N2IwNDc3M2I3NTU2YjlmNTczYzQzMmM -VOLC_SMS_SECRET_ACCESS_KEY=TXpjM01HUTFZMlV5TUdKbE5Ea3lNRGhqTUdSak16UTFOV0ptTW1SaE5XRQ== -VOLC_SMS_ACCOUNT=8b3f6ca8 -VOLC_SMS_TEMPLATE_ID=S1T_1y2pb9ej2g9vm -VOLC_SMS_SIGN=纬佳网络科技 \ No newline at end of file +CAPTCHA_ENABLED=true + +BASE_URL=http://localhost:8000 + +CORS_ORIGINS=["http://localhost:5173", "http://localhost:3000"] + +RATE_LIMIT_ENABLED=true + +CELERY_BROKER_URL= +CELERY_RESULT_BACKEND= +CHATAPI_REQUEST_TIMEOUT_SECONDS=120 +CHATAPI_VIDEO_FPS=0.5 +CHATAPI_ASYNC_MAX_RETRIES=3 +CHATAPI_ASYNC_RETRY_BACKOFF_SECONDS=30 +CHATAPI_ASYNC_POLL_INTERVAL_SECONDS=30 +CHATAPI_ASYNC_IMAGE_DEADLINE_MINUTES=10 +CHATAPI_ASYNC_VIDEO_DEADLINE_MINUTES=30 + +ARK_CHAT_PROMPT_MAX_CONCURRENCY=20 +ARK_IMAGE_CREATE_MAX_CONCURRENCY=10 +ARK_VIDEO_CREATE_MAX_CONCURRENCY=10 +ARK_IMAGE_POLL_MAX_CONCURRENCY=50 +ARK_VIDEO_POLL_MAX_CONCURRENCY=50 +RESULT_DOWNLOAD_MAX_CONCURRENCY=10 +PROVIDER_LIMIT_WAIT_TIMEOUT_SECONDS=30.0 +PROVIDER_LIMIT_TOKEN_TTL_SECONDS=600 + +CELERY_DB_POOL_SIZE=1 +CELERY_DB_MAX_OVERFLOW=1 +CELERY_DB_POOL_TIMEOUT=30 +CELERY_DB_POOL_RECYCLE=1800 + +DOWNLOAD_TASK_PRIORITY_NORMAL=5 +DOWNLOAD_TASK_PRIORITY_RECOVER=0 +DOWNLOAD_TASK_MAX_ATTEMPTS=3 +DOWNLOAD_TASK_RETRY_BACKOFF_SECONDS=30 +DOWNLOAD_TASK_LEASE_SECONDS=600 +DOWNLOAD_TASK_QUEUE_TIMEOUT_SECONDS=300 +DOWNLOAD_RECOVERY_BATCH_SIZE=100 +DOWNLOAD_RECOVERY_STARTUP_DELAY_SECONDS=3 +DOWNLOAD_ACTIVE_REDIS_HASH_KEY=vg:celery:download:active +DOWNLOAD_ACTIVE_REDIS_ZSET_KEY=vg:celery:download:active_index + +RESOURCE_SIGN_SECRET=resource-signature-secret-key-for-API-authentication +RESOURCE_SIGN_EXPIRE_SECONDS=60 +RESOURCE_SIGN_ARG_EXPIRE=exp +RESOURCE_SIGN_ARG_SIGNATURE=sign diff --git a/video-gen-api/app/api/v1/test.py b/video-gen-api/app/api/v1/test.py index 02961f89..7a38e1d9 100644 --- a/video-gen-api/app/api/v1/test.py +++ b/video-gen-api/app/api/v1/test.py @@ -1,10 +1,24 @@ -from fastapi import APIRouter, Depends, HTTPException, Request - +from fastapi import APIRouter, HTTPException +import asyncio +from redis.asyncio import Redis +from app.config import settings +from app.utils.redis import get_redis router = APIRouter(prefix="/test", tags=["test"]) @router.get("/index") async def test(): - return {"message": "test","code":200} - + redis = None + try: + redis = get_redis() + await redis.set("test", "123", ex=60) + value = await redis.get("test") + return {"message": "redis测试成功", "value": value} + except Exception as e: + if redis: + try: + await redis.close() + except: + pass + return {"message": f"redis测试失败: {str(e)}"} \ 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 bc8e052f..6579f7e0 100644 --- a/video-gen-api/app/api/v1/upload_material.py +++ b/video-gen-api/app/api/v1/upload_material.py @@ -1,21 +1,25 @@ import os +import json +import uuid from typing import Any, Optional, Dict -from fastapi import APIRouter, Depends, HTTPException, status +from fastapi import APIRouter, Depends, HTTPException, status, Query from pydantic import BaseModel, Field from sqlalchemy.ext.asyncio import AsyncSession +from sqlalchemy import select, func from app.dependencies import get_current_user, get_db from app.models.user import User from app.models.pre_test_template import PreTestTemplate +from app.models.upload_task import UploadTask from app.services.upload_material_service import upload_material_to_platform -from sqlalchemy import select +from app.services.upload_queue import upload_queue from app.utils.douyinApi import DouyinApi router = APIRouter(prefix="/upload-material", tags=["上传素材"]) -class UploadTask(BaseModel): +class UploadTaskRequest(BaseModel): advertiser_ids: list[str] = Field(..., description="广告主id数组，支持多条") resource_ids: list[str] = Field(..., description="资源id数组（generated_resources表主键）") oauth_id: str = Field(..., description="授权表id") @@ -24,7 +28,7 @@ class UploadTask(BaseModel): class BatchUploadRequest(BaseModel): - tasks: list[UploadTask] = Field(..., description="批量上传任务列表") + tasks: list[UploadTaskRequest] = Field(..., description="批量上传任务列表") @router.post( @@ -37,22 +41,6 @@ async def batch_upload_material( current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ) -> Any | dict: - # try: - # params = { - # "advertiser_id": 1836693172153543, - # "video_ids": ["tos-cn-i-sd07hgqsbj/2db9a53ee3ff4b6f8a2763b3bb463047"], - # "diagnose_config": {"platform": "AD", "external_action": "AD_APP_ACTIVATE"}, - # } - # response = await DouyinApi().pre_test_material("0019ecab9b8bc57d964", params) - # return response - # except Exception as e: - # import traceback - # return { - # "code": 500, - # "message": "请求失败", - # "error": str(e), - # "traceback": traceback.format_exc() - # } if not req.tasks: return { "code": 0, @@ -214,4 +202,174 @@ async def batch_upload_material( "total_requested": sum(len(t.resource_ids) * len(t.advertiser_ids) for t in req.tasks), }, "details": all_results, + } + + +@router.post( + "/async-batch-upload", + summary="异步批量上传素材到平台", + description="支持批量上传多个授权账户下的资源到素材库，提交后立即返回，后台异步处理", +) +async def async_batch_upload_material( + req: BatchUploadRequest, + current_user: User = Depends(get_current_user), + db: AsyncSession = Depends(get_db), +) -> Any | dict: + if not req.tasks: + return { + "code": 0, + "message": "上传任务列表不能为空", + "task_ids": [], + } + + task_ids = [] + + for task in req.tasks: + if not task.advertiser_ids or not task.resource_ids: + continue + + if task.is_pre_test == "1" and not task.pre_test_template: + continue + + for advertiser_id in task.advertiser_ids: + for resource_id in task.resource_ids: + task_id = str(uuid.uuid4()).replace("-", "")[:32] + + upload_task = UploadTask( + id=task_id, + user_id=current_user.id, + oauth_id=task.oauth_id, + advertiser_id=advertiser_id, + resource_id=resource_id, + status=1, + note=None, + ) + + db.add(upload_task) + await upload_queue.enqueue(task_id) + task_ids.append(task_id) + + await db.commit() + + return { + "code": 0, + "message": "上传任务已提交，请通过查询接口查看进度", + "task_ids": task_ids, + } + + +@router.get( + "/upload-status", + summary="查询上传任务状态", + description="查询单个上传任务的执行状态和结果", +) +async def get_upload_status( + task_id: str = Query(..., description="上传任务ID"), + current_user: User = Depends(get_current_user), + db: AsyncSession = Depends(get_db), +) -> Any | dict: + result = await db.execute( + select(UploadTask).where( + UploadTask.id == task_id, + UploadTask.user_id == current_user.id, + UploadTask.deleted_at.is_(None), + ) + ) + task = result.scalar_one_or_none() + + if not task: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail="任务不存在或无权限查看", + ) + + status_map = { + 1: "待上传", + 2: "上传中", + 3: "上传成功", + 4: "上传失败", + } + + return { + "code": 0, + "data": { + "task_id": task.id, + "status": task.status, + "status_text": status_map.get(task.status, "未知"), + "advertiser_id": task.advertiser_id, + "resource_id": task.resource_id, + "note": task.note, + "created_at": task.created_at, + "updated_at": task.updated_at, + }, + } + + +@router.get( + "/upload-history", + summary="查询上传任务历史", + description="查询当前用户的上传任务历史列表", +) +async def get_upload_history( + page: int = Query(1, ge=1, description="页码"), + page_size: int = Query(20, ge=1, le=100, description="每页数量"), + status: Optional[int] = Query(None, description="上传状态筛选：1待上传，2上传中，3上传成功，4上传失败"), + current_user: User = Depends(get_current_user), + db: AsyncSession = Depends(get_db), +) -> Any | dict: + offset = (page - 1) * page_size + + query = select(UploadTask).where( + UploadTask.user_id == current_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.scalars().all() + + count_query = select(func.count(UploadTask.id)).where( + UploadTask.user_id == current_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: "上传失败", + } + + return { + "code": 0, + "data": [ + { + "task_id": task.id, + "status": task.status, + "status_text": status_map.get(task.status, "未知"), + "advertiser_id": task.advertiser_id, + "resource_id": task.resource_id, + "note": task.note, + "created_at": task.created_at, + "updated_at": task.updated_at, + } + for task in tasks + ], + "pagination": { + "page": page, + "page_size": page_size, + "total": total, + }, } \ No newline at end of file diff --git a/video-gen-api/app/main.py b/video-gen-api/app/main.py index 7acc1321..e59afbbc 100644 --- a/video-gen-api/app/main.py +++ b/video-gen-api/app/main.py @@ -63,6 +63,11 @@ async def lifespan(app: FastAPI): from app.tasks.token_refresh_task import token_refresh_scheduler token_refresh_task = asyncio.create_task(token_refresh_scheduler()) + # 启动上传队列（异步处理素材上传） + from app.services.upload_queue import upload_queue + await upload_queue.recover() + upload_queue_task = asyncio.create_task(upload_queue.run()) + # 启动时立即同步一次未支付订单 asyncio.create_task(asyncio.sleep(5)) # 等待5秒后再同步，让系统完全启动 async def startup_sync(): @@ -87,6 +92,8 @@ async def lifespan(app: FastAPI): await queue_task 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/models/upload_material.py b/video-gen-api/app/models/upload_material.py new file mode 100644 index 00000000..77e5f26b --- /dev/null +++ b/video-gen-api/app/models/upload_material.py @@ -0,0 +1,27 @@ +from sqlalchemy import String, Text, Integer +from sqlalchemy.orm import Mapped, mapped_column + +from app.models.base import Base, TimestampMixin, SoftDeleteMixin + + +class UploadMaterial(Base, TimestampMixin, SoftDeleteMixin): + __tablename__ = "upload_material" + + id: Mapped[str] = mapped_column( + String(32), primary_key=True, comment="主键" + ) + user_id: Mapped[str] = mapped_column( + String(64), nullable=False, index=True, comment="用户登录id" + ) + advertiser_id: Mapped[str] = mapped_column( + String(64), nullable=False, index=True, comment="广告主id" + ) + resource_id: Mapped[str] = mapped_column( + String(64), nullable=False, index=True, comment="资源id" + ) + status: Mapped[int] = mapped_column( + Integer, nullable=False, default=1, comment="上传状态：1待上传，2上传中，3上传成功，4上传失败" + ) + note: Mapped[str | None] = mapped_column( + Text, nullable=True, comment="上传备注" + ) diff --git a/video-gen-api/app/services/upload_queue.py b/video-gen-api/app/services/upload_queue.py new file mode 100644 index 00000000..5ade0627 --- /dev/null +++ b/video-gen-api/app/services/upload_queue.py @@ -0,0 +1,386 @@ +import asyncio +import logging +from datetime import datetime, timezone + +from sqlalchemy import select, update +from sqlalchemy.orm import selectinload + +from app.models.base import async_session +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.utils.id_gen import generate_id +from app.utils.douyinApi import DouyinApi + +import os +import hashlib + +logger = logging.getLogger("upload_queue") +douyin_api = DouyinApi() + + +class UploadQueue: + def __init__(self): + self.queue: asyncio.Queue[str] = asyncio.Queue() + self.running = False + + async def enqueue(self, task_id: str): + """Add a task to the queue.""" + await self.queue.put(task_id) + + async def recover(self): + """Recover pending tasks from DB on startup.""" + async with async_session() as db: + result = await db.execute( + select(UploadTask).where( + UploadTask.status.in_([1, 2]), + UploadTask.deleted_at.is_(None), + ) + ) + tasks = result.scalars().all() + for task in tasks: + await self.queue.put(task.id) + logger.info(f"Recovered upload task: {task.id}") + + async def run(self): + """Main processing loop.""" + self.running = True + logger.info("Upload queue started") + + while self.running: + try: + task_id = await asyncio.wait_for(self.queue.get(), timeout=5.0) + except asyncio.TimeoutError: + continue + + try: + await self._process(task_id) + except Exception as e: + logger.error(f"Error processing upload task {task_id}: {e}") + finally: + self.queue.task_done() + + logger.info("Upload queue stopped") + + async def _process(self, task_id: str): + """Process a single upload task.""" + async with async_session() as db: + result = await db.execute( + select(UploadTask).where( + UploadTask.id == task_id, + UploadTask.deleted_at.is_(None), + ).with_for_update() + ) + task = result.scalar_one_or_none() + + if not task: + logger.warning(f"Upload task {task_id} not found or deleted") + return + + if task.status == 2: + logger.warning(f"Upload task {task_id} is already running") + return + + task.status = 2 + await db.commit() + + try: + result = await _upload_single_material( + task.id, + task.user_id, + task.oauth_id, + task.advertiser_id, + task.resource_id, + db=None + ) + + async with async_session() as db: + if result.get("success"): + await db.execute( + update(UploadTask).where(UploadTask.id == task_id).values( + status=3, + note=result.get("message", "上传成功"), + ) + ) + else: + await db.execute( + update(UploadTask).where(UploadTask.id == task_id).values( + status=4, + note=result.get("error", "上传失败"), + ) + ) + await db.commit() + + logger.info(f"Upload task {task_id} completed: {'success' if result.get('success') else 'failed'}") + + except Exception as e: + async with async_session() as db: + await db.execute( + update(UploadTask).where(UploadTask.id == task_id).values( + status=4, + note=f"上传失败: {str(e)}", + ) + ) + await db.commit() + + logger.error(f"Upload task {task_id} failed: {e}") + + def stop(self): + """Stop the queue.""" + self.running = False + + +async def _upload_single_material( + task_id: str, + user_id: str, + oauth_id: str, + advertiser_id: str, + resource_id: str, + db=None +) -> dict: + if db is None: + from app.models.base import async_session + db = async_session() + + try: + oauth = await db.execute( + select(UserOAuth).where( + UserOAuth.id == oauth_id, + UserOAuth.deleted_at.is_(None), + UserOAuth.user_id == user_id, + ) + ) + oauth = oauth.scalar_one_or_none() + if not oauth: + return { + "success": False, + "error": "授权记录不存在" + } + + resource = await db.execute( + select(GeneratedResource).where( + GeneratedResource.id == resource_id, + GeneratedResource.user_id == user_id, + GeneratedResource.deleted_at.is_(None), + ) + ) + resource = resource.scalar_one_or_none() + + if not resource: + return { + "success": False, + "error": f"资源 {resource_id} 不存在或不属于当前用户" + } + + resource_type = resource.resource_type + storage_path = resource.storage_path + + if resource_type not in ["image", "video"]: + return { + "success": False, + "error": f"不支持的资源类型: {resource_type}，仅支持image和video", + } + + if not storage_path: + return { + "success": False, + "error": "资源本地存储路径为空", + } + + return await _upload_to_juliang( + oauth_id, storage_path, resource_type, advertiser_id, resource, db, user_id + ) + finally: + if db is not None: + await db.close() + + +async def _upload_to_juliang( + oauth_id: str, + storage_path: str, + resource_type: str, + advertiser_id: str, + resource: GeneratedResource, + db, + current_user_id: str, +) -> dict: + 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, + )) + await db.commit() + + 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, + )) + await db.commit() + + 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}", + } + + +upload_queue = UploadQueue() \ No newline at end of file diff --git a/video-gen-api/app/utils/redis.py b/video-gen-api/app/utils/redis.py index 13ffa65b..75d5fd68 100644 --- a/video-gen-api/app/utils/redis.py +++ b/video-gen-api/app/utils/redis.py @@ -9,7 +9,7 @@ async def init_redis() -> None: return try: from redis.asyncio import Redis - redis_client = Redis.from_url(settings.REDIS_URL, decode_responses=True) + redis_client = Redis.from_url(settings.REDIS_URL, decode_responses=True, protocol=2) await redis_client.ping() except Exception: redis_client = None