898 lines
32 KiB
Python
898 lines
32 KiB
Python
import json
|
||
import logging
|
||
import os
|
||
from datetime import datetime
|
||
|
||
from fastapi import APIRouter, Depends, HTTPException, Query, Request, UploadFile, File, status
|
||
from fastapi.responses import RedirectResponse
|
||
from sqlalchemy import select, func
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from app.config import settings
|
||
from app.dependencies import get_db, get_current_user
|
||
from app.models.user import User
|
||
from app.models.project import Project
|
||
from app.models.generation_record import GenerationRecord
|
||
from app.schemas.generation import (
|
||
OptimizeParams,
|
||
GenerateParams,
|
||
GenerationRecordOut,
|
||
GenerationRecordPageListOut,
|
||
OptimizeResult,
|
||
UpdatePromptRequest,
|
||
GenerationType,
|
||
DURATIONS,
|
||
ASPECT_RATIOS,
|
||
RESOLUTIONS,
|
||
IMAGE_SIZES,
|
||
)
|
||
from app.services.credits import deduct_credits, calc_text_credits
|
||
from app.services.llm import optimize_prompt
|
||
from app.services.video_url import generate_temp_url, validate_and_get_record_id, get_video_stream_url
|
||
from app.services.resource_accounting_service import (
|
||
record_generation_record_generated_resource,
|
||
safe_file_size,
|
||
)
|
||
from app.services.resource_signed_url_service import build_resource_signed_url
|
||
from app.services.resource_capacity_service import assert_user_resource_capacity_available
|
||
from app.services.generation_billing_service import (
|
||
CHARGE_TEXT_PROMPT,
|
||
OWNER_GENERATION_RECORD,
|
||
build_credit_biz_key,
|
||
charge_generation_media_by_params,
|
||
charge_generation_media_for_record,
|
||
get_next_credit_attempt_no,
|
||
)
|
||
from app.services.generation_refund_service import mark_generation_record_failed_and_refund_once
|
||
from app.services.media_token_usage_snapshot_service import sync_generation_record_media_token_snapshot
|
||
from app.services.credit_record_meta_service import build_generation_record_prompt_meta
|
||
from app.services.video_cover_service import async_create_video_cover_for_local_video
|
||
from app.enums.audio_reference import (
|
||
AUDIO_ALLOWED_EXTENSIONS,
|
||
AUDIO_ALLOWED_MIME_TYPES,
|
||
AUDIO_MAX_FILE_SIZE_MB,
|
||
)
|
||
from app.utils.id_gen import generate_id
|
||
from app.utils.exceptions import InsufficientCreditsError, RecordNotFoundError, InvalidStatusError
|
||
|
||
router = APIRouter(prefix="/generation-records", tags=["generation"])
|
||
logger = logging.getLogger("videogen")
|
||
|
||
|
||
def _record_to_out(record: GenerationRecord, project_name: str) -> GenerationRecordOut:
|
||
refs = None
|
||
if record.media_references:
|
||
try:
|
||
refs = json.loads(record.media_references)
|
||
except (json.JSONDecodeError, TypeError):
|
||
refs = None
|
||
|
||
error_message = record.error_message
|
||
if error_message:
|
||
from app.services.error_codes import ARK_ERRORS
|
||
import re
|
||
match = re.search(r"code='([^']+)'", error_message)
|
||
if match:
|
||
code = match.group(1)
|
||
if code in ARK_ERRORS:
|
||
error_message = ARK_ERRORS[code]
|
||
else:
|
||
parts = error_message.split(":")
|
||
if len(parts) >= 2 and parts[1].strip() in ARK_ERRORS:
|
||
error_message = ARK_ERRORS[parts[1].strip()]
|
||
|
||
return GenerationRecordOut(
|
||
id=record.id,
|
||
project_id=record.project_id,
|
||
project_name=project_name,
|
||
original_prompt=record.original_prompt,
|
||
optimized_prompt=record.optimized_prompt,
|
||
gen_type=record.gen_type,
|
||
duration=record.duration,
|
||
aspect_ratio=record.aspect_ratio,
|
||
resolution=record.resolution,
|
||
image_size=record.image_size,
|
||
image_proportion=record.image_proportion,
|
||
image_px=record.image_px,
|
||
status=record.status,
|
||
video_url=build_resource_signed_url(record.video_url) if record.video_url else '',
|
||
video_cover_url=build_resource_signed_url(record.video_cover_url) if record.video_cover_url else '',
|
||
image_url=build_resource_signed_url(record.image_url) if record.image_url else '',
|
||
references=refs,
|
||
text_credits_cost=round(record.text_credits_cost or 0.00, 2),
|
||
# text_tokens_used=record.text_tokens_used or 0,
|
||
credits_cost=round(record.credits_cost or 0.00, 2),
|
||
# video_tokens_used=record.video_tokens_used or 0,
|
||
# image_tokens_used=record.image_tokens_used or 0,
|
||
error_message=error_message,
|
||
created_at=record.created_at,
|
||
generated_at=record.generated_at,
|
||
)
|
||
|
||
|
||
@router.get("", response_model=GenerationRecordPageListOut)
|
||
async def list_records(
|
||
project_id: str | None = Query(
|
||
None,
|
||
alias="project_id",
|
||
description="查询单个项目的生成记录",
|
||
),
|
||
status: str | None = Query(
|
||
None,
|
||
description="查询状态，可以不传。prompt_optimized:待生成 | generating:生成中 | failed:失败 | completed:成功",
|
||
examples=["completed"],
|
||
),
|
||
page: int = Query(
|
||
1,
|
||
ge=1,
|
||
description="分页页码，从1开始",
|
||
examples=[1],
|
||
),
|
||
page_size: int = Query(
|
||
10,
|
||
ge=1,
|
||
le=100,
|
||
description="每页返回的生成记录数量，范围 1~100",
|
||
examples=[10],
|
||
),
|
||
record_ids: list[str] | None = Query(
|
||
None,
|
||
description="对应记录ID数组",
|
||
examples=[["0019e8c55ddd1429b86", "0019e8c54dfc1e13262"]],
|
||
),
|
||
current_user: User = Depends(get_current_user),
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
allowed_statuses = {
|
||
"prompt_optimized",
|
||
"generating",
|
||
"failed",
|
||
"completed",
|
||
}
|
||
|
||
if status and status not in allowed_statuses:
|
||
raise HTTPException(
|
||
status_code=400,
|
||
detail="状态参数错误，仅支持：prompt_optimized、generating、failed、completed",
|
||
)
|
||
|
||
offset = (page - 1) * page_size
|
||
|
||
conditions = [
|
||
GenerationRecord.user_id == current_user.id,
|
||
GenerationRecord.deleted_at.is_(None),
|
||
Project.deleted_at.is_(None),
|
||
]
|
||
|
||
if project_id:
|
||
conditions.append(GenerationRecord.project_id == project_id)
|
||
|
||
if status:
|
||
conditions.append(GenerationRecord.status == status)
|
||
|
||
if record_ids:
|
||
conditions.append(GenerationRecord.id.in_(record_ids))
|
||
|
||
total_result = await db.execute(
|
||
select(func.count(GenerationRecord.id))
|
||
.join(Project, GenerationRecord.project_id == Project.id)
|
||
.where(*conditions)
|
||
)
|
||
total = total_result.scalar_one() or 0
|
||
|
||
query = (
|
||
select(GenerationRecord, Project.name)
|
||
.join(Project, GenerationRecord.project_id == Project.id)
|
||
.where(*conditions)
|
||
.order_by(GenerationRecord.created_at.desc())
|
||
.offset(offset)
|
||
.limit(page_size)
|
||
)
|
||
|
||
result = await db.execute(query)
|
||
rows = result.all()
|
||
|
||
return {
|
||
"total": int(total),
|
||
"page": page,
|
||
"page_size": page_size,
|
||
"items": [
|
||
_record_to_out(record, project_name)
|
||
for record, project_name in rows
|
||
],
|
||
}
|
||
|
||
@router.post("/optimize", response_model=OptimizeResult)
|
||
async def optimize(
|
||
req: OptimizeParams,
|
||
current_user: User = Depends(get_current_user),
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
record = None
|
||
# Validate parameters based on generation type
|
||
if req.gen_type == GenerationType.video:
|
||
if req.duration not in DURATIONS:
|
||
raise HTTPException(status_code=400, detail=f"视频时长必须为{DURATIONS}秒之一")
|
||
if not req.duration:
|
||
raise HTTPException(status_code=400, detail="视频生成需要指定时长")
|
||
elif req.gen_type == GenerationType.image:
|
||
if req.image_size not in IMAGE_SIZES:
|
||
raise HTTPException(status_code=400, detail=f"图片分辨率必须为{IMAGE_SIZES}之一")
|
||
if not req.image_size:
|
||
raise HTTPException(status_code=400, detail="图片生成需要指定画面分辨率")
|
||
|
||
# Idempotency check: if key provided, return existing record if found
|
||
if req.idempotency_key:
|
||
existing = await db.execute(
|
||
select(GenerationRecord, Project.name)
|
||
.join(Project, GenerationRecord.project_id == Project.id)
|
||
.where(
|
||
GenerationRecord.user_id == current_user.id,
|
||
GenerationRecord.deleted_at.is_(None),
|
||
Project.deleted_at.is_(None),
|
||
GenerationRecord.idempotency_key == req.idempotency_key,
|
||
GenerationRecord.gen_type == req.gen_type,
|
||
GenerationRecord.status == "prompt_optimized",
|
||
)
|
||
.order_by(GenerationRecord.created_at.desc())
|
||
.limit(1)
|
||
)
|
||
row = existing.first()
|
||
if row:
|
||
record, project_name = row
|
||
return OptimizeResult(
|
||
optimized_prompt=record.optimized_prompt or "",
|
||
text_credits_cost=record.text_credits_cost or 0.00,
|
||
text_tokens_used=record.text_tokens_used or 0,
|
||
record=_record_to_out(record, project_name),
|
||
)
|
||
|
||
# Check project exists and belongs to user
|
||
proj_result = await db.execute(
|
||
select(Project).where(
|
||
Project.id == req.project_id,
|
||
Project.user_id == current_user.id,
|
||
Project.deleted_at.is_(None),
|
||
)
|
||
.limit(1)
|
||
)
|
||
project = proj_result.scalar_one_or_none()
|
||
if not project:
|
||
raise HTTPException(status_code=404, detail="项目不存在")
|
||
|
||
# Optimize prompt via LLM with type-specific context
|
||
try:
|
||
optimized, token_usage = await optimize_prompt(
|
||
db, req.prompt,
|
||
user_id=current_user.id,
|
||
industry_key=project.industry,
|
||
duration=req.duration if req.gen_type == GenerationType.video else None,
|
||
image_size=req.image_size if req.gen_type == GenerationType.image else None,
|
||
image_proportion=req.image_proportion if req.gen_type == GenerationType.image else None,
|
||
image_px=req.image_px if req.gen_type == GenerationType.image else None,
|
||
references=req.references,
|
||
gen_type=req.gen_type,
|
||
)
|
||
# Create record BEFORE LLM call so it's visible if user refreshes
|
||
record = GenerationRecord(
|
||
id=generate_id(),
|
||
user_id=current_user.id,
|
||
project_id=req.project_id,
|
||
original_prompt=req.prompt,
|
||
gen_type=req.gen_type,
|
||
duration=req.duration,
|
||
image_size=req.image_size,
|
||
image_proportion=req.image_proportion,
|
||
image_px=req.image_px,
|
||
status="optimizing",
|
||
credits_cost=0,
|
||
text_credits_cost=0,
|
||
text_tokens_used=0,
|
||
media_references=json.dumps(req.references) if req.references else None,
|
||
idempotency_key=req.idempotency_key,
|
||
)
|
||
db.add(record)
|
||
await db.flush()
|
||
await db.commit()
|
||
except Exception as e:
|
||
from app.services.error_codes import extract_error_message
|
||
if record:
|
||
record.status = "failed"
|
||
record.error_message = extract_error_message(e, "提示词")
|
||
await db.flush()
|
||
await db.commit()
|
||
error_message = extract_error_message(e, "提示词")
|
||
raise HTTPException(
|
||
status_code=502,
|
||
detail=f"AI模型调用失败: {error_message}"
|
||
)
|
||
|
||
text_credits = await calc_text_credits(
|
||
db, token_usage["input_tokens"], token_usage["output_tokens"],
|
||
)
|
||
|
||
failed_record_id = record.id
|
||
failed_user_id = current_user.id
|
||
try:
|
||
prompt_attempt_no = 1
|
||
prompt_biz_key = build_credit_biz_key(
|
||
owner_type=OWNER_GENERATION_RECORD,
|
||
owner_id=record.id,
|
||
attempt_no=prompt_attempt_no,
|
||
charge_kind=CHARGE_TEXT_PROMPT,
|
||
action="charge",
|
||
)
|
||
prompt_meta = await build_generation_record_prompt_meta(
|
||
db,
|
||
record_id=record.id,
|
||
attempt_no=prompt_attempt_no,
|
||
charge_kind=CHARGE_TEXT_PROMPT,
|
||
usage=token_usage,
|
||
)
|
||
await deduct_credits(
|
||
db, current_user.id, text_credits,
|
||
f"提示词优化 - {project.name}",
|
||
related_id=record.id,
|
||
biz_key=prompt_biz_key,
|
||
record_meta=prompt_meta,
|
||
)
|
||
except InsufficientCreditsError as e:
|
||
# /optimize 阶段只处理提示词优化扣费。
|
||
# 提示词积分不足时，之前已落库的 optimizing 记录必须改为 failed，避免前端长期显示生成中。
|
||
# 此阶段没有媒体生成扣费，不调用生成失败退款逻辑。
|
||
await db.rollback()
|
||
result = await db.execute(
|
||
select(GenerationRecord)
|
||
.where(
|
||
GenerationRecord.id == failed_record_id,
|
||
GenerationRecord.user_id == failed_user_id,
|
||
GenerationRecord.deleted_at.is_(None),
|
||
)
|
||
.with_for_update()
|
||
.limit(1)
|
||
)
|
||
failed_record = result.scalar_one_or_none()
|
||
if failed_record:
|
||
failed_record.status = "failed"
|
||
failed_record.error_message = e.detail
|
||
failed_record.optimized_prompt = None
|
||
failed_record.text_credits_cost = 0
|
||
failed_record.credits_cost = 0
|
||
failed_record.text_tokens_used = token_usage.get("total_tokens", 0)
|
||
await db.flush()
|
||
# 这里必须主动提交，否则后续抛出 402 后 get_db 会 rollback，failed 状态会被回滚。
|
||
await db.commit()
|
||
raise e
|
||
|
||
record.optimized_prompt = optimized
|
||
record.status = "prompt_optimized"
|
||
record.text_credits_cost = round(text_credits, 2)
|
||
record.text_tokens_used = token_usage["total_tokens"]
|
||
await db.flush()
|
||
|
||
return OptimizeResult(
|
||
optimized_prompt=optimized,
|
||
text_credits_cost=round(text_credits, 2),
|
||
# text_tokens_used=token_usage["total_tokens"],
|
||
record=_record_to_out(record, project.name),
|
||
)
|
||
|
||
|
||
@router.post("/{record_id}/generate")
|
||
async def generate(
|
||
record_id: str,
|
||
req: GenerateParams,
|
||
current_user: User = Depends(get_current_user),
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
result = await db.execute(
|
||
select(GenerationRecord, Project.name)
|
||
.join(Project, GenerationRecord.project_id == Project.id)
|
||
.where(
|
||
GenerationRecord.id == record_id,
|
||
GenerationRecord.user_id == current_user.id,
|
||
GenerationRecord.deleted_at.is_(None),
|
||
Project.deleted_at.is_(None),
|
||
)
|
||
.with_for_update()
|
||
)
|
||
row = result.first()
|
||
if not row:
|
||
raise RecordNotFoundError()
|
||
|
||
record, project_name = row
|
||
if record.status not in ("prompt_optimized", "failed"):
|
||
raise InvalidStatusError("当前状态不允许生成")
|
||
|
||
await assert_user_resource_capacity_available(db, current_user.id)
|
||
|
||
attempt_no = await get_next_credit_attempt_no(
|
||
db,
|
||
owner_type=OWNER_GENERATION_RECORD,
|
||
owner_id=record.id,
|
||
)
|
||
|
||
if record.gen_type == GenerationType.video:
|
||
# Video generation
|
||
if req.aspect_ratio not in ASPECT_RATIOS:
|
||
raise HTTPException(status_code=400, detail="不支持的画面比例")
|
||
if req.resolution not in RESOLUTIONS:
|
||
raise HTTPException(status_code=400, detail="不支持的分辨率")
|
||
|
||
duration = record.duration or 5
|
||
media_billing = await charge_generation_media_by_params(
|
||
db,
|
||
user_id=current_user.id,
|
||
record_id=record.id,
|
||
gen_type="video",
|
||
duration=duration,
|
||
resolution=req.resolution,
|
||
project_name=project_name,
|
||
description_prefix=project_name+"-",
|
||
owner_type=OWNER_GENERATION_RECORD,
|
||
attempt_no=attempt_no,
|
||
)
|
||
|
||
record.aspect_ratio = req.aspect_ratio
|
||
record.resolution = req.resolution
|
||
record.credits_cost = round(float(record.credits_cost or 0) + media_billing.total_charged, 2)
|
||
record.status = "generating"
|
||
record.error_message = None
|
||
record.video_url = None
|
||
record.video_cover_url = None
|
||
record.image_url = None
|
||
record.seedance_task_id = None
|
||
await db.flush()
|
||
|
||
try:
|
||
from app.services.video_gen import get_active_engine, submit_video_task
|
||
from app.services.error_codes import extract_error_message
|
||
from app.services.video_queue import task_queue
|
||
|
||
engine = await get_active_engine(db)
|
||
task_id = await submit_video_task(
|
||
db,
|
||
engine,
|
||
record,
|
||
include_media_references=False,
|
||
)
|
||
record.seedance_task_id = task_id
|
||
await db.flush()
|
||
await task_queue.enqueue(record_id)
|
||
except Exception as e:
|
||
await mark_generation_record_failed_and_refund_once(
|
||
db,
|
||
record=record,
|
||
error_message=extract_error_message(e, "视频"),
|
||
)
|
||
await db.flush()
|
||
|
||
elif record.gen_type == GenerationType.image:
|
||
image_size = req.image_size or record.image_size or "2K"
|
||
media_billing = await charge_generation_media_by_params(
|
||
db,
|
||
user_id=current_user.id,
|
||
record_id=record.id,
|
||
gen_type="image",
|
||
image_size=image_size,
|
||
project_name=project_name,
|
||
description_prefix=project_name+"-",
|
||
owner_type=OWNER_GENERATION_RECORD,
|
||
attempt_no=attempt_no,
|
||
)
|
||
|
||
record.image_size = image_size
|
||
record.credits_cost = round(float(record.credits_cost or 0) + media_billing.total_charged, 2)
|
||
record.status = "generating"
|
||
record.error_message = None
|
||
record.image_url = None
|
||
record.video_url = None
|
||
record.video_cover_url = None
|
||
record.seedance_task_id = None
|
||
await db.flush()
|
||
|
||
try:
|
||
from app.services.video_queue import task_queue
|
||
await task_queue.enqueue(record_id)
|
||
except Exception as e:
|
||
await mark_generation_record_failed_and_refund_once(
|
||
db,
|
||
record=record,
|
||
error_message=f"图片任务队列投递失败: {e}",
|
||
)
|
||
await db.flush()
|
||
|
||
return _record_to_out(record, project_name)
|
||
|
||
|
||
@router.post("/{record_id}/retry")
|
||
async def retry_generation(
|
||
record_id: str,
|
||
current_user: User = Depends(get_current_user),
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
result = await db.execute(
|
||
select(GenerationRecord, Project.name)
|
||
.join(Project, GenerationRecord.project_id == Project.id)
|
||
.where(
|
||
GenerationRecord.id == record_id,
|
||
GenerationRecord.user_id == current_user.id,
|
||
GenerationRecord.deleted_at.is_(None),
|
||
Project.deleted_at.is_(None),
|
||
)
|
||
.with_for_update()
|
||
)
|
||
row = result.first()
|
||
if not row:
|
||
raise RecordNotFoundError()
|
||
|
||
record, project_name = row
|
||
if record.status != "failed":
|
||
raise InvalidStatusError("只有失败的记录可以重试")
|
||
|
||
await assert_user_resource_capacity_available(db, current_user.id)
|
||
|
||
attempt_no = await get_next_credit_attempt_no(
|
||
db,
|
||
owner_type=OWNER_GENERATION_RECORD,
|
||
owner_id=record.id,
|
||
)
|
||
media_billing = await charge_generation_media_for_record(
|
||
db,
|
||
record=record,
|
||
project_name=project_name,
|
||
description_prefix="视频重试",
|
||
attempt_no=attempt_no,
|
||
)
|
||
|
||
record.status = "generating"
|
||
record.error_message = None
|
||
record.video_url = None
|
||
record.video_cover_url = None
|
||
record.image_url = None
|
||
record.seedance_task_id = None
|
||
record.generated_at = None
|
||
record.credits_cost = round(float(record.credits_cost or 0) + media_billing.total_charged, 2)
|
||
await db.flush()
|
||
|
||
try:
|
||
from app.services.video_queue import task_queue
|
||
if record.gen_type == GenerationType.video:
|
||
from app.services.video_gen import get_active_engine, submit_video_task, extract_error_message
|
||
engine = await get_active_engine(db)
|
||
task_id = await submit_video_task(
|
||
db,
|
||
engine,
|
||
record,
|
||
include_media_references=False,
|
||
)
|
||
record.seedance_task_id = task_id
|
||
await db.flush()
|
||
await task_queue.enqueue(record_id)
|
||
except Exception as e:
|
||
from app.services.error_codes import extract_error_message
|
||
await mark_generation_record_failed_and_refund_once(
|
||
db,
|
||
record=record,
|
||
error_message=extract_error_message(e, "重试"),
|
||
)
|
||
await db.flush()
|
||
|
||
return _record_to_out(record, project_name)
|
||
|
||
|
||
@router.put("/{record_id}/prompt")
|
||
async def update_prompt(
|
||
record_id: str,
|
||
req: UpdatePromptRequest,
|
||
current_user: User = Depends(get_current_user),
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
result = await db.execute(
|
||
select(GenerationRecord).where(
|
||
GenerationRecord.id == record_id,
|
||
GenerationRecord.user_id == current_user.id,
|
||
GenerationRecord.deleted_at.is_(None),
|
||
)
|
||
.limit(1)
|
||
)
|
||
record = result.scalar_one_or_none()
|
||
if not record:
|
||
raise RecordNotFoundError()
|
||
if record.status != "prompt_optimized":
|
||
raise InvalidStatusError("只有待生成状态可以修改提示词")
|
||
|
||
record.optimized_prompt = req.optimized_prompt
|
||
await db.flush()
|
||
return {"message": "ok"}
|
||
|
||
|
||
@router.get("/{record_id}/video")
|
||
async def get_video(
|
||
record_id: str,
|
||
token: str = Query(...),
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
"""Validate temp token and redirect to video URL."""
|
||
validated_id = await validate_and_get_record_id(token)
|
||
if validated_id != record_id:
|
||
raise HTTPException(status_code=403, detail="无效的视频链接")
|
||
|
||
video_url = await get_video_stream_url(db, record_id)
|
||
if not video_url:
|
||
raise HTTPException(status_code=404, detail="视频不存在")
|
||
|
||
return RedirectResponse(url=video_url)
|
||
|
||
|
||
@router.get("/{record_id}/queue-status")
|
||
async def get_queue_status(
|
||
record_id: str,
|
||
current_user: User = Depends(get_current_user),
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
"""Get queue position, estimated wait time, and current status for a generation record."""
|
||
from sqlalchemy import func
|
||
|
||
result = await db.execute(
|
||
select(GenerationRecord).where(
|
||
GenerationRecord.id == record_id,
|
||
GenerationRecord.user_id == current_user.id,
|
||
GenerationRecord.deleted_at.is_(None),
|
||
)
|
||
.limit(1)
|
||
)
|
||
record = result.scalar_one_or_none()
|
||
if not record:
|
||
raise RecordNotFoundError()
|
||
|
||
queue_position = None
|
||
estimated_wait_seconds = None
|
||
|
||
if record.status == "generating":
|
||
ahead_result = await db.execute(
|
||
select(func.count(GenerationRecord.id)).where(
|
||
GenerationRecord.status == "generating",
|
||
GenerationRecord.deleted_at.is_(None),
|
||
GenerationRecord.created_at < record.created_at,
|
||
)
|
||
)
|
||
ahead = ahead_result.scalar() or 0
|
||
queue_position = ahead + 1
|
||
estimated_wait_seconds = ahead * 60
|
||
|
||
return {
|
||
"record_id": record.id,
|
||
"status": record.status,
|
||
"queue_position": queue_position,
|
||
"estimated_wait_seconds": estimated_wait_seconds,
|
||
}
|
||
|
||
|
||
@router.post("/callbacks/seedance")
|
||
async def seedance_callback(request: Request, db: AsyncSession = Depends(get_db)):
|
||
"""Receive async callback from Seedance API."""
|
||
data = await request.json()
|
||
task_id = data.get("id")
|
||
task_status = data.get("status")
|
||
|
||
if not task_id:
|
||
return {"message": "ignored"}
|
||
|
||
result = await db.execute(
|
||
select(GenerationRecord).where(
|
||
GenerationRecord.seedance_task_id == task_id,
|
||
GenerationRecord.deleted_at.is_(None),
|
||
)
|
||
.with_for_update()
|
||
.limit(1)
|
||
)
|
||
record = result.scalar_one_or_none()
|
||
if not record:
|
||
return {"message": "record not found"}
|
||
|
||
if task_status == "succeeded":
|
||
remote_url = data.get("content", {}).get("video_url", "")
|
||
record.status = "completed"
|
||
storage_path = None
|
||
file_size_bytes = 0
|
||
# Download video to local storage
|
||
if settings.STORAGE_TYPE == "local" and remote_url:
|
||
try:
|
||
from app.services.video_gen import download_video
|
||
date_dir = datetime.now().strftime("%Y/%m/%d")
|
||
dest_dir = os.path.join(settings.STORAGE_LOCAL_PATH, date_dir)
|
||
os.makedirs(dest_dir, exist_ok=True)
|
||
dest = os.path.join(dest_dir, f"{record.id}.mp4")
|
||
await download_video(remote_url, dest)
|
||
record.video_url = f"/generate/videos/{date_dir}/{record.id}.mp4"
|
||
cover_url, _cover_storage_path = await async_create_video_cover_for_local_video(
|
||
record_id=record.id,
|
||
video_path=dest,
|
||
date_dir=date_dir,
|
||
log_prefix=f"SeedanceCallback视频封面生成 record_id={record.id}",
|
||
)
|
||
record.video_cover_url = cover_url
|
||
storage_path = dest
|
||
file_size_bytes = safe_file_size(dest)
|
||
except Exception as e:
|
||
logger.warning(f"Callback download failed, using remote URL: {e}")
|
||
record.video_url = remote_url
|
||
else:
|
||
record.video_url = remote_url
|
||
record.generated_at = datetime.now()
|
||
if record.video_url:
|
||
await record_generation_record_generated_resource(
|
||
db,
|
||
record,
|
||
resource_url=record.video_url,
|
||
storage_path=storage_path,
|
||
file_size_bytes=file_size_bytes,
|
||
remote_url=remote_url,
|
||
generated_at=record.generated_at,
|
||
)
|
||
# Extract video token usage from callback
|
||
usage = data.get("usage", {})
|
||
if usage:
|
||
record.video_tokens_used = usage.get("total_tokens", 0)
|
||
await sync_generation_record_media_token_snapshot(db, record, provider_response=data)
|
||
# Log callback response
|
||
from app.services.video_gen import _log_video_response
|
||
_log_video_response(record.id, data)
|
||
# Notify user
|
||
from app.services.notification import create_notification
|
||
from app.api.v1.notifications import push_notification_to_user
|
||
notif = await create_notification(
|
||
db, record.user_id, "视频生成完成",
|
||
"您的视频已生成完成，可以查看了。", "video", record.id,
|
||
)
|
||
await push_notification_to_user(record.user_id, notif)
|
||
elif task_status == "failed":
|
||
error_message = data.get("error", "视频生成失败")
|
||
await mark_generation_record_failed_and_refund_once(
|
||
db,
|
||
record=record,
|
||
error_message=error_message,
|
||
)
|
||
# Log callback response
|
||
from app.services.video_gen import _log_video_response
|
||
_log_video_response(record.id, data, error=record.error_message)
|
||
# Notify user
|
||
from app.services.notification import create_notification
|
||
from app.api.v1.notifications import push_notification_to_user
|
||
notif = await create_notification(
|
||
db, record.user_id, "视频生成失败",
|
||
f"视频生成失败：{record.error_message}", "video", record.id,
|
||
)
|
||
await push_notification_to_user(record.user_id, notif)
|
||
|
||
await db.flush()
|
||
return {"message": "ok"}
|
||
|
||
|
||
@router.post("/upload-image")
|
||
async def upload_image(
|
||
file: UploadFile = File(...),
|
||
current_user: User = Depends(get_current_user),
|
||
gen_type: str = Query("video", description="生成类型：video-视频，image-图片"),
|
||
):
|
||
"""Upload an image for generation reference."""
|
||
import os
|
||
import uuid
|
||
from app.config import settings
|
||
from datetime import datetime
|
||
|
||
if not file.content_type or not file.content_type.startswith("image/"):
|
||
raise HTTPException(status_code=400, detail="仅支持图片文件")
|
||
|
||
ext = os.path.splitext(file.filename or ".png")[1] or ".png"
|
||
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
|
||
safe_name = f"{gen_type}_img_{current_user.id}_{timestamp}_{uuid.uuid4().hex[:8]}{ext}"
|
||
date_dir = datetime.now().strftime("%Y/%m/%d")
|
||
dir_path = os.path.join(settings.UPLOAD_LOCAL_PATH, "images", date_dir)
|
||
os.makedirs(dir_path, exist_ok=True)
|
||
file_path = os.path.join(dir_path, safe_name)
|
||
|
||
content = await file.read()
|
||
if len(content) > 10 * 1024 * 1024:
|
||
raise HTTPException(status_code=400, detail="图片大小不能超过10MB")
|
||
|
||
with open(file_path, "wb") as f:
|
||
f.write(content)
|
||
|
||
url = f"/uploads/images/{date_dir}/{safe_name}"
|
||
return {"url": url, "filename": file.filename or safe_name, "type": "image", "gen_type": gen_type}
|
||
|
||
|
||
@router.post("/upload-video")
|
||
async def upload_video(
|
||
file: UploadFile = File(...),
|
||
current_user: User = Depends(get_current_user),
|
||
):
|
||
"""Upload a video for generation reference."""
|
||
import os
|
||
import uuid
|
||
from app.config import settings
|
||
from datetime import datetime
|
||
|
||
if not file.content_type or not file.content_type.startswith("video/"):
|
||
raise HTTPException(status_code=400, detail="仅支持视频文件")
|
||
|
||
ext = os.path.splitext(file.filename or ".mp4")[1] or ".mp4"
|
||
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
|
||
safe_name = f"video_ref_{current_user.id}_{timestamp}_{uuid.uuid4().hex[:8]}{ext}"
|
||
date_dir = datetime.now().strftime("%Y/%m/%d")
|
||
dir_path = os.path.join(settings.UPLOAD_LOCAL_PATH, "videos", date_dir)
|
||
os.makedirs(dir_path, exist_ok=True)
|
||
file_path = os.path.join(dir_path, safe_name)
|
||
|
||
content = await file.read()
|
||
if len(content) > 100 * 1024 * 1024:
|
||
raise HTTPException(status_code=400, detail="视频大小不能超过100MB")
|
||
|
||
with open(file_path, "wb") as f:
|
||
f.write(content)
|
||
|
||
url = f"/uploads/videos/{date_dir}/{safe_name}"
|
||
return {"url": url, "filename": file.filename or safe_name, "type": "video"}
|
||
|
||
|
||
@router.post("/upload-audio")
|
||
async def upload_audio(
|
||
file: UploadFile = File(...),
|
||
current_user: User = Depends(get_current_user),
|
||
):
|
||
"""Upload an audio file for AI creation reference."""
|
||
import os
|
||
import uuid
|
||
from app.config import settings
|
||
from datetime import datetime
|
||
|
||
ext = os.path.splitext(file.filename or "")[1].lower().lstrip(".")
|
||
if ext not in AUDIO_ALLOWED_EXTENSIONS:
|
||
raise HTTPException(status_code=400, detail="仅支持 mp3、wav 音频文件")
|
||
|
||
expected_mime = AUDIO_ALLOWED_MIME_TYPES.get(ext)
|
||
if not file.content_type or file.content_type != expected_mime:
|
||
raise HTTPException(status_code=400, detail=f"音频 MIME 类型错误，{ext} 必须为 {expected_mime}")
|
||
|
||
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
|
||
safe_name = f"audio_ref_{current_user.id}_{timestamp}_{uuid.uuid4().hex[:8]}.{ext}"
|
||
date_dir = datetime.now().strftime("%Y/%m/%d")
|
||
dir_path = os.path.join(settings.UPLOAD_LOCAL_PATH, "audios", date_dir)
|
||
os.makedirs(dir_path, exist_ok=True)
|
||
file_path = os.path.join(dir_path, safe_name)
|
||
|
||
content = await file.read()
|
||
max_bytes = AUDIO_MAX_FILE_SIZE_MB * 1024 * 1024
|
||
if len(content) > max_bytes:
|
||
raise HTTPException(status_code=400, detail=f"音频大小不能超过{AUDIO_MAX_FILE_SIZE_MB}MB")
|
||
|
||
with open(file_path, "wb") as f:
|
||
f.write(content)
|
||
|
||
url = f"/uploads/audios/{date_dir}/{safe_name}"
|
||
return {"url": url, "filename": file.filename or safe_name, "type": "audio"}
|
||
|
||
|
||
@router.post("/delete-file")
|
||
async def delete_upload(
|
||
url: str = Query(..., description="文件URL，如 /uploads/images/2024/01/01/video_img_xxx.png"),
|
||
current_user: User = Depends(get_current_user),
|
||
):
|
||
"""Delete an uploaded file by URL."""
|
||
import os
|
||
from app.config import settings
|
||
|
||
if not url.startswith("/uploads/"):
|
||
raise HTTPException(status_code=400, detail="无效的文件路径")
|
||
|
||
if current_user.id not in url:
|
||
raise HTTPException(status_code=403, detail="无权删除此文件")
|
||
|
||
file_path = os.path.join(settings.UPLOAD_LOCAL_PATH, url.replace("/uploads/", ""))
|
||
if os.path.exists(file_path):
|
||
os.remove(file_path)
|
||
|
||
return {"message": "ok"}
|